frigate/frigate/config/profile_manager.py

"""Profile manager for activating/deactivating named config profiles."""

import copy
import json
import logging
from collections.abc import Callable
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

from frigate.config.camera.updater import (
    CameraConfigUpdateEnum,
    CameraConfigUpdatePublisher,
    CameraConfigUpdateTopic,
)
from frigate.config.camera.zone import ZoneConfig
from frigate.const import CONFIG_DIR
from frigate.util.builtin import deep_merge
from frigate.util.config import apply_section_update

logger = logging.getLogger(__name__)

PROFILE_SECTION_UPDATES: dict[str, CameraConfigUpdateEnum] = {
    "audio": CameraConfigUpdateEnum.audio,
    "birdseye": CameraConfigUpdateEnum.birdseye,
    "detect": CameraConfigUpdateEnum.detect,
    "face_recognition": CameraConfigUpdateEnum.face_recognition,
    "lpr": CameraConfigUpdateEnum.lpr,
    "motion": CameraConfigUpdateEnum.motion,
    "notifications": CameraConfigUpdateEnum.notifications,
    "objects": CameraConfigUpdateEnum.objects,
    "record": CameraConfigUpdateEnum.record,
    "review": CameraConfigUpdateEnum.review,
    "snapshots": CameraConfigUpdateEnum.snapshots,
    "zones": CameraConfigUpdateEnum.zones,
}

# Retained MQTT switch topics per profile section, with a payload getter.
# Republished on profile change so MQTT/HA don't show a stale toggle.
SECTION_STATE_TOPICS: dict[str, list[tuple[str, Callable[[Any], Any]]]] = {
    "audio": [("audio", lambda c: "ON" if c.audio.enabled else "OFF")],
    "birdseye": [
        ("birdseye", lambda c: "ON" if c.birdseye.enabled else "OFF"),
        (
            "birdseye_mode",
            lambda c: c.birdseye.mode.value.upper() if c.birdseye.enabled else "OFF",
        ),
    ],
    "detect": [("detect", lambda c: "ON" if c.detect.enabled else "OFF")],
    "motion": [
        ("motion", lambda c: "ON" if c.motion.enabled else "OFF"),
        ("improve_contrast", lambda c: "ON" if c.motion.improve_contrast else "OFF"),
        ("motion_threshold", lambda c: c.motion.threshold),
        ("motion_contour_area", lambda c: c.motion.contour_area),
    ],
    "notifications": [
        ("notifications", lambda c: "ON" if c.notifications.enabled else "OFF"),
    ],
    "objects": [
        ("object_descriptions", lambda c: "ON" if c.objects.genai.enabled else "OFF"),
    ],
    "record": [("recordings", lambda c: "ON" if c.record.enabled else "OFF")],
    "review": [
        ("review_alerts", lambda c: "ON" if c.review.alerts.enabled else "OFF"),
        (
            "review_detections",
            lambda c: "ON" if c.review.detections.enabled else "OFF",
        ),
        (
            "review_descriptions",
            lambda c: "ON" if c.review.genai.enabled else "OFF",
        ),
    ],
    "snapshots": [("snapshots", lambda c: "ON" if c.snapshots.enabled else "OFF")],
}

PERSISTENCE_FILE = Path(CONFIG_DIR) / ".profiles"


