Behavision/behavision/config.py

"""Typed configuration loaded from YAML with ${ENV} expansion.

Secrets never live in YAML: the YAML references environment variables
(populated from `.env`), so the config file is safe to commit.
"""
from __future__ import annotations

import os
import re
from pathlib import Path
from typing import Optional
from urllib.parse import quote

import yaml
from pydantic import BaseModel, Field, model_validator

_ENV_RE = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}")


def _expand_env(text: str) -> str:
    return _ENV_RE.sub(lambda m: os.environ.get(m.group(1), ""), text)


class CameraTuning(BaseModel):
    """Per-camera overrides for the recognition gates. None = use the global.

    These are per-camera because they describe a *view*, not a preference: a
    gate measured on an entrance camera at head height does not describe an
    overhead corridor camera, and a real deployment has both at one site.
    Measured on Office1: genuine faces score 0.32-0.45 there against a global
    gate of 0.65, so every visitor was discarded — while the same gate is
    correct for a frontal camera where real faces score 0.70-0.82.

    Note the asymmetry before overriding the similarity thresholds. Quality is
    purely local — it only asks whether THIS view is good enough to store.
    match/enroll are not: every camera writes into one shared gallery, so a
    camera set loose can merge two people into an identity that a stricter
    camera then trusts. Loosen quality per camera freely; loosen match only
    with measured cross-person data from that camera.
    """
    min_enroll_quality: Optional[float] = None
    match_threshold: Optional[float] = None
    enroll_threshold: Optional[float] = None


class CameraConfig(BaseModel):
    id: str
    url: str = ""
    host: str = ""
    port: int = 554
    path: str = "/"
    username: str = ""
    password: str = ""
    webcam: Optional[int] = None
    max_width: int = 1280  # frames wider than this are downscaled at ingest
    tuning: CameraTuning = CameraTuning()

    def source(self) -> "str | int":
        """Resolved capture source: webcam index, explicit URL, or a URL
        built from parts with percent-encoded credentials."""
        if self.webcam is not None:
            return self.webcam
        if self.url:
            return self.url
        if not self.host:
            raise ValueError(f"camera '{self.id}': set url, host or webcam")
        auth = ""
        if self.username:
            auth = quote(self.username, safe="")
            if self.password:
                auth += ":" + quote(self.password, safe="")
            auth += "@"
        path = self.path if self.path.startswith("/") else "/" + self.path
        return f"rtsp://{auth}{self.host}:{self.port}{path}"

    def safe_url(self) -> str:
        """Loggable form with the password masked."""
        src = self.source()
        if isinstance(src, int):
            return f"webcam:{src}"
        return re.sub(r"(rtsp://[^:/@]+:)[^@]*@", r"\1*****@", src)


class AppSection(BaseModel):
    data_dir: Path = Path("data")
    models_dir: Path = Path("models")
    log_level: str = "INFO"
    # Save every aligned chip the encoder sees to data/debug/ — diagnostic
    # only, off in normal operation (it writes face images to disk).
    debug_faces: bool = False
    # Write one face image per resolved visit to data/outbox/ for the agent to
    # upload. OFF by default, and that default is the product's privacy
    # position rather than an oversight: with it off this machine holds
    # templates and timestamps and nothing resembling a photograph. Turning it
    # on changes what the system is under GDPR and India's DPDP, so it has to
    # be a decision somebody makes rather than one they inherit.
    store_faces: bool = False


class ApiSection(BaseModel):
    host: str = "0.0.0.0"
    port: int = 8010
    # HTTP Basic credentials. Blank + loopback host = open (unreachable from
    # off-box anyway); blank + routable host = generated, see
    # ensure_api_credentials(). Never hardcode these — they come from .env.
    username: str = ""
    password: str = ""

    @model_validator(mode="before")
    @classmethod
    def _normalize_blanks(cls, values):
        # Unset ${ENV} placeholders parse as YAML null — treat as "".
        if isinstance(values, dict):
            values = {k: ("" if v is None else v) for k, v in values.items()}
            if values.get("port") == "":
                values["port"] = 8010
        return values

    @property
    def auth_enabled(self) -> bool:
        return bool(self.username and self.password)

    @property
    def is_loopback(self) -> bool:
        return self.host in ("127.0.0.1", "::1", "localhost", "")


class DetectionSection(BaseModel):
    # Measured on the deployment site: frosted-glass false positives pass
    # 0.75, real faces score higher. Keep in step with config/default.yaml.
    score_threshold: float = 0.82
    nms_threshold: float = 0.3
    min_face_px: int = 48
    max_faces: int = 20


class RecognitionSection(BaseModel):
    model_file: str = ""  # pin a specific model filename; empty = auto
    # Override the channel order the encoder feeds the model. Empty =
    # inferred from the model family (ArcFace RGB, AdaFace BGR).
    color_order: str = ""
    match_threshold: float = 0.42
    enroll_threshold: float = 0.32
    reinforce_threshold: float = 0.55
    max_embeddings_per_identity: int = 5
    auto_enroll: bool = True
    min_enroll_quality: float = 0.65  # real frontal faces 0.70-0.82, glass blurs <=0.54
    sighting_cooldown_seconds: float = 30.0

    @model_validator(mode="after")
    def _sane(self) -> "RecognitionSection":
        if not (0 < self.enroll_threshold < self.match_threshold < 1):
            raise ValueError("need 0 < enroll_threshold < match_threshold < 1")
        return self

    def merged(self, tuning: "CameraTuning | None") -> "RecognitionSection":
        """This section with one camera's overrides applied.

        Returns a validated copy, so a per-camera pair that inverts
        enroll/match is rejected here rather than silently driving decisions
        that contradict each other.
        """
        if tuning is None:
            return self
        overrides = {k: v for k, v in tuning.model_dump().items()
                     if v is not None}
        if not overrides:
            return self
        return RecognitionSection.model_validate(
            {**self.model_dump(), **overrides})


