bottleneck_detection

bottleneck_detection

BottleneckDetectionEvents

BottleneckDetectionEvents(
    dataframe: DataFrame,
    *,
    time_column: str = "systime",
    event_uuid: str = "prod:bottleneck",
    value_column: str = "value_bool"
)

Bases: Base

Production: Bottleneck Detection

Identify which station constrains a production line by analyzing utilization of multiple boolean run-state signals.

Methods: - station_utilization: Per-station uptime percentage per window. - detect_bottleneck: Identify the bottleneck station per window. - shifting_bottleneck: Track when the bottleneck moves between stations. - throughput_constraint_summary: Summary statistics across all stations.

station_utilization

station_utilization(
    station_uuids: list[str],
    window: str = "1h",
    *,
    time_weighted: bool = True
) -> pd.DataFrame

Per-station uptime percentage per time window.

Parameters:

Name	Type	Description	Default
station_uuids	list[str]	List of UUID strings identifying station run-state signals.	required
window	str	Resample window (e.g. '1h', '30m').	'1h'
time_weighted	bool	When True (default), utilization is time-weighted: each sample is weighted by the gap to the next sample (as in RuntimeAccountingEvents), so it reflects wall-clock time in the running state on change-encoded and irregularly sampled signals. The trailing sample of a station has no successor and carries no weight. When False, utilization is the mean of the boolean samples in the window, which equals the time share only under uniform sampling.	True

Returns:

Type	Description
DataFrame	DataFrame with columns: start, end (start + window),
DataFrame	duration_seconds, uuid (the station uuid), source_uuid (the
DataFrame	station uuid), utilization_pct.

detect_bottleneck

detect_bottleneck(
    station_uuids: list[str],
    window: str = "1h",
    *,
    time_weighted: bool = True
) -> pd.DataFrame

Identify the bottleneck station per window.

The bottleneck is the station with the highest utilization — it is always running because it is the constraint.

Parameters:

Name	Type	Description	Default
station_uuids	list[str]	List of station run-state UUIDs.	required
window	str	Resample window.	'1h'
time_weighted	bool	Forwarded to :meth:station_utilization (default True).	True

Returns:

Type	Description
DataFrame	DataFrame with columns: start, end, duration_seconds, uuid (the
DataFrame	detector's event uuid), source_uuid (the bottleneck station uuid),
DataFrame	bottleneck_uuid, utilization_pct.

shifting_bottleneck

shifting_bottleneck(
    station_uuids: list[str],
    window: str = "1h",
    *,
    time_weighted: bool = True
) -> pd.DataFrame

Track when the bottleneck identity changes between stations.

Parameters:

Name	Type	Description	Default
station_uuids	list[str]	List of station run-state UUIDs.	required
window	str	Resample window.	'1h'
time_weighted	bool	Forwarded to :meth:station_utilization.	True

Returns:

Type	Description
DataFrame	DataFrame with columns: systime, uuid (the detector's event uuid),
DataFrame	source_uuid ("<from_uuid>,<to_uuid>"), from_uuid, to_uuid,
DataFrame	previous_utilization, new_utilization.

throughput_constraint_summary

throughput_constraint_summary(
    station_uuids: list[str],
    window: str = ...,
    *,
    time_weighted: bool = ...,
    as_frame: Literal[True] = ...
) -> pd.DataFrame

throughput_constraint_summary(
    station_uuids: list[str],
    window: str = ...,
    *,
    time_weighted: bool = ...,
    as_frame: Literal[False]
) -> dict[str, Any]

throughput_constraint_summary(
    station_uuids: list[str],
    window: str = "1h",
    *,
    time_weighted: bool = True,
    as_frame: bool = True
) -> pd.DataFrame | dict[str, Any]

Summary statistics for bottleneck analysis.

Parameters:

Name	Type	Description	Default
station_uuids	list[str]	List of station run-state UUIDs.	required
window	str	Resample window.	'1h'
time_weighted	bool	Forwarded to :meth:station_utilization.	True
as_frame	bool	When True (default), return one row per station (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:

Type	Description
DataFrame | dict[str, Any]	With as_frame=True (default), a summary DataFrame with one row
DataFrame | dict[str, Any]	per station that has data, in station_uuids order, columns:
DataFrame | dict[str, Any]	start, end, duration_seconds: the analysed span (first window start to last window end), the same on every row
DataFrame | dict[str, Any]	uuid, source_uuid: the station uuid
DataFrame | dict[str, Any]	bottleneck_count: windows in which the station was the bottleneck (bottleneck_counts of the dict form)
DataFrame | dict[str, Any]	bottleneck_pct: share of windows, 0-100 (bottleneck_percentages)
DataFrame | dict[str, Any]	avg_utilization_pct: mean utilization over the windows (avg_utilization_per_station)
DataFrame | dict[str, Any]	most_frequent_bottleneck: True on the station that was the bottleneck most often
DataFrame | dict[str, Any]	With as_frame=False (deprecated), a dict with:
DataFrame | dict[str, Any]	bottleneck_counts, bottleneck_percentages,
DataFrame | dict[str, Any]	most_frequent_bottleneck, avg_utilization_per_station.