State-Machine Design for RUN/IDLE/DOWN Event Classification

Most OEE pipelines eventually collapse dozens of PLC bits — run permissive, fault relay, E-stop chain, guard interlock — into three states an operator actually recognizes: the machine is running, it is briefly not running, or it is down. Doing that collapse correctly is a finite-state-machine design problem, not a filter chain, and it is the layer directly beneath event-to-downtime mapping that decides which raw signals are even allowed to become a state transition. This page specifies a minimal RUN/IDLE/DOWN model built for the general case: a deterministic transition table, hysteresis that stops a flickering run bit from generating dozens of spurious events per minute, and idempotent handling of out-of-order or duplicated telemetry so replaying a shift’s events twice never double-counts a single second of downtime.

The model is deliberately smaller than the five-state FSM used for full downtime mapping (which also tracks Maintenance and Stopped); RUN/IDLE/DOWN is the core engine that a richer state set builds on, and it is the same three states that both microstop thresholds for conveyor motor jams and every other asset-specific threshold page ultimately feed events into.

The state model and transition table #Permalink to this section

A RUN/IDLE/DOWN machine has exactly six legal transitions and no others; anything else arriving from upstream is a data-quality event, not a state change. IDLE is the entry hub — every asset boots into IDLE until proven running — and DOWN is reachable only from IDLE (a fault confirmed after sitting idle) or directly from RUN for a small set of safety-critical triggers that must not wait for a debounce window. Critically, there is no DOWN → RUN edge: recovery always passes back through IDLE first, so an operator clearing an E-stop cannot accidentally skip the confirmation that the run permissive is actually re-established.

Time spent in DOWN is exactly the Downtime term the Availability factor consumes, so the state machine’s correctness is a direct precondition for the OEE calculation:

$A = \frac{\text{Planned Production Time} - \sum \text{DOWN dwell}}{\text{Planned Production Time}}$

A spurious DOWN transition inflates that sum and understates Availability; a missed one does the opposite, which is why every transition here is guarded rather than inferred.

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from typing import Callable


class MachineState(str, Enum):
    RUN = "RUN"
    IDLE = "IDLE"
    DOWN = "DOWN"


@dataclass(frozen=True)
class Sample:
    """A single conditioned telemetry sample for one asset."""
    asset_id: str
    seq: int
    ts_utc: float  # epoch seconds
    run_bit: bool
    fault_bit: bool
    critical_fault: bool
    idle_dwell_sec: float
    max_threshold_sec: float  # asset-specific, e.g. 45.0 for conveyors


Guard = Callable[[Sample], bool]

# (from_state) -> ordered list of (guard, to_state); first matching guard wins.
# Order matters: safety-critical transitions must be evaluated before
# ordinary ones so an E-stop is never delayed behind a slower rule.
TRANSITIONS: dict[MachineState, list[tuple[Guard, MachineState]]] = {
    MachineState.RUN: [
        (lambda s: s.critical_fault, MachineState.DOWN),
        (lambda s: not s.run_bit, MachineState.IDLE),
    ],
    MachineState.IDLE: [
        (lambda s: s.run_bit, MachineState.RUN),
        (
            lambda s: s.fault_bit and s.idle_dwell_sec >= s.max_threshold_sec,
            MachineState.DOWN,
        ),
    ],
    MachineState.DOWN: [
        (lambda s: not s.fault_bit and not s.critical_fault, MachineState.IDLE),
    ],
}


def next_state(current: MachineState, sample: Sample) -> MachineState:
    """Pure function: (state, sample) -> state. No side effects, no I/O."""
    for guard, target in TRANSITIONS.get(current, []):
        if guard(sample):
            return target
    return current  # no guard matched: self-loop, state unchanged

Keeping next_state pure — no clocks read internally, no mutation — is what makes the machine replayable: given the same (state, sample) pair it always returns the same result, which is the determinism contract the whole event-to-downtime mapping layer depends on.

Debounce and hysteresis to stop flapping #Permalink to this section

The transition table above is stateless and reacts on every sample, which means a chattering run bit — contact bounce, a marginal proximity sensor, PLC scan-cycle jitter — fires a transition on every single flicker. The fix is hysteresis: a proposed transition is only committed once its triggering condition has held continuously for a minimum dwell, and a class wraps the pure next_state function with that timing memory.

from dataclasses import dataclass, field


