"""PIPECORE-01/FRAMERESULT-01: SDK callback은 독립 값만 bounded queue에 넘긴다.""" from collections import Counter, deque from dataclasses import replace from app.region_scenario import alerted, event_ids, members, visible as scenario_visible import threading import time from app.contracts.frames import DetectionObservation, PoseObservation from app.pose_config import PERSON_MODELS, POSE_MODEL from app.fire_config import FIRE_MODEL, FIRE_NAMES from app.crowd_config import CROWD_MODEL from app.contracts.visualization import VisualizationCommand, FrameResult from app.metadata_samples import MetadataSamples class DetResultPath: def __init__(self, ingress, tracker, model_key, threshold, collector, is_current, on_error, *, publisher=None, capacity=32, max_age=1.0, clock=time.monotonic, display_threshold=None, visual=None, intrusion=None, scenario_revision=None, display_categories=None, debug_all_candidates=False, pose_config=None, camera_id=None, fire=None, fall=None, fight=None, crowd=None): if model_key not in PERSON_MODELS | {FIRE_MODEL, CROWD_MODEL}: raise ValueError('Canonical Det path requires COCO RFDETRMedium.') if model_key == FIRE_MODEL and (intrusion is not None or pose_config is not None): raise ValueError('Fire cannot consume person Scenario/Pose settings.') if fire is not None and model_key != FIRE_MODEL: raise ValueError('FIRE scenario requires the Fire provider.') if fire is not None and (intrusion is not None or tracker is not None): raise ValueError('FIRE scenario cannot consume person Scenario/Tracking state.') if fall is not None and (model_key != POSE_MODEL or tracker is None or fire is not None): raise ValueError('FALL requires one Pose provider and shared Tracking.') if fight is not None and (model_key != POSE_MODEL or tracker is None or fire is not None): raise ValueError('FIGHT requires one Pose provider and shared Tracking.') if (crowd is not None) != (model_key == CROWD_MODEL) or (crowd is not None and any(x is not None for x in (tracker,intrusion,fall,fight,fire,pose_config))): raise ValueError('CROWD requires standalone provider without Tracking.') if not 0 <= threshold <= 1 or type(capacity) is not int or not 1 <= capacity <= 256: raise ValueError('Invalid Det queue settings.') if not 0 < max_age < float('inf'): raise ValueError('Invalid Det queue age.') if tracker is None and intrusion is not None: raise ValueError('INTRUSION requires Tracking On.') self.ingress, self.tracker, self.model_key = ingress, tracker, model_key self.threshold, self.collector = threshold, collector self.display_threshold = threshold if display_threshold is None else display_threshold if type(self.display_threshold) not in (int, float) or not 0 <= self.display_threshold <= 1: raise ValueError('Invalid display threshold.') self.visual = visual self.pose_config = pose_config self.intrusion = intrusion self.fire = fire self.fall = fall self.fight, self.crowd = fight, crowd self.camera_id = camera_id self.artifacts = None self.scenario_revision = scenario_revision from app.stream_settings import display_categories as validate_categories selected = validate_categories(display_categories) self.display_categories = frozenset(selected) if selected is not None else None self.debug_all_candidates = debug_all_candidates self.is_current, self.on_error, self.publisher = is_current, on_error, publisher self.capacity, self.max_age, self.clock = capacity, max_age, clock self._cv = threading.Condition() self._queue = deque() self._closed = False self._last_seq = -1 self._metrics = Counter() self._latest = None self._error = None self._thread = None self._samples = MetadataSamples.from_environment() def start(self): with self._cv: if self._closed or self._thread is not None: raise RuntimeError('Det result worker 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.') self._thread = threading.Thread(target=self._run, name='det-results', daemon=True) self._thread.start() def submit(self, frame, copied, handle=None, *, gap_before=False): if (frame.server_id, frame.source_id, frame.stream_id, frame.run_id, frame.run_generation) != self.ingress.identity: raise ValueError('Stale or foreign Frame.') if (frame.frame_width, frame.frame_height) != (copied.width, copied.height): raise ValueError('Native/ingress source dimensions differ.') if len(copied.detections) > 300: raise ValueError('Native candidate limit exceeded.') if self.model_key == FIRE_MODEL and any(d.class_id not in (0,1) for d in copied.detections): raise ValueError('Invalid Fire class id.') observations = tuple(DetectionObservation(frame, self.model_key, d.class_id, d.score, d.bbox_xyxy) for d in copied.detections) poses = tuple(PoseObservation(frame,self.model_key,d.score,d.bbox_xyxy,d.keypoints) if d.keypoints is not None else None for d in copied.detections) if any((p is not None) != (self.model_key == POSE_MODEL) for p in poses): raise ValueError('Provider output contract mismatch.') with self._cv: if self._closed or not self.is_current(): self._metrics['stale'] += 1 return False if frame.frame_seq <= self._last_seq: self._metrics['out_of_order'] += 1 return False self._last_seq = frame.frame_seq if (self.fall is not None or self.fight is not None or self.crowd is not None) and hasattr(self.ingress, 'result_gap'): gap_before = self.ingress.result_gap(frame) or gap_before if gap_before: self._metrics['fall_upstream_gap'] += 1 if len(self._queue) == self.capacity: self._queue.popleft() self._metrics['queue_drop'] += 1 if self._queue: self._queue[0] = (*self._queue[0][:-1], True) else: gap_before = True self._queue.append((self.clock(), frame, copied, observations, poses, handle, gap_before)) self._cv.notify() return True def _run(self): skipped_before = False try: while True: with self._cv: self._cv.wait_for(lambda: self._closed or self._queue, .25 if self.intrusion or self.crowd else None) if self._closed: return item = self._queue.popleft() if self._queue else None # Scenario state is worker-owned; close() clears it only after join. if item is None: if self.crowd: self.crowd.expire_idle() if self.intrusion: self.intrusion.expire() continue submitted, frame, copied, observations, poses, handle, gap_before = item if not self.is_current() or self.clock() - submitted > self.max_age: if self.intrusion: self.intrusion.expire() with self._cv: self._metrics['stale'] += 1 skipped_before = True continue # This accepted Frame finishes event handoff even if Stop arrives during evaluation. # close() joins this worker before closing capture or switching to a new run. with self._cv: if self._closed or not self.is_current(): continue gap_before = gap_before or skipped_before skipped_before = False metrics = getattr(self.ingress, 'metrics', None) started_ns = time.monotonic_ns() detail = metrics is not None and metrics.detail_enabled() started_cpu = time.thread_time_ns() if detail else 0 if metrics: metrics.duration('queue', int((self.clock() - submitted) * 1e9)) if self.crowd is not None: result = self.crowd.process(frame, observations, gap_before=gap_before) if result is None: continue visible = {id(c.detection) for c in result.commands} kept = tuple(id(o) in visible for o in observations) ids, events = [None]*len(observations), result.events elif self.fire is not None: scenario_start, scenario_cpu = time.monotonic_ns(), time.thread_time_ns() from app.fire_result import evaluate_result as evaluate_fire_result result = evaluate_fire_result(self.fire, frame, observations, camera_id=self.camera_id, scenario_revision=self.scenario_revision) if result is None: continue visible = {id(command.detection) for command in result.commands} kept = tuple(id(observation) in visible for observation in observations) ids = [None] * len(observations) events = result.events if metrics: metrics.duration('scenario_cpu', time.thread_time_ns() - scenario_cpu) metrics.duration('scenario', time.monotonic_ns() - scenario_start) else: selected=tuple(range(len(observations))) if self.pose_config and self.pose_config.nms_enabled: from app.pose_nms import pose_nms_indices nms_start=time.monotonic_ns() selected=pose_nms_indices(observations,self.pose_config.nms_iou_threshold) with self._cv: self._metrics['nms_calls']+=1 self._metrics['nms_elapsed_ns']+=time.monotonic_ns()-nms_start self._metrics['nms_suppressed']+=len(observations)-len(selected) selected_set=set(selected) kept=tuple(i in selected_set for i in range(len(observations))) selected_observations=tuple(observations[i] for i in selected) tracking_start=time.monotonic_ns() tracking_cpu = time.thread_time_ns() selected_ids = (self.tracker.update(selected_observations) if self.tracker is not None else [None] * len(selected_observations)) ids=[None]*len(observations) if len(selected_ids)!=len(selected): raise ValueError('Tracker candidate association is incomplete.') for i,tid in zip(selected,selected_ids):ids[i]=tid tracking_cpu = time.thread_time_ns() - tracking_cpu if metrics: metrics.duration('tracking', time.monotonic_ns() - tracking_start) metrics.duration('tracking_cpu', tracking_cpu) if len(ids) != len(observations): raise ValueError('Tracker candidate association is incomplete.') decisions, events = ((None,)*len(observations), ()) if self.intrusion: scenario_start, scenario_cpu = time.monotonic_ns(), time.thread_time_ns() selected_decisions, events = self.intrusion.evaluate(frame, selected_observations, selected_ids) decisions=[None]*len(observations) for i,d in zip(selected,selected_decisions):decisions[i]=d events=tuple(replace(e,object_ref=str(selected[int(e.object_ref)])) for e in events) if metrics: metrics.duration('scenario_cpu', time.thread_time_ns() - scenario_cpu) metrics.duration('scenario', time.monotonic_ns() - scenario_start) for scenario in (self.fall, self.fight): if scenario is None: continue scenario_start, scenario_cpu = time.monotonic_ns(), time.thread_time_ns() fall_decisions, fall_events = scenario.evaluate(frame, tuple(poses[i] for i in selected), selected_ids, gap_before=gap_before) decisions = list(decisions) for index, decision in zip(selected, fall_decisions): decisions[index] = (*members(decisions[index]), *members(decision)) events += tuple(replace(e, object_ref=str(selected[int(e.object_ref)])) for e in fall_events) if metrics: metrics.duration('scenario_cpu', time.thread_time_ns() - scenario_cpu) metrics.duration('scenario', time.monotonic_ns() - scenario_start) commands = tuple(VisualizationCommand(o, self._label(o.class_id), color='#FF3030' if alerted(d) else '#00CC66', visible=keep and (scenario_visible(o.score, d) if d else (o.score >= self.pose_config.person_score if self.pose_config else o.score > self.display_threshold) and (self.display_categories is None or self._label(o.class_id) in self.display_categories)), track_id=tid, event_ids=event_ids(d), scenario=d, pose=p) for o, tid, d, p, keep in zip(observations, ids, decisions, poses, kept)) configs = members(self.intrusion.config if self.intrusion 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, commands, events, configs if len(configs) > 1 else configs[0] if configs else None, self.scenario_revision, self.debug_all_candidates, self.pose_config, kept if self.pose_config else None, provider_model=self.model_key, camera_id=self.camera_id) with self._cv: if self._closed or not self.is_current(): if self.fall is None and self.fight is None and not (self.crowd and result.events): continue else: self._latest = result self._metrics['events_created'] += len(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: accepted = self.artifacts.submit(result) if accepted is False and any(e.payload()['scenario'] in ('FALL','FIGHT','CROWD') for e in events): raise RuntimeError('FALL Event handoff rejected.') items = [dict(d.downstream(), track_id=tid) for d, tid, keep in zip(copied.detections, ids, kept) if keep and d.score > self.threshold] # Det-only compatibility collector. Pose join은 이 key를 사용하지 않는다. collector_frames = self.collector.add('pose' if self.model_key==POSE_MODEL else 'det', 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=1000, metrics=metrics) if detail: publish_start, publish_cpu = time.monotonic_ns(), time.thread_time_ns() if self.publisher is not None: if metrics: metrics.mark('result', (copied.frame_number, copied.pts)) accepted = self.publisher.channel(frame.source_id, frame.run_id, frame.run_generation, 'RUNNING') if not accepted or not self.publisher.submit(result.snapshot()): with self._cv: self._metrics['publisher_rejected'] += 1 elif metrics: metrics.mark('publisher_submit', (copied.frame_number, copied.pts)) elif metrics: metrics.mark('result', (copied.frame_number, copied.pts)) if detail: metrics.duration('result_publish', time.monotonic_ns() - publish_start) metrics.duration('result_publish_cpu', time.thread_time_ns() - publish_cpu) metrics.duration('result_work_cpu', time.thread_time_ns() - started_cpu) if metrics: metrics.duration('result_work', time.monotonic_ns() - started_ns) if detail: lifecycle_start, lifecycle_cpu = time.monotonic_ns(), time.thread_time_ns() if handle is not None: handle.mark_source_stage('inference_frame_completed') handle.mark_frame(detection_count=len(items)) if detail: metrics.duration('result_lifecycle', time.monotonic_ns() - lifecycle_start) metrics.duration('result_lifecycle_cpu', time.thread_time_ns() - lifecycle_cpu) with self._cv: if type(collector_frames) is int: self._metrics['collector_frames'] = collector_frames self._metrics['processed'] += 1 except Exception: with self._cv: self._error = 'DET_RESULT_FAILED' self._queue.clear() self._closed = True self.on_error() def _label(self, class_id): if self.model_key == FIRE_MODEL: return FIRE_NAMES[class_id] from app.class_catalog import coco_name return coco_name(class_id) def snapshot(self): result = self.latest_result() return result.snapshot(visible_only=False) if result else None def latest_result(self): with self._cv: return self._latest def status(self): with self._cv: return dict(state='ERROR' if self._error else ('STOPPED' if self._closed else ('CONNECTED' if self._metrics['processed'] else 'PREPARED')), error=self._error, display_threshold=self.display_threshold, visual=self.visual.status() if self.visual else None, 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, sample_error=self._samples.error if self._samples else None, performance=self.ingress.metrics.snapshot() if hasattr(self.ingress, 'metrics') else None, fire_metrics=dict(self.fire.metrics) if self.fire is not None else None, queued=len(self._queue), **self._metrics) def close(self, timeout): deadline = time.monotonic() + max(0, timeout) with self._cv: self._closed = True self._queue.clear() self._latest = None self._cv.notify_all() if self._thread is not None: self._thread.join(max(0, timeout)) if self._thread.is_alive(): raise RuntimeError('Det result worker cleanup timed out.') if self.tracker is not None: self.tracker.reset() if self.intrusion: self.intrusion.clear() if self.fire: self.fire.clear() if self.fall: self.fall.clear() if self.fight: self.fight.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 - time.monotonic())): raise RuntimeError('Visual worker cleanup timed out.') if self.artifacts is not None and not self.artifacts.close(max(0, deadline - time.monotonic())): 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 - time.monotonic())): raise RuntimeError('Metadata channel worker cleanup failed or timed out.')