124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
"""Загрузка и валидация конфигурации 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")
|
|
# Толерантно к отсутствию/повреждению: если файла нет, на его месте каталог
|
|
# (битый bind-mount) или YAML невалиден — стартуем без камер (добавляются из UI),
|
|
# а не падаем с ошибкой на старте.
|
|
raw: dict = {}
|
|
if os.path.isfile(path):
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
raw = yaml.safe_load(fh) or {}
|
|
except (OSError, yaml.YAMLError):
|
|
raw = {}
|
|
|
|
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()
|