@dataclass
class RunIdleDownStateMachine:
    """Stateful wrapper around next_state that debounces flapping inputs."""

    asset_id: str
    hysteresis_hold_sec: float = 0.5   # IDLE -> RUN confirmation
    debounce_dwell_sec: float = 1.5    # RUN -> IDLE confirmation
    recovery_hold_sec: float = 2.0     # DOWN -> IDLE confirmation
    state: MachineState = MachineState.IDLE
    _pending_target: MachineState | None = field(default=None, repr=False)
    _pending_since: float | None = field(default=None, repr=False)

    def _required_hold(self, target: MachineState) -> float:
        if target is MachineState.RUN:
            return self.hysteresis_hold_sec
        if target is MachineState.IDLE and self.state is MachineState.RUN:
            return self.debounce_dwell_sec
        if target is MachineState.IDLE and self.state is MachineState.DOWN:
            return self.recovery_hold_sec
        return 0.0  # RUN -> DOWN critical fault: immediate, no hold

    def ingest(self, sample: Sample) -> MachineState | None:
        """Feed one conditioned sample. Returns the new state on a
        committed transition, or None if nothing changed."""
        proposed = next_state(self.state, sample)
        if proposed == self.state:
            self._pending_target = None
            return None

        required_hold = self._required_hold(proposed)
        if required_hold == 0.0:
            # Safety-critical: commit immediately, no debounce window.
            self.state = proposed
            self._pending_target = None
            return proposed

        if self._pending_target != proposed:
            self._pending_target = proposed
            self._pending_since = sample.ts_utc
            return None

        assert self._pending_since is not None
        if sample.ts_utc - self._pending_since >= required_hold:
            self.state = proposed
            self._pending_target = None
            return proposed
        return None

Note that the RUN → DOWN critical-fault edge has a required_hold of zero: safety events must never wait behind a debounce timer, while ordinary run/idle flicker and fault recovery both get an explicit hold. This mirrors the microstop thresholds for conveyor motor jams debounce logic at a more general level — that page’s debounce_hold_sec is exactly this machine’s debounce_dwell_sec applied to one asset class.

Gap and out-of-order handling with idempotent replay #Permalink to this section

Two failure modes corrupt a naive streaming state machine: samples arriving out of order after an edge-buffer replay, and the same sample being delivered twice by an at-least-once MQTT broker. Both are solved by treating seq as the source of truth rather than arrival order, buffering briefly to reorder, and refusing to apply any sample whose sequence number has already been committed.

from collections import deque


class IdempotentReplayBuffer:
    """Reorders bounded-lateness samples and drops already-applied duplicates.

    `watermark_grace_sec` bounds how long a late sample is still accepted;
    anything older is routed to a dead-letter table instead of silently
    dropped or silently misordering the state machine.
    """

    def __init__(
        self,
        machine: RunIdleDownStateMachine,
        *,
        watermark_grace_sec: float = 5.0,
    ) -> None:
        self._machine = machine
        self._grace = watermark_grace_sec
        self._buffer: deque[Sample] = deque()
        self._watermark_ts: float = float("-inf")
        self._last_applied_seq: int = -1

    def push(self, sample: Sample) -> list[MachineState]:
        """Admit one sample; return any state transitions it triggers
        once buffered samples clear the watermark."""
        if sample.seq <= self._last_applied_seq:
            return []  # duplicate delivery: idempotent no-op

        self._buffer.append(sample)
        self._watermark_ts = max(self._watermark_ts, sample.ts_utc - self._grace)

        ready = sorted(
            (s for s in self._buffer if s.ts_utc <= self._watermark_ts),
            key=lambda s: s.seq,
        )
        transitions: list[MachineState] = []
        for ready_sample in ready:
            self._buffer.remove(ready_sample)
            if ready_sample.seq <= self._last_applied_seq:
                continue  # a duplicate slipped in while buffered
            result = self._machine.ingest(ready_sample)
            self._last_applied_seq = ready_sample.seq
            if result is not None:
                transitions.append(result)
        return transitions

Sequence numbers, not timestamps, gate idempotency here on purpose: two samples can legitimately share a timestamp under a fast PLC scan, but seq is assigned once per event at the edge and is safe to compare with a plain <=. Applying MQTT QoS 1 semantics correctly means the pipeline, not the broker, is responsible for exactly-once effect even though delivery is only at-least-once — this buffer is that responsibility made explicit rather than assumed away.

Gotchas & anti-patterns #Permalink to this section

• Zero required hold on RUN → IDLE. Treating the run-to-idle debounce as optional “to be safe” turns every scan-cycle flicker into a logged microstop, inflating Performance-loss event counts by an order of magnitude on noisy lines.
• Allowing DOWN → RUN directly. Skipping the IDLE confirmation after a fault clear risks resuming production on a machine whose run permissive was never actually re-verified — always route recovery through IDLE.
• Timestamp-only deduplication. Two genuinely distinct events can share a timestamp on a fast scan cycle; only a monotonic per-asset seq is safe to use for the idempotency check.
• Unbounded reorder buffers. A watermark grace period that is too generous (or absent) lets state resolution lag arbitrarily behind real time on a busy shift; bound it and dead-letter anything older, matching the grace-period discipline in event-to-downtime mapping.
• A shared max_threshold_sec across asset classes. The IDLE → DOWN guard’s threshold is not a machine-wide constant; a conveyor and a CNC cell escalate to downtime at very different dwell times, as their respective threshold-tuning pages establish independently.

Quick reference #Permalink to this section

Related #Permalink to this section

• Event-to-Downtime Mapping — the parent pipeline this state model feeds
• Calculating OEE with Overlapping Maintenance Windows — resolving simultaneous DOWN-triggering alarms
• Microstop Thresholds for Conveyor Motor Jams — a concrete max_threshold_sec and debounce instance of this machine
• Threshold Tuning for Microstops — where the IDLE-band duration boundaries are chosen