"""RTSP SourceConfig generation and nvurisrcbin property ownership. The public API keeps a complete RTSP URI. This module owns only the typed pipeline behavior that must be translated to NVIDIA properties. """ from __future__ import annotations from contextlib import nullcontext import os import re import secrets import threading from dataclasses import dataclass, field from pathlib import Path from typing import Literal from urllib.parse import urlsplit import yaml from pydantic import BaseModel, ConfigDict, Field SOURCE_CONFIG_ROOT = Path("/run/contextra/sources") _SAFE_ROLE = re.compile(r"^[A-Za-z0-9_-]{1,32}$") class RtspSourceOptions(BaseModel): """Vendor-neutral RTSP behavior exposed by the Contextra API.""" model_config = ConfigDict(extra="forbid", frozen=True) transport: Literal["tcp", "auto"] = Field( default="tcp", description="RTSP 전송 방식. 첫 Smoke의 재현 가능한 기본값은 tcp다.", ) latency_ms: int = Field( default=1000, ge=0, le=60_000, description="RTSP jitter buffer 지연 시간(ms).", ) disable_audio: bool = Field( default=True, description="영상 분석 Pipeline이 사용하지 않는 audio track을 비활성화한다.", ) reconnect_policy: Literal["none"] = Field( default="none", description="자동 재연결 정책. 첫 Vertical slice에서는 retry하지 않는다.", ) first_frame_timeout_s: int = Field( default=60, ge=5, le=300, description="모든 required role의 첫 Frame을 기다리는 최대 시간(초).", ) def nvidia_properties(self, gpu_id: int) -> dict[str, object]: """Translate the stable API contract to nvurisrcbin properties.""" protocol = 4 if self.transport == "tcp" else 7 return { "type": 2, "file-loop": False, "select-rtp-protocol": protocol, "latency": self.latency_ms, "disable-audio": self.disable_audio, "gpu-id": int(gpu_id), "init-rtsp-reconnect-interval": 0, "rtsp-reconnect-interval": 0, "rtsp-reconnect-attempts": 0, } class RtspSourceAdapterError(RuntimeError): """Bounded source preparation error that never contains the input URI.""" def __init__(self, failure_stage: str): self.failure_stage = str(failure_stage) super().__init__("RTSP source preparation failed.") @dataclass class PreparedRtspSource: """Ephemeral SourceConfig plus safe hooks needed by one Pipeline role.""" config_path: Path source_node: str stage_identity: str nvidia_properties: dict[str, object] = field(repr=False) _lock: threading.Lock = field(default_factory=threading.Lock, repr=False) _cleaned: bool = field(default=False, repr=False) def build_capture(self, flow, pipeline, decoded_probe=None, rtp_time=False, diagnostic=None, frame_size=None): """Build the official SourceConfig path and apply properties Flow omits. PyServiceMaker 9.1 reads ``source-bin`` and ``file-loop`` from the YAML, but does not copy every ``source-config.properties`` item to each static source node. RTSPINPUT-02~04 therefore keep the mapping inside this adapter and apply it before Pipeline start. """ span = diagnostic.span if diagnostic is not None else lambda _: nullcontext() dimensions = {} if frame_size is None else dict(width=frame_size[0], height=frame_size[1]) with span('batch_capture'): capture = flow.batch_capture( str(self.config_path), gpu_id=int(self.nvidia_properties["gpu-id"]), live_source=True, file_loop=False, **dimensions, ) with span('source_property'): pipeline[self.source_node].set(dict(self.nvidia_properties)) if rtp_time: with span('rtp_attach'): capture = capture.attach('contextra_rtp_time', 'rtp-time', properties={'source-node': self.source_node}) if decoded_probe is not None: with span('ingress_probe'): capture = capture.attach(decoded_probe) return capture def cleanup(self) -> None: """Idempotently remove the credential-bearing ephemeral config.""" with self._lock: if self._cleaned: return try: self.config_path.unlink(missing_ok=True) finally: self._cleaned = True def __enter__(self): return self def __exit__(self, _exc_type, _exc, _traceback): self.cleanup() # RTSPINPUT-02~04: 외부 URI와 NVIDIA property를 분리하고, 실제 Probe로 확인한 단계만 기록한다. class RtspSourceAdapter: """Prepare a private official ServiceMaker SourceConfig for one RTSP role.""" def __init__( self, uri: str, options: RtspSourceOptions | dict | None = None, *, gpu_id: int = 0, config_root: str | os.PathLike[str] = SOURCE_CONFIG_ROOT, ): self._validate_uri(uri) self._uri = uri self.options = ( options if isinstance(options, RtspSourceOptions) else RtspSourceOptions.model_validate(options or {}) ) self.gpu_id = int(gpu_id) if self.gpu_id < 0: raise ValueError("gpu_id must be non-negative") self.config_root = Path(config_root) def __repr__(self) -> str: return ( f"RtspSourceAdapter(options={self.options!r}, gpu_id={self.gpu_id}, " f"config_root={str(self.config_root)!r})" ) def prepare(self, role: str, run_generation: int) -> PreparedRtspSource: """Atomically create a private YAML file for one role generation.""" role = str(role) if not _SAFE_ROLE.fullmatch(role): raise RtspSourceAdapterError("source_identity") generation = int(run_generation) if generation < 0: raise RtspSourceAdapterError("source_identity") root = self._prepare_root() token = secrets.token_hex(8) source_node = f"rtsp-{role}-g{generation}-{token}" config_path = root / f"{source_node}.yaml" properties = self.options.nvidia_properties(self.gpu_id) document = { "source-list": [ { "uri": self._uri, "sensor-id": source_node, "sensor-name": f"contextra-{role}-g{generation}", } ], "source-config": { "source-bin": "nvurisrcbin", "properties": properties, }, } flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL if hasattr(os, "O_NOFOLLOW"): flags |= os.O_NOFOLLOW fd = None try: fd = os.open(config_path, flags, 0o600) with os.fdopen(fd, "w", encoding="utf-8") as handle: fd = None yaml.safe_dump(document, handle, sort_keys=False, allow_unicode=False) handle.flush() os.fsync(handle.fileno()) os.chmod(config_path, 0o600) except Exception as error: if fd is not None: os.close(fd) config_path.unlink(missing_ok=True) raise RtspSourceAdapterError("source_config_create") from error return PreparedRtspSource( config_path=config_path, source_node=source_node, stage_identity=f"rtsp:{role}:g{generation}", nvidia_properties=properties, ) def _prepare_root(self) -> Path: if not self.config_root.is_absolute(): raise RtspSourceAdapterError("source_config_root") try: self.config_root.mkdir(mode=0o700, parents=True, exist_ok=True) if self.config_root.is_symlink() or not self.config_root.is_dir(): raise OSError("unsafe source config root") os.chmod(self.config_root, 0o700) return self.config_root.resolve(strict=True) except Exception as error: raise RtspSourceAdapterError("source_config_root") from error @staticmethod def _validate_uri(uri: str) -> None: try: parsed = urlsplit(str(uri)) except Exception as error: raise RtspSourceAdapterError("source_uri_validation") from error if parsed.scheme.lower() != "rtsp" or not parsed.hostname: raise RtspSourceAdapterError("source_uri_validation")