--- app/fall_runtime.py.before +++ app/fall_runtime.py @@ -18,6 +18,7 @@ policy: FallConfig geometry: IntrusionConfig engine: str = 'legacy' + r2: object | None = None scenario_name = 'FALL' @classmethod @@ -36,9 +37,10 @@ if default not in ('true', 'false'): raise ValueError('Invalid FALL site default.') values.setdefault('test_skip_down_hold', default == 'true') + r2=None if engine=='r2': - from app.fall_r2.config import block,score_floor - if 'r2' in value:block(value['r2']) + from app.fall_r2.config import R2Config,score_floor + r2=R2Config.parse(value['r2']) if 'r2' in value else R2Config() values=dict(enabled=value.get('enabled',False),person_score=score_floor(value.get('person_score',.25),.25), joint_score=score_floor(value.get('joint_score',.10),.10)) elif 'r2' in value:raise ValueError('R2 settings require the AI-owned r2 engine selection.') @@ -48,7 +50,7 @@ roi=((0., 0.), (1., 0.), (1., 1.), (0., 1.)) if roi is None or roi == [] else roi, roi_id=value.get('roi_id', 'primary'), roi_name=value.get('roi_name', 'Fall ROI'), debug_roi=value.get('debug_roi', True), point_x=.5, point_y=1.) - return cls(policy, geometry, engine) + return cls(policy, geometry, engine, r2) @property def enabled(self): @@ -77,7 +79,7 @@ def settings_payload(self): value=dict(self.policy.payload(), **{k: v for k, v in self.geometry.payload().items() if k in ('roi', 'roi_id', 'roi_name', 'debug_roi')}) - if self.engine=='r2':value.update(engine='r2',r2={'schema_version':1}) + if self.engine=='r2':value.update(engine='r2',r2=self.r2.payload()) return value --- app/stream_settings.py.before +++ app/stream_settings.py @@ -130,6 +130,10 @@ current = FallSettings.parse(spec.get('fall') or {}).settings_payload() merged=dict(current, **value) if value.get('engine')=='legacy':merged.pop('r2',None) + elif 'r2' in value and merged.get('engine')=='r2': + from app.fall_r2.config import R2Config + R2Config.parse(value['r2']) + merged['r2']=dict(current.get('r2') or {},**value['r2']) configured = FallSettings.parse(merged).settings_payload() if configured != current: candidate[field] = configured --- app/fall_r2/config.py.before +++ app/fall_r2/config.py @@ -1,8 +1,56 @@ -"""FALL-R2-WEB-01: version marker only; geometry and timing constants are fixed.""" +"""FALL-R2-CONFIG-02: bounded, local angle overrides; timing remains fixed.""" +from dataclasses import asdict, dataclass import math + +ANGLE_LIMITS = { + 'r1_torso_min_deg': (50., 60.), + 'r1_leg_min_deg': (50., 60.), + 'd1_knee_min_deg': (140., 160.), +} class R2ConfigError(ValueError): pass + + +def angle_value(value, minimum, maximum): + try: + valid = type(value) in (int, float) and math.isfinite(value) and minimum <= value <= maximum + except OverflowError: + valid = False + if not valid: + raise R2ConfigError('FALL_R2_ANGLE_INVALID') + return float(value) + + +@dataclass(frozen=True) +class R2Config: + schema_version: int = 1 + r1_torso_min_deg: float = 50. + r1_leg_min_deg: float = 50. + d1_knee_min_deg: float = 140. + + def __post_init__(self): + if type(self.schema_version) is not int or self.schema_version != 1: + raise R2ConfigError('FALL_R2_SCHEMA_VERSION_UNSUPPORTED') + for key, bounds in ANGLE_LIMITS.items(): + angle_value(getattr(self, key), *bounds) + + def payload(self): + value = asdict(self) + value.update({key: float(value[key]) for key in ANGLE_LIMITS}) + return value + + @classmethod + def parse(cls, value): + block(value) + return cls(**value) + + def patched(self, value): + # The existing Web reset owns common scores/cooldown, never these angles. + if block(value, request=True): + return self + return self.parse(dict(self.payload(), **value)) + def score_floor(value, minimum): try:valid=type(value) in (int,float) and math.isfinite(value) and 0<=value<=1 @@ -11,9 +59,14 @@ return max(float(value),minimum) def block(value, *, request=False): - allowed={'schema_version','reset_to_defaults'} if request else {'schema_version'} + allowed={'schema_version', *ANGLE_LIMITS} + if request:allowed.add('reset_to_defaults') if type(value) is not dict or set(value)-allowed:raise R2ConfigError('FALL_R2_FIELDS_INVALID') if 'schema_version' not in value:raise R2ConfigError('FALL_R2_SCHEMA_VERSION_REQUIRED') if type(value['schema_version']) is not int or value['schema_version']!=1:raise R2ConfigError('FALL_R2_SCHEMA_VERSION_UNSUPPORTED') - if 'reset_to_defaults' in value and value['reset_to_defaults'] is not True:raise R2ConfigError('FALL_R2_RESET_INVALID') + if 'reset_to_defaults' in value: + if value['reset_to_defaults'] is not True or set(value)&set(ANGLE_LIMITS): + raise R2ConfigError('FALL_R2_RESET_INVALID') + for key, bounds in ANGLE_LIMITS.items(): + if key in value:angle_value(value[key], *bounds) return value.get('reset_to_defaults',False) --- app/fall_r2/web.py.before +++ app/fall_r2/web.py @@ -2,7 +2,7 @@ import logging import math import re -from .config import R2ConfigError,score_floor,block +from .config import ANGLE_LIMITS,R2ConfigError,score_floor,block LOG=logging.getLogger(__name__) def apply_parameters(spec,request): @@ -17,6 +17,7 @@ explicit='fall_r2' in raw;cfg=current.settings_payload() cooldown=(spec.get('event_cooldowns') or {}).get('FALL',10.) reset=block(raw['fall_r2'],request=True) if explicit else False + if explicit:cfg['r2']=current.r2.patched(raw['fall_r2']).payload() common={'person_confidence','pose_confidence','cooldown_seconds'} if reset: if common & set(raw):raise R2ConfigError('FALL_R2_RESET_INVALID') @@ -42,6 +43,13 @@ reason='LEGACY_NOT_APPLIED' if key in ('down_angle_deg','recovery_hold_seconds') else 'UNKNOWN_PARAMETER' LOG.info('fall_r2_parameter_ignored parameter=%s reason=%s',safe,reason) cfg=FallSettings.parse(cfg).settings_payload() + if explicit: + for key in ANGLE_LIMITS: + if key in raw['fall_r2']: + LOG.info('fall_r2_parameter parameter=%s requested=%s effective=%s corrected=False', + key,raw['fall_r2'][key],cfg['r2'][key]) spec['fall']=cfg;spec.setdefault('event_cooldowns',{})['FALL']=cooldown - LOG.info('fall_r2_effective version=1 person_score=%s joint_score=%s cooldown=%s',cfg['person_score'],cfg['joint_score'],cooldown) + LOG.info('fall_r2_effective version=1 person_score=%s joint_score=%s cooldown=%s torso=%s leg=%s knee=%s', + cfg['person_score'],cfg['joint_score'],cooldown,cfg['r2']['r1_torso_min_deg'], + cfg['r2']['r1_leg_min_deg'],cfg['r2']['d1_knee_min_deg']) return True --- app/fall_r2/core.py.before +++ app/fall_r2/core.py @@ -4,13 +4,18 @@ from .reference import features, SpatialState from .geometry import measure from .decision import decide +from .config import R2Config class R2Core: - def __init__(self, person_score=.25, joint_score=.10): + def __init__(self, person_score=.25, joint_score=.10, *, config=None): root=Path(__file__).parent self.base=json.loads((root/'base.json').read_text()) self.rules=json.loads((root/'rules.json').read_text()) + config=R2Config() if config is None else config + if not isinstance(config,R2Config):raise TypeError('Expected R2Config.') self.base.update(person_score=person_score,joint_score=joint_score) + self.base.update(torso_down_deg=config.r1_torso_min_deg,leg_down_deg=config.r1_leg_min_deg) + self.rules['D1']['knee_angle_min']=config.d1_knee_min_deg self.clear() def clear(self): --- app/fall_r2/runtime.py.before +++ app/fall_r2/runtime.py @@ -11,7 +11,7 @@ class R2Runtime: def __init__(self,config,event_id_factory,wall_ns,contains=None): self.config=config;self.event_id_factory=event_id_factory;self.wall_ns=wall_ns;self.contains=contains - self.core=R2Core(config.policy.person_score,config.policy.joint_score) + self.core=R2Core(config.policy.person_score,config.policy.joint_score,config=config.r2) self.metrics=Counter();self.clear() def clear(self):