Marius Mutu
9e8cbafbd4
- 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>
112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
"""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}"
|
|
)
|