"""실제 처리 Frame과 동일 송신 결과의 opt-in 이미지 증거. 재추론/최신 이미지 대입 금지. CONTEXTRA_METADATA_VISUAL_DIR: 사용자가 준비한 mode700 task 디렉터리. capture.enabled 파일이 있을 때 실행별 최대3장/60초만 수집한다. 제거하면 새 수집이 중단된다. 이미지 추출은 기존 SDK callback에서, 인코딩/파일 I/O는 별도 bounded worker에서 수행한다. """ from collections import Counter, OrderedDict import json import os from pathlib import Path import stat import threading import time import uuid class MetadataVisual: MAX_IMAGE_BYTES = 8 * 1024 * 1024 MAX_OUTPUT_BYTES = 16 * 1024 * 1024 def __init__(self, root, *, clock=time.monotonic, display_threshold=None, require_prepared=False): self.root = Path(root) if (not self.root.is_absolute() or self.root.is_symlink() or not self.root.is_dir() or stat.S_IMODE(self.root.stat().st_mode) != 0o700): raise ValueError('Private visual directory is required.') self.clock, self.started = clock, clock() self.display_threshold = display_threshold self._cv = threading.Condition() self._pending = OrderedDict() self._closed = False self._metrics = Counter() self._attempts = 0 self._captures = 0 self._bytes = 0 self._output = None self._capture_enabled = (self.root / 'capture.enabled').is_file() self._last_capture = None self.pts_interval_s = None self._last_capture_pts = None self._capture_timing = [] self._save_timing = [] from app.capture_pixels import preparation_status self._preparation = preparation_status() self._ready = not require_prepared or self._preparation.get('ready', False) self._thread = threading.Thread(target=self._run, name='metadata-visual', daemon=True) self._thread.start() @classmethod def from_environment(cls, *, display_threshold=None): root = os.environ.get('CONTEXTRA_METADATA_VISUAL_DIR') return cls(root, display_threshold=display_threshold, require_prepared=True) if root else None def enabled(self): return (self._ready and not self._closed and self._attempts < 3 and self.clock() - self.started <= 60 and self._capture_enabled) def wants_image(self, frame): return (self.enabled() and self._captures < 3 and frame.frame_width * frame.frame_height * 3 <= self.MAX_IMAGE_BYTES) def reserve_image(self, frame, surface_size=None): if not self._cv.acquire(blocking=False): self._metrics['busy_drop'] += 1 return False try: if not self.wants_image(frame): return False if self._last_capture is not None and self.clock()-self._last_capture < .5: return False if self.pts_interval_s is not None: if frame.pts_ns is None or frame.pts_ns < 0: return False if self._last_capture_pts is not None and frame.pts_ns-self._last_capture_pts < self.pts_interval_s*1e9: return False self._captures += 1 self._last_capture = self.clock() self._last_capture_pts = frame.pts_ns width, height = surface_size or (frame.frame_width, frame.frame_height) if width*height*3 > self.MAX_IMAGE_BYTES or width*frame.frame_height != height*frame.frame_width: self._metrics['geometry_mismatch'] += 1 return False return True finally: self._cv.release() def _submit(self, frame, kind, value): if not self._cv.acquire(blocking=False): self._metrics['busy_drop'] += 1 return False try: if not self.enabled(): return False if frame not in self._pending and len(self._pending) >= 32: self._pending.popitem(last=False) self._metrics['unmatched_drop'] += 1 entry = self._pending.setdefault(frame, [self.clock(), {}]) entry[1][kind] = value self._cv.notify() return True finally: self._cv.release() def submit_result(self, result): return self._submit(result.frame, 'result', result) def submit_image(self, frame, hwc, surface_size=None): if not self.enabled(): return False width, height = surface_size or (frame.frame_width, frame.frame_height) if (tuple(hwc.shape) != (height, width, 3) or str(hwc.dtype) != 'uint8' or width*height*3 > self.MAX_IMAGE_BYTES or width*frame.frame_height != height*frame.frame_width): self.note_failure('geometry_mismatch') return False # SDK/기존 이미지 cache 소유권과 분리한다. 최대8MiB, GPU surface는 보관하지 않는다. return self._submit(frame, 'image', (hwc.tobytes(), width, height)) def note_failure(self, code): if self._cv.acquire(blocking=False): try: self._metrics[code if code in {'geometry_mismatch', 'extract_failed', 'unsupported_batch'} else 'capture_failed'] += 1 finally: self._cv.release() else: self._metrics['busy_drop'] += 1 def note_capture(self, seq, started, extracted, copied): if self._cv.acquire(blocking=False): try: if len(self._capture_timing) < 3: self._capture_timing.append(dict(frame_seq=seq, sampled_ns=started, extract_ms=(extracted-started)/1e6, copy_ms=(copied-extracted)/1e6)) finally: self._cv.release() def _run(self): while True: try: self._capture_enabled = (self.root / 'capture.enabled').is_file() except OSError: self._capture_enabled = False with self._cv: self._cv.wait_for(lambda: self._closed or any(len(x[1]) == 2 for x in self._pending.values()), .1) if self._closed: self._pending.clear() return for frame, (at, entry) in list(self._pending.items()): if self.clock() - at > 2 or not self.enabled(): del self._pending[frame] self._metrics['expired'] += 1 ready = next(((f, e[1]) for f, e in self._pending.items() if len(e[1]) == 2), None) if ready is None: continue frame, entry = ready del self._pending[frame] self._attempts += 1 try: self._save(frame, entry['image'], entry['result']) with self._cv: self._metrics['saved'] += 1 except Exception: with self._cv: self._metrics['write_failed'] += 1 def _save(self, frame, pixels, result): started = time.monotonic_ns() import io import numpy as np from app.metadata_render import render_rgb if result.debug_all_candidates: from dataclasses import replace result = replace(result, commands=tuple(replace(c, visible=True) for c in result.commands)) snapshot = result.snapshot() if result.debug_all_candidates: snapshot['debugScope'] = 'ALL_CANDIDATES_NOT_NORMAL_TRANSPORT' pixels, width, height = pixels pixels = np.frombuffer(pixels, dtype=np.uint8).reshape(height, width, 3) if (width, height) != (frame.frame_width, frame.frame_height): from PIL import Image # Known same-aspect mux surface; no crop/padding transform is inferred. pixels = np.asarray(Image.fromarray(pixels).resize( (frame.frame_width, frame.frame_height), Image.Resampling.BILINEAR)) image, stats = render_rgb(result, pixels, display_threshold=self.display_threshold) rendered = time.monotonic_ns() encoded = io.BytesIO() image.save(encoded, format='JPEG', quality=90) encoded_at = time.monotonic_ns() payload = json.dumps(snapshot, ensure_ascii=False, allow_nan=False).encode() stats.pop('label_rects') stats['capture_surface'] = [width, height] summary = json.dumps(dict(frameSeq=frame.frame_seq, **stats), allow_nan=False).encode() jpg = encoded.getvalue() if self._bytes + len(jpg) + len(payload) + len(summary) > self.MAX_OUTPUT_BYTES: raise ValueError('Visual output budget exceeded.') if self._output is None: self._output = self.root / ('capture-' + uuid.uuid4().hex) self._output.mkdir(mode=0o700) for suffix, data in (('json', payload), ('jpg', jpg), ('stats', summary)): target = self._output / f'frame-{frame.frame_seq}.{suffix}' fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, 'wb') as output: output.write(data) self._bytes += len(data) with self._cv: if len(self._save_timing) < 3: self._save_timing.append(dict(frame_seq=frame.frame_seq, render_ms=(rendered-started)/1e6, encode_ms=(encoded_at-rendered)/1e6, json_write_ms=(time.monotonic_ns()-encoded_at)/1e6)) def status(self): with self._cv: return dict(self._metrics, queued=len(self._pending), attempts=self._attempts, captures=self._captures, preparation=self._preparation, capture_timing=list(self._capture_timing), save_timing=list(self._save_timing), output_bytes=self._bytes, worker_alive=self._thread.is_alive()) def close(self, timeout=1): with self._cv: self._closed = True self._pending.clear() self._cv.notify_all() self._thread.join(max(0, timeout)) return not self._thread.is_alive()