class TrackingSection(BaseModel):
    iou_threshold: float = 0.3
    max_misses: int = 25
    min_hits_for_id: int = 4
    min_embeddings_for_id: int = 3
    min_quality_to_encode: float = 0.35
    max_id_attempts: int = 8
    # Ambiguous tracks keep accumulating embeddings every frame but
    # only re-decide this often, so max_id_attempts spans seconds of
    # genuinely different frames rather than one burst.
    id_retry_interval_seconds: float = 0.5
    # A resolved track keeps contributing views for the rest of the
    # visit, so an identity does not stay stuck on the single embedding
    # it was born with. Sampled this often; each view is still subject
    # to the reinforce/quality/cap gates in Gallery.
    reinforce_during_track: bool = True
    reinforce_interval_seconds: float = 1.0


class AttributesSection(BaseModel):
    enabled: bool = True
    # Gate for collecting a per-frame age/gender/emotion sample. Deliberately
    # NOT recognition.min_enroll_quality, which it used to borrow: that gate
    # is 0.65 and guards minting a permanent identity, while genuine faces on
    # an overhead camera measure 0.32-0.45. Sharing it meant no track ever
    # collected the multiple samples the median is computed from, so the
    # aggregate silently collapsed to a single frame — the exact instability
    # the median was added to remove.
    min_quality: float = 0.35


class EmailSection(BaseModel):
    smtp_host: str = ""
    smtp_port: int = 587
    username: str = ""
    password: str = ""
    to: str = ""
    min_interval_seconds: float = 300.0

    @model_validator(mode="before")
    @classmethod
    def _normalize_blanks(cls, values):
        # Unset ${ENV} placeholders parse as YAML null — treat as "".
        if isinstance(values, dict):
            values = {k: ("" if v is None else v) for k, v in values.items()}
            if values.get("smtp_port") == "":
                values["smtp_port"] = 587
        return values

    @property
    def enabled(self) -> bool:
        return bool(self.smtp_host and self.username and self.to)


class EventsSection(BaseModel):
    webhook_url: str = ""
    email: EmailSection = Field(default_factory=EmailSection)

    @model_validator(mode="before")
    @classmethod
    def _normalize_blanks(cls, values):
        if isinstance(values, dict) and values.get("webhook_url") is None:
            values["webhook_url"] = ""
        return values


class Config(BaseModel):
    app: AppSection = Field(default_factory=AppSection)
    api: ApiSection = Field(default_factory=ApiSection)
    cameras: list[CameraConfig] = Field(default_factory=list)
    detection: DetectionSection = Field(default_factory=DetectionSection)
    recognition: RecognitionSection = Field(default_factory=RecognitionSection)
    tracking: TrackingSection = Field(default_factory=TrackingSection)
    attributes: AttributesSection = Field(default_factory=AttributesSection)
    events: EventsSection = Field(default_factory=EventsSection)


def ensure_api_credentials(cfg: "Config") -> "tuple[bool, bool]":
    """Make sure a routable API is never served unauthenticated.

    A live face feed plus a biometric gallery must not be readable by anyone
    who can reach the port. But failing to boot mid-deployment is its own
    outage, so instead of refusing to start we mint a credential, persist it
    0600 under data/, and log it. Returns (auth_enabled, was_generated).
    """
    import os
    import secrets

    if cfg.api.auth_enabled:
        return True, False
    if cfg.api.is_loopback:
        return False, False  # not reachable off-box; leave it open

    cred_file = cfg.app.data_dir / "api_credentials.txt"
    if cred_file.exists():
        parsed = dict(
            line.split("=", 1) for line in
            cred_file.read_text(encoding="utf-8").splitlines() if "=" in line)
        cfg.api.username = parsed.get("username", "").strip()
        cfg.api.password = parsed.get("password", "").strip()
        if cfg.api.auth_enabled:
            return True, False

    cfg.api.username = "behavision"
    cfg.api.password = secrets.token_urlsafe(16)
    cred_file.write_text(
        f"username={cfg.api.username}\npassword={cfg.api.password}\n",
        encoding="utf-8")
    try:
        os.chmod(cred_file, 0o600)
    except OSError:  # best effort (Windows)
        pass
    return True, True


def load_config(path: "Path | str | None" = None) -> Config:
    """Load .env, then YAML with ${ENV} expansion, into a validated Config.

    Relative `data_dir` / `models_dir` resolve against the **state root**, not
    the code: installed, the code lives under `Program Files` where nothing may
    write, and the database, logs and downloaded models still have to go
    somewhere that survives an upgrade. In a checkout the two are the same
    directory, so development is unaffected.
    """
    from dotenv import load_dotenv

    from .paths import ensure_config, env_file, state_root

    env = env_file()
    if env is not None:
        load_dotenv(env)

    cfg_path = Path(path) if path else ensure_config()
    raw = yaml.safe_load(_expand_env(cfg_path.read_text(encoding="utf-8"))) or {}
    cfg = Config.model_validate(raw)

    root = state_root()
    for key in ("data_dir", "models_dir"):
        p = getattr(cfg.app, key)
        if not p.is_absolute():
            setattr(cfg.app, key, root / p)
    cfg.app.data_dir.mkdir(parents=True, exist_ok=True)
    cfg.app.models_dir.mkdir(parents=True, exist_ok=True)
    return cfg