import json
import time
import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch
from app.fight_config import FightConfig
from app.fight_scenario import FightEngine
from app.fight_runtime import FightSettings, FightScenario
from app.crowd_config import CrowdConfig, CROWD_MODEL
from app.crowd_scenario import Crowd
from app.crowd_result import CrowdResultProcessor
from app.secure_control import ControlService, Start, ControlError
from app.pose_config import execution_plan, POSE_MODEL
from app.stream_settings import settings_view, patch_settings, revision
from app.medium_preprocess import medium_preprocess_config
from app.det_result_path import DetResultPath
from app.metadata_samples import MetadataSamples
from test_fight import frame, person, warm, sample, episode
from test_loitering_settings import FakeManager, spec, request
from test_loitering_connection import CopiedDetection, Sink


class IntegrationTests(unittest.TestCase):
    def test_far_frame_preserves_episode(self):
        e=FightEngine(FightConfig(enabled=True));warm(e);self.assertEqual(len(episode(e)),1)
        sample(e,1.3,far=True)
        self.assertIn((1,2),e.latched_pairs())
        for i in range(14,30):
            self.assertFalse(sample(e,i/10,a=540 if i%5==0 else 460).event_candidates)

    def test_invalid_pts_breaks_close_and_quiet_time(self):
        e=FightEngine(FightConfig(enabled=True));warm(e)
        f=frame(.7,pts_ns=None);e.evaluate(f,(person(f,450),person(f,650)),(1,2))
        sample(e,.8)
        self.assertEqual(e._pairs[(1,2)].close_ns,0)
        self.assertEqual(e._pairs[(1,2)].quiet_ns,0)

    def test_web_fight_shared_pose_and_crowd_switch(self):
        m=FakeManager();service=ControlService(m,'server',spec())
        s=service.start_camera(Start(json.dumps(request('FIGHT','FALL','INTRUSION','LOITERING'))))
        self.assertEqual(execution_plan(s)['model'],POSE_MODEL)
        self.assertFalse(s['secure_control_projection']['skippedScenarios'])
        m=FakeManager(s);service=ControlService(m,'server',spec())
        crowd=service.start_camera(Start(json.dumps(request('CROWD'))))
        self.assertEqual(execution_plan(crowd)['model'],CROWD_MODEL)
        self.assertFalse(crowd['tracking']['enabled'])
        self.assertEqual([c['key'] for c in settings_view(crowd)['categories']],['person','head'])
        back=ControlService(FakeManager(crowd),'server',spec()).start_camera(Start(json.dumps(request('FIGHT'))))
        self.assertEqual(execution_plan(back)['model'],POSE_MODEL)
        self.assertTrue(back['tracking']['enabled'])

    def test_cas_noop_and_unsupported_combinations(self):
        s=dict(spec(),fight={'enabled':True})
        changed=patch_settings(s,{'fight':{'person_score':.3}})
        self.assertEqual(revision(changed),revision(patch_settings(changed,{'fight':{'person_score':.3}})))
        for combination in (('FIGHT','FIRE'),('FIGHT','CROWD')):
            m=FakeManager(spec())
            with self.assertRaises(ControlError):
                ControlService(m,'server',spec()).start_camera(Start(json.dumps(request(*combination))))
            self.assertFalse(m.applied)

    def test_fight_current_pair_event_and_roi(self):
        a=FightScenario(FightSettings.parse({'enabled':True}),event_id_factory=lambda:'fight-event')
        emitted=[]
        for i in range(13):
            f=frame(i/10);p=(person(f,450,540 if i in (7,12) else 460),person(f,650))
            rows,events=a.evaluate(f,p,(1,2));emitted.extend(events)
        self.assertEqual(len(emitted),1)
        self.assertEqual(emitted[0].object_ref,'0');self.assertEqual(emitted[0].track_id,1)
        self.assertTrue(all(any(d.state=='FIGHT' for d in row) for row in rows))
        a.clear();self.assertFalse(a.engine.latched_pairs())

    def test_crowd_worker_no_tracking_or_events(self):
        sink=Sink();errors=Mock()
        c=CrowdResultProcessor(Crowd(CrowdConfig(enabled=True)),camera_id=77)
        with patch.object(MetadataSamples,'from_environment',return_value=None):
            p=DetResultPath(SimpleNamespace(identity=('test','77','stream','run',1)),None,CROWD_MODEL,.5,
                Mock(add=Mock(return_value=1)),lambda:True,errors,crowd=c,publisher=sink)
        try:
            p.start()
            for i in range(4):
                f=frame(i);d=CopiedDetection(score=.9,bbox_xyxy=(.1,.1,.2,.4),keypoints=None)
                copied=SimpleNamespace(width=1280,height=720,detections=(d,),frame_number=i,pts=f.pts_ns)
                n=p.status().get('processed',0);p.submit(f,copied)
                end=time.monotonic()+2
                while p.status().get('processed',0)==n and not p.status()['error'] and time.monotonic()<end:time.sleep(.001)
                self.assertIsNone(p.status()['error'])
            result=p.latest_result().snapshot()
            self.assertFalse(result['events'])
            self.assertIsNone(result['objects'][0]['trackId'])
            self.assertIn('P≈',result['objects'][0]['labelText'])
        finally:p.close(2)
        self.assertEqual(c.engine.diagnostics()['samples'],0)

    def test_preprocess_contract_is_separate(self):
        c=medium_preprocess_config(1,fp16_input=True,resolution=512)
        self.assertIn('1;3;512;512',c);self.assertIn('libcontextra_crowd_preprocess.so',c)
        self.assertIn('1;3;576;576',medium_preprocess_config(1,fp16_input=True))


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