flow_constraints

flow_constraints ¤

FlowConstraintEvents ¤

FlowConstraintEvents(
    dataframe: DataFrame,
    *,
    time_column: str = "systime",
    event_uuid: str = "prod:flow"
)

Bases: Base

Production: Flow Constraints

Both detectors work on two boolean run signals only (no buffer level or part-presence signal), so they flag the run-state pattern, not its cause:

blocked_events: upstream running while downstream is stopped -- the upstream station keeps producing into a stopped consumer and its output buffer fills.
starved_events: downstream running while upstream is stopped -- the downstream station keeps consuming with no upstream supply and drains its input buffer. This is the onset of starvation, not the common "station idle for lack of input" state: once the downstream station actually runs dry and stops, both are stopped and the interval is no longer reported.

blocked_events ¤

blocked_events(
    *,
    roles: dict[str, str],
    tolerance: str = "200ms",
    tolerance_before: str | None = None,
    tolerance_after: str | None = None,
    min_duration: str = "0s",
    interval_end: str = "next_change"
) -> pd.DataFrame

Blocked: upstream_run=True while downstream_run=False.

Flags intervals where the upstream station runs while its downstream consumer is stopped, so the buffer between them fills.

Parameters:

Name	Type	Description	Default
`roles`	`dict[str, str]`	Dictionary mapping role names to UUIDs. Expected keys: 'upstream_run', 'downstream_run'	required
`tolerance`	`str`	Default tolerance for time alignment (used if directional tolerances not provided)	`'200ms'`
`tolerance_before`	`str \| None`	Tolerance for looking backward in time during alignment	`None`
`tolerance_after`	`str \| None`	Tolerance for looking forward in time during alignment	`None`
`min_duration`	`str`	Minimum duration for an event to be included	`'0s'`
`interval_end`	`str`	`"next_change"` (default) ends an interval at the first aligned sample where the condition no longer holds (the transition), so a condition seen at a single sample still lasts until the next sample; a condition that holds until the end of the data ends at the last sample. `"last_sample"` ends it at the last sample where the condition holds (the behaviour before this option; a single-sample event then has zero duration).	`'next_change'`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: start, end, uuid, source_uuid, is_delta, type,
`DataFrame`	time_alignment_quality, duration, severity

Raises:

Type	Description
`ValueError`	If `interval_end` is not `"next_change"` or `"last_sample"`.

Example

roles = {'upstream_run': 'uuid1', 'downstream_run': 'uuid2'}

starved_events ¤

starved_events(
    *,
    roles: dict[str, str],
    tolerance: str = "200ms",
    tolerance_before: str | None = None,
    tolerance_after: str | None = None,
    min_duration: str = "0s",
    interval_end: str = "next_change"
) -> pd.DataFrame

Starved: downstream_run=True while upstream_run=False.

Flags intervals where the downstream station runs while its upstream supplier is stopped, i.e. downstream is consuming its input buffer with no resupply. A downstream station that has already stopped for lack of input (both stations stopped) is not reported; with run signals alone that state cannot be told apart from any other joint stop.

Parameters:

Name	Type	Description	Default
`roles`	`dict[str, str]`	Dictionary mapping role names to UUIDs. Expected keys: 'upstream_run', 'downstream_run'	required
`tolerance`	`str`	Default tolerance for time alignment (used if directional tolerances not provided)	`'200ms'`
`tolerance_before`	`str \| None`	Tolerance for looking backward in time during alignment	`None`
`tolerance_after`	`str \| None`	Tolerance for looking forward in time during alignment	`None`
`min_duration`	`str`	Minimum duration for an event to be included	`'0s'`
`interval_end`	`str`	`"next_change"` (default) ends an interval at the first aligned sample where the condition no longer holds (the transition), so a condition seen at a single sample still lasts until the next sample; a condition that holds until the end of the data ends at the last sample. `"last_sample"` ends it at the last sample where the condition holds (the behaviour before this option; a single-sample event then has zero duration).	`'next_change'`

Returns:

Type	Description
`DataFrame`	DataFrame with columns: start, end, uuid, source_uuid, is_delta, type,
`DataFrame`	time_alignment_quality, duration, severity

Raises:

Type	Description
`ValueError`	If `interval_end` is not `"next_change"` or `"last_sample"`.

Example

roles = {'upstream_run': 'uuid1', 'downstream_run': 'uuid2'}

flow_constraint_analytics ¤

flow_constraint_analytics(
    *,
    roles: dict[str, str],
    tolerance: str = ...,
    tolerance_before: str | None = ...,
    tolerance_after: str | None = ...,
    min_duration: str = ...,
    warning_threshold: str = ...,
    critical_threshold: str = ...,
    interval_end: str = ...,
    as_frame: Literal[True] = ...
) -> pd.DataFrame

