"""CPU-only FIRE v2 evidence tests.""" import unittest from dataclasses import replace from itertools import count from app.contracts.frames import FrameEnvelope, DetectionObservation from app.fire_config import FIRE_MODEL, FireConfig from app.fire_scenario import Fire, FireScenarioConfig, FULL_FRAME from app.fire_result import evaluate_result def frame(seq, seconds, **changes): value = FrameEnvelope('test-server','77','test-stream','run-a',1,seq, None if seconds is None else round(seconds*1e9),1_700_000_000_000,1280,720) return replace(value, **changes) def observation(f, kind, score=.9, box=(.1,.1,.3,.3)): return DetectionObservation(f,FIRE_MODEL,kind,score,box) class FireConfigTest(unittest.TestCase): def test_defaults_and_roundtrip(self): cfg=FireScenarioConfig() self.assertEqual((cfg.flame_confidence,cfg.smoke_confidence),(.5,.5)) self.assertEqual((cfg.flame_window_seconds,cfg.smoke_window_seconds),(2.,2.)) self.assertEqual((cfg.flame_positive_ratio,cfg.smoke_positive_ratio),(.6,.6)) self.assertEqual(cfg,FireScenarioConfig.parse(cfg.settings_payload())) def test_saved_legacy_settings_migrate_to_flame(self): cfg=FireScenarioConfig.parse(dict(enabled=True,confidence=.7,window_seconds=3, positive_ratio=.8)) self.assertEqual((cfg.flame_confidence,cfg.flame_window_seconds, cfg.flame_positive_ratio),(.7,3,.8)) self.assertEqual(cfg.smoke_confidence,.5) def test_public_payload_keeps_compact_wire(self): payload=FireScenarioConfig().payload() self.assertIn('confidence',payload) self.assertFalse(set(payload)&{'flame_window_seconds','smoke_window_seconds', 'min_observation_seconds','max_observation_gap_seconds','clear_seconds'}) def test_invalid_values(self): for field in ('flame_confidence','smoke_confidence', 'flame_positive_ratio','smoke_positive_ratio'): with self.subTest(field=field),self.assertRaises(ValueError): FireScenarioConfig(**{field:float('nan')}) for field in ('flame_window_seconds','smoke_window_seconds', 'min_observation_seconds','clear_seconds','max_observation_gap_seconds'): with self.subTest(field=field),self.assertRaises(ValueError): FireScenarioConfig(**{field:0}) def test_provider_toggle_stays_distinct(self): self.assertEqual(FireConfig(enabled=True).payload(),{'enabled':True}) self.assertFalse(FireConfig(enabled=True).event_enabled) with self.assertRaises(ValueError): FireConfig(scenario={'enabled':True}) def test_roi_empty_is_full_frame(self): for value in (None,[],()): self.assertEqual(FireScenarioConfig(roi=value).roi,FULL_FRAME) class FireEvidenceTest(unittest.TestCase): def setUp(self): self.ids=count(1) self.engine=Fire(FireScenarioConfig(enabled=True), event_id_factory=lambda:'fire-'+str(next(self.ids)), wall_ns=lambda:1_700_000_000_000_000_000) self.seq=0 def step(self, seconds, smoke=True, flame=False, smoke_score=.9, flame_score=.9): f=frame(self.seq,seconds);self.seq+=1 rows=[] if smoke: rows.append(observation(f,0,smoke_score)) if flame: rows.append(observation(f,1,flame_score,(.5,.5,.8,.8))) return self.engine.evaluate(f,tuple(rows)) def warm(self, *, smoke=True, flame=False): result=None for i in range(11): result=self.step(i/10,smoke=smoke,flame=flame) return result def test_smoke_alone_can_trigger(self): result=self.warm(smoke=True,flame=False) self.assertEqual(len(result.events),1) event=result.events[0] self.assertEqual((event.class_id,event.category_key,event.trigger_reason), (0,'smoke_visible','SMOKE')) def test_flame_alone_can_trigger(self): result=self.warm(smoke=False,flame=True) event=result.events[0] self.assertEqual((event.class_id,event.category_key,event.trigger_reason), (1,'flame_visible','FLAME')) def test_both_trigger_once_and_highest_score_is_representative(self): for i in range(10): self.step(i/10,smoke=True,flame=True,smoke_score=.8,flame_score=.95) result=self.step(1,smoke=True,flame=True,smoke_score=.8,flame_score=.95) self.assertEqual(len(result.events),1) event=result.events[0] self.assertEqual(event.trigger_reason,'FLAME_AND_SMOKE') self.assertEqual(event.class_id,1) self.assertEqual(event.object_ref,'1') def test_current_object_required_for_new_event(self): for i in range(10): self.step(i/10,smoke=True) result=self.step(1,smoke=False,flame=False) self.assertFalse(result.events) self.assertIsNone(self.engine.event_id) def test_clear_requires_both_negative(self): self.warm(smoke=True,flame=False) for i in range(12,70): # Smoke remains positive, therefore flame-negative alone never clears. self.step(i/10,smoke=True,flame=False) self.assertIsNotNone(self.engine.event_id) start=self.seq base=7.0 for i in range(51): self.step(base+i/10,smoke=False,flame=False) self.assertIsNone(self.engine.event_id) def test_gap_is_not_negative_evidence(self): self.warm(smoke=True,flame=False) event_id=self.engine.event_id self.step(20,smoke=False,flame=False) self.assertEqual(self.engine.event_id,event_id) self.assertEqual(self.engine.clear_from,20_000_000_000) def test_class_specific_confidence(self): engine=Fire(FireScenarioConfig(enabled=True,smoke_confidence=.7,flame_confidence=.3)) f=frame(0,0) result=engine.evaluate(f,(observation(f,0,.69),observation(f,1,.3))) self.assertEqual(result.indices,(1,)) def test_visualization_colors_and_compacted_reference(self): for i in range(10): self.step(i/10,smoke=True,flame=True,smoke_score=.8,flame_score=.95) f=frame(self.seq,1);self.seq+=1 obs=(observation(f,0,.8),observation(f,1,.95,(.5,.5,.8,.8))) result=evaluate_result(self.engine,f,obs) self.assertEqual([c.color for c in result.commands],['#A855F7','#FF3030']) self.assertEqual(result.events[0].object_ref,'1') self.assertEqual(result.events[0].payload()['categoryKey'],'flame_visible') def test_stale_sequence_rejected(self): f=frame(2,.3);self.engine.evaluate(f,(observation(f,0),)) before=self.engine.metrics.copy() f=frame(2,.4) self.assertFalse(self.engine.evaluate(f,(observation(f,0),)).accepted) self.assertEqual(self.engine.metrics['stale_frame'],before['stale_frame']+1) if __name__=='__main__': unittest.main()