"""Загрузка и валидация конфигурации NVR из cameras.yaml.""" from __future__ import annotations import os from dataclasses import dataclass, field from functools import lru_cache import yaml @dataclass(frozen=True) class Camera: id: str name: str enabled: bool ip: str user: str password: str main_path: str sub_path: str | None = None def rtsp_url(self, path: str) -> str: return f"rtsp://{self.user}:{self.password}@{self.ip}:554{path}" @property def record_url(self) -> str: """Поток для записи (основной, H.265, копированием).""" return self.rtsp_url(self.main_path) @dataclass(frozen=True) class Storage: path: str = "/records" quota_gb: int = 50 retention_days: int = 7 segment_seconds: int = 600 @dataclass(frozen=True) class Live: go2rtc_url: str = "http://localhost:1984" @dataclass class Config: storage: Storage live: Live cameras: list[Camera] = field(default_factory=list) def camera(self, cam_id: str) -> Camera | None: return next((c for c in self.cameras if c.id == cam_id), None) @property def enabled_cameras(self) -> list[Camera]: return [c for c in self.cameras if c.enabled] def load_config(path: str | None = None) -> Config: path = path or os.environ.get("NVR_CONFIG", "cameras.yaml") with open(path, "r", encoding="utf-8") as fh: raw = yaml.safe_load(fh) or {} storage = Storage(**(raw.get("storage") or {})) live = Live(**(raw.get("live") or {})) cameras = [ Camera( id=c["id"], name=c.get("name", c["id"]), enabled=c.get("enabled", True), ip=c["ip"], user=c["user"], password=c["password"], main_path=c["main_path"], sub_path=c.get("sub_path"), ) for c in (raw.get("cameras") or []) ] return Config(storage=storage, live=live, cameras=cameras) def camera_to_yaml(c: Camera) -> dict: return { "id": c.id, "name": c.name, "enabled": c.enabled, "ip": c.ip, "user": c.user, "password": c.password, "main_path": c.main_path, "sub_path": c.sub_path, } def save_config(config: Config, path: str | None = None) -> None: """Сохраняет конфиг обратно в cameras.yaml (комментарии теряются).""" path = path or os.environ.get("NVR_CONFIG", "cameras.yaml") data = { "storage": { "path": config.storage.path, "quota_gb": config.storage.quota_gb, "retention_days": config.storage.retention_days, "segment_seconds": config.storage.segment_seconds, }, "live": {"go2rtc_url": config.live.go2rtc_url}, "cameras": [camera_to_yaml(c) for c in config.cameras], } # пишем на месте: cameras.yaml примонтирован как отдельный файл (bind mount), # переименование поверх точки монтирования даёт EBUSY. with open(path, "w", encoding="utf-8") as fh: yaml.safe_dump(data, fh, allow_unicode=True, sort_keys=False) @lru_cache(maxsize=1) def get_config() -> Config: return load_config()