flow_constraint_analytics(
    *,
    roles: dict[str, str],
    tolerance: str = ...,
    tolerance_before: str | None = ...,
    tolerance_after: str | None = ...,
    min_duration: str = ...,
    warning_threshold: str = ...,
    critical_threshold: str = ...,
    interval_end: str = ...,
    as_frame: Literal[False]
) -> dict[str, Any]

flow_constraint_analytics(
    *,
    roles: dict[str, str],
    tolerance: str = "200ms",
    tolerance_before: str | None = None,
    tolerance_after: str | None = None,
    min_duration: str = "0s",
    warning_threshold: str = "5s",
    critical_threshold: str = "30s",
    interval_end: str = "next_change",
    as_frame: bool = True
) -> pd.DataFrame | dict[str, Any]

Generate comprehensive analytics for flow constraints (blockages and starvations).

Parameters:

Name	Type	Description	Default
`roles`	`dict[str, str]`	Dictionary mapping role names to UUIDs. Expected keys: 'upstream_run', 'downstream_run'	required
`tolerance`	`str`	Default tolerance for time alignment (used if directional tolerances not provided)	`'200ms'`
`tolerance_before`	`str \| None`	Tolerance for looking backward in time during alignment	`None`
`tolerance_after`	`str \| None`	Tolerance for looking forward in time during alignment	`None`
`min_duration`	`str`	Minimum duration for an event to be included	`'0s'`
`warning_threshold`	`str`	Duration below which an event is 'minor'.	`'5s'`
`critical_threshold`	`str`	Duration below which an event is 'moderate' (at or above it, 'severe').	`'30s'`
`interval_end`	`str`	Forwarded to :meth:`blocked_events` and :meth:`starved_events`: `"next_change"` (default) ends an event at the transition, `"last_sample"` at the last sample where the condition holds.	`'next_change'`
`as_frame`	`bool`	When True (default), return a summary DataFrame (see below). `as_frame=False` returns the dictionary form; it is deprecated, emits a `DeprecationWarning` and will be removed in the next major release.	`True`

Returns:

Name	Type	Description
	`DataFrame \| dict[str, Any]`	With `as_frame=True` (default), a summary DataFrame with two
	`DataFrame \| dict[str, Any]`	rows, `constraint_type` `"blocked"` then `"starved"`, and
`columns`	`DataFrame \| dict[str, Any]`
	`DataFrame \| dict[str, Any]`	start, end, duration_seconds: first event start to last event end of that type (NaT / 0 when there is none)
	`DataFrame \| dict[str, Any]`	uuid: the event uuid; source_uuid: the upstream (blocked) or downstream (starved) run signal
	`DataFrame \| dict[str, Any]`	constraint_type
	`DataFrame \| dict[str, Any]`	event_count, total_duration_seconds, avg_duration_seconds (0 when there is no event)
	`DataFrame \| dict[str, Any]`	minor_count, moderate_count, severe_count: the severity breakdown
	`DataFrame \| dict[str, Any]`	time_alignment_quality: share of samples aligned between the two signals (0 when there is no event)
	`DataFrame \| dict[str, Any]`	The totals of the dict form are the column sums:
	`DataFrame \| dict[str, Any]`	`total_constraint_events` is `event_count.sum()` and
	`DataFrame \| dict[str, Any]`	`overall_alignment_quality` is `time_alignment_quality.mean()`.
	`DataFrame \| dict[str, Any]`	The events themselves come from :meth:`blocked_events` and
	`DataFrame \| dict[str, Any]`	meth:`starved_events`.
	`DataFrame \| dict[str, Any]`	With `as_frame=False` (deprecated), a dictionary containing
	`DataFrame \| dict[str, Any]`	analytics for both blocked and starved events:
	`DataFrame \| dict[str, Any]`	blocked_events: DataFrame of blocked events
	`DataFrame \| dict[str, Any]`	starved_events: DataFrame of starved events
	`DataFrame \| dict[str, Any]`	summary: Dictionary with statistics including: blocked_count: Total number of blocked events starved_count: Total number of starved events blocked_total_duration: Total duration of blocked events starved_total_duration: Total duration of starved events blocked_avg_duration: Average duration of blocked events starved_avg_duration: Average duration of starved events blocked_severity_breakdown: Count by severity level starved_severity_breakdown: Count by severity level overall_alignment_quality: Average alignment quality across both types

Example

import pandas as pd, ts_shape t = pd.date_range("2025-01-01", periods=6, freq="10s") up = pd.DataFrame({"systime": t, "uuid": "up", "value_bool": True}) down = up.assign(uuid="down", value_bool=[True, False, False, True, True, True]) flow = ts_shape.FlowConstraintEvents(pd.concat([up, down])) out = flow.flow_constraint_analytics(roles={"upstream_run": "up", "downstream_run": "down"}) out[["constraint_type", "event_count", "total_duration_seconds"]] constraint_type event_count total_duration_seconds 0 blocked 1 20.0 1 starved 0 0.0