|
1 | | -"""Pydantic config models: AppConfig, CameraConfig, PTZ presets, layouts. |
| 1 | +"""Pydantic config models for AutoPTZ v2. |
2 | 2 |
|
3 | | -Phase 1 implementation target. |
| 3 | +All config objects are immutable (frozen) pydantic models so they can be |
| 4 | +passed safely between threads and processes without defensive copying. |
| 5 | +
|
| 6 | +Cameras are addressed by stable UUID (`CameraConfig.id`) everywhere — never |
| 7 | +by list position or a "current active" global. |
4 | 8 | """ |
5 | 9 | from __future__ import annotations |
6 | 10 |
|
7 | | -# TODO(phase-1): AppConfig, CameraConfig, PTZPreset, Layout pydantic models |
| 11 | +import uuid |
| 12 | +from datetime import UTC, datetime |
| 13 | +from typing import Literal |
| 14 | + |
| 15 | +from pydantic import BaseModel, Field, field_validator |
| 16 | + |
| 17 | +# ── Helpers ─────────────────────────────────────────────────────────────────── |
| 18 | + |
| 19 | +def _new_id() -> str: |
| 20 | + return str(uuid.uuid4()) |
| 21 | + |
| 22 | + |
| 23 | +def _now() -> datetime: |
| 24 | + return datetime.now(UTC).replace(tzinfo=None) |
| 25 | + |
| 26 | + |
| 27 | +# ── Hardware / EP prefs ─────────────────────────────────────────────────────── |
| 28 | + |
| 29 | +class HardwarePrefs(BaseModel, frozen=True): |
| 30 | + force_ep: str | None = None # "CoreMLExecutionProvider", etc.; None = auto |
| 31 | + model_tier: Literal["nano", "small", "medium", "large"] = "nano" |
| 32 | + max_workers: int = Field(default=4, ge=1, le=32) |
| 33 | + |
| 34 | + |
| 35 | +# ── Theme ───────────────────────────────────────────────────────────────────── |
| 36 | + |
| 37 | +class ThemeConfig(BaseModel, frozen=True): |
| 38 | + name: Literal["dark", "light", "system"] = "dark" |
| 39 | + accent: str = "#3d9bff" # hex colour token |
| 40 | + |
| 41 | + |
| 42 | +# ── Source ──────────────────────────────────────────────────────────────────── |
| 43 | + |
| 44 | +class SourceConfig(BaseModel, frozen=True): |
| 45 | + type: Literal["usb", "rtsp", "onvif", "ndi"] = "usb" |
| 46 | + address: str = "" # index (USB), URL (RTSP/ONVIF), name (NDI) |
| 47 | + username: str = "" |
| 48 | + password: str = "" |
| 49 | + substream: bool = False |
| 50 | + fps: float = Field(default=30.0, gt=0.0, le=240.0) |
| 51 | + |
| 52 | + |
| 53 | +# ── Reconnect ───────────────────────────────────────────────────────────────── |
| 54 | + |
| 55 | +class ReconnectConfig(BaseModel, frozen=True): |
| 56 | + backoff_initial_s: float = Field(default=1.0, gt=0.0) |
| 57 | + backoff_max_s: float = Field(default=30.0, gt=0.0) |
| 58 | + stall_timeout_s: float = Field(default=5.0, gt=0.0) |
| 59 | + |
| 60 | + |
| 61 | +# ── Tracking ────────────────────────────────────────────────────────────────── |
| 62 | + |
| 63 | +class TrackingConfig(BaseModel, frozen=True): |
| 64 | + tracker: Literal["botsort", "deepocsort", "bytetrack"] = "botsort" |
| 65 | + detect_interval: int = Field(default=1, ge=1, le=30) |
| 66 | + reid_enabled: bool = False |
| 67 | + reid_threshold_hi: float = Field(default=0.70, ge=0.0, le=1.0) |
| 68 | + reid_threshold_lo: float = Field(default=0.45, ge=0.0, le=1.0) |
| 69 | + coast_window_ms: int = Field(default=1500, ge=0) |
| 70 | + face_confirm: bool = False |
| 71 | + quality_floor: Literal["auto", "high", "balanced", "low"] = "auto" |
| 72 | + |
| 73 | + |
| 74 | +# ── PTZ ─────────────────────────────────────────────────────────────────────── |
| 75 | + |
| 76 | +class PanTiltZoomLimits(BaseModel, frozen=True): |
| 77 | + pan_min: float = -1.0 |
| 78 | + pan_max: float = 1.0 |
| 79 | + tilt_min: float = -1.0 |
| 80 | + tilt_max: float = 1.0 |
| 81 | + zoom_min: float = 0.0 |
| 82 | + zoom_max: float = 1.0 |
| 83 | + |
| 84 | + |
| 85 | +class PTZConfig(BaseModel, frozen=True): |
| 86 | + backend: Literal["auto", "ndi", "visca_ip", "visca_usb", "onvif"] = "auto" |
| 87 | + address: str | None = None |
| 88 | + max_pan_speed: float = Field(default=0.5, ge=0.0, le=1.0) |
| 89 | + max_tilt_speed: float = Field(default=0.5, ge=0.0, le=1.0) |
| 90 | + max_zoom_speed: float = Field(default=0.3, ge=0.0, le=1.0) |
| 91 | + invert_pan: bool = False |
| 92 | + invert_tilt: bool = False |
| 93 | + deadzone_x: float = Field(default=0.05, ge=0.0, le=0.5) |
| 94 | + deadzone_y: float = Field(default=0.05, ge=0.0, le=0.5) |
| 95 | + kp: float = Field(default=0.6, ge=0.0) |
| 96 | + kd: float = Field(default=0.05, ge=0.0) |
| 97 | + kv: float = Field(default=0.1, ge=0.0) |
| 98 | + auto_zoom: bool = True |
| 99 | + zoom_framing: Literal["tight", "medium", "wide"] = "medium" |
| 100 | + soft_limits: PanTiltZoomLimits | None = None |
| 101 | + |
| 102 | + |
| 103 | +class PTZPreset(BaseModel, frozen=True): |
| 104 | + id: str = Field(default_factory=_new_id) |
| 105 | + camera_id: str = "" |
| 106 | + idx: int |
| 107 | + name: str |
| 108 | + pan: float = 0.0 |
| 109 | + tilt: float = 0.0 |
| 110 | + zoom: float = 0.0 |
| 111 | + native_preset: int | None = None |
| 112 | + |
| 113 | + @field_validator("name") |
| 114 | + @classmethod |
| 115 | + def _name_nonempty(cls, v: str) -> str: |
| 116 | + if not v.strip(): |
| 117 | + raise ValueError("PTZPreset.name must not be blank") |
| 118 | + return v |
| 119 | + |
| 120 | + |
| 121 | +# ── Target / framing intent ─────────────────────────────────────────────────── |
| 122 | + |
| 123 | +class TargetConfig(BaseModel, frozen=True): |
| 124 | + mode: Literal["identity", "manual", "off"] = "off" |
| 125 | + identity_id: str | None = None |
| 126 | + default_on_start: int | None = None # preset idx to recall on startup |
| 127 | + |
| 128 | + |
| 129 | +# ── Camera ──────────────────────────────────────────────────────────────────── |
| 130 | + |
| 131 | +class CameraConfig(BaseModel, frozen=True): |
| 132 | + id: str = Field(default_factory=_new_id) |
| 133 | + name: str |
| 134 | + source: SourceConfig = Field(default_factory=SourceConfig) |
| 135 | + enabled: bool = True |
| 136 | + tracking: TrackingConfig = Field(default_factory=TrackingConfig) |
| 137 | + ptz: PTZConfig = Field(default_factory=PTZConfig) |
| 138 | + presets: list[PTZPreset] = Field(default_factory=list) |
| 139 | + target: TargetConfig = Field(default_factory=TargetConfig) |
| 140 | + reconnect: ReconnectConfig = Field(default_factory=ReconnectConfig) |
| 141 | + |
| 142 | + @field_validator("name") |
| 143 | + @classmethod |
| 144 | + def _name_nonempty(cls, v: str) -> str: |
| 145 | + if not v.strip(): |
| 146 | + raise ValueError("CameraConfig.name must not be blank") |
| 147 | + return v |
| 148 | + |
| 149 | + |
| 150 | +# ── Layout ──────────────────────────────────────────────────────────────────── |
| 151 | + |
| 152 | +class TilePlacement(BaseModel, frozen=True): |
| 153 | + camera_id: str |
| 154 | + x: int = 0 |
| 155 | + y: int = 0 |
| 156 | + w: int = 1 |
| 157 | + h: int = 1 |
| 158 | + z: int = 0 |
| 159 | + visible: bool = True |
| 160 | + |
| 161 | + |
| 162 | +class Layout(BaseModel, frozen=True): |
| 163 | + id: str = Field(default_factory=_new_id) |
| 164 | + name: str |
| 165 | + tiles: list[TilePlacement] = Field(default_factory=list) |
| 166 | + |
| 167 | + @field_validator("name") |
| 168 | + @classmethod |
| 169 | + def _name_nonempty(cls, v: str) -> str: |
| 170 | + if not v.strip(): |
| 171 | + raise ValueError("Layout.name must not be blank") |
| 172 | + return v |
| 173 | + |
| 174 | + |
| 175 | +# ── Identity ────────────────────────────────────────────────────────────────── |
| 176 | + |
| 177 | +class IdentityRecord(BaseModel, frozen=True): |
| 178 | + id: str = Field(default_factory=_new_id) |
| 179 | + name: str |
| 180 | + embeddings: list[bytes] = Field(default_factory=list) |
| 181 | + thumbnail: bytes | None = None |
| 182 | + created_at: datetime = Field(default_factory=_now) |
| 183 | + updated_at: datetime = Field(default_factory=_now) |
| 184 | + |
| 185 | + @field_validator("name") |
| 186 | + @classmethod |
| 187 | + def _name_nonempty(cls, v: str) -> str: |
| 188 | + if not v.strip(): |
| 189 | + raise ValueError("IdentityRecord.name must not be blank") |
| 190 | + return v |
| 191 | + |
| 192 | + |
| 193 | +# ── Top-level app config ────────────────────────────────────────────────────── |
| 194 | + |
| 195 | +CURRENT_SCHEMA_VERSION = 1 |
| 196 | + |
| 197 | + |
| 198 | +class AppConfig(BaseModel, frozen=True): |
| 199 | + schema_version: int = CURRENT_SCHEMA_VERSION |
| 200 | + theme: ThemeConfig = Field(default_factory=ThemeConfig) |
| 201 | + active_layout_id: str = "" |
| 202 | + hardware: HardwarePrefs = Field(default_factory=HardwarePrefs) |
| 203 | + cameras: list[CameraConfig] = Field(default_factory=list) |
| 204 | + |
| 205 | + def with_camera(self, cam: CameraConfig) -> AppConfig: |
| 206 | + """Return a new AppConfig with *cam* upserted (replace by id or append).""" |
| 207 | + cameras = [c for c in self.cameras if c.id != cam.id] + [cam] |
| 208 | + return self.model_copy(update={"cameras": cameras}) |
| 209 | + |
| 210 | + def without_camera(self, camera_id: str) -> AppConfig: |
| 211 | + """Return a new AppConfig with the given camera removed.""" |
| 212 | + return self.model_copy( |
| 213 | + update={"cameras": [c for c in self.cameras if c.id != camera_id]} |
| 214 | + ) |
0 commit comments