silver_pose/v1/config.py

"""Validated, credential-safe configuration loading for Silver Pose V1."""

import hashlib
import json
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict
from urllib.parse import quote


class ConfigError(ValueError):
    """Raised when a configuration file cannot safely start V1."""


# Hikvision RTSP path: channel "101" = channel 1 main stream, "102" = sub stream.
_HIK_PATH = "/Streaming/Channels/{0}"


def build_rtsp_url(
    host: str,
    port: int,
    username: str,
    password: str,
    channel: str = "101",
    path_template: str = _HIK_PATH,
) -> str:
    """Build an RTSP URL, percent-encoding credentials per RFC 3986.

    Encoding the username and password means special characters such as ``@``,
    ``:`` and ``/`` in a password no longer break URL parsing. The returned URL
    holds credentials and must never be logged or persisted.
    """

    user = quote(str(username), safe="")
    secret = quote(str(password), safe="")
    path = path_template.format(quote(str(channel), safe=""))
    return "rtsp://{0}:{1}@{2}:{3}{4}".format(user, secret, host, int(port), path)


@dataclass(frozen=True)
class EventConfig:
    keypoint_confidence_threshold: float
    suspect_window_seconds: float
    confirm_window_seconds: float
    recovery_window_seconds: float
    cooldown_seconds: float


@dataclass(frozen=True)
class AppConfig:
    source_id: str
    source_url: str
    model_path: Path
    model_sha256: str
    confidence_threshold: float
    event: EventConfig
    event_dir: Path
    source_mode: str = "stream"

    @property
    def runtime_config_version(self) -> str:
        """Return a stable, non-secret identifier for the active event settings."""

        payload = {
            "source_id": self.source_id,
            "model_sha256": self.model_sha256,
            "confidence_threshold": self.confidence_threshold,
            "event": {
                "keypoint_confidence_threshold": self.event.keypoint_confidence_threshold,
                "suspect_window_seconds": self.event.suspect_window_seconds,
                "confirm_window_seconds": self.event.confirm_window_seconds,
                "recovery_window_seconds": self.event.recovery_window_seconds,
                "cooldown_seconds": self.event.cooldown_seconds,
            },
        }
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return "cfg-" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()


_ENVIRONMENT_NAME = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
_SHA256 = re.compile(r"^[0-9a-fA-F]{64}$")


def _mapping(value: Any, field_name: str) -> Dict[str, Any]:
    if not isinstance(value, dict):
        raise ConfigError("{0} must be an object".format(field_name))
    return value


def _text(value: Any, field_name: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise ConfigError("{0} must be a non-empty string".format(field_name))
    return value.strip()


def _number(value: Any, field_name: str, minimum: float, maximum: float) -> float:
    try:
        result = float(value)
    except (TypeError, ValueError):
        raise ConfigError("{0} must be numeric".format(field_name))
    if not minimum <= result <= maximum:
        raise ConfigError(
            "{0} must be between {1} and {2}".format(field_name, minimum, maximum)
        )
    return result


def _resolve_path(config_path: Path, value: Any, field_name: str) -> Path:
    raw_path = Path(_text(value, field_name))
    if raw_path.is_absolute():
        return raw_path
    return (config_path.parent / raw_path).resolve()


def load_config(path: Path) -> AppConfig:
    """Load one public/local configuration pair without persisting credentials.

    The config file can name an environment variable but must not embed a source
    address. The resolved address remains in memory only.
    """

    config_path = Path(path).resolve()
    try:
        raw = json.loads(config_path.read_text(encoding="utf-8"))
    except OSError as exc:
        raise ConfigError("cannot read config: {0}".format(exc))
    except json.JSONDecodeError as exc:
        raise ConfigError("invalid JSON config: {0}".format(exc))

    root = _mapping(raw, "config")
    source = _mapping(root.get("source"), "source")
    if "url" in source or "rtsp_url" in source:
        raise ConfigError("source must define structured host fields or rtsp_url_env, not a full URL")

    if "host" in source:
        host = _text(source.get("host"), "source.host")
        port = int(_number(source.get("port", 554), "source.port", 1.0, 65535.0))
        channel = str(source.get("channel", "101")).strip()
        if not channel:
            raise ConfigError("source.channel must be a non-empty value")
        username = _text(source.get("username"), "source.username")
        password = source.get("password")
        if not isinstance(password, str) or password == "":
            raise ConfigError("source.password must be a non-empty string")
        source_url = build_rtsp_url(host, port, username, password, channel)
    else:
        environment_name = _text(source.get("rtsp_url_env"), "source.rtsp_url_env")
        if not _ENVIRONMENT_NAME.match(environment_name):
            raise ConfigError("source.rtsp_url_env must be an environment variable name")
        source_url = os.environ.get(environment_name)
        if not source_url:
            raise ConfigError("missing RTSP environment variable: {0}".format(environment_name))

    source_mode = source.get("mode", "stream")
    if source_mode not in ("replay", "stream"):
        raise ConfigError("source.mode must be 'replay' or 'stream'")

    model = _mapping(root.get("model"), "model")
    model_sha256 = _text(model.get("sha256"), "model.sha256").lower()
    if not _SHA256.match(model_sha256):
        raise ConfigError("model.sha256 must be a 64-character SHA-256 value")

    event_raw = _mapping(root.get("event"), "event")
    event = EventConfig(
        keypoint_confidence_threshold=_number(
            event_raw.get("keypoint_confidence_threshold"),
            "event.keypoint_confidence_threshold",
            0.0,
            1.0,
        ),
        suspect_window_seconds=_number(
            event_raw.get("suspect_window_seconds"),
            "event.suspect_window_seconds",
            0.0,
            30.0,
        ),
        confirm_window_seconds=_number(
            event_raw.get("confirm_window_seconds"),
            "event.confirm_window_seconds",
            1.0,
            3.0,
        ),
        recovery_window_seconds=_number(
            event_raw.get("recovery_window_seconds"),
            "event.recovery_window_seconds",
            0.0,
            300.0,
        ),
        cooldown_seconds=_number(
            event_raw.get("cooldown_seconds"),
            "event.cooldown_seconds",
            0.0,
            3600.0,
        ),
    )

    artifacts = _mapping(root.get("artifacts"), "artifacts")
    return AppConfig(
        source_id=_text(source.get("id"), "source.id"),
        source_url=source_url,
        model_path=_resolve_path(config_path, model.get("path"), "model.path"),
        model_sha256=model_sha256,
        confidence_threshold=_number(
            model.get("confidence_threshold"), "model.confidence_threshold", 0.0, 1.0
        ),
        event=event,
        event_dir=_resolve_path(config_path, artifacts.get("event_dir"), "artifacts.event_dir"),
        source_mode=source_mode,
    )