"""FRAMERESULT-01: 실제 Frame 참조와 AI 표시 의미만 직렬화한다.""" from dataclasses import dataclass import re from app.contracts.frames import DetectionObservation, FrameEnvelope @dataclass(frozen=True) class VisualizationCommand: detection: DetectionObservation label_text: str color: str = "#00CC66" visible: bool = True track_id: str | int | None = None event_ids: tuple[str, ...] = () scenario: object | None = None pose: object | None = None def __post_init__(self): if not isinstance(self.label_text, str) or len(self.label_text) > 128: raise ValueError("Invalid display label.") if not re.fullmatch(r"#[0-9A-Fa-f]{6}", self.color) or type(self.visible) is not bool: raise ValueError("Invalid display command.") if self.track_id is not None and type(self.track_id) not in (str, int): raise ValueError("Invalid track reference.") ids = tuple(self.event_ids) if len(ids) > 16 or any(type(x) is not str or not x or len(x) > 100 for x in ids): raise ValueError("Invalid event reference.") object.__setattr__(self, "event_ids", ids) if self.pose is not None and (self.pose.frame != self.detection.frame or self.pose.model_key != self.detection.model_key or self.pose.bbox_xyxy != self.detection.bbox_xyxy or self.pose.score != self.detection.score): raise ValueError('Pose and detection differ.') @dataclass(frozen=True) class FrameResult: """FRAMERESULT-01: 최신 디버그 참조와 소비자가 공유하는 불변 기준 결과.""" frame: FrameEnvelope commands: tuple[VisualizationCommand, ...] events: tuple = () scenario_config: object | None = None scenario_revision: str | None = None debug_all_candidates: bool = False pose_config: object | None = None nms_kept: tuple[bool, ...] | None = None provider_model: str | None = None camera_id: str | int | None = None crowd_regions: tuple | None = None def __post_init__(self): commands = tuple(self.commands) if len(commands) > 300 or any(c.detection.frame != self.frame for c in commands): raise ValueError("Invalid Frame result.") object.__setattr__(self, "commands", commands) if self.provider_model is not None and (not self.provider_model or any(c.detection.model_key != self.provider_model for c in commands)): raise ValueError('Frame provider differs from observations.') if len(self.events) > 900 or any(e.frame != self.frame for e in self.events): raise ValueError('Invalid Event Frame.') object.__setattr__(self, 'events', tuple(self.events)) if self.crowd_regions is not None: from app.crowd_event import CrowdRegion regions = tuple(self.crowd_regions) if len(regions) > 1 or any(not isinstance(r, CrowdRegion) for r in regions): raise ValueError('Invalid Crowd regions.') object.__setattr__(self, 'crowd_regions', regions) if self.nms_kept is not None: mask=tuple(self.nms_kept) if len(mask)!=len(commands) or any(type(v) is not bool for v in mask): raise ValueError('Invalid NMS selection.') object.__setattr__(self,'nms_kept',mask) def snapshot(self, *, visible_only=True): camera_id = self.frame.source_id if self.camera_id is None else self.camera_id value = frame_snapshot(self.frame, camera_id, self.commands, visible_only=visible_only) if self.scenario_config is not None: from app.region_scenario import name, members value['scenarioConfig'] = {name(c): c.payload() for c in members(self.scenario_config)} value['scenarioRevision'] = self.scenario_revision value['events'] = [dict(e.payload(), cameraId=camera_id) for e in self.events] if self.pose_config is not None: value['poseDisplay'] = self.pose_config.payload() from app.pose_config import JOINT_NAMES value['keypointSchema'] = dict(name='COCO_17',values=['x','y','score'], names=list(JOINT_NAMES),coordinateSpace='SOURCE_FRAME',normalized=True, scoreMeaning='per_joint_findability',sideConvention='ANATOMICAL') if self.nms_kept is not None: value['nms'] = dict(input=len(self.commands),kept=sum(self.nms_kept), suppressed=len(self.commands)-sum(self.nms_kept),iouThreshold=self.pose_config.nms_iou_threshold, enabled=self.pose_config.nms_enabled) if not visible_only: for item in value['objects']: item['nmsSuppressed']=not self.nms_kept[int(item['objectRef'])] if self.crowd_regions is not None: value['crowd'] = dict(regions=[r.payload() for r in self.crowd_regions]) return value def frame_snapshot(frame: FrameEnvelope, camera_id, commands=(), *, visible_only=False): if type(camera_id) not in (str, int) or not str(camera_id): raise ValueError("Invalid camera reference.") if str(camera_id) != frame.source_id: raise ValueError("Camera and source differ.") time = frame.frame_time objects = [] for index, command in enumerate(commands): if index >= 300: raise ValueError("Frame display limit exceeded.") detection = command.detection if detection.frame != frame: raise ValueError("Observation belongs to another frame.") if visible_only and not command.visible: continue objects.append({ "objectRef": str(index), "classId": detection.class_id, "modelKey": detection.model_key, "confidence": detection.score, "bbox": list(detection.bbox_xyxy), "trackId": command.track_id, "eventIds": list(command.event_ids), "labelText": command.label_text, "color": command.color, "visible": command.visible, }) if command.scenario is not None: from app.region_scenario import members objects[-1]['scenarioResults'] = [d.payload() for d in members(command.scenario)] if command.pose is not None: objects[-1]['keypoints'] = [list(k) for k in command.pose.keypoints] objects[-1]['keypointSchema'] = 'COCO_17' objects[-1]['keypointInFrame'] = [0<=k[0]<=1 and 0<=k[1]<=1 for k in command.pose.keypoints] return { "schemaVersion": "1", "cameraId": camera_id, "sourceId": frame.source_id, "serverId": frame.server_id, "streamId": frame.stream_id, "runId": frame.run_id, "runGeneration": frame.run_generation, "frameSeq": frame.frame_seq, "ptsNs": frame.pts_ns, "ingestTimeWallMs": frame.ingest_time_wall_ms, "frameWidth": frame.frame_width, "frameHeight": frame.frame_height, "coordinateSpace": "SOURCE_FRAME", "format": "XYXY", "normalized": True, "origin": "TOP_LEFT", "configVersion": frame.config_version, "frameTime": { "anchorId": time.anchor_id if time else None, "startWallTimeMs": time.anchor_wall_time_ms if time else None, "anchorRtpTimestamp": time.anchor_rtp_timestamp if time else None, "rtpClockRate": time.rtp_clock_rate if time else None, "rtpTimestamp": time.source_rtp_timestamp if time else None, "rtpTimestampExt": time.source_rtp_timestamp_ext if time else None, "detectedAtWallMs": time.source_time_wall_ms if time else None, "syncStatus": time.sync_status if time else "UNAVAILABLE", "timeSource": time.time_source if time else None, }, "objects": objects, }