"""multi_pipeline: 3개 모델 독립 파이프라인 + Python 병합 (검증된 아키텍처). 구조: pipeline_det: capture -> infer(RFDETRMedium, uid=1, custom-lib) -> retrieve pipeline_pose: capture -> infer(RFDETRKeypointPreview, uid=7) -> retrieve pipeline_seg: capture -> infer(RFDETRSegMedium, uid=8) -> retrieve 각 파이프라인은 별도 스레드에서 p.start().wait() 로 구동. Det는 Native Object Metadata를 복사한다. Pose/Seg는 기존 tensor parser를 사용한다. 결과를 buffer_pts 키로 dict에 병합 (동일 파일/RTSP 기준 pts 일치 가정, 불일치 시 ±1 frame_number 근접 매칭). Object Metadata wrapper는 iterator 밖으로 전달하지 않는다. Pipeline Runtime 검증은 별도다. """ import json import threading import time import os from typing import Optional, Any import numpy as np os.environ["GST_PLUGIN_SCANNER"] = "/usr/lib/x86_64-linux-gnu/gstreamer1.0/gstreamer1.0/gst-plugin-scanner" import gi gi.require_version('Gst', '1.0') from gi.repository import Gst, GLib Gst.init(None) from pyservicemaker import BufferOperator, BufferRetriever, Flow, Pipeline, Probe from app.rfdetr_postprocess import ( parse_boxes_labels, parse_pose_keypoints, squeeze_batch, ) from app.source_adapters.rtsp import RtspSourceAdapterError from app.native_metadata import copy_detection_frame, copy_pose_frame, failure_diagnostic from app.start_diagnostics import StartDiagnostics from app.stream_lifecycle import ( RoleRunHandle, classify_pipeline_failure, make_stream_failure, ) DET_CFG = "/opt/contextra/app/nvinferserver_grpc_det.txt" # RFDETRMedium POSE_CFG = "/opt/contextra/app/nvinferserver_grpc_pose.txt" # RFDETRKeypointPreview SEG_CFG = "/opt/contextra/app/nvinferserver_grpc_seg.txt" # RFDETRSegMedium # pts 불일치 허용 폭. frame_number 키 사용 시 폴백만. PTS_TOL_NS = 5_000_000 # 5ms class _Cap: def __init__(self, c, dev): self._c = c self._dev = dev def __dlpack__(self): return self._c def __dlpack_device__(self): return self._dev class _DecodedFrameObserver(BufferOperator): """Observe the decoded/muxed Frame without copying its pixel payload.""" def __init__(self, role_handle: RoleRunHandle, stage="decoded_frame_received", *, drop_on_stop=False, ingress=None, metrics=None): super().__init__() self._role_handle = role_handle self._stage = stage self._drop_on_stop = drop_on_stop self._ingress = ingress self._metrics = metrics def handle_buffer(self, _buffer): self._role_handle.diagnostic.count(self._stage) if self._drop_on_stop and self._role_handle.stop_requested: self._role_handle.diagnostic.count("dropped_before_infer") return False if self._ingress is not None: try: if not self._ingress.stamp(_buffer.batch_meta): return False except Exception as error: self._role_handle.mark_failure(make_stream_failure( 'ROLE_RUNTIME_FAILED', 'frame_identity', 'det', error)) return False if self._metrics is not None: stage = {'preprocess_output': 'preprocess', 'infer_output': 'infer'}[self._stage] for frame in _buffer.batch_meta.frame_items: self._metrics.mark(stage, (int(frame.frame_number), int(frame.buffer_pts))) self._role_handle.mark_source_stage(self._stage) return True def _dl(t): return np.from_dlpack(_Cap(t.__dlpack__(None), t.__dlpack_device__())) class ResultCollector: """3개 파이프라인 결과를 frame_number 키로 병합 (동일 소스 기준). buffer_pts는 이 영상에서 신뢰할 수 없음(0, 11111ns, ... — 프레임 단위가 아님). frame_number(0,1,2,…)가 자연 정렬 키. frame_number 부재 시 pts±TOL 폴백. """ def __init__(self, required_roles=None): self.lock = threading.Lock() self.frames = {} # key -> {"det": ..., "pose": ..., "seg": ...} self.merged = 0 self.per_task = {} # task -> count self.completed_tasks = set() self.required = set(required_roles or ("det", "pose", "seg")) def _key(self, fn: int, pts: int): if fn >= 0: return fn # pts 폴백 (frame_number 부재) for k, f in self.frames.items(): if isinstance(f.get("_pts"), int) and abs(f["_pts"] - pts) <= PTS_TOL_NS: return k return pts def add(self, task: str, fn: int, pts: int, data: dict, max_frames: int = 0, metrics=None): detail = metrics is not None and metrics.detail_enabled() wait_start = time.monotonic_ns() if detail else 0 eviction = None with self.lock: if detail: acquired, cpu_start = time.monotonic_ns(), time.thread_time_ns() key = self._key(fn, pts) f = self.frames.get(key) if f is None: f = {"_fn": fn, "_pts": pts} self.frames[key] = f # 무한 루프 모드: dict 무제한 팽창 방지 (최근 100프레임만) if max_frames > 0 and len(self.frames) > max_frames: if detail: evict_start, evict_cpu = time.monotonic_ns(), time.thread_time_ns() excess = len(self.frames) - max_frames # The usual one-Frame overflow needs no full-history sort. rank = lambda x: x[1].get("_fn", 0) oldest = ([min(self.frames.items(), key=rank)] if excess == 1 else sorted(self.frames.items(), key=rank)[:excess]) for ok, _ in oldest: self.frames.pop(ok, None) if detail: eviction = (time.monotonic_ns() - evict_start, time.thread_time_ns() - evict_cpu) f[task] = data self.per_task[task] = self.per_task.get(task, 0) + 1 if self.required <= set(f.keys()): self.merged += 1 count = len(self.frames) if detail: work, cpu = time.monotonic_ns() - acquired, time.thread_time_ns() - cpu_start # Record outside the collector lock; diagnostics must not lengthen its critical section. if detail: metrics.duration('collector_wait', acquired - wait_start) metrics.duration('collector_work', work) metrics.duration('collector_work_cpu', cpu) if eviction is not None: metrics.duration('collector_evict', eviction[0]) metrics.duration('collector_evict_cpu', eviction[1]) return count def task_done(self, task: str): with self.lock: self.completed_tasks.add(task) # COCO 80 foreground class names (index 0..79 == COCO 1..80). # Native Det는 명시적 COCO mapping을 마친 zero-based class_id를 전달한다. # COCO 클래스 카탈로그 — app/class_catalog.py(SSOT)에서 재사용. # multi_pipeline 내 정의 제거: capability 계산 등 가벼운 경로가 GStreamer # import 없이 class_catalog만 사용하도록 분리 (v2 구동 정책). from app.class_catalog import COCO_NAMES, coco_name as _coco_name def _frame_to_hwc(t): """SDK RGB tensor -> independent CPU RGB HWC, not BGR (CAPTURE-RGB-01).""" from app.capture_pixels import copy_sdk_rgb return copy_sdk_rgb(t) class StorageWriter: """병합 프레임을 metadata_storage (PostgreSQL + Redis Streams)에 기록. main.py의 API 경로와 동일한 계약 준수: - private_key = {site_id}_{cam}_{now_ms}_{fn:05d} - insert_frame -> insert_detections (동일 pkey, frames.id FK) - Redis publish_frame / publish_detections (fire-and-forget) bbox는 픽셀 xyxy, bbox_normalized는 정규화 xyxy로 변환. 저장 실패는 파이프라인을 막지 않는다 (non-fatal, 로그만). """ def __init__(self, site_id="SITE01", stream_id="CAM_01", store=True, persist_every=1, scenario_runner=None, al_writer=None, gate=None, required_roles=None, tracker=None): self.site_id = site_id self.stream_id = stream_id self.scenario_runner = scenario_runner self.al_writer = al_writer # Phase 3 — Python-side IoU 트래커 (track_id 실체화). None이면 track_id=None 유지. self.tracker = tracker self._required = set(required_roles or ("det", "pose", "seg")) # 이벤트 게이트 (persist_mode=event_gated) — 지정 시 PG 직접 기록 대신 # 프레임은 gate 버퍼(메모리)로, 이벤트 발생 시 ±윈도우만 PG 플러시. self.gate = gate self.cam = stream_id.replace("CAM_", "") self.persist_every = max(1, int(persist_every)) self.store = bool(store) self.persisted = set() self._seq = 0 self._storage = None self._redis = None self.FrameMeta = None self.DetectionMeta = None if self.store: try: from app.metadata_storage import ( storage, redis_publisher, FrameMeta, DetectionMeta, ) self._storage = storage self._redis = redis_publisher self.FrameMeta = FrameMeta self.DetectionMeta = DetectionMeta except Exception as e: print(f"[STORAGE] disabled (non-fatal): {e}", flush=True) self.store = False def _flush(self): for x in (self._storage, self._redis): try: if x is not None and hasattr(x, "close"): x.close() except Exception: pass def persist(self, fn_key, f: dict, w: int, h: int) -> bool: if not self.store: return False if fn_key in self.persisted: return False self._seq += 1 if (self._seq % self.persist_every) != 0: self.persisted.add(fn_key) return False now_ms = int(time.time() * 1000) pkey = f"{self.site_id}_{self.cam}_{now_ms}_{int(fn_key):05d}" # 이벤트 게이트 모드: PG 직접 기록 대신 메모리 버퍼 유지. # 시나리오 엔진은 여기서 그대로 실행 (이벤트는 항상 PG에 저장) → # 발생 이벤트를 gate.on_events()로 전달해 ±윈도우 플러시를 예약한다. # AL raw frame 캡처는 det 파이프라인(al_writer)에서 수행 — 여기서는 # meta JSON/Redis 프레임 발행만 스킵 (이벤트 Redis 발행은 runner가 담당). if self.gate is not None: try: raw_dets = [] for d in f.get("det", {}).get("items", []): cx, cy, bw, bh = (float(v) for v in d["box"]) raw_dets.append({ "track_id": d.get("track_id"), # Phase 3 — IoU 트래커 id (None 가능) "class_name": _coco_name(d["class_id"]), "confidence": float(d["score"]), "bbox": ((cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h), }) if self.scenario_runner is not None and raw_dets: evs = self.scenario_runner.run(raw_dets, pkey, self.stream_id, self.site_id, now_ms, frame_size=(w, h)) if evs: self.gate.on_events(evs) self.gate.offer(fn_key, f, w, h, pkey) self.persisted.add(fn_key) return True except Exception as ge: print(f"[STORAGE] gate persist failed fn={fn_key} (non-fatal): {ge}", flush=True) self.persisted.add(fn_key) return False try: frame_meta = self.FrameMeta( private_key=pkey, stream_id=self.stream_id, site_id=self.site_id, timestamp_utc_ms=now_ms, frame_sequence_id=int(fn_key), inference_fps=15.0, width=w, height=h, ) self._storage.insert_frame(frame_meta) dets = [] raw_dets = [] for d in f.get("det", {}).get("items", []): cx, cy, bw, bh = (float(v) for v in d["box"]) dets.append(self.DetectionMeta( private_key=pkey, track_id=d.get("track_id"), # Phase 3 — IoU 트래커 id (None 가능) class_label=_coco_name(d["class_id"]), confidence=float(d["score"]), bbox=( (cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h, ), bbox_normalized=( cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2, ), )) raw_dets.append({ "track_id": d.get("track_id"), # Phase 3 — IoU 트래커 id (None 가능) "class_name": _coco_name(d["class_id"]), "confidence": float(d["score"]), "bbox": ((cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h), }) # pose(keypoints) — 동일 detections 테이블에 person 행으로 저장 # keypoints = [[x,y,conf]x17] 정규화 (추론/파싱은 이미 수행, 저장만 추가) for p in f.get("pose", {}).get("items", []): cx, cy, bw, bh = (float(v) for v in p["box"]) dets.append(self.DetectionMeta( private_key=pkey, track_id=None, class_label="person", confidence=float(p["score"]), bbox=( (cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h, ), bbox_normalized=( cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2, ), keypoints=p.get("keypoints"), )) # seg — 동일 detections 테이블에 seg 행으로 저장 (mask 요약) # 풀 마스크(144x144)는 저장 부담이 크므로 fg_px/비중/해상도 요약만 for sg in f.get("seg", {}).get("items", []): cx, cy, bw, bh = (float(v) for v in sg["box"]) total_px = int((f.get("seg", {}).get("mask_w", 144)) * (f.get("seg", {}).get("mask_h", 144))) or 20736 dets.append(self.DetectionMeta( private_key=pkey, track_id=None, class_label=_coco_name(sg["class_id"]), confidence=float(sg["score"]), bbox=( (cx - bw / 2) * w, (cy - bh / 2) * h, (cx + bw / 2) * w, (cy + bh / 2) * h, ), bbox_normalized=( cx - bw / 2, cy - bh / 2, cx + bw / 2, cy + bh / 2, ), source="seg", seg={ "fg_px": int(sg["fg_px"]), "fg_ratio": round(int(sg["fg_px"]) / total_px, 6), "mask_w": int(f.get("seg", {}).get("mask_w", 144)), "mask_h": int(f.get("seg", {}).get("mask_h", 144)), }, )) if dets: self._storage.insert_detections(dets) # 시나리오 엔진 (배회/침입) — det 기반으로 이벤트 생성 (non-fatal) if self.scenario_runner is not None and raw_dets: try: self.scenario_runner.run(raw_dets, pkey, self.stream_id, self.site_id, now_ms, frame_size=(w, h)) except Exception as se: print(f"[SCENARIO] run failed (non-fatal): {se}", flush=True) if self._redis is not None: try: self._redis.publish_frame(frame_meta) if dets: self._redis.publish_detections(dets) except Exception as re: print(f"[STORAGE] redis publish failed (non-fatal): {re}", flush=True) # Active Learning metadata JSON backup (non-fatal) if self.al_writer is not None: try: self.al_writer.submit_meta(pkey, { "private_key": pkey, "stream_id": self.stream_id, "site_id": self.site_id, "timestamp_utc_ms": now_ms, "frame_sequence_id": int(fn_key), "width": w, "height": h, "det": f.get("det", {}), "pose": f.get("pose", {}), "seg": f.get("seg", {}), }) except Exception as ae: print(f"[AL] meta submit failed (non-fatal): {ae}", flush=True) self.persisted.add(fn_key) n_pose = len(f.get("pose", {}).get("items", [])) n_seg = len(f.get("seg", {}).get("items", [])) print(f"[STORAGE] frame fn={fn_key} pkey={pkey} dets={len(dets)} pose_kp={n_pose} seg={n_seg}", flush=True) return True except Exception as e: print(f"[STORAGE] frame fn={fn_key} persist failed (non-fatal): {e}", flush=True) self.persisted.add(fn_key) # 재시도 폭주 방지 return False def persist_new(self, collector: "ResultCollector"): """collector의 병합 프레임 중 미기록 프레임을 기록. 반환=이번 기록 수.""" if not self.store: return 0 written = 0 with collector.lock: items = list(collector.frames.items()) for key, f in items: if not (self._required <= set(f.keys())): continue w = f.get("det", {}).get("w", 1920) h = f.get("det", {}).get("h", 1080) if self.persist(key, f, w, h): written += 1 return written class ScenarioRunner: """Compatibility rejection: use canonical Stream settings for new INTRUSION.""" def __init__(self, *args, **kwargs): raise ValueError('Legacy scenarios are retired; use canonical INTRUSION settings.') def _make_retriever(task: str, cfg: str, src: str, collector: ResultCollector, max_buffers: int, state: dict, al_writer=None, score_threshold: float = 0.5, tracker=None, role_handle: RoleRunHandle | None = None, component_id: int | None = None, is_current=None, result_path=None): def active(): return ((role_handle is None or not role_handle.stop_requested) and (is_current is None or is_current())) class R(BufferRetriever): def consume(self, buffer): callback_at = time.monotonic_ns() if role_handle is not None: role_handle.diagnostic.count("consume_entered") if not active(): return 1 if max_buffers > 0 and state["n"] >= max_buffers: return 1 state["n"] += 1 n = state["n"] if role_handle is not None: role_handle.mark_source_stage("consume_entered") try: visual = result_path.visual if result_path is not None else None frame_count = sum(1 for _ in buffer.batch_meta.frame_items) if visual is not None else 0 # SDK Frame wrapper는 iterator 수명 안에서 소비한다. list로 보관하지 않는다. for fm in buffer.batch_meta.frame_items: from app.pipeline_diagnostics import diagnostics diagnostics.mark('callback', (int(fm.frame_number), int(fm.buffer_pts)), callback_at) if not active(): break if task == 'det' or (task == 'pose' and result_path is not None): copied = (copy_pose_frame if task=='pose' else copy_detection_frame)(fm, component_id) envelope = result_path.ingress.copy(fm) if result_path is not None else None if result_path is not None: result_path.ingress.metrics.mark('native', (int(fm.frame_number), int(fm.buffer_pts))) if not active(): break if role_handle is not None: role_handle.mark_source_stage("native_completed") layers = None else: titems = list(fm.tensor_items) if not titems: continue layers = titems[0].as_tensor_output().get_layers() pts = int(getattr(fm, "buffer_pts", 0) or 0) fn = int(getattr(fm, "frame_number", -1)) w = int(getattr(fm, "source_width", 1920) or 1920) h = int(getattr(fm, "source_height", 1080) or 1080) if task == 'pose' and layers is not None and "keypoints" in layers: dets = squeeze_batch(_dl(layers["dets"]), 2) labels = squeeze_batch(_dl(layers["labels"]), 2) kp = squeeze_batch(_dl(layers["keypoints"]), 3) r = parse_boxes_labels(dets, labels, score_threshold) kpdec = parse_pose_keypoints(kp, r["model_classes"]) idx = np.where(r["keep"])[0] persons = [{ "box": tuple(float(v) for v in dets[i]), "score": float(r["scores"][i]), "keypoints": kpdec[i].tolist(), # [17,3] normalized } for i in idx] collector.add(task, fn, pts, { "w": w, "h": h, "count": len(persons), "items": persons, }, max_frames=(0 if max_buffers > 0 else 1000)) if role_handle is not None: role_handle.mark_frame(detection_count=len(persons)) print(f"[{task.upper()}] buf{n} fn={fn} pts={pts}: {len(persons)} persons", flush=True) elif task == 'seg' and "masks" in layers: dets = squeeze_batch(_dl(layers["dets"]), 2) labels = squeeze_batch(_dl(layers["labels"]), 2) masks = squeeze_batch(_dl(layers["masks"]), 3) r = parse_boxes_labels(dets, labels, score_threshold) idx = np.where(r["keep"])[0] segs = [] for i in idx: m = masks[i].astype(np.float64) mprob = 1.0 / (1.0 + np.exp(-np.clip(m, -500, 500))) segs.append({ "box": tuple(float(v) for v in dets[i]), "class_id": int(r["classes"][i]), "score": float(r["scores"][i]), "fg_px": int((mprob > 0.5).sum()), }) collector.add(task, fn, pts, { "w": w, "h": h, "count": len(segs), "items": segs, "mask_w": int(masks.shape[-1]), "mask_h": int(masks.shape[-2]), }, max_frames=(0 if max_buffers > 0 else 1000)) if role_handle is not None: role_handle.mark_frame(detection_count=len(segs)) print(f"[{task.upper()}] buf{n} fn={fn} pts={pts}: {len(segs)} segs (mask {int(masks.shape[-2])}x{int(masks.shape[-1])})", flush=True) elif task == 'det' or (task == 'pose' and result_path is not None): fn, pts, w, h = (copied.frame_number, copied.pts, copied.width, copied.height) # THRESH-01: Native 후보는 보존하고 기존 profile의 소비자 필터만 적용한다. dets_out = [item.downstream() for item in copied.detections if item.score > score_threshold] if dets_out and not state.get('native_geometry_reported'): state['native_geometry_reported'] = True sample = [dict(class_id=d.class_id, score=d.score, bbox_xyxy=d.bbox_xyxy) for d in copied.detections if d.score > score_threshold][:3] print('[NATIVE_BRIDGE_SAMPLE] ' + json.dumps({ 'source': [w, h], 'surface': [copied.bbox_width, copied.bbox_height], 'detections': sample, }), flush=True) if result_path is not None: # THRESH-01: ByteTrack에는 표시 threshold 이전의 Native 후보 전체 전달. result_path.submit(envelope, copied, role_handle) diagnostics.mark('submitted', (copied.frame_number, copied.pts)) else: # 기존 CLI/미설정 Stream 호환. canonical 경로의 fallback으로 사용하지 않는다. from app.tracker import assign_track_ids assign_track_ids(tracker, dets_out) collector.add(task, fn, pts, { "w": w, "h": h, "count": len(dets_out), "items": dets_out, }, max_frames=(0 if max_buffers > 0 else 1000)) if role_handle is not None and result_path is None: # 탐지 0건도 정상 처리된 Frame이므로 lifecycle 첫 Frame으로 인정한다. role_handle.mark_source_stage("inference_frame_completed") role_handle.mark_frame(detection_count=len(dets_out)) # Active Learning: capture the decoded frame ONLY in the # det pipeline (single extraction per frame) -> background writer. _hwc = None if al_writer is not None and task == "det" and fn >= 0: try: _hwc = _frame_to_hwc(buffer.extract(0)) al_writer.submit_frame(fn, _hwc, dets_out) except Exception as aexc: print(f"[AL] frame capture failed fn={fn} (non-fatal): {aexc}", flush=True) # 같은 callback의 identity/독립 CPU 이미지로만 결합한다. 인코딩은 worker 소유. visual = result_path.visual if result_path is not None else None if visual is not None and frame_count != 1: visual.note_failure('unsupported_batch') elif visual is not None and active() and visual.reserve_image( envelope, (copied.bbox_width, copied.bbox_height)): try: _start = time.monotonic_ns() if _hwc is None: _tensor = buffer.extract(0) _extracted = time.monotonic_ns() _hwc = _frame_to_hwc(_tensor) del _tensor else: _extracted = _start visual.note_capture(envelope.frame_seq, _start, _extracted, time.monotonic_ns()) visual.submit_image(envelope, _hwc, (copied.bbox_width, copied.bbox_height)) except Exception: visual.note_failure('extract_failed') except Exception as error: if task == 'det' and not state.get('native_error_reported'): state['native_error_reported'] = True print('[NATIVE_METADATA_DIAGNOSTIC] ' + json.dumps(failure_diagnostic(error)), flush=True) failure = make_stream_failure( "ROLE_RUNTIME_FAILED", "native_metadata" if task == 'det' else "tensor_parse", task, error ) if role_handle is not None: role_handle.mark_failure(failure) print( f"[{task.upper()}] pipeline error " f"code={failure.error_code} stage={failure.failure_stage}", flush=True, ) return -1 if max_buffers > 0 and n >= max_buffers: def _s(): state["pipe"].stop() return False GLib.timeout_add(300, _s) # SDK Receiver expects consumed buffer count, not GST_FLOW_OK (0 retries after 1s). return 1 R._cfg = cfg R._src = src return R def _handle_pipeline_message(handle, message, *, finite=False): """SDK가 공개한 종류만 처리한다. 원문 message/name/URI는 읽지 않는다.""" kind = type(message).__name__ if kind == 'EOSMessage' and not handle.stop_requested and not finite: handle.mark_failure(make_stream_failure('ROLE_RUNTIME_FAILED', 'bus_eos', handle.role)) elif kind == 'StateTransitionMessage': handle.mark_source_stage('pipeline_state_changed') def _pipeline_message_callback(handle, *, finite=False): # 설치 SDK binding/내부 예제는 docstring과 달리 message 한 개를 전달한다. def on_message(message): _handle_pipeline_message(handle, message, finite=finite) return on_message def run_pipeline(name, cfg, src, collector, max_buffers, al_writer=None, pipes=None, score_threshold: float = 0.5, tracker=None, run_generation: int = 0, on_role_update=None, source_adapter=None, component_id: int | None = None, is_current=None, drop_on_stop=False, result_path=None): state = {"n": 0, "pipe": None} handle = RoleRunHandle(name, run_generation, on_role_update) def _worker(): diagnostic = StartDiagnostics(name) prepared_source = None decoded_probe = None stage_probes = [] try: if handle.stop_requested: return with diagnostic.span("pipeline"): p = Pipeline(name=name) state["pipe"] = p handle.bind_pipeline(p) if handle.stop_requested: return if pipes is not None: pipes[name] = p R = _make_retriever(name, cfg, src, collector, max_buffers, state, al_writer, score_threshold=score_threshold, tracker=tracker, role_handle=handle, component_id=component_id, is_current=is_current, result_path=result_path) diagnostic.mark("flow", "enter") preprocess_cfg = str(cfg) + '.preprocess' frame_size = None if os.path.isfile(preprocess_cfg): import configparser pre = configparser.ConfigParser() pre.read(preprocess_cfg) frame_size = (pre.getint('property','processing-width'),pre.getint('property','processing-height')) if source_adapter is not None: with diagnostic.span("source_prepare"): prepared_source = source_adapter.prepare(name, run_generation) decoded_probe = Probe( f"decoded-frame-{name}-{run_generation}", _DecodedFrameObserver(handle, drop_on_stop=drop_on_stop, ingress=result_path.ingress if result_path else None), ) capture = prepared_source.build_capture( Flow(p), p, decoded_probe=decoded_probe, rtp_time=bool(result_path and (result_path.ingress.frame_time is not None or result_path.artifacts is not None)), diagnostic=diagnostic, frame_size=frame_size, ) else: # File validation must terminate at EOS; RTSP has its own adapter. dimensions = {} if frame_size is None else dict(width=frame_size[0], height=frame_size[1]) capture = Flow(p).batch_capture([src], **{"file-loop": False}, **dimensions) if result_path is not None: decoded_probe = Probe('frame-identity', _DecodedFrameObserver( handle, drop_on_stop=drop_on_stop, ingress=result_path.ingress)) capture = capture.attach(decoded_probe) if os.path.isfile(preprocess_cfg): # SDK Flow 기본 target ID=1이 렌더링한 infer ID를 덮어쓰지 않도록 명시한다. with diagnostic.span("preprocess"): capture = capture.preprocess(preprocess_cfg, target_unique_ids=component_id) stage_probes.append(Probe('preprocess-stage', _DecodedFrameObserver( handle, 'preprocess_output', drop_on_stop=drop_on_stop, metrics=result_path.ingress.metrics if result_path else None))) capture = capture.attach(stage_probes[-1]) with diagnostic.span("infer"): inferred = capture.infer(cfg, with_triton=True) stage_probes.append(Probe('infer-stage', _DecodedFrameObserver(handle, 'infer_output', metrics=result_path.ingress.metrics if result_path else None))) with diagnostic.span("retrieve"): # Offline validation follows media PTS, rather than decoding as fast as possible. inferred.attach(stage_probes[-1]).retrieve( R(), **({'sync': True} if src.startswith('file://') else {})) diagnostic.mark("flow", "returned") if handle.stop_requested: return on_message = _pipeline_message_callback(handle, finite=max_buffers > 0) with diagnostic.span("sdk_start"): waiter = handle.start_pipeline(on_message) if waiter is None: return handle.mark_initialized() with diagnostic.span("sdk_wait"): handle.diagnostic.mark("wait_enter") try: waiter.wait() except BaseException: handle.diagnostic.mark("wait_raised") raise else: handle.diagnostic.mark("wait_returned") if not handle.stop_requested and max_buffers == 0: handle.mark_failure(make_stream_failure( "ROLE_RUNTIME_FAILED", "pipeline_exit", name )) except RtspSourceAdapterError as error: failure = make_stream_failure( "PIPELINE_INIT_FAILED", error.failure_stage, name, error ) handle.mark_failure(failure) print( f"[{name}] pipeline error " f"code={failure.error_code} stage={failure.failure_stage}", flush=True, ) except Exception as error: failure = classify_pipeline_failure(error, name, handle.initialized) handle.mark_failure(failure) print( f"[{name}] pipeline error " f"code={failure.error_code} stage={failure.failure_stage}", flush=True, ) finally: handle.diagnostic.mark("worker_finally") diagnostic.close() if prepared_source is not None: prepared_source.cleanup() if handle.failure is None and not handle.stop_requested: handle.mark_stopped() collector.task_done(name) handle.diagnostic.mark("worker_cleanup_returned") t = threading.Thread(target=_worker, name=f"pipe-{name}", daemon=True) handle.bind_thread(t) t.start() return handle def main(): import argparse ap = argparse.ArgumentParser(description="Contextra multi-model pipeline (det+pose+seg, tensor-based)") ap.add_argument("--rtsp-url", help="RTSP 스트림 URL (무한 루프)") ap.add_argument("--file", help="로컬/파일 URI (file:///path) — 검증용, max-buffers 후 자동 종료") ap.add_argument("--max-buffers", type=int, default=8, help="파일 모드 버퍼 수 (RTSP 모드 무시)") ap.add_argument("--store", dest="store", action="store_true", default=True, help="병합 프레임을 PostgreSQL/Redis(metadata_storage)에 기록") ap.add_argument("--no-store", dest="store", action="store_false", help="DB 기록 비활성화 (파이프라인 실행만)") ap.add_argument("--persist-every", type=int, default=1, help="N프레임마다 1프레임 기록 (RTSP 모드 부하 절감)") ap.add_argument("--site-id", default="SITE01", help="site_id (메타데이터)") ap.add_argument("--stream-id", default="CAM_01", help="stream_id (메타데이터)") ap.add_argument("--scenarios", default="all", help="활성 시나리오 (쉼표 구분, 예: loitering,intrusion / all / off)") ap.add_argument("--active-learning", dest="al", action="store_true", default=True, help="Active Learning artifact 저장 (raw_frames/crop/annotations/metadata_json)") ap.add_argument("--no-active-learning", dest="al", action="store_false", help="Active Learning 저장 비활성화") ap.add_argument("--persist-mode", default="all", choices=["all", "event_gated"], help="all=모든 프레임 저장(기본) | event_gated=이벤트 ±윈도우만 저장") ap.add_argument("--profile-json", default=None, help="inference_profile JSON 문자열 (예: '{\"det_model\":\"RFDETRLarge\",\"thresholds\":{\"score\":0.4}}')") args = ap.parse_args() if not args.rtsp_url and not args.file: ap.error("--rtsp-url 또는 --file 중 하나는 필수") src = args.rtsp_url or args.file continuous = bool(args.rtsp_url) max_buffers = 0 if continuous else args.max_buffers print( f"[MP] source_mode={'rtsp' if continuous else 'file'} " f"continuous={continuous} max_buffers={max_buffers}", flush=True, ) # Per-stream inference profile (Phase 1) — 없으면 기본값(roles 전부 + 카탈로그 primary). from app.inference_profiles import resolve_from_spec, render_cfg, ProfileError, inference_component_id prof = None if args.profile_json: try: _praw = json.loads(args.profile_json) except Exception as e: ap.error(f"--profile-json invalid JSON: {e}") try: prof = resolve_from_spec(args.stream_id, {"inference_profile": _praw}) except ProfileError as pe: ap.error(str(pe)) else: try: prof = resolve_from_spec(args.stream_id, {}) except ProfileError as pe: print(f"[MP] profile resolution failed (non-fatal): {pe}", flush=True) required_roles = set(prof.roles) if prof is not None else {"det", "pose", "seg"} collector = ResultCollector(required_roles=list(prof.roles) if prof else None) # Phase 3 — per-stream IoU 트래커 (track_id 실체화). from app.tracker import IoUTracker tracker = IoUTracker() scenario_runner = None if args.scenarios and args.scenarios.lower() != "off": try: active = ( [s.strip() for s in args.scenarios.split(",") if s.strip()] if args.scenarios.lower() != "all" else None ) scenario_runner = ScenarioRunner(active=active, stream_id=args.stream_id, profile=prof) print(f"[MP] scenarios=ON active={list(scenario_runner.engines.keys())}", flush=True) except Exception as se: print(f"[MP] scenarios disabled (non-fatal): {se}", flush=True) scenario_runner = None al_writer = None if args.al: try: from app.active_learning import ActiveLearningWriter al_writer = ActiveLearningWriter(site_id=args.site_id, stream_id=args.stream_id, enabled=True) except Exception as ale: print(f"[MP] active_learning disabled (non-fatal): {ale}", flush=True) al_writer = None # 이벤트 게이트 영속화 (CLI 경로) — event_gated 시 프레임은 메모리만, # 이벤트 발생 시 ±윈도우(전 10s/후 5s)만 PG에 플러시. gate = None if args.persist_mode == "event_gated": try: from app.event_gate import EventGate from app.metadata_storage import storage as _storage gate = EventGate(_storage, stream_id=args.stream_id, site_id=args.site_id) gate.start() print("[MP] persist_mode=event_gated (window -10s/+5s)", flush=True) except Exception as ge: print(f"[MP] event_gated disabled (non-fatal): {ge}", flush=True) writer = StorageWriter(site_id=args.site_id, stream_id=args.stream_id, store=args.store, persist_every=args.persist_every, scenario_runner=scenario_runner, al_writer=al_writer, gate=gate, required_roles=list(prof.roles) if prof else None, tracker=tracker) print(f"[MP] storage={'ON' if writer.store else 'OFF'} persist_every={writer.persist_every} " f"active_learning={'ON' if al_writer and al_writer.enabled else 'OFF'} tracker=IoU", flush=True) # Per-stream cfg 렌더링 (Phase 1) — 프로파일 리졸브 실패 시만 레거시 하드코딩 cfg 폴백. if prof is not None: role_cfgs = {} for _role in prof.roles: try: role_cfgs[_role] = render_cfg(args.stream_id, _role, prof.resolved_models[_role]) except Exception as ce: print(f"[MP] cfg render failed for {_role} (non-fatal): {ce}", flush=True) if not role_cfgs: raise SystemExit("[MP] no pipeline configs rendered — aborting") threads = [ run_pipeline(_role, role_cfgs[_role], src, collector, max_buffers, al_writer if _role == "det" else None, score_threshold=prof.score_threshold, tracker=tracker if _role == "det" else None, component_id=inference_component_id(args.stream_id, _role)) for _role in ("det", "pose", "seg") if _role in role_cfgs ] else: threads = [ run_pipeline("det", DET_CFG, src, collector, max_buffers, al_writer, tracker=tracker, component_id=1), run_pipeline("pose", POSE_CFG, src, collector, max_buffers, al_writer), run_pipeline("seg", SEG_CFG, src, collector, max_buffers, al_writer), ] if continuous: # RTSP: 무한 루프. 주기적으로 병합 프레임 요약 출력. print("[MP] RTSP continuous mode — merged frames logged every 5s. Ctrl+C to stop.", flush=True) try: while True: time.sleep(5) try: writer.persist_new(collector) except Exception as we: print(f"[STORAGE] persist_new failed (non-fatal): {we}", flush=True) with collector.lock: n_frames = len(collector.frames) per_task = dict(collector.per_task) # 최신 병합 프레임 1개 샘플 with collector.lock: latest = None for k, f in sorted(collector.frames.items(), key=lambda x: x[1].get("_fn", 0)): if required_roles <= set(f.keys()): latest = f if latest is not None: # roles 제한 시 존재하는 역할만 요약 (KeyError 방지). d = latest.get("det") or {} po = latest.get("pose") or {} s = latest.get("seg") or {} person_cls = sum(1 for x in d.get("items", []) if x["class_id"] == 0) print(f"[MP] frames={n_frames} per_task={per_task} | latest fn={latest.get('_fn')} " f"det={d.get('count',0)}(person={person_cls}) pose={po.get('count',0)} seg={s.get('count',0)}", flush=True) except KeyboardInterrupt: print("[MP] stopping...", flush=True) for t in threads: t.join(timeout=5) writer._flush() if gate is not None: gate.close() # pending 이벤트 윈도우 즉시 플러시 후 정리 if al_writer is not None: al_writer.close() return # 파일 모드: 3개 파이프라인 전부 완료 대기 (180s hard cap) t0 = time.time() while time.time() - t0 < 180: with collector.lock: done_tasks = set(collector.completed_tasks) if required_roles <= done_tasks: break if all(not t.is_alive() for t in threads): break time.sleep(1) for t in threads: t.join(timeout=10) written = writer.persist_new(collector) print(f"[MP] storage written={written}", flush=True) print("\n[MP] ===== MERGED FRAMES =====", flush=True) with collector.lock: items = sorted((k, f) for k, f in collector.frames.items()) per_task = dict(collector.per_task) shown = 0 for key, f in items: if not (required_roles <= set(f.keys())): continue # roles 제한 시 존재하는 역할만 요약 (KeyError 방지). d = f.get("det") or {} po = f.get("pose") or {} s = f.get("seg") or {} person_cls = sum(1 for x in d.get("items", []) if x["class_id"] == 0) print(f"[MP] frame fn={f.get('_fn')}: det={d.get('count',0)} (person={person_cls}) " f"pose_persons={po.get('count',0)} seg={s.get('count',0)}", flush=True) shown += 1 if shown >= max_buffers: break print(f"[MP] done: merged={collector.merged} frames={len(items)} per_task={per_task}", flush=True) writer._flush() if gate is not None: gate.close() # pending 이벤트 윈도우 즉시 플러시 후 정리 (파일 모드 종료) if al_writer is not None: al_writer.close() if __name__ == "__main__": main()