"""FALL-R2-CONFIG-02: public tuning contracts and branch-local overrides.""" from copy import deepcopy from dataclasses import FrozenInstanceError import json import math import tempfile import unittest from unittest.mock import patch from app.fall_r2.config import ANGLE_LIMITS, R2Config, R2ConfigError, block from app.fall_r2.core import R2Core from app.fall_r2.decision import decide from app.fall_r2.reference import features from app.fall_runtime import FallSettings, FallScenario from app.secure_control import ControlService, Start, command_error from app.stream_settings import patch_settings, revision, settings_view from test_fall_r2 import saved, start from test_loitering_settings import FakeManager, request, spec def body(parameters, *enabled): value = request(*(enabled or ('FALL',))) next(s for s in value['scenarios'] if s['scenarioCode'] == 'FALL')['parameters'] = parameters return value class ConfigTests(unittest.TestCase): def test_defaults_bounds_and_numeric_types(self): self.assertEqual(R2Config.parse({'schema_version': 1}).payload(), dict(schema_version=1, r1_torso_min_deg=50., r1_leg_min_deg=50., d1_knee_min_deg=140.)) for key, (lo, hi) in ANGLE_LIMITS.items(): for value in (lo, hi, (lo + hi) / 2 + .25): with self.subTest(key=key, value=value): self.assertEqual(getattr(R2Config.parse({'schema_version': 1, key: value}), key), value) for value in (None, True, str(lo), [], {}, lo - .01, hi + .01, float('nan'), float('inf'), 10**400): with self.subTest(key=key, value=repr(value)), self.assertRaisesRegex(R2ConfigError, 'FALL_R2_ANGLE_INVALID'): R2Config.parse({'schema_version': 1, key: value}) for value in ({}, {'schema_version': True}, {'schema_version': 1.0}, {'schema_version': 2}, {'schema_version': 1, 'bbox_aspect': 1.1}, None, []): with self.subTest(value=value), self.assertRaises(R2ConfigError): R2Config.parse(value) def test_immutable_partial_and_reset(self): old = R2Config(1, 55., 56., 150.) new = old.patched({'schema_version': 1, 'r1_leg_min_deg': 58.5}) self.assertEqual(new, R2Config(1, 55., 58.5, 150.)) self.assertIs(old.patched({'schema_version': 1, 'reset_to_defaults': True}), old) with self.assertRaises(FrozenInstanceError): old.r1_leg_min_deg = 60. for key in ANGLE_LIMITS: with self.subTest(key=key), self.assertRaisesRegex(R2ConfigError, 'FALL_R2_RESET_INVALID'): old.patched({'schema_version': 1, 'reset_to_defaults': True, key: None}) with self.assertRaisesRegex(R2ConfigError, 'FALL_R2_FIELDS_INVALID'): block({'schema_version': 1, 'reset_to_defaults': True}) def test_marker_and_explicit_defaults_same_settings(self): a = FallSettings.parse({'engine': 'r2'}) b = FallSettings.parse({'engine': 'r2', 'r2': R2Config().payload()}) self.assertEqual(a.settings_payload(), b.settings_payload()) self.assertEqual(command_error(R2ConfigError('FALL_R2_ANGLE_INVALID')), 'FALL_R2_ANGLE_INVALID') class StartTests(unittest.TestCase): def tuned(self): return start({'fall_r2': R2Config(1, 55., 56., 150.).payload(), 'person_confidence': .4, 'pose_confidence': .3, 'cooldown_seconds': 12})[1] def test_omission_partial_reset_and_explicit_restore(self): old = self.tuned() for value in ({}, {'fall_r2': {'schema_version': 1}}, {'down_angle_deg': 20, 'recovery_hold_seconds': 99}): self.assertEqual(start(value, old)[1], old) _, changed = start({'fall_r2': {'schema_version': 1, 'r1_leg_min_deg': 58}}, old) self.assertEqual(changed['fall']['r2'], R2Config(1, 55., 58., 150.).payload()) _, reset = start({'fall_r2': {'schema_version': 1, 'reset_to_defaults': True}}, changed) self.assertEqual(reset['fall']['r2'], changed['fall']['r2']) self.assertEqual((reset['fall']['person_score'], reset['fall']['joint_score'], reset['event_cooldowns']['FALL']), (.25, .1, 10.)) _, restored = start({'fall_r2': R2Config().payload()}, changed) self.assertEqual(restored['fall']['r2'], R2Config().payload()) self.assertEqual((restored['fall']['person_score'], restored['fall']['joint_score'], restored['event_cooldowns']['FALL']), (.4, .3, 12.)) for name in ('roi', 'enabled', 'engine'): self.assertEqual(restored['fall'][name], changed['fall'][name]) def test_rejection_is_atomic_even_after_other_projection(self): old = self.tuned() bad = [dict(schema_version=1, r1_leg_min_deg=49), dict(schema_version=1, d1_knee_min_deg='150'), dict(schema_version=1, reset_to_defaults=True, r1_leg_min_deg=50), dict(schema_version=1, surprise=50), dict(r1_leg_min_deg=55)] for config in bad: with self.subTest(config=config): manager = FakeManager(old); before = deepcopy(manager.group.spec) req = body({'fall_r2': config}) with self.assertRaises(ValueError): ControlService(manager, 'server', spec()).start_camera(Start(json.dumps(req))) self.assertFalse(manager.applied) self.assertEqual(manager.group.spec, before) with self.assertRaisesRegex(R2ConfigError, 'FALL_R2_RESET_INVALID'): start({'fall_r2': {'schema_version': 1, 'reset_to_defaults': True}, 'person_confidence': .4}, old) with self.assertRaisesRegex(R2ConfigError, 'FALL_R2_NOT_ENABLED'): start({'fall_r2': R2Config().payload()}, spec()) def test_partial_private_patch_and_legacy_rollback(self): old = self.tuned() changed = patch_settings(old, {'fall': {'r2': {'schema_version': 1, 'r1_leg_min_deg': 59}}}) self.assertEqual(changed['fall']['r2'], R2Config(1, 55., 59., 150.).payload()) self.assertEqual(patch_settings(changed, {'fall': {'r2': {'schema_version': 1}}}), changed) self.assertNotIn('r2', patch_settings(changed, {'fall': {'engine': 'legacy'}})['fall']) with self.assertRaises(R2ConfigError): patch_settings(changed, {'fall': {'r2': {'schema_version': 1, 'reset_to_defaults': True}}}) def test_fight_and_other_scenarios_preserved(self): old = self.tuned(); old['fight']['engine'] = 'r5' old['fight']['r5'] = dict(schema_version=1, window_seconds=1., min_samples=2, contact_min_samples=3) _, old = start({}, old) _, new = start({'fall_r2': {'schema_version': 1, 'd1_knee_min_deg': 160}}, old) for key in old: if key != 'fall': self.assertEqual(new[key], old[key], key) self.assertEqual(old['fall']['person_score'], new['fall']['person_score']) def test_real_manager_mock_start_persistence_noop_and_changes(self): from app.stream_manager import StreamManager from app.stream_config_store import ConfigPersistenceError with tempfile.TemporaryDirectory() as root: manager = StreamManager(root+'/base', root+'/overlay') group = manager.add(self.tuned()) service = ControlService(manager, 'server', spec()) req = body({'fall_r2': {'schema_version': 1}}) before = manager.settings('stream') with patch.object(group, 'start', return_value=group), \ patch.object(group, 'restart', side_effect=AssertionError('No-op must not restart')), \ patch.object(manager.config_store, 'update', side_effect=AssertionError('No-op must not write')): service.start_camera(Start(json.dumps(req))) self.assertEqual(manager.settings('stream'), before) change = {'fall': {'r2': {'schema_version': 1, 'd1_knee_min_deg': 155}}} with patch.object(group, 'start', side_effect=AssertionError('No live Start')): out = manager.patch_settings('stream', before['configured']['revision'], change) self.assertEqual(out['configured']['fall']['r2']['d1_knee_min_deg'], 155.) restarted = StreamManager(root+'/base', root+'/overlay') restarted.load_and_autostart() self.assertEqual(restarted.settings('stream')['effective']['fall']['r2'], out['configured']['fall']['r2']) rev = out['configured']['revision'] with patch.object(manager.config_store, 'update', side_effect=ConfigPersistenceError()): with self.assertRaises(ConfigPersistenceError): manager.patch_settings('stream', rev, {'fall': {'r2': {'schema_version': 1, 'r1_leg_min_deg': 60}}}) self.assertEqual(manager.settings('stream'), out) for variant in ('uri', 'anchor', 'roi', 'config'): candidate = deepcopy(req) if variant == 'uri':candidate['rtspUrl'] = 'rtsp://fixture.invalid/other' elif variant == 'anchor':candidate['startWallTimeMs'] += 100 elif variant == 'roi': next(s for s in candidate['scenarios'] if s['scenarioCode']=='FALL')['rois'] = [ dict(label='small', points=[[0.,0.],[.5,0.],[.5,.5],[0.,.5]])] else: next(s for s in candidate['scenarios'] if s['scenarioCode']=='FALL')['parameters']['fall_r2']['r1_torso_min_deg'] = 60 with self.subTest(variant=variant), patch.object(group, 'restart') as restart: service.start_camera(Start(json.dumps(candidate))) self.assertEqual(restart.call_count, 1) def person(torso=55, leg=55, ankles=True): points = [[0., 0., 0.] for _ in range(17)] def offset(p, length, angle): return [p[0]+length*math.sin(math.radians(angle)), p[1]+length*math.cos(math.radians(angle)), .9] for shoulder, hip, knee, ankle, delta in ((5,11,13,15,0), (6,12,14,16,10)): points[shoulder] = [80., 70.+delta, .9] points[hip] = offset(points[shoulder], 100, torso) points[knee] = offset(points[hip], 60, 70) if ankles:points[ankle] = offset(points[hip], 150, leg) return dict(query=0, track_id=None, score=.9, bbox=[0.,0.,400.,250.], keypoints=points) class BranchTests(unittest.TestCase): def test_local_constants_and_independent_instances(self): a = R2Core(config=R2Config(1,60,60,160)); b = R2Core() self.assertEqual((b.base['torso_down_deg'], b.base['leg_down_deg'], b.rules['D1']['knee_angle_min']), (50,50,140)) self.assertEqual(a.rules['D2'], b.rules['D2']) for key in b.base: if key not in ('torso_down_deg','leg_down_deg'):self.assertEqual(a.base[key], b.base[key]) for key in b.rules['D1']: if key != 'knee_angle_min':self.assertEqual(a.rules['D1'][key], b.rules['D1'][key]) engine = FallScenario(FallSettings.parse(dict(engine='r2', enabled=True, r2=R2Config(1,55,56,150).payload()))) self.assertEqual(engine.engine.core.base['leg_down_deg'], 56) def test_r1_torso_leg_and_thigh_scope(self): base = R2Core().base self.assertTrue(features(person(), base)['hit']) self.assertFalse(features(person(), R2Core(config=R2Config(r1_torso_min_deg=60)).base)['hit']) self.assertFalse(features(person(), R2Core(config=R2Config(r1_leg_min_deg=60)).base)['hit']) p = person(torso=70, ankles=False) self.assertEqual(features(p, base)['rule'], 'F2_THIGH_ONLY') self.assertEqual(features(p, R2Core(config=R2Config(r1_leg_min_deg=60)).base)['rule'], 'F2_THIGH_ONLY') def test_knee_threshold_does_not_block_other_branches_or_people(self): default = R2Core(); strict = R2Core(config=R2Config(1,60,60,160)) base = dict(hit=False, eligible=True, torso={}) extra = dict(available=True, aspect=1.2, torso_angle=30, torso_ratio=.2, shoulder_below_hip=-.1, legs=[dict(knee_angle=150., length_ratio=.3, ankle_drop=.1, knee_drop=.05)]) self.assertEqual(decide(base, extra, default.rules)['added_rules'], ['D1_COMPACT_EXTENDED']) self.assertFalse(decide(base, extra, strict.rules)['hit']) self.assertTrue(decide(dict(base,hit=True), extra, strict.rules)['hit']) second = deepcopy(extra); second['legs'][0]['knee_angle'] = 170 self.assertTrue(decide(base, second, strict.rules)['hit']) reverse = deepcopy(extra); reverse['shoulder_below_hip'] = .1 reverse['legs'][0].update(ankle_drop=-.1, knee_drop=0., knee_angle=130.) self.assertEqual(decide(base, reverse, strict.rules)['added_rules'], ['D2_REVERSED_ORDER']) both = deepcopy(extra); both['legs'].append(dict(extra['legs'][0], knee_angle=130.)) self.assertFalse(decide(base, both, default.rules)['hit']) missing = dict(extra, legs=[]) self.assertTrue(decide(base, missing, default.rules)['insufficient']) if __name__ == '__main__': unittest.main()