stateDiagram-v2
    [*] --> IDLE
    IDLE --> FEEDING: start && zone_clear
    FEEDING --> TRANSPORTING: one part fed
    TRANSPORTING --> DETECTED: sensor active
    DETECTED --> COMPLETE: pulse valid && count=1
    FEEDING --> FAULT: timeout / double feed
    TRANSPORTING --> FAULT: timeout / lost part
    DETECTED --> FAULT: stuck / inconsistent
    FAULT --> IDLE: manual ack + cause removed + area clear
    COMPLETE --> IDLE: new cycle reset

# safe-smoke
from enum import Enum

class State(str, Enum):
    IDLE = "IDLE"
    FAULT = "FAULT"

def recover(state: State, *, operator_ack: bool, cause_removed: bool, area_clear: bool) -> State:
    if state is not State.FAULT:
        return state
    return State.IDLE if operator_ack and cause_removed and area_clear else State.FAULT

assert recover(State.FAULT, operator_ack=False, cause_removed=True, area_clear=True) is State.FAULT
assert recover(State.FAULT, operator_ack=True, cause_removed=False, area_clear=True) is State.FAULT
assert recover(State.FAULT, operator_ack=True, cause_removed=True, area_clear=True) is State.IDLE
print("safe-smoke: manual recovery gate PASS")

{
  "mode": "simulation-only",
  "feed_timeout_ticks": 3,
  "transport_timeout_ticks": 4,
  "sensor_min_on_ticks": 1,
  "sensor_max_on_ticks": 2
}

[
  {"id":"normal","feed_delay":2,"sensor_delay":3,"sensor_on_ticks":1,"part_count":1,"double_feed":false},
  {"id":"feed-boundary","feed_delay":3,"sensor_delay":4,"sensor_on_ticks":2,"part_count":1,"double_feed":false},
  {"id":"feeder-jam","feed_delay":4,"sensor_delay":1,"sensor_on_ticks":1,"part_count":1,"double_feed":false},
  {"id":"conveyor-jam","feed_delay":1,"sensor_delay":5,"sensor_on_ticks":1,"part_count":1,"double_feed":false},
  {"id":"double-feed","feed_delay":1,"sensor_delay":1,"sensor_on_ticks":1,"part_count":2,"double_feed":true},
  {"id":"lost-part","feed_delay":1,"sensor_delay":2,"sensor_on_ticks":1,"part_count":0,"double_feed":false},
  {"id":"sensor-stuck","feed_delay":1,"sensor_delay":2,"sensor_on_ticks":3,"part_count":1,"double_feed":false}
]

from __future__ import annotations

import argparse
import json
from dataclasses import asdict, dataclass, field
from enum import Enum
from pathlib import Path


class State(str, Enum):
    IDLE = "IDLE"
    FEEDING = "FEEDING"
    TRANSPORTING = "TRANSPORTING"
    DETECTED = "DETECTED"
    COMPLETE = "COMPLETE"
    FAULT = "FAULT"


@dataclass
class Result:
    case_id: str
    state: str
    fault: str | None
    feeder_output: str
    conveyor_output: str
    transitions: list[str] = field(default_factory=list)


class CellFSM:
    def __init__(self, cfg: dict) -> None:
        if cfg.get("mode") != "simulation-only":
            raise ValueError("hardware mode is forbidden")
        self.cfg = cfg
        self.state = State.IDLE
        self.feeder = "STOP"
        self.conveyor = "STOP"
        self.fault_code: str | None = None
        self.log: list[str] = []

    def move(self, target: State, reason: str) -> None:
        self.log.append(f"{self.state.value}->{target.value}:{reason}")
        self.state = target

    def safe_stop(self) -> None:
        self.feeder = "STOP"
        self.conveyor = "STOP"

    def fail(self, code: str) -> None:
        self.safe_stop()
        self.fault_code = code
        self.move(State.FAULT, code)

    def run(self, case: dict) -> Result:
        if self.state is not State.IDLE:
            raise RuntimeError("new cycle requires IDLE")
        self.move(State.FEEDING, "start")
        self.feeder = "RUN"
        feed_delay = int(case["feed_delay"])
        if bool(case["double_feed"]) or int(case["part_count"]) > 1:
            self.fail("double-feed")
        elif feed_delay > int(self.cfg["feed_timeout_ticks"]):
            self.fail("feeder-timeout")
        else:
            self.feeder = "STOP"
            self.conveyor = "RUN"
            self.move(State.TRANSPORTING, f"feed-confirmed@{feed_delay}")

        if self.state is State.TRANSPORTING:
            sensor_delay = int(case["sensor_delay"])
            if sensor_delay > int(self.cfg["transport_timeout_ticks"]):
                self.fail("transport-timeout")
            elif int(case["part_count"]) != 1:
                self.fail("part-count-invalid")
            else:
                self.conveyor = "STOP"
                self.move(State.DETECTED, f"sensor-active@{sensor_delay}")

        if self.state is State.DETECTED:
            pulse = int(case["sensor_on_ticks"])
            low = int(self.cfg["sensor_min_on_ticks"])
            high = int(self.cfg["sensor_max_on_ticks"])
            if not low <= pulse <= high:
                self.fail("sensor-pulse-invalid")
            else:
                self.safe_stop()
                self.move(State.COMPLETE, f"pulse-valid:{pulse}")

        assert self.state not in (State.COMPLETE, State.FAULT) or (
            self.feeder == "STOP" and self.conveyor == "STOP"
        )
        return Result(
            str(case["id"]), self.state.value, self.fault_code,
            self.feeder, self.conveyor, list(self.log),
        )

    def recover(self, operator_ack: bool, cause_removed: bool, area_clear: bool) -> bool:
        if self.state is not State.FAULT:
            return False
        if not (operator_ack and cause_removed and area_clear):
            return False
        self.fault_code = None
        self.safe_stop()
        self.move(State.IDLE, "manual-recovery")
        return True


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", type=Path, required=True)
    parser.add_argument("--cases", type=Path, required=True)
    parser.add_argument("--out", type=Path, required=True)
    args = parser.parse_args()
    cfg = json.loads(args.config.read_text(encoding="utf-8"))
    cases = json.loads(args.cases.read_text(encoding="utf-8"))
    results: list[Result] = []
    for case in cases:
        fsm = CellFSM(cfg)
        result = fsm.run(case)
        results.append(result)
        if result.state == "FAULT":
            assert not fsm.recover(False, True, True)
            assert fsm.state is State.FAULT
            assert fsm.recover(True, True, True)
            assert fsm.state is State.IDLE
    states = {result.case_id: result.state for result in results}
    assert states["normal"] == states["feed-boundary"] == "COMPLETE"
    for case_id in ("feeder-jam", "conveyor-jam", "double-feed", "lost-part", "sensor-stuck"):
        assert states[case_id] == "FAULT"
    args.out.write_text(
        json.dumps([asdict(result) for result in results], ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(f"PASS: complete={sum(r.state == 'COMPLETE' for r in results)} fault={sum(r.state == 'FAULT' for r in results)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python lab20_fsm.py --config lab20-config.json --cases lab20-cases.json --out lab20-results.json