"""CPU-only FIRE evidence tests. No SDK, network, DB or wall-clock sleeps."""
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 flame(f, score=.9, box=(.1, .1, .3, .3), kind=1):
    return DetectionObservation(f, FIRE_MODEL, kind, score, box)


class FireConfigTest(unittest.TestCase):
    def test_defaults(self):
        c = FireScenarioConfig()
        self.assertFalse(c.enabled)
        self.assertEqual((c.confidence, c.window_seconds, c.min_observation_seconds,
            c.positive_ratio, c.max_observation_gap_seconds, c.clear_seconds), (.5, 2., 1., .6, .5, 5.))

    def test_roundtrip(self):
        c = FireScenarioConfig(enabled=True)
        self.assertEqual(c, FireScenarioConfig.parse(c.settings_payload()))

    def test_no_public_temporal_fields(self):
        c = FireScenarioConfig()
        self.assertFalse(set(c.payload()) & {'window_seconds','min_observation_seconds','positive_ratio',
            'max_observation_gap_seconds','clear_seconds','max_samples'})

    def test_empty_roi_is_full_frame(self):
        for value in (None, [], ()):
            self.assertEqual(FireScenarioConfig(roi=value).roi, FULL_FRAME)

    def test_invalid_polygon(self):
        for roi in ([[0,0]], [[0,0],[1,1],[0,1],[1,0]], [[0,0],[2,0],[1,1]], [[0,0]]*3):
            with self.subTest(roi=roi), self.assertRaises(ValueError): FireScenarioConfig(roi=roi)

    def test_nonfinite_and_boolean_numbers(self):
        for field in ('confidence','window_seconds','min_observation_seconds','positive_ratio',
                      'clear_seconds','max_observation_gap_seconds'):
            for val in (float('nan'), float('inf'), -float('inf'), True, '0.5'):
                with self.subTest(field=field,val=val), self.assertRaises(ValueError):
                    FireScenarioConfig(**{field:val})

    def test_timer_bounds(self):
        for field in ('window_seconds','min_observation_seconds','clear_seconds','max_observation_gap_seconds'):
            for val in (0, -1, 1e-12, 3601):
                with self.subTest(field=field,val=val), self.assertRaises(ValueError):
                    FireScenarioConfig(**{field:val})

    def test_ratio_bounds(self):
        for val in (0,-.1,1.01):
            with self.assertRaises(ValueError): FireScenarioConfig(positive_ratio=val)

    def test_window_relationship(self):
        for changes in ({'min_observation_seconds':3},{'max_observation_gap_seconds':3}):
            with self.assertRaises(ValueError): FireScenarioConfig(**changes)

    def test_unknown_fields(self):
        for value in (None, [], {'random':1}):
            with self.assertRaises(ValueError): FireScenarioConfig.parse(value)

    def test_sample_capacity(self):
        for value in (1,4097,2.5,True):
            with self.assertRaises(ValueError): FireScenarioConfig(max_samples=value)

    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_nested_patch_preserves_timers(self):
        a = FireConfig(True, {'enabled':True,'clear_seconds':8.})
        b = a.patched({'scenario':{'confidence':.7}})
        self.assertEqual(b.scenario.clear_seconds, 8.)
        self.assertEqual(a.scenario.confidence,.5)
        for bad in (None, [], {'unknown':1}):
            with self.assertRaises(ValueError): a.patched(bad)


