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


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


@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

    @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 rtsp_url_env, not an embedded address")
    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))

    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"),
    )