"""15 grouped CPU tests of the temporary rule, NOT real fight accuracy.""" from dataclasses import replace from copy import deepcopy import unittest from app.contracts.frames import FrameEnvelope, PoseObservation from app.pose_config import POSE_MODEL from app.fight_config import FightConfig from app.fight_scenario import FightEngine, _body def frame(t, **kw): value = FrameEnvelope("test", "77", "stream", "run", 1, round(t * 10), round(t * 1e9), 1, 1280, 720) return replace(value, **kw) def person(f, x, wrist=None, *, missing=(), score=.9): points = [(0., 0., 0.) for _ in range(17)] coordinates = {5: (x-30, 260), 6: (x+30, 260), 11: (x-25, 460), 12: (x+25, 460), 9: (x+10 if wrist is None else wrist, 360), 10: (x-10, 360)} for i, (px, py) in coordinates.items(): points[i] = (px / f.frame_width, py / f.frame_height, 0. if i in missing else .9) return PoseObservation(f, POSE_MODEL, score, ((x-60)/1280, 200/720, (x+60)/1280, 650/720), tuple(points)) def sample(engine, t, a=460, b=660, *, reverse=False, far=False, miss=(), gap=False): f = frame(t) poses = (person(f, 450, a, missing=miss), person(f, 950 if far else 650, b)) ids = (1, 2) if reverse: poses, ids = poses[::-1], ids[::-1] return engine.evaluate(f, poses, ids, gap_before=gap) def warm(engine): for i in range(7): sample(engine, i / 10) def episode(engine, *, bilateral=False, offset=0.): out = [] for i in range(7, 13): a = 540 if i == 7 or (i == 12 and not bilateral) else 460 b = 580 if i == 12 and bilateral else 660 result = sample(engine, offset + i/10, a, b) out.extend(result.event_candidates) return out class FightTest(unittest.TestCase): def engine(self, **kw): return FightEngine(FightConfig(enabled=True, **kw)) def test_01_defaults_validation_and_off(self): self.assertFalse(FightConfig().enabled) self.assertEqual(FightConfig().required_bursts, 2) self.assertEqual(FightConfig.parse({}).payload()["pair_distance_ratio"], 1.2) for value in ({"enabled":1}, {"extra":1}, {"joint_score":float("nan")}, {"arm_motion_ratio":float("inf")}, {"required_bursts":True}, {"pair_close_hold_seconds":0}, {"pair_retention_seconds":1}): with self.subTest(value=value), self.assertRaises(ValueError): FightConfig.parse(value) engine = FightEngine() self.assertEqual(sample(engine, 0).reason, "DISABLED") self.assertEqual(engine.snapshot()["pair_states"], 0) def test_02_geometry_scores_and_debug_gate_independence(self): f = frame(0) p = person(f, 450) body = _body(p, 1, "0", FightConfig()) self.assertAlmostEqual(body.scale, 200) self.assertEqual(body.center, (450, 360)) self.assertEqual(sum(k[2] >= .15 for k in p.keypoints), 6) self.assertIsNotNone(_body(replace(p, score=.25), 1, "0", FightConfig())) self.assertIsNone(_body(replace(p, score=.249), 1, "0", FightConfig())) self.assertEqual(_body(person(f, 450, missing=(5,)), 1, "0", FightConfig()).torso_source, "right") self.assertIsNone(_body(person(f, 450, missing=(6,11)), 1, "0", FightConfig())) def test_03_far_pair_never_emits(self): engine = self.engine() for i in range(30): self.assertFalse(sample(engine, i/10, a=540 if i%2 else 460, far=True).event_candidates) self.assertEqual(engine.snapshot()["pair_states"], 0) def test_04_near_stationary_pair_never_emits(self): engine = self.engine() for i in range(30): self.assertFalse(sample(engine, i/10).event_candidates) self.assertEqual(engine.snapshot()["arm_bursts"], 0) def test_05_motion_away_from_opponent_never_emits(self): engine = self.engine() for i in range(20): self.assertFalse(sample(engine, i/10, a=460-12*i).event_candidates) def test_06_one_burst_is_not_an_event(self): engine = self.engine() warm(engine) events = [] for i in range(7, 15): events.extend(sample(engine, i/10, a=540 if i>=7 else 460).event_candidates) self.assertFalse(events) self.assertEqual(engine.snapshot()["arm_bursts"], 1) def test_07_two_bursts_current_pair_candidate(self): engine = self.engine() warm(engine) events = episode(engine) self.assertEqual(len(events), 1) e = events[0] self.assertEqual((e.track_ids, e.object_refs, e.scenario), ((1,2), ("0","1"), "FIGHT")) self.assertEqual(e.frame, frame(1.2)) self.assertEqual(e.bboxes[0], person(e.frame, 450, 540).bbox_xyxy) self.assertFalse(hasattr(e, "event_id")) def test_08_one_burst_per_actor_counts_for_pair(self): engine = self.engine() warm(engine) self.assertEqual(len(episode(engine, bilateral=True)), 1) def test_09_sustained_qualifying_motion_is_not_multiple_bursts(self): engine = self.engine() warm(engine) for i in range(7, 13): sample(engine, i/10, a=480+20*(i-7)) self.assertEqual(engine.snapshot()["arm_bursts"], 1) self.assertEqual(engine.snapshot().get("event_candidates", 0), 0) def test_10_latch_and_observed_quiet_rearm(self): engine = self.engine() warm(engine) self.assertEqual(len(episode(engine)), 1) for i in range(13, 27): self.assertFalse(sample(engine, i/10, a=540 if i in (17,22) else 460).event_candidates) for i in range(27, 57): sample(engine, i/10) self.assertEqual(engine.snapshot()["rearmed"], 1) self.assertEqual(len(episode(engine, offset=5.)), 1) def test_11_missing_does_not_create_motion_or_quiet_evidence(self): engine = self.engine() warm(engine) self.assertEqual(len(episode(engine)), 1) for i in range(13, 43): f = frame(i/10) result = engine.evaluate(f, (person(f,450,missing=(9,10)), person(f,650,missing=(9,10))), (1,2)) self.assertFalse(result.event_candidates) self.assertTrue(result.pairs[0].alert) self.assertFalse(engine.evaluate(frame(4.3), (), ()).pairs) self.assertEqual(engine.snapshot()["pair_states"], 1) engine.evaluate(frame(9.3), (), ()) self.assertEqual(engine.snapshot()["pair_states"], 0) def test_12_pair_identity_reordering_and_isolation(self): engine = self.engine() for i in range(7): sample(engine, i/10, reverse=bool(i%2)) all_events = [] for i in range(7, 13): f = frame(i/10) poses = (person(f,1100), person(f,650), person(f,450,540 if i in (7,12) else 460)) r = engine.evaluate(f, poses, (3,2,1)) all_events.extend(r.event_candidates) self.assertEqual(len(all_events), 1) self.assertEqual(all_events[0].track_ids, (1,2)) self.assertEqual(all_events[0].object_refs, ("2","1")) def test_13_pts_run_isolation_and_clear(self): engine = self.engine() warm(engine) for pts in (None, -1, 600_000_000, 500_000_000, float("nan")): f = frame(.7, pts_ns=pts) out = engine.evaluate(f, (person(f,450),person(f,650)), (1,2)) self.assertFalse(out.accepted) self.assertFalse(out.event_candidates) f = frame(.8, run_id="another") with self.assertRaises(ValueError): engine.evaluate(f, (person(f,450),person(f,650)), (1,2)) engine.clear() self.assertTrue(engine.evaluate(f, (), ()).accepted) def test_14_input_validation_and_bounded_state(self): engine = self.engine(max_pairs=1, max_people=3, history_capacity=4) f = frame(0) poses = (person(f,450), person(f,550), person(f,650)) with self.assertRaises(ValueError): engine.evaluate(f, poses, (1,1,2)) with self.assertRaises(ValueError): engine.evaluate(f, (replace(poses[0],model_key="other"),), (1,)) with self.assertRaises(ValueError): engine.evaluate(f, poses, (1,)) for i in range(20): f = frame(i/10) engine.evaluate(f, tuple(person(f,x) for x in (450,550,650)), (1,2,3)) self.assertEqual(engine.snapshot()["pair_states"], 1) self.assertGreater(engine.snapshot()["pair_capacity_rejected"], 0) self.assertLessEqual(len(next(iter(engine._pairs.values())).history), 4) def test_15_gap_single_wrist_and_raw_pose_preservation(self): engine = self.engine() warm(engine) sample(engine, .7, a=540) self.assertFalse(sample(engine, .8, a=460, gap=True).event_candidates) for i in range(9,15): self.assertFalse(sample(engine, i/10, a=540).event_candidates) f = frame(1.5) poses = (person(f,450,missing=(9,)),person(f,650)) saved = deepcopy(poses) engine.evaluate(f, poses, (1,2)) self.assertEqual(poses, saved) self.assertEqual(poses[0].keypoints[9][2], 0) if __name__ == "__main__": unittest.main()