class FireEvidenceTest(unittest.TestCase):
    def setUp(self):
        self.ids = count(1)
        self.engine = self.make()
        self.seq = 0

    def make(self, **options):
        return Fire(FireScenarioConfig(enabled=True, **options),
            event_id_factory=lambda: 'fire-'+str(next(self.ids)), wall_ns=lambda:1_700_000_000_000_000_000)

    def step(self, seconds, positive=True, score=.9, boxes=None, kind=1, **changes):
        f = frame(self.seq, seconds, **changes); self.seq += 1
        obs = tuple(flame(f, score, box, kind) for box in (boxes or [(.1,.1,.3,.3)])) if positive else ()
        return self.engine.evaluate(f, obs)

    def confirmed(self):
        out = [self.step(i/10) for i in range(11)]
        events = [e for r in out for e in r.events]
        self.assertEqual(len(events),1)
        self.assertEqual(events[0].frame.pts_ns, 1_000_000_000)
        return events[0]

    def test_first_positive_not_event(self):
        self.assertFalse(self.step(0).events)

    def test_minimum_one_second(self):
        for i in range(10): self.assertFalse(self.step(i/10).events)
        self.assertEqual(len(self.step(1).events),1)

    def test_current_negative_cannot_confirm(self):
        for i in range(10): self.step(i/10)
        self.assertFalse(self.step(1,False).events)
        self.assertEqual(len(self.step(1.1).events),1)

    def test_ratio_exactly_sixty_percent(self):
        for t,p in zip((0,.25,.5,.75),(True,False,True,False)): self.step(t,p)
        self.assertEqual(len(self.step(1).events),1)

    def test_ratio_below_threshold(self):
        for t,p in zip((0,.25,.5,.75),(False,False,True,False)): self.step(t,p)
        self.assertFalse(self.step(1).events)

    def test_ratio_is_sample_count_not_time_weighted(self):
        for t,p in zip((0,.45,.5,.95),(True,False,True,False)): self.step(t,p)
        self.assertEqual(len(self.step(1).events),1)
        self.assertEqual(sum(p for _,p in self.engine.samples),3)

    def test_exact_gap_allowed(self):
        self.step(0); self.step(.5)
        self.assertEqual(len(self.step(1).events),1)

    def test_gap_breaks_confirmation(self):
        self.step(0); self.step(.4)
        self.assertFalse(self.step(1).events)
        self.assertEqual(len(self.engine.samples),1)

    def test_long_gap_does_not_warmup(self):
        self.step(0)
        self.assertFalse(self.step(2).events)

    def test_missing_pts_excluded_and_breaks_continuity(self):
        self.step(0); self.step(.4)
        r=self.step(None)
        self.assertFalse(r.events)
        self.assertFalse(self.engine.samples)
        self.assertEqual(r.decisions[0].state,'TIME_UNAVAILABLE')
        self.assertFalse(self.step(.8).events)
        self.assertEqual(len(self.engine.samples),1)

    def test_invalid_pts_excluded(self):
        for pts in (-1,True,1.5,2**63):
            f=frame(self.seq,0,pts_ns=pts); self.seq+=1
            self.assertFalse(self.engine.evaluate(f,(flame(f),)).events)
        self.assertEqual(self.engine.metrics['samples'],0)

    def test_reversed_pts_breaks_evidence(self):
        self.step(.5); self.step(.8)
        self.assertFalse(self.step(.6).events)
        self.assertFalse(self.engine.samples)
        self.assertFalse(self.step(.9).events)

    def test_equal_pts_not_double_counted(self):
        self.step(0); self.step(0)
        self.assertEqual(self.engine.metrics['samples'],1)
        self.assertFalse(self.engine.samples)

    def test_stale_sequence_rejected(self):
        f=frame(2,.3); self.engine.evaluate(f,(flame(f),))
        before=tuple(self.engine.samples)
        for seq in (2,1):
            f=frame(seq,.4)
            self.assertFalse(self.engine.evaluate(f,(flame(f),)).accepted)
        self.assertEqual(tuple(self.engine.samples),before)

    def test_skipped_sequence_is_not_negative_sample(self):
        for seq,t in ((0,0),(2,.5),(9,1)):
            f=frame(seq,t); r=self.engine.evaluate(f,(flame(f),))
        self.assertEqual(len(r.events),1)
        self.assertEqual(len(self.engine.samples),3)

    def test_foreign_run_rejected(self):
        self.step(0)
        for changes in ({'run_id':'other'},{'run_generation':2},{'source_id':'88'}):
            with self.assertRaises(ValueError): self.step(.1,**changes)

    def test_observation_association_rejected(self):
        f=frame(0,0); other=frame(1,.1)
        for obs in ((flame(other),),(replace(flame(f),model_key='RFDETRMedium'),),(replace(flame(f),class_id=2),)):
            with self.assertRaises(ValueError): self.engine.evaluate(f,obs)

    def test_smoke_only_is_negative_not_flame(self):
        for i in range(11): self.assertFalse(self.step(i/10,kind=0).events)
        self.assertEqual(self.engine.metrics['positive_samples'],0)

    def test_confidence_equality_and_low(self):
        self.assertEqual(self.step(0,score=.5).indices,(0,))
        self.assertEqual(self.step(.1,score=.499).indices,())

    def test_multiple_flames_one_positive_sample(self):
        r=self.step(0,boxes=[(.1,.1,.3,.3),(.5,.5,.8,.8)])
        self.assertEqual(len(r.indices),2)
        self.assertEqual(self.engine.metrics['positive_samples'],1)

    def test_highest_score_representative_and_tie(self):
        for i in range(10): self.step(i/10)
        f=frame(self.seq,1)
        obs=(flame(f,.7),flame(f,.95,(.4,.4,.6,.6)),flame(f,.95,(.7,.7,.9,.9)))
        event=self.engine.evaluate(f,obs).events[0]
        self.assertEqual(event.object_ref,'1')
        self.assertEqual(event.bbox,obs[1].bbox_xyxy)
        self.assertIsNone(event.track_id)
        self.assertEqual(event.ai_recorded_at_wall_ms,1_700_000_000_000)

    def test_episode_once(self):
        self.confirmed()
        for i in range(11,100): self.assertFalse(self.step(i/10).events)
        self.assertEqual(self.engine.metrics['events_created'],1)

    def test_clear_requires_observed_five_seconds(self):
        self.confirmed()
        for i in range(12,62): self.step(i/10,False)
        self.assertIsNotNone(self.engine.event_id)
        self.step(6.2,False)
        self.assertIsNone(self.engine.event_id)

    def test_long_silence_does_not_clear(self):
        original=self.confirmed().event_id
        self.step(20,False)
        self.assertEqual(self.engine.event_id,original)
        self.assertEqual(self.engine.clear_from,20_000_000_000)

    def test_negative_gap_restarts_clear(self):
        self.confirmed()
        for i in range(12,42): self.step(i/10,False)
        self.step(8,False)
        self.assertIsNotNone(self.engine.event_id)
        self.assertEqual(self.engine.clear_from,8_000_000_000)

    def test_positive_resets_clear(self):
        self.confirmed(); self.step(1.2,False); self.step(1.5,False)
        self.step(1.6)
        self.assertIsNone(self.engine.clear_from)
        self.step(1.7,False)
        self.assertEqual(self.engine.clear_from,1_700_000_000)

    def test_invalid_time_does_not_clear_latch(self):
        e=self.confirmed(); self.step(1.2,False); self.step(None,False)
        self.assertEqual(self.engine.event_id,e.event_id)
        self.assertIsNone(self.engine.clear_from)

    def test_new_episode_after_clear(self):
        first=self.confirmed().event_id
        for i in range(12,63): self.step(i/10,False)
        events=[]
        for i in range(63,91): events.extend(self.step(i/10).events)
        self.assertEqual(len(events),1)
        self.assertNotEqual(first,events[0].event_id)

    def test_wall_clock_not_evidence(self):
        self.engine.clock=lambda: (_ for _ in ()).throw(AssertionError('wall clock used for evidence'))
        self.confirmed(); self.engine.expire()
        self.assertIsNotNone(self.engine.event_id)

    def test_bounded_rolling_history(self):
        for i in range(1000): self.step(i/100)
        self.assertLessEqual(len(self.engine.samples),201)
        self.assertLessEqual(self.engine.samples[-1][0]-self.engine.samples[0][0],2_000_000_000)

    def test_sample_capacity_resets_conservatively(self):
        self.engine=self.make(max_samples=2)
        self.step(0); self.step(.4); r=self.step(.8)
        self.assertEqual(len(self.engine.samples),1)
        self.assertFalse(r.events)
        self.assertEqual(self.engine.metrics['sample_capacity_reset'],1)

    def test_roi_center_not_bottom_center(self):
        self.engine=self.make(roi=((0,0),(1,0),(1,.5),(0,.5)))
        self.assertEqual(self.step(0,boxes=[(.1,.1,.3,.8)]).indices,(0,))
        self.assertFalse(self.step(.1,boxes=[(.1,.5,.3,.8)]).indices)

    def test_roi_boundary_inside(self):
        self.engine=self.make(roi=((0,0),(.5,0),(.5,1),(0,1)))
        self.assertEqual(self.step(0,boxes=[(.4,.1,.6,.3)]).indices,(0,))

    def test_clear_allows_new_run(self):
        self.confirmed(); self.engine.clear()
        self.assertFalse(self.engine.samples)
        self.assertIsNone(self.engine.event_id)
        self.step(0,run_id='new-run',run_generation=2)

    def test_invalid_event_factory_is_fatal(self):
        self.engine.event_id_factory=lambda:''
        for i in range(10): self.step(i/10)
        with self.assertRaises(ValueError): self.step(1)

    def test_adapter_compacts_reference_and_never_reuses_bbox(self):
        for i in range(10): self.step(i/10)
        f=frame(self.seq,1)
        obs=(flame(f,.9,kind=0),flame(f,.6),flame(f,.9,(.5,.5,.8,.8)))
        r=evaluate_result(self.engine,f,obs)
        self.assertEqual(len(r.commands),2)
        self.assertEqual(r.events[0].object_ref,'1')
        self.assertTrue(all(c.color=='#FF3030' and c.track_id is None for c in r.commands))
        negative=evaluate_result(self.engine,frame(self.seq+1,1.1),())
        self.assertFalse(negative.commands)
        self.assertFalse(negative.events)
        self.assertIsNotNone(self.engine.event_id)


if __name__=='__main__': unittest.main()