feat: calibration/ corpus + scenarii regresie FSM

- calibration/frames/: 16 PNG-uri ground-truth numite {ts}_{color}.png,
  copiate din logs/fires (izolate de samples/ și logs/fires/ care se pot goli)
- calibration/calibration_labels.json: mutat din samples/, curățat de entries
  cu fișiere inexistente, extins la acoperire completă 7 culori → 16/16 PASS
- calibration/scenarios.json: 8 secvențe FSM (BUY/SELL full cycle, phase_skip,
  catchup, post-fire suppression) pe frame-uri reale
- tests/test_scenarios_regression.py: parametrizat pe scenarios.json, asertează
  color+state+reason+trigger+alerts+scheduler prin pipeline-ul
  Detector → _handle_tick
- docs: README + CLAUDE reflectă noua structură, incidentul 2026-04-20/21
  (pixel saturat UNKNOWN → FSM blocat în PRIMED → polling continuu) +
  troubleshooting pentru trigger UNKNOWN

Pytest: 184 → 192 passed (+8 scenarii regresie).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-04-21 08:32:11 +03:00
parent bed79fcc35
commit 9e8cbafbd4
24 changed files with 578 additions and 78 deletions

View File

@@ -0,0 +1,111 @@
"""Image-backed regression scenarios.
Each scenario in `calibration/scenarios.json` is a sequence of real PNG frames
fed through the full Detector → _handle_tick pipeline. Asserts per step:
- detector classifies the exact expected color (accepted=True)
- FSM transition reason/state + trigger match
- notifier receives exactly the expected new alert kinds
- scheduler-running flag (mirroring _handle_fsm_result) matches
Frames live in calibration/frames/ (self-contained, survives logs/fires/ purges).
"""
from __future__ import annotations
import json
from pathlib import Path
import cv2
import pytest
from atm.config import Config
from atm.detector import Detector
from atm.main import _handle_tick
from atm.state_machine import StateMachine
from tests.test_handle_tick import FakeNotifier, FakeAudit
_SCENARIOS_PATH = Path("calibration/scenarios.json")
_CONFIGS_DIR = Path("configs")
# Reasons that stop the screenshot scheduler (mirrors main.py:_handle_fsm_result).
_SCHEDULER_STOP_REASONS = {"fire", "cooled", "phase_skip", "opposite_rearm"}
def _load_scenarios() -> list[dict]:
return json.loads(_SCENARIOS_PATH.read_text(encoding="utf-8"))
@pytest.fixture(scope="module")
def cfg() -> Config:
return Config.load_current(_CONFIGS_DIR)
@pytest.mark.parametrize(
"scenario", _load_scenarios(), ids=lambda s: s["id"]
)
def test_scenario(scenario: dict, cfg: Config) -> None:
fsm = StateMachine(lockout_s=cfg.lockout_s)
notif = FakeNotifier()
audit = FakeAudit()
detector = Detector(cfg=cfg, capture=lambda: None)
scheduler_running = False
first_accepted = True
for i, step in enumerate(scenario["steps"]):
frame_path = Path(step["frame"])
assert frame_path.exists(), f"{scenario['id']}[{i}]: missing frame {frame_path}"
frame = cv2.imread(str(frame_path))
assert frame is not None, f"{scenario['id']}[{i}]: cv2.imread failed"
res = detector.step(ts=float(i), frame=frame)
assert res.accepted, (
f"{scenario['id']}[{i}]: detector rejected {frame_path.name} "
f"(match={res.match.name if res.match else None}, "
f"d={res.match.distance if res.match else None}, rgb={res.rgb})"
)
assert res.color == step["expected_color"], (
f"{scenario['id']}[{i}]: color mismatch — expected "
f"{step['expected_color']}, got {res.color}"
)
alerts_before = len(notif.alerts)
tr = _handle_tick(
fsm, res.color, float(i), notif, audit,
first_accepted=first_accepted, cfg=cfg,
)
first_accepted = False
assert tr is not None, f"{scenario['id']}[{i}]: _handle_tick returned None"
assert tr.reason == step["expected_reason"], (
f"{scenario['id']}[{i}]: reason mismatch — expected "
f"{step['expected_reason']}, got {tr.reason}"
)
assert tr.next.value == step["expected_state"], (
f"{scenario['id']}[{i}]: state mismatch — expected "
f"{step['expected_state']}, got {tr.next.value}"
)
assert tr.trigger == step["expected_trigger"], (
f"{scenario['id']}[{i}]: trigger mismatch — expected "
f"{step['expected_trigger']}, got {tr.trigger}"
)
new_alerts = [a.kind for a in notif.alerts[alerts_before:]]
assert new_alerts == step["expected_new_alerts"], (
f"{scenario['id']}[{i}]: alert mismatch — expected "
f"{step['expected_new_alerts']}, got {new_alerts}"
)
# Scheduler lifecycle (mirrors _handle_fsm_result main.py:953-957)
if tr.reason == "prime" and not scheduler_running:
scheduler_running = True
elif tr.reason in _SCHEDULER_STOP_REASONS and scheduler_running:
scheduler_running = False
# Also stops on trigger fire (main.py:960-964)
if tr.trigger and not tr.locked and scheduler_running:
scheduler_running = False
assert scheduler_running == step["expected_scheduler_running"], (
f"{scenario['id']}[{i}]: scheduler_running mismatch — expected "
f"{step['expected_scheduler_running']}, got {scheduler_running}"
)