operating-room-monitor-server/app/config.py

from __future__ import annotations

from pathlib import Path
from urllib.parse import quote_plus
from typing import Any, Literal

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

from app.baked import algorithm as baked_algorithm
from app.or_site_config import OrSiteConfig


class _SettingsGroup:
    """按主题分组的 Settings 视图；属性访问代理回主 Settings 实例。"""

    _FIELDS: tuple[str, ...] = ()

    def __init__(self, root: "Settings") -> None:
        object.__setattr__(self, "_root", root)

    def __getattr__(self, name: str) -> Any:
        if name not in self._FIELDS:
            raise AttributeError(
                f"{type(self).__name__} has no field '{name}'; "
                f"available: {self._FIELDS}"
            )
        return getattr(self._root, name)

    def __setattr__(self, name: str, value: Any) -> None:
        if name in self._FIELDS:
            setattr(self._root, name, value)
        else:
            object.__setattr__(self, name, value)


class _VideoGroup(_SettingsGroup):
    """仅含 RTSP 解析与按路后端；抽帧/推理/耗材等见 app.baked.pipeline / algorithm。"""

    _FIELDS = (
        "video_default_backend",
        "video_camera_backend_overrides_json",
        "video_rtsp_url_template",
        "or_site_config_json_file",
    )


class _VoiceGroup(_SettingsGroup):
    _FIELDS = ()


class _HikvisionGroup(_SettingsGroup):
    _FIELDS = (
        "hikvision_lib_dir",
        "hikvision_sdk_enabled",
        "hikvision_device_ip",
        "hikvision_device_port",
        "hikvision_user",
        "hikvision_password",
        "hikvision_channel",
        "hikvision_preview_rtsp_template",
        "hikvision_camera_rtsp_urls_json",
        "hikvision_sdk_fallback_to_rtsp",
    )


class _MinioGroup(_SettingsGroup):
    _FIELDS = (
        "minio_endpoint",
        "minio_access_key",
        "minio_secret_key",
        "minio_bucket",
        "minio_secure",
        "minio_region",
    )


class _BaiduGroup(_SettingsGroup):
    _FIELDS = (
        "baidu_speech_app_id",
        "baidu_speech_api_key",
        "baidu_speech_secret_key",
        "baidu_speech_connection_timeout_ms",
        "baidu_speech_socket_timeout_ms",
        "baidu_speech_asr_dev_pid",
    )


class _DemoGroup(_SettingsGroup):
    _FIELDS = (
        "demo_cors_enabled",
        "demo_cors_origins",
        "demo_orchestrator_enabled",
        "demo_orchestrator_rtsp_port",
        "demo_orchestrator_rtsp_json_host",
    )


class _DatabaseGroup(_SettingsGroup):
    _FIELDS = (
        "database_url",
        "postgres_user",
        "postgres_password",
        "postgres_db",
        "postgres_host",
        "postgres_port",
    )


class _ServerGroup(_SettingsGroup):
    _FIELDS = (
        "server_host",
        "server_port",
    )

_PACKAGE_DIR = Path(__file__).resolve().parent
_REPO_ROOT = _PACKAGE_DIR.parent
_DEFAULT_ENV_FILE = _REPO_ROOT / ".env"


