"""Helpers shared by the operation modules.""" from __future__ import annotations import threading from dataclasses import dataclass from typing import Any import numpy as np from .. import media_io, paths from ..config import Settings from ..errors import BadRequest from ..media_io import VideoInfo from ..models import Registry @dataclass class Ctx: settings: Settings registry: Registry def open_input( ctx: Ctx, path: str, cancel: threading.Event, need_video: bool = True ) -> tuple[str, VideoInfo]: """Resolve, size-check, probe, and duration-check an input file.""" real = paths.input_file(ctx.settings.media_root, path, ctx.settings.max_input_bytes) info = media_io.probe(real, cancel) media_io.check_duration(info, ctx.settings.max_seconds) if need_video and not info.has_video: raise BadRequest("input has no video stream") return real, info def clamp01(value: float) -> float: return float(min(max(value, 0.0), 1.0)) def norm_box(x: float, y: float, w: float, h: float, width: int, height: int) -> dict[str, float]: """Pixel box -> normalized box clipped to the frame.""" x0, y0 = clamp01(x / width), clamp01(y / height) x1, y1 = clamp01((x + w) / width), clamp01((y + h) / height) return {"x": x0, "y": y0, "w": max(x1 - x0, 0.0), "h": max(y1 - y0, 0.0)} def norm_point(x: float, y: float, width: int, height: int) -> dict[str, float]: return {"x": clamp01(x / width), "y": clamp01(y / height)} def unique_sorted(times: list[int]) -> list[int]: return sorted(set(int(t) for t in times)) def jsonable(value: Any) -> Any: if isinstance(value, dict): return {k: jsonable(v) for k, v in value.items()} if isinstance(value, (list, tuple)): return [jsonable(v) for v in value] if isinstance(value, np.generic): return value.item() return value