"""PR13 bounded multi-provider Frame coordinator. The coordinator owns Frame acceptance/finalization and scenario state. It stores only immutable copied metadata, never SDK Frame wrappers or pixels. Provider results arriving after a Frame deadline/finalization/eviction are health responses but cannot mutate Tracking, Scenario, Event or Metadata state. """ from __future__ import annotations from collections import Counter, OrderedDict from dataclasses import dataclass, field, replace import os import threading import time from app.contracts.frames import DetectionObservation, PoseObservation from app.contracts.visualization import FrameResult, VisualizationCommand from app.region_scenario import alerted, event_ids, members, visible as scenario_visible @dataclass class _OpenFrame: frame: object admitted_at: float deadline: float required: set[str] results: dict state: str = 'OPEN' gaps: set = field(default_factory=set) claimed: bool = False selected_at: float | None = None def public_positions(commands, events, capacity=300): """Keep Event refs and each visible provider's representative before fill.""" from app.crowd_event import CrowdEvent required = {int(e.object_ref) for e in events if not isinstance(e, CrowdEvent)} if any(i < 0 or i >= len(commands) for i in required): raise ValueError('Event object reference is outside provider commands.') if len(commands) <= capacity: return list(range(len(commands))) representatives = {} for i, command in enumerate(commands): if not command.visible: continue d = command.detection separate_classes = d.model_key in ('RFDETRFireSmoke', 'Awiros_person_and_head_detection') group = (d.model_key, d.class_id if separate_classes else None) # CROWD's count label must survive ahead of its plain boxes. rank = (int(' | P' in command.label_text or ' | H' in command.label_text), int(alerted(command.scenario)), d.score, -i) if group not in representatives or rank > representatives[group][0]: representatives[group] = (rank, i) required.update(i for _, i in representatives.values()) if len(required) > capacity: raise RuntimeError('Mandatory metadata object capacity exceeded.') ranked = sorted((i for i in range(len(commands)) if i not in required), key=lambda i: (not commands[i].visible, not (alerted(commands[i].scenario) or commands[i].event_ids), -commands[i].detection.score, i)) return sorted(required | set(ranked[:capacity-len(required)])) class MultiProviderCoordinator: def __init__(self, spec, ingress, provider_plan, collector, is_current, on_fatal, *, publisher=None, on_provider_frame=None, clock=time.monotonic): self.spec = spec self.ingress = ingress self.plan = provider_plan self.collector = collector self.is_current = is_current self.on_fatal = on_fatal self.publisher = publisher self.on_provider_frame = on_provider_frame self.clock = clock self.freshness_s = float(os.environ.get( 'CONTEXTRA_PROVIDER_RESULT_FRESHNESS_MS', '100')) / 1000.0 self.stall_s = float(os.environ.get('CONTEXTRA_PROVIDER_STALL_SECONDS', '2.0')) self.startup_s = float(os.environ.get('CONTEXTRA_PROVIDER_STARTUP_SECONDS', '10.0')) flag = os.environ.get('CONTEXTRA_PROVIDER_FRESHNESS_ENABLED', 'true').lower() if flag not in ('true', 'false'): raise ValueError('Invalid freshness flag.') self.freshness_enabled = flag == 'true' if not self.freshness_enabled and len(provider_plan.providers) != 1: raise ValueError('Freshness OFF requires one provider per coordinator.') if (not 0 < self.freshness_s <= 10 or not 0 < self.stall_s <= 300 or not self.stall_s <= self.startup_s <= 300): raise ValueError('Invalid multi-provider timing settings.') self.capacity = 2 if self.freshness_enabled else 3 from app.stream_settings import revision, display_score, visual_options self.scenario_revision = (spec.get('_provider_runtime') or {}).get('revision') or revision(spec) self.display_threshold = display_score( spec, ((spec.get('inference_profile') or {}).get('thresholds') or {}).get('score', .5)) self.display_categories, self.debug_all_candidates = visual_options(spec) self.display_categories = (frozenset(self.display_categories) if self.display_categories is not None else None) self.camera_id = spec.get('camera_id') from app.region_scenario import compose from app.fall_runtime import FallSettings, FallScenario from app.fight_runtime import FightSettings, FightScenario from app.fire_config import build_engine from app.crowd_config import CrowdConfig from app.crowd_scenario import Crowd from app.crowd_result import CrowdResultProcessor from app.pose_config import PoseConfig self.region = compose(spec) fall_cfg = FallSettings.parse(spec.get('fall') or {}) fight_cfg = FightSettings.parse(spec.get('fight') or {}) self.fall = FallScenario(fall_cfg) if fall_cfg.enabled else None self.fight = FightScenario(fight_cfg) if fight_cfg.enabled else None self.fire = build_engine(spec) from app.event_cooldown import EventCooldown self.event_cooldown = EventCooldown(clock=self.clock, overrides=spec.get('event_cooldowns')) engines = getattr(self.region, 'engines', (self.region,)) for engine in (*engines, self.fall.engine if self.fall else None, self.fight.engine if self.fight else None, self.fire): if engine is not None: engine.event_admission = self.event_cooldown.allow crowd_cfg = CrowdConfig.parse(spec.get('crowd') or {}) self.crowd = (CrowdResultProcessor( Crowd(crowd_cfg), camera_id=self.camera_id, scenario_revision=self.scenario_revision) if crowd_cfg.enabled else None) if self.crowd is not None: self.crowd.engine.event_admission = self.event_cooldown.allow self.pose_config = (PoseConfig.parse(spec.get('pose') or {}) if any(p.key == 'pose' for p in provider_plan.providers) else None) from app.byte_tracking import ByteTracker, TrackingSettings tracking = dict(spec.get('tracking') or {}) tracking['frame_rate'] = ingress.sampler.fps tracking_cfg = TrackingSettings.parse(tracking) self.tracker = (ByteTracker(tracking_cfg, metrics=ingress.metrics) if provider_plan.person_owner and tracking_cfg.enabled else None) self.visual = None self.artifacts = None self._samples = None self._cv = threading.Condition() self._frames = OrderedDict() self._closed = False self._thread = None self._last_admitted = -1 self._last_published = -1 self._latest = None self._error = None self._metrics = Counter() now = self.clock() self._providers = { p.key: dict(state='ACTIVE', reason=None, first_input_at=None, last_input_at=None, last_response_at=None, responses=0, late=0, busy_drop=0) for p in provider_plan.providers } self._dependencies = { scenario: provider.key for provider in provider_plan.providers for scenario in provider.scenarios } self._requested = set(provider_plan.requested_scenarios) self._started_at = now self._person_gap_pending = False self._gap_pending = {p.key: False for p in provider_plan.providers} self._finalizing = None self._branch_release = {} self._timings = {} self._retired = OrderedDict() def _notify_fatal(self, reason, error=None): # Preserve the first failure boundary before StreamGroup clears the path. import json import sys import traceback from pathlib import Path with self._cv: diagnostic = dict(code='MULTI_PROVIDER_FATAL', reason=reason, monotonic_ns=time.monotonic_ns(), error=self._error, providers={k: {n: v[n] for n in ('state', 'reason', 'responses', 'late')} for k, v in self._providers.items()}) if error is not None: diagnostic['exception_type'] = type(error).__name__ diagnostic['locations'] = [dict(file=Path(f.filename).name, line=f.lineno, function=f.name) for f in traceback.extract_tb(error.__traceback__)[-4:]] try: print(json.dumps(diagnostic), file=sys.stderr, flush=True) except OSError: pass self.on_fatal() def _timing_locked(self, provider, stage, seconds): if seconds is None or seconds < 0: return ms = seconds * 1000 row = self._timings.setdefault(provider, {}).setdefault( stage, dict(count=0, sum_ms=0., max_ms=0., over_100ms=0)) row['count'] += 1 row['sum_ms'] += ms row['max_ms'] = max(row['max_ms'], ms) row['over_100ms'] += int(ms > 100) def provider_gate_time(self, provider, seq, entered_at): with self._cv: state = self._frames.get(seq) if (state is None or state.state != 'OPEN' or self._closed or not self.is_current() or self._providers[provider]['state'] != 'ACTIVE' or (self.freshness_enabled and entered_at > state.deadline)): return None state.claimed = True self._providers[provider].setdefault('first_request_at', entered_at) self._timing_locked(provider, 'selected_to_gate', entered_at-state.admitted_at) return state.admitted_at def _retire_locked(self, seq, reason): self._retired[seq] = reason while len(self._retired) > 64: self._retired.popitem(last=False) def register_branch_release(self, key, release): with self._cv: self._branch_release[key] = release disabled = self._closed or self._providers[key]['state'] != 'ACTIVE' if disabled: release() def stop_admission(self): with self._cv: self._closed = True self._frames.clear() releases = list(self._branch_release.values()) self._cv.notify_all() for release in releases: release() def _evict_locked(self, state): self._retire_locked(state.frame.frame_seq, 'evicted') state.state = 'EVICTED' for key in self._gap_pending: self._gap_pending[key] = True self._metrics['frame_evicted'] += 1 def start(self): with self._cv: if self._closed or self._thread is not None: raise RuntimeError('Multi-provider coordinator cannot start twice.') if self.publisher is not None: if not self.publisher.channel(self.ingress.identity[1], self.ingress.identity[3], self.ingress.identity[4], 'STARTING'): raise RuntimeError('Metadata channel worker admission rejected.') from app.metadata_samples import MetadataSamples from app.metadata_visual import MetadataVisual from app.event_capture import EventCapture self._samples = MetadataSamples.from_environment() if os.environ.get('CONTEXTRA_METADATA_VISUAL_ENABLED', 'true').lower() == 'true': self.visual = MetadataVisual.from_environment(display_threshold=self.display_threshold) self.artifacts = EventCapture.from_environment(self.spec) self._thread = threading.Thread( target=self._run, name='multi-provider-results', daemon=True) self._thread.start() def admit(self, frame, admitted_at=None): admitted_at = self.clock() if admitted_at is None else float(admitted_at) fatal = False with self._cv: if self._closed or not self.is_current() or frame.frame_seq <= self._last_admitted: self._metrics['admission_stale'] += 1 return False self._last_admitted = frame.frame_seq active = {k for k, v in self._providers.items() if v['state'] == 'ACTIVE'} for key in active: status = self._providers[key] if status['first_input_at'] is None: status['first_input_at'] = admitted_at status['last_input_at'] = admitted_at while len(self._frames) >= self.capacity: evictable = next((seq for seq, s in self._frames.items() if self.freshness_enabled or not (s.claimed or s.results)), None) if evictable is None: self._metrics['admission_capacity_drop'] += 1 return False oldest = self._frames.pop(evictable) self._evict_locked(oldest) self._frames[frame.frame_seq] = _OpenFrame( frame, admitted_at, admitted_at+self.freshness_s, active, {}) self._metrics['admitted'] += 1 fatal = self._check_all_unavailable_locked() self._cv.notify_all() if fatal: self._notify_fatal('admission_unavailable') return not fatal def is_open(self, frame_seq): with self._cv: state = self._frames.get(frame_seq) return bool(not self._closed and self.is_current() and state and state.state == 'OPEN' and (not self.freshness_enabled or self.clock() <= state.deadline)) def provider_available(self, key): with self._cv: return self._providers.get(key, {}).get('state') == 'ACTIVE' def note_provider_progress(self, key, stage): if stage not in ('infer_output', 'callback'): raise ValueError('Invalid provider progress stage.') with self._cv: status = self._providers[key] if not self._closed and status['state'] == 'ACTIVE': status['last_progress_at'] = self.clock() status['last_progress_stage'] = stage def note_busy_drop(self, key): with self._cv: if key in self._providers: self._providers[key]['busy_drop'] += 1 self._metrics['latest_busy_drop'] += 1 def submit(self, provider, frame, copied, response_at=None, *, timing=None): response_at = self.clock() if response_at is None else float(response_at) lock_started = self.clock() accepted = False with self._cv: status = self._providers.get(provider) if status is None: raise ValueError('Unknown provider.') # Even a late or empty result proves provider liveness. status['last_response_at'] = response_at status.setdefault('first_response_at', response_at) status['responses'] += 1 state = self._frames.get(frame.frame_seq) acceptance_at = self.clock() if state is not None: self._timing_locked(provider, 'selected_to_response', response_at-state.admitted_at) if acceptance_at > state.deadline: self._metrics['over_freshness_' + provider] += 1 self._timing_locked(provider, 'submit_lock_wait', acceptance_at-lock_started) if timing is not None: selected, gate, infer, callback, copy_done = timing for stage, start, end in ( ('gate_to_infer', gate, infer), ('infer_to_callback', infer, callback), ('callback_copy', callback, copy_done), ('selected_to_callback', selected, callback), ('selected_to_accept_check', selected, acceptance_at)): if start is not None and end is not None: self._timing_locked(provider, stage, end-start) if (status['state'] != 'ACTIVE' or state is None or state.state != 'OPEN' or (self.freshness_enabled and acceptance_at > state.deadline) or frame.frame_seq <= self._last_published): status['late'] += 1 self._metrics['late_result'] += 1 reason = ('provider_disabled' if status['state'] != 'ACTIVE' else self._retired.get(frame.frame_seq, 'missing_slot') if state is None else 'closed_frame' if state.state != 'OPEN' else 'deadline' if acceptance_at > state.deadline else 'backward') self._metrics['rejected_' + provider + '_' + reason] += 1 return False if provider not in state.required or provider in state.results: self._metrics['duplicate_or_unrequired_result'] += 1 return False if ((frame.frame_width, frame.frame_height) != (copied.width, copied.height) or len(copied.detections) > 300): raise ValueError('Provider result association differs.') state.results[provider] = copied accepted = True self._metrics['accepted_result'] += 1 from app.pipeline_diagnostics import diagnostics if diagnostics.enabled: diagnostics.for_provider(provider).mark('accepted') self._cv.notify_all() if self.on_provider_frame is not None: self.on_provider_frame(provider, time.time()) return accepted def mark_provider_unavailable(self, key, reason='provider_unavailable'): fatal = False with self._cv: status = self._providers.get(key) if status is None or status['state'] != 'ACTIVE': return False status['state'] = 'UNAVAILABLE' status['reason'] = str(reason) self._metrics['provider_unavailable'] += 1 for state in self._frames.values(): state.required.discard(key) state.results.pop(key, None) self._gap_pending[key] = True fatal = self._check_all_unavailable_locked() self._cv.notify_all() release = self._branch_release.get(key) if release is not None: release() if fatal: self._notify_fatal('provider_unavailable') return True def _check_all_unavailable_locked(self): if not self._requested: return not any(v['state'] == 'ACTIVE' for v in self._providers.values()) active = {s for s in self._requested if self._providers[self._dependencies[s]]['state'] == 'ACTIVE'} return not active def _ready_locked(self, now): for seq, state in self._frames.items(): if state.state != 'OPEN': continue if state.required <= set(state.results) or (self.freshness_enabled and now >= state.deadline): return seq if not self.freshness_enabled and state.claimed: return None return None def _next_wait_locked(self, now): if not self.freshness_enabled: return .1 deadlines = [s.deadline for s in self._frames.values() if s.state == 'OPEN'] wait = max(0., min(deadlines)-now) if deadlines else .1 return min(wait, .1) def _check_stalls_locked(self, now): unavailable = [] input_recent = any(v['last_input_at'] is not None and now-v['last_input_at'] <= self.stall_s for v in self._providers.values()) if not input_recent: return unavailable for key, status in self._providers.items(): if status['state'] != 'ACTIVE' or status['first_input_at'] is None: continue responded = status['last_response_at'] is not None reference = status['last_response_at'] if responded else status['first_input_at'] if responded: reference = max(reference, status.get('last_progress_at', reference)) if now-reference >= (self.stall_s if responded else self.startup_s): unavailable.append(key) return unavailable def _run(self): try: while True: state = None stalls = [] with self._cv: if self._closed or not self.is_current(): return now = self.clock() stalls = self._check_stalls_locked(now) for key in stalls: status = self._providers[key] status['state'] = 'UNAVAILABLE' status['reason'] = 'provider_stalled' self._metrics['provider_stalled'] += 1 for open_state in self._frames.values(): open_state.required.discard(key) open_state.results.pop(key, None) self._gap_pending[key] = True fatal = self._check_all_unavailable_locked() seq = self._ready_locked(now) if seq is not None and not fatal: for older in [k for k in self._frames if k < seq]: self._evict_locked(self._frames.pop(older)) state = self._frames.pop(seq) if seq <= self._last_published: self._evict_locked(state) self._metrics['backward_drop'] += 1 state = None else: # Claim before any temporal mutation; accepted results # are frozen even if Stop/disable arrives during handoff. state.state = 'FINALIZING' self._retire_locked(seq, 'finalizing') for key in self._gap_pending: if key in state.results: if self._gap_pending[key]: state.gaps.add(key) self._gap_pending[key] = False else: self._gap_pending[key] = True self._finalizing = state self._metrics['finalizing_claims'] += 1 if state is None and not fatal: self._cv.wait(self._next_wait_locked(now)) for key in stalls: release = self._branch_release.get(key) if release is not None: release() if fatal: self._notify_fatal('provider_stalled') return if state is not None: self._finalize(state) with self._cv: state.state = 'FINALIZED' self._retire_locked(seq, 'finalized') self._finalizing = None except Exception as error: with self._cv: self._error = 'MULTI_PROVIDER_RESULT_FAILED' self._closed = True self._frames.clear() self._notify_fatal('result_exception', error) def _observations(self, provider, frame, copied): observations = tuple( DetectionObservation(frame, provider.model, d.class_id, d.score, d.bbox_xyxy) for d in copied.detections) poses = tuple( PoseObservation(frame, provider.model, d.score, d.bbox_xyxy, d.keypoints) if d.keypoints is not None else None for d in copied.detections) from app.pose_config import POSE_MODEL if provider.model == POSE_MODEL and any(p is None for p in poses): raise ValueError('Pose provider returned a person without keypoints.') if provider.model != POSE_MODEL and any(p is not None for p in poses): raise ValueError('Non-Pose provider returned unexpected keypoints.') return observations, poses def _person_result(self, provider, frame, copied, gap_before): from app.pose_config import POSE_MODEL from app.class_catalog import coco_name from app.pipeline_diagnostics import diagnostics metric = diagnostics.for_provider(provider.key) with metric.phase('result_observations'): observations, poses = self._observations(provider, frame, copied) selected = tuple(i for i, o in enumerate(observations) if o.class_id == 0) if provider.model == POSE_MODEL and self.pose_config and self.pose_config.nms_enabled: from app.pose_nms import pose_nms_indices with metric.phase('result_nms'): selected = tuple(i for i in self.pose_nms(observations) if observations[i].class_id == 0) selected_obs = tuple(observations[i] for i in selected) with metric.phase('result_tracking'): selected_ids = (self.tracker.update(selected_obs) if self.tracker is not None else [None] * len(selected_obs)) ids = [None] * len(observations) for i, tid in zip(selected, selected_ids): ids[i] = tid decisions = [None] * len(observations) events = () if self.region is not None: with metric.phase('result_region'): gap_options = ({'gap_before': gap_before} if getattr(self.ingress, 'source_latest', False) else {}) rows, emitted = self.region.evaluate(frame, selected_obs, selected_ids, **gap_options) for i, row in zip(selected, rows): decisions[i] = row events += tuple(replace(e, object_ref=str(selected[int(e.object_ref)])) for e in emitted) if provider.model == POSE_MODEL: selected_poses = tuple(poses[i] for i in selected) for scenario in (self.fall, self.fight): if scenario is None: continue with metric.phase('result_'+('fall' if scenario is self.fall else 'fight')): rows, emitted = scenario.evaluate( frame, selected_poses, selected_ids, gap_before=gap_before) for i, row in zip(selected, rows): decisions[i] = (*members(decisions[i]), *members(row)) events += tuple(replace(e, object_ref=str(selected[int(e.object_ref)])) for e in emitted) selected_set = set(selected) commands = [] for i, (observation, pose) in enumerate(zip(observations, poses)): decision = decisions[i] label = coco_name(observation.class_id) category_visible = ( self.display_categories is None or label in self.display_categories) if observation.class_id == 0: visible = ( i in selected_set and (scenario_visible(observation.score, decision) if decision else (observation.score >= self.pose_config.person_score if provider.model == POSE_MODEL and self.pose_config else observation.score > self.display_threshold)) and category_visible) else: # Medium may own person Tracking and still provide future non-person # COCO classes. Those objects stay display-only here. visible = observation.score > self.display_threshold and category_visible commands.append(VisualizationCommand( observation, label, color='#FF3030' if alerted(decision) else '#00CC66', visible=visible, track_id=ids[i], event_ids=event_ids(decision), scenario=decision, pose=pose)) configs = members(self.region.config if self.region else None) if self.fall is not None: configs += (self.fall.config,) if self.fight is not None: configs += (self.fight.config,) result = FrameResult( frame, tuple(commands), events, configs if len(configs) > 1 else configs[0] if configs else None, self.scenario_revision, self.debug_all_candidates, self.pose_config if provider.model == POSE_MODEL else None, tuple(i in selected_set for i in range(len(observations))) if provider.model == POSE_MODEL else None, provider_model=provider.model, camera_id=self.camera_id) return result, ids def pose_nms(self, observations): from app.pose_nms import pose_nms_indices return pose_nms_indices(observations, self.pose_config.nms_iou_threshold) def _plain_medium_result(self, provider, frame, copied): from app.class_catalog import coco_name observations, _ = self._observations(provider, frame, copied) commands = [] for observation in observations: if self.plan.person_owner == 'pose' and observation.class_id == 0: continue label = coco_name(observation.class_id) commands.append(VisualizationCommand( observation, label, color='#00CC66', visible=(observation.score > self.display_threshold and (self.display_categories is None or label in self.display_categories)))) return FrameResult(frame, tuple(commands), provider_model=provider.model, camera_id=self.camera_id) def _finalize(self, state): from app.pipeline_diagnostics import diagnostics metric = diagnostics.for_provider(self.plan.providers[0].key) with metric.phase('result_total'): self._finalize_result(state, metric) def _finalize_result(self, state, metric): frame = state.frame plan_by_key = {p.key: p for p in self.plan.providers} parts = [] person_ids = None person_key = self.plan.person_owner if person_key and person_key in state.results: gap_before = person_key in state.gaps if hasattr(self.ingress, 'result_gap'): gap_before = self.ingress.result_gap(frame) or gap_before result, person_ids = self._person_result( plan_by_key[person_key], frame, state.results[person_key], gap_before) parts.append(result) with metric.phase('result_collector'): self._collect_person(frame, plan_by_key[person_key], state.results[person_key], person_ids) if 'medium' in state.results and person_key != 'medium': parts.append(self._plain_medium_result( plan_by_key['medium'], frame, state.results['medium'])) if 'fire' in state.results and self.fire is not None: from app.fire_result import evaluate_result observations, _ = self._observations( plan_by_key['fire'], frame, state.results['fire']) result = evaluate_result( self.fire, frame, observations, camera_id=self.camera_id, scenario_revision=self.scenario_revision, gap_before='fire' in state.gaps) if result is not None: parts.append(result) if 'crowd' in state.results and self.crowd is not None: observations, _ = self._observations( plan_by_key['crowd'], frame, state.results['crowd']) gap = 'crowd' in state.gaps if hasattr(self.ingress, 'result_gap'): gap = self.ingress.result_gap(frame) or gap result = self.crowd.process(frame, observations, gap_before=gap) if result is not None: parts.append(result) with metric.phase('result_composition'): result = self._compose_result(frame, parts, person_key, state) with self._cv: if state.state != 'FINALIZING' or frame.frame_seq <= self._last_published: raise RuntimeError('Invalid finalization ownership.') self._last_published = frame.frame_seq self._latest = result self._metrics['finalized'] += 1 self._metrics['events_created'] += len(result.events) if self._samples is not None and self._samples.enabled(): self._samples.capture(result.snapshot(visible_only=False)) if self.visual is not None: self.visual.submit_result(result) if self.artifacts is not None: with metric.phase('result_artifact_handoff'): accepted = self.artifacts.submit(result) if accepted is False and result.events: raise RuntimeError('Scenario Event handoff rejected.') if self.publisher is not None: with metric.phase('result_publish'): self.publisher.channel(frame.source_id, frame.run_id, frame.run_generation, 'RUNNING') note = getattr(self.publisher, 'note_frame_timing', None) if note is not None and state.selected_at is not None: note(frame, round(state.selected_at*1e9), round(state.admitted_at*1e9)) if not self.publisher.submit(result.snapshot()): with self._cv: self._metrics['publisher_rejected'] += 1 def _compose_result(self, frame, parts, person_key, state): if len(self.plan.providers) == 1 and len(parts) == 1: result = parts[0] if result.frame != frame: raise ValueError('Foreign provider result frame.') public_positions(result.commands, result.events) fields = dict(scenario_revision=self.scenario_revision, debug_all_candidates=self.debug_all_candidates, pose_config=self.pose_config if person_key == 'pose' else None, camera_id=self.camera_id) changes = {key: value for key, value in fields.items() if getattr(result, key) != value} return replace(result, **changes) if changes else result # Only the shared-provider path needs composition and reference remapping. raw_commands, raw_events, configs = [], [], [] from app.crowd_event import CrowdEvent crowd_regions = None for part in parts: if part.crowd_regions is not None: crowd_regions = part.crowd_regions offset = len(raw_commands) raw_commands.extend(part.commands) raw_events.extend( event if isinstance(event, CrowdEvent) else replace(event, object_ref=str(int(event.object_ref)+offset)) for event in part.events) configs.extend(members(part.scenario_config)) selected_positions = public_positions(raw_commands, raw_events) if len(raw_commands) > len(selected_positions): with self._cv: self._metrics['command_cap_drop'] += len(raw_commands)-len(selected_positions) remap = {old_index: new_index for new_index, old_index in enumerate(selected_positions)} commands = [raw_commands[i] for i in selected_positions] events = [event if isinstance(event, CrowdEvent) else replace(event, object_ref=str(remap[int(event.object_ref)])) for event in raw_events] return FrameResult( frame, tuple(commands), tuple(events), tuple(configs) if len(configs) > 1 else configs[0] if configs else None, self.scenario_revision, self.debug_all_candidates, self.pose_config if person_key == 'pose' and person_key in state.results else None, camera_id=self.camera_id, crowd_regions=crowd_regions) def _collect_person(self, frame, provider, copied, ids): if self.collector is None: return items = [] for detection, tid in zip(copied.detections, ids): if detection.class_id != 0: continue item = dict(detection.downstream(), track_id=tid) if detection.keypoints is not None: item['keypoints'] = [list(k) for k in detection.keypoints] items.append(item) task = 'pose' if provider.key == 'pose' else 'det' self.collector.add( task, frame.frame_seq, frame.pts_ns or 0, {'w': frame.frame_width, 'h': frame.frame_height, 'count': len(items), 'items': items, 'frame': frame}, max_frames=32, metrics=self.ingress.metrics) def scenario_status(self): with self._cv: active, disabled = [], [] for scenario in sorted(self._requested): key = self._dependencies[scenario] status = self._providers[key] if status['state'] == 'ACTIVE': active.append(scenario) else: disabled.append({'name': scenario, 'reason': status['reason']}) return {'active': active, 'disabled': disabled} def status(self): with self._cv: providers = { key: { **{k: v for k, v in value.items() if k not in ('first_input_at', 'last_input_at', 'last_response_at')}, 'last_response_age_s': ( round(self.clock()-value['last_response_at'], 3) if value['last_response_at'] is not None else None), 'first_input_at': value['first_input_at'], 'first_response_at': value.get('first_response_at'), } for key, value in self._providers.items() } return dict( source_latest_trial=getattr(self.ingress, 'source_latest', False), event_cooldown=self.event_cooldown.status(), state='ERROR' if self._error else ('STOPPED' if self._closed else ('CONNECTED' if self._metrics['finalized'] else 'PREPARED')), error=self._error, freshness_ms=round(self.freshness_s*1000), freshness_enabled=self.freshness_enabled, startup_seconds=self.startup_s, stall_seconds=self.stall_s, queued=len(self._frames), capacity=self.capacity, inflight_age_s=max((self.clock()-s.admitted_at for s in self._frames.values() if s.claimed and not s.results), default=0), pending_age_s=max((self.clock()-s.admitted_at for s in self._frames.values() if not s.claimed), default=0), providers=providers, scenarios=self.scenario_status(), provider_timing={key: {stage: dict(row) for stage, row in stages.items()} for key, stages in self._timings.items()}, artifacts=self.artifacts.status() if self.artifacts else None, fight_notification=self.fight.notification_status() if self.fight else None, fall_notification=self.fall.notification_status() if self.fall else None, visual=self.visual.status() if self.visual else None, performance=self.ingress.metrics.snapshot(), collector_frames=self._collector_frame_count(), finalizing=int(self._finalizing is not None), **self._metrics) def _collector_frame_count(self): if self.collector is None: return 0 with self.collector.lock: return len(self.collector.frames) def latest_result(self): with self._cv: return self._latest def snapshot(self): result = self.latest_result() return result.snapshot(visible_only=False) if result else None def close(self, timeout): deadline = self.clock() + max(0, timeout) self.stop_admission() with self._cv: self._closed = True self._frames.clear() self._cv.notify_all() if self._thread is not None and self._thread is not threading.current_thread(): self._thread.join(max(0, deadline-self.clock())) if self._thread.is_alive(): raise RuntimeError('Multi-provider coordinator cleanup timed out.') if self.tracker is not None: self.tracker.reset() if self.region: self.region.clear() if self.fall: self.fall.clear() if self.fight: self.fight.clear() if self.fire: self.fire.clear() if self.crowd: self.crowd.close() if hasattr(self.ingress, 'metrics'): self.ingress.metrics.close() if self.visual is not None and not self.visual.close(max(0, deadline-self.clock())): raise RuntimeError('Visual worker cleanup timed out.') if self.artifacts is not None and not self.artifacts.close(max(0, deadline-self.clock())): raise RuntimeError('Artifact capture cleanup timed out.') if self.publisher is not None: self.publisher.channel( self.ingress.identity[1], self.ingress.identity[3], self.ingress.identity[4], 'ERROR' if self._error else 'STOPPED') if not self.publisher.wait_channel( self.ingress.identity[1], self.ingress.identity[3], self.ingress.identity[4], max(0, deadline-self.clock())): raise RuntimeError('Metadata channel worker cleanup failed or timed out.')