Skip to content

machine_state

machine_state ¤

MachineStateEvents ¤

MachineStateEvents(
    dataframe: DataFrame,
    run_state_uuid: str,
    *,
    event_uuid: str = "prod:run_idle",
    value_column: str = "value_bool",
    time_column: str = "systime",
    value_range: (
        tuple[float | None, float | None] | None
    ) = None,
    specs: Mapping[str, SignalSpec] | None = None
)

Bases: Base

Production: Machine State

Detect run/idle transitions and intervals from a boolean state signal.

  • MachineStateEvents: Run/idle state intervals and transitions.
  • detect_run_idle: Intervalize run/idle states with optional min duration filter.
  • transition_events: Point events on state changes (idle->run, run->idle).
Example

import pandas as pd, ts_shape t = pd.date_range("2025-01-01 08:00", periods=6, freq="10min") df = pd.DataFrame({"systime": t, "uuid": "run", ... "value_bool": [True, True, False, True, True, False]}) out = ts_shape.MachineStateEvents(df, "run").detect_run_idle() out[["start", "state", "duration_seconds"]] start state duration_seconds 0 2025-01-01 08:00:00 run 1200.0 1 2025-01-01 08:20:00 idle 600.0 2 2025-01-01 08:30:00 run 1200.0 3 2025-01-01 08:50:00 idle 0.0

detect_run_idle ¤

detect_run_idle(
    min_duration: str = "0s",
    *,
    interval_end: str | None = None,
    data_end: Timestamp | None = None
) -> pd.DataFrame

Return intervals labeled as 'run' or 'idle'.

  • min_duration: discard intervals shorter than this duration.
  • interval_end: how to close each interval. "next_change" closes each interval at the next transition instant, which measures both dense and change-encoded (is_delta) signals. "last_sample" ends an interval at its last sample; on a change-encoded signal (one row per transition) every interval then has zero duration. Default (None): follow the signal's spec when specs= was given (held states close at the next change, and gaps longer than the spec's max_gap are "no data"), otherwise "next_change".
  • data_end: optional bound for the final, still-open interval when intervals close at the next change (e.g. the end of the data window). Columns: start, end, uuid, source_uuid, is_delta, state, duration_seconds

transition_events ¤

transition_events() -> pd.DataFrame

Return point events at state transitions.

Columns: systime, uuid, source_uuid, is_delta, transition ('idle_to_run'|'run_to_idle'), time_since_last_transition_seconds

detect_rapid_transitions ¤

detect_rapid_transitions(
    threshold: str = "5s", min_count: int = 3
) -> pd.DataFrame

Identify suspicious rapid state changes.

  • threshold: time window to look for rapid transitions
  • min_count: minimum number of transitions within threshold to be considered rapid Returns: DataFrame with systime (= start, the first transition of the burst), uuid (event uuid), source_uuid (run-state uuid), start, end, transition_count, duration_seconds

state_quality_metrics ¤

state_quality_metrics(
    *,
    interval_end: str | None = ...,
    data_end: Timestamp | None = ...,
    as_frame: Literal[True] = ...
) -> pd.DataFrame

state_quality_metrics(
    *,
    interval_end: str | None = ...,
    data_end: Timestamp | None = ...,
    as_frame: Literal[False]
) -> dict[str, Any]

state_quality_metrics(
    *,
    interval_end: str | None = None,
    data_end: Timestamp | None = None,
    as_frame: bool = True
) -> pd.DataFrame | dict[str, Any]

Return quality metrics for the state data.

Parameters:

Name Type Description Default
interval_end str | None

Forwarded to :meth:detect_run_idle. Default None resolves exactly as in :meth:detect_run_idle: follow the signal's spec when specs= was given, otherwise "next_change" (run/idle durations are measured to the next transition).

 None
data_end Timestamp | None

Forwarded to :meth:detect_run_idle; optional bound for the final, still-open interval.

 None
as_frame bool

When True (default), return a one-row 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:

Type Description
DataFrame | dict[str, Any]

With as_frame=True (default), a one-row summary DataFrame with
DataFrame | dict[str, Any]

columns start, end, duration_seconds (the span of the state
DataFrame | dict[str, Any]

signal; NaT / 0 without data), uuid (event uuid), source_uuid
DataFrame | dict[str, Any]

(run-state uuid), followed by the metrics below.
DataFrame | dict[str, Any]

With as_frame=False (deprecated), a dictionary with:
DataFrame | dict[str, Any]
  • total_transitions: total number of state transitions
DataFrame | dict[str, Any]
  • avg_run_duration: average duration of run states in seconds
DataFrame | dict[str, Any]
  • avg_idle_duration: average duration of idle states in seconds
DataFrame | dict[str, Any]
  • run_idle_ratio: ratio of total run time to total idle time; NaN when there is no idle time (including no intervals at all), since the ratio is undefined
DataFrame | dict[str, Any]
  • data_gaps_detected: number of data gaps found
DataFrame | dict[str, Any]
  • rapid_transitions_detected: number of rapid transition events