"""FALL-WIRE-01: local ROI/decision adapter; one owner, no tracker or delivery I/O.""" from dataclasses import dataclass, fields import os import time import uuid from app.fall_config import FallConfig from app.fall_scenario import FallEngine from app.intrusion import IntrusionConfig, IntrusionEvent, inside # September demo defaults for existing settings keys only; explicit values win. # Keep FallConfig defaults as the reproducible pre-demo temporal policy. DEMO_FALL_DEFAULTS = dict(person_score=.15, joint_score=.10, down_angle_deg=50.) @dataclass(frozen=True) class FallSettings: policy: FallConfig geometry: IntrusionConfig scenario_name = 'FALL' @classmethod def parse(cls, value): if type(value) is not dict: raise ValueError('Invalid FALL settings.') names = {f.name for f in fields(FallConfig)} roi_names = {'roi', 'roi_id', 'roi_name', 'debug_roi'} if set(value) - names - roi_names: raise ValueError('Invalid FALL settings fields.') values = dict(DEMO_FALL_DEFAULTS, **{k: v for k, v in value.items() if k in names}) # Site default applies only to absent values; an explicit per-camera Off wins. default = os.environ.get('CONTEXTRA_FALL_SKIP_DOWN_HOLD_DEFAULT', 'true') if default not in ('true', 'false'): raise ValueError('Invalid FALL site default.') values.setdefault('test_skip_down_hold', default == 'true') policy = FallConfig.parse(values) roi = value.get('roi') geometry = IntrusionConfig(enabled=policy.enabled, confidence=policy.person_score, 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) @property def enabled(self): return self.policy.enabled @property def confidence(self): return self.policy.person_score @property def roi(self): return self.geometry.roi @property def roi_name(self): return self.geometry.roi_name @property def debug_roi(self): return self.geometry.debug_roi def payload(self): return {k: v for k, v in self.geometry.payload().items() if k in ('enabled', 'confidence', 'roi', 'roi_id', 'roi_name', 'debug_roi', 'point_x', 'point_y')} def settings_payload(self): return dict(self.policy.payload(), **{k: v for k, v in self.geometry.payload().items() if k in ('roi', 'roi_id', 'roi_name', 'debug_roi')}) @dataclass(frozen=True) class FallDisplay: point: tuple inside: bool state: str confidence: float event_id: str | None = None def payload(self): return dict(scenario='FALL', point=list(self.point), inside=self.inside, state=self.state, confidence=self.confidence, eventId=self.event_id) @dataclass(frozen=True) class FallEvent(IntrusionEvent): def payload(self): return dict(super().payload(), scenario='FALL') class FallScenario: def __init__(self, config, *, wall_ns=time.time_ns, event_id_factory=lambda: str(uuid.uuid4()), direct_down=None, direct_down_cooldown_seconds=None, contains=None): # AI-private: never add these options to Settings/Metadata/Event payloads. if direct_down is None: value = os.environ.get('CONTEXTRA_FALL_DEMO_DIRECT_DOWN', 'true') if value not in ('true', 'false'): raise ValueError('Invalid FALL direct-down toggle.') direct_down = value == 'true' if direct_down_cooldown_seconds is None: direct_down_cooldown_seconds = float(os.environ.get( 'CONTEXTRA_FALL_DIRECT_DOWN_COOLDOWN_SECONDS', '2.0')) self.config = config self.contains = contains self.engine = FallEngine(config.policy, direct_down=direct_down, direct_down_cooldown_seconds=direct_down_cooldown_seconds) self.wall_ns = wall_ns self.event_id_factory = event_id_factory self._events = {} def evaluate(self, frame, poses, track_ids, *, gap_before=False): points = tuple(((p.bbox_xyxy[0]+p.bbox_xyxy[2])/2, p.bbox_xyxy[3]) for p in poses) selected = tuple(i for i, point in enumerate(points) if (self.contains(point) if self.contains else inside(point, self.config.roi))) result = self.engine.evaluate(frame, tuple(poses[i] for i in selected), tuple(track_ids[i] for i in selected), gap_before=gap_before) if not result.accepted: return tuple(FallDisplay(p, i in selected, 'TIME_UNAVAILABLE', self.config.confidence) for i, p in enumerate(points)), () self._events = {tid: eid for tid, eid in self._events.items() if (self.engine.inspect(tid) or {}).get('latched')} events = [] for candidate in result.event_candidates: original = selected[int(candidate.object_ref)] event_id = self.event_id_factory() if type(event_id) is not str or not event_id or len(event_id) > 100: raise ValueError('Invalid FALL Event ID.') self._events[candidate.track_id] = event_id events.append(FallEvent(event_id, frame, candidate.track_id, self.config.geometry.roi_id, str(original), candidate.person_score, candidate.bbox_xyxy, poses[original].model_key, self.wall_ns()//1_000_000)) decisions = [FallDisplay(p, False, 'OUTSIDE', self.config.confidence) for p in points] for index, decision in zip(selected, result.decisions): decisions[index] = FallDisplay(points[index], True, decision.state, self.config.confidence, self._events.get(decision.track_id) if decision.alert else None) return tuple(decisions), tuple(events) def clear(self): self.engine.clear() self._events.clear() def project_fall(spec, scenarios): request = next(s for s in scenarios if s['scenarioCode'] == 'FALL') if not request['enabled'] and 'fall' not in spec: return cfg = dict(spec.get('fall') or {}) roi = request['rois'][0] if request['rois'] else None cfg.update(enabled=request['enabled'], person_score=DEMO_FALL_DEFAULTS['person_score'] if request['confidence'] is None else request['confidence'], roi=roi['points'] if roi else [], roi_name=roi['label'] if roi else 'Fall ROI') spec['fall'] = FallSettings.parse(cfg).settings_payload() def validate(spec): config = FallSettings.parse(spec.get('fall') or {}) if config.enabled: if not spec.get('server_id') or not spec.get('source_id'): raise ValueError('FALL requires canonical identity.') if not (spec.get('tracking') or {}).get('enabled', True): raise ValueError('FALL requires Tracking On.') return config