class ProfileManager:
    """Manages profile activation, persistence, and config application."""

    def __init__(
        self,
        config,
        config_updater: CameraConfigUpdatePublisher,
        dispatcher=None,
    ):
        from frigate.config.config import FrigateConfig

        self.config: FrigateConfig = config
        self.config_updater = config_updater
        self.dispatcher = dispatcher
        self._base_configs: dict[str, dict[str, dict]] = {}
        self._base_api_configs: dict[str, dict[str, dict]] = {}
        self._base_enabled: dict[str, bool] = {}
        self._base_zones: dict[str, dict[str, ZoneConfig]] = {}
        self._snapshot_base_configs()

    def _snapshot_base_configs(self) -> None:
        """Snapshot each camera's current section configs, enabled, and zones."""
        for cam_name, cam_config in self.config.cameras.items():
            self._base_configs[cam_name] = {}
            self._base_api_configs[cam_name] = {}
            self._base_enabled[cam_name] = cam_config.enabled
            self._base_zones[cam_name] = copy.deepcopy(cam_config.zones)
            for section in PROFILE_SECTION_UPDATES:
                section_value = getattr(cam_config, section, None)
                if section_value is None:
                    continue

                if section == "zones":
                    # zones is a dict of ZoneConfig models
                    self._base_configs[cam_name][section] = {
                        name: zone.model_dump() for name, zone in section_value.items()
                    }
                    self._base_api_configs[cam_name][section] = {
                        name: {
                            **zone.model_dump(
                                mode="json",
                                warnings="none",
                                exclude_none=True,
                            ),
                            "color": zone.color,
                        }
                        for name, zone in section_value.items()
                    }
                else:
                    self._base_configs[cam_name][section] = section_value.model_dump()
                    self._base_api_configs[cam_name][section] = (
                        section_value.model_dump(
                            mode="json",
                            warnings="none",
                            exclude_none=True,
                        )
                    )

    def update_config(self, new_config) -> None:
        """Update config reference after config/set replaces the in-memory config.

        Preserves active profile state: re-snapshots base configs from the new
        (freshly parsed) config, then re-applies profile overrides if a profile
        was active.

        Deliberately does not clear the dispatcher's runtime overrides. This is
        the config-save path, not a profile switch: the save only invalidates
        the toggles it rewrote in yaml, which /api/config/set already clears by
        key. The broad wipe belongs to activate_profile alone.
        """
        current_active = self.config.active_profile
        self.config = new_config

        # Re-snapshot base configs from the new config (which has base values)
        self._base_configs.clear()
        self._base_api_configs.clear()
        self._base_enabled.clear()
        self._base_zones.clear()
        self._snapshot_base_configs()

        # Re-apply profile overrides without publishing ZMQ updates
        # (the config/set caller handles its own ZMQ publishing)
        if current_active is not None:
            if current_active in self.config.profiles:
                changed: dict[str, set[str]] = {}
                self._apply_profile_overrides(current_active, changed)
                self.config.active_profile = current_active
            else:
                # Profile was deleted — deactivate
                self.config.active_profile = None
                self._persist_active_profile(None)

    def _validate_profile_name(self, profile_name: str | None) -> str | None:
        """Return an error message if the name is not a defined profile."""
        if profile_name is not None and profile_name not in self.config.profiles:
            return f"Profile '{profile_name}' is not defined in the profiles section"

        return None

    def _apply_to_config(
        self, profile_name: str | None
    ) -> tuple[dict[str, set[str]], str | None]:
        """Reset every camera to base, then apply the named profile on top.

        Returns the changed camera/section pairs, plus an error message if
        applying the profile failed partway through.
        """
        changed: dict[str, set[str]] = {}

        self._reset_to_base(changed)

        if profile_name is not None:
            err = self._apply_profile_overrides(profile_name, changed)
            if err:
                return changed, err

        return changed, None

    def apply_profile_to_config(self, profile_name: str | None) -> str | None:
        """Apply a profile to the in-memory config, without publishing it.

        Safe to call ahead of activate_profile: both reset to the base config
        first, so the later call re-derives the same state and still reports
        every section as changed.

        Returns:
            None on success, or an error message string on failure.
        """
        err = self._validate_profile_name(profile_name)

        if err:
            return err

        return self._apply_to_config(profile_name)[1]

    def _persisted_profile_to_restore(self) -> str | None:
        """Return the persisted profile name, if it still applies to a camera."""
        persisted = self.load_persisted_profile()

        if not persisted or not any(
            persisted in cam.profiles for cam in self.config.cameras.values()
        ):
            return None

        return persisted

    def restore_persisted_profile_to_config(self) -> None:
        """Restore the persisted profile into the config, without publishing.

        Called before worker processes start, so they are handed a config that
        already carries the profile rather than relying on the broadcast that
        restore_persisted_profile() sends later.
        """
        persisted = self._persisted_profile_to_restore()

        if persisted is None:
            return

        err = self.apply_profile_to_config(persisted)

        if err:
            logger.error("Failed to apply persisted profile '%s': %s", persisted, err)

    def restore_persisted_profile(self) -> None:
        """Re-activate the persisted profile once subscribers are connected.

        The config already carries the profile; this pass publishes it for the
        processes that start before the config can be corrected, and for the
        retained MQTT states.
        """
        persisted = self._persisted_profile_to_restore()

        if persisted is None:
            return

        logger.info("Restoring persisted profile '%s'", persisted)
        # runtime overrides are layered on top by the dispatcher's replay
        self.activate_profile(persisted, clear_runtime_overrides=False)

    def activate_profile(
        self,
        profile_name: str | None,
        clear_runtime_overrides: bool = True,
    ) -> str | None:
        """Activate a profile by name, or deactivate if None.

        Args:
            profile_name: Profile name to activate, or None to deactivate.
            clear_runtime_overrides: When True (the default, for user-initiated
                activations) drop the dispatcher's runtime override file because
                the layer below changed. Startup callers that are replaying a
                persisted profile pass False so the runtime state stays
                available for the subsequent replay step.

        Returns:
            None on success, or an error message string on failure.
        """
        err = self._validate_profile_name(profile_name)

        if err:
            return err

        # Track which camera/section pairs get changed for ZMQ publishing
        changed, err = self._apply_to_config(profile_name)

        if err:
            return err

        # Publish ZMQ updates only for sections that actually changed
        self._publish_updates(changed)

        self.config.active_profile = profile_name
        self._persist_active_profile(profile_name)

        # a profile switch invalidates the steady-state runtime overrides
        if clear_runtime_overrides and self.dispatcher is not None:
            self.dispatcher.clear_runtime_state()

        logger.info(
            "Profile %s",
            f"'{profile_name}' activated" if profile_name else "deactivated",
        )
        return None

    def _reset_to_base(self, changed: dict[str, set[str]]) -> None:
        """Reset all cameras to their base (no-profile) config."""
        for cam_name, cam_config in self.config.cameras.items():
            # Restore enabled state
            base_enabled = self._base_enabled.get(cam_name)
            if base_enabled is not None and cam_config.enabled != base_enabled:
                cam_config.enabled = base_enabled
                changed.setdefault(cam_name, set()).add("enabled")

            # Restore zones (always restore from snapshot; direct Pydantic
            # comparison fails when ZoneConfig contains numpy arrays)
            base_zones = self._base_zones.get(cam_name)
            if base_zones is not None:
                cam_config.zones = copy.deepcopy(base_zones)
                changed.setdefault(cam_name, set()).add("zones")

            # Restore section configs (zones handled above)
            base = self._base_configs.get(cam_name, {})
            for section in PROFILE_SECTION_UPDATES:
                if section == "zones":
                    continue
                base_data = base.get(section)
                if base_data is None:
                    continue
                err = apply_section_update(cam_config, section, base_data)
                if err:
                    logger.error(
                        "Failed to reset section '%s' on camera '%s': %s",
                        section,
                        cam_name,
                        err,
                    )
                else:
                    changed.setdefault(cam_name, set()).add(section)

    def _apply_profile_overrides(
        self, profile_name: str, changed: dict[str, set[str]]
    ) -> str | None:
        """Apply profile overrides for all cameras that have the named profile."""
        for cam_name, cam_config in self.config.cameras.items():
            profile = cam_config.profiles.get(profile_name)
            if profile is None:
                continue

            # Apply enabled override
            if profile.enabled is not None and cam_config.enabled != profile.enabled:
                cam_config.enabled = profile.enabled
                changed.setdefault(cam_name, set()).add("enabled")

            # Apply zones override — merge profile zones into base zones
            if profile.zones is not None:
                base_zones = self._base_zones.get(cam_name, {})
                merged_zones = copy.deepcopy(base_zones)
                merged_zones.update(profile.zones)
                # Profile zone objects are parsed without colors or contours
                # (those are set during CameraConfig init / post-validation).
                # Inherit the base zone's color when available, and ensure
                # every zone has a valid contour for rendering.
                for name, zone in merged_zones.items():
                    if zone.contour.size == 0:
                        zone.generate_contour(cam_config.frame_shape)
                    if zone.color == (0, 0, 0) and name in base_zones:
                        zone._color = base_zones[name].color
                cam_config.zones = merged_zones
                changed.setdefault(cam_name, set()).add("zones")

            base = self._base_configs.get(cam_name, {})

            for section in PROFILE_SECTION_UPDATES:
                if section == "zones":
                    continue
                profile_section = getattr(profile, section, None)
                if profile_section is None:
                    continue

                overrides = profile_section.model_dump(exclude_unset=True)
                if not overrides:
                    continue

                base_data = base.get(section, {})
                merged = deep_merge(overrides, base_data)

                err = apply_section_update(cam_config, section, merged)
                if err:
                    return f"Failed to apply profile '{profile_name}' section '{section}' on camera '{cam_name}': {err}"

                changed.setdefault(cam_name, set()).add(section)

        return None

    def _publish_updates(self, changed: dict[str, set[str]]) -> None:
        """Publish ZMQ config updates only for sections that changed."""
        for cam_name, sections in changed.items():
            cam_config = self.config.cameras.get(cam_name)
            if cam_config is None:
                continue

            for section in sections:
                if section == "enabled":
                    self.config_updater.publish_update(
                        CameraConfigUpdateTopic(
                            CameraConfigUpdateEnum.enabled, cam_name
                        ),
                        cam_config.enabled,
                    )
                    if self.dispatcher is not None:
                        self.dispatcher.publish(
                            f"{cam_name}/enabled/state",
                            "ON" if cam_config.enabled else "OFF",
                            retain=True,
                        )
                    continue

                if section == "zones":
                    self.config_updater.publish_update(
                        CameraConfigUpdateTopic(CameraConfigUpdateEnum.zones, cam_name),
                        cam_config.zones,
                    )
                    continue

                update_enum = PROFILE_SECTION_UPDATES.get(section)
                if update_enum is None:
                    continue
                settings = getattr(cam_config, section, None)
                if settings is not None:
                    self.config_updater.publish_update(
                        CameraConfigUpdateTopic(update_enum, cam_name),
                        settings,
                    )

                # republish MQTT switch states
                if self.dispatcher is not None:
                    for suffix, get_payload in SECTION_STATE_TOPICS.get(section, ()):
                        self.dispatcher.publish(
                            f"{cam_name}/{suffix}/state",
                            get_payload(cam_config),
                            retain=True,
                        )

    def _persist_active_profile(self, profile_name: str | None) -> None:
        """Persist the active profile state to disk as JSON."""
        try:
            data = self._load_persisted_data()
            data["active"] = profile_name
            if profile_name is not None:
                data.setdefault("last_activated", {})[profile_name] = datetime.now(
                    UTC
                ).timestamp()
            PERSISTENCE_FILE.write_text(json.dumps(data))
        except OSError:
            logger.exception("Failed to persist active profile")

    @staticmethod
    def _load_persisted_data() -> dict:
        """Load the full persisted profile data from disk."""
        try:
            if PERSISTENCE_FILE.exists():
                raw = PERSISTENCE_FILE.read_text().strip()
                if raw:
                    return json.loads(raw)
        except (OSError, json.JSONDecodeError):
            logger.exception("Failed to load persisted profile data")
        return {"active": None, "last_activated": {}}

    @staticmethod
    def load_persisted_profile() -> str | None:
        """Load the persisted active profile name from disk."""
        data = ProfileManager._load_persisted_data()
        name = data.get("active")
        return name if name else None

    def get_base_configs_for_api(self, camera_name: str) -> dict[str, dict]:
        """Return base (pre-profile) section configs for a camera.

        These are JSON-serializable dicts suitable for direct inclusion in
        the /api/config response, with None values already excluded.
        """
        return self._base_api_configs.get(camera_name, {})

    def get_available_profiles(self) -> list[dict[str, str]]:
        """Get list of all profile definitions from the top-level config."""
        return [
            {"name": name, "friendly_name": defn.friendly_name}
            for name, defn in sorted(self.config.profiles.items())
        ]

    def get_profile_info(self) -> dict:
        """Get profile state info for API responses."""
        data = self._load_persisted_data()
        return {
            "profiles": self.get_available_profiles(),
            "active_profile": self.config.active_profile,
            "last_activated": data.get("last_activated", {}),
        }