class Settings(BaseSettings):
    """Application configuration loaded from environment / .env.
    算法与管线默认可调项见 ``app.baked.algorithm`` / ``app.baked.pipeline``。
    """

    database_url: str | None = None
    postgres_user: str = "postgres"
    postgres_password: str = "postgres"
    postgres_db: str = "operation_room"
    postgres_host: str = "localhost"
    postgres_port: int = 35432

    server_host: str = "0.0.0.0"
    server_port: int = Field(default=38080, ge=1, le=65535)

    video_default_backend: Literal["rtsp", "hikvision_sdk", "auto"] = "rtsp"
    video_camera_backend_overrides_json: str = ""
    video_rtsp_url_template: str = ""
    #: 手术室站点配置（UTF-8 JSON）：须含 video_rtsp_urls 与 voice_or_room_bindings，见 or_site_config.sample.json
    or_site_config_json_file: str = ""

    hikvision_lib_dir: str = "/opt/hikvision/lib"
    hikvision_sdk_enabled: bool = False
    hikvision_device_ip: str = ""
    hikvision_device_port: int = Field(default=8000, ge=1, le=65535)
    hikvision_user: str = ""
    hikvision_password: str = ""
    hikvision_channel: int = Field(default=1, ge=1, le=512)
    hikvision_preview_rtsp_template: str = ""
    hikvision_camera_rtsp_urls_json: str = ""
    hikvision_sdk_fallback_to_rtsp: bool = True

    baidu_speech_app_id: str = Field(default="", validation_alias="BAIDU_APP_ID")
    baidu_speech_api_key: str = Field(default="", validation_alias="BAIDU_API_KEY")
    baidu_speech_secret_key: str = Field(default="", validation_alias="BAIDU_SECRET_KEY")
    baidu_speech_connection_timeout_ms: int | None = Field(
        default=None, validation_alias="BAIDU_CONNECTION_TIMEOUT_MS"
    )
    baidu_speech_socket_timeout_ms: int | None = Field(
        default=None, validation_alias="BAIDU_SOCKET_TIMEOUT_MS"
    )
    baidu_speech_asr_dev_pid: int = Field(
        default=1537, ge=1000, le=99999, validation_alias="BAIDU_ASR_DEV_PID"
    )

    minio_endpoint: str = ""
    minio_access_key: str = ""
    minio_secret_key: str = ""
    minio_bucket: str = "operation-room-voice"
    minio_secure: bool = False
    minio_region: str = ""

    demo_cors_enabled: bool = True
    demo_cors_origins: str = "*"
    demo_orchestrator_enabled: bool = False
    demo_orchestrator_rtsp_port: int = Field(default=18554, ge=1, le=65535)
    demo_orchestrator_rtsp_json_host: str = "host.docker.internal"

    def parsed_demo_cors_origins(self) -> list[str]:
        raw = (self.demo_cors_origins or "").strip()
        if not raw:
            return []
        if raw == "*":
            return ["*"]
        return [item.strip() for item in raw.split(",") if item.strip()]

    model_config = SettingsConfigDict(
        env_file=(str(_DEFAULT_ENV_FILE),),
        env_file_encoding="utf-8",
        extra="ignore",
        env_ignore_empty=True,
        populate_by_name=True,
    )

    @property
    def sqlalchemy_database_url(self) -> str:
        component_values = (
            self.postgres_user,
            self.postgres_password,
            self.postgres_db,
            self.postgres_host,
            self.postgres_port,
        )
        default_component_values = (
            "postgres",
            "postgres",
            "operation_room",
            "localhost",
            35432,
        )

        if component_values != default_component_values or not self.database_url:
            user = quote_plus(self.postgres_user)
            password = quote_plus(self.postgres_password)
            database = quote_plus(self.postgres_db)
            return (
                "postgresql+asyncpg://"
                f"{user}:{password}@{self.postgres_host}:{self.postgres_port}/{database}"
            )

        return self.database_url

    @property
    def baidu_speech_configured(self) -> bool:
        return bool(
            self.baidu_speech_app_id.strip()
            and self.baidu_speech_api_key.strip()
            and self.baidu_speech_secret_key.strip()
        )

    @property
    def minio_configured(self) -> bool:
        return bool(
            self.minio_endpoint.strip()
            and self.minio_access_key.strip()
            and self.minio_secret_key.strip()
            and self.minio_bucket.strip()
        )

    def load_or_site_config(self) -> OrSiteConfig | None:
        """解析 ``or_site_config_json_file``；未配置路径时返回 ``None``。"""
        from app.or_site_config import load_or_site_config_from_path

        path_raw = (self.or_site_config_json_file or "").strip()
        if not path_raw:
            return None
        path = Path(path_raw).expanduser()
        if not path.is_file():
            raise ValueError(f"OR_SITE_CONFIG_JSON_FILE is set but file not found: {path}")
        return load_or_site_config_from_path(path)

    def video_rtsp_url_map(self) -> dict[str, str]:
        cfg = self.load_or_site_config()
        if cfg is None:
            return {}
        return dict(cfg.video_rtsp_urls)

    @property
    def or_site_config_sample_path(self) -> str:
        return baked_algorithm.default_or_site_config_sample_path()

    @property
    def video(self) -> _VideoGroup:
        return _VideoGroup(self)

    @property
    def voice(self) -> _VoiceGroup:
        return _VoiceGroup(self)

    @property
    def hikvision(self) -> _HikvisionGroup:
        return _HikvisionGroup(self)

    @property
    def minio(self) -> _MinioGroup:
        return _MinioGroup(self)

    @property
    def baidu(self) -> _BaiduGroup:
        return _BaiduGroup(self)

    @property
    def demo(self) -> _DemoGroup:
        return _DemoGroup(self)

    @property
    def database(self) -> _DatabaseGroup:
        return _DatabaseGroup(self)

    @property
    def server(self) -> _ServerGroup:
        return _ServerGroup(self)


settings = Settings()