blakeblackshear.frigate/frigate/config/profile_manager.py

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

import copy
import logging
from pathlib import Path
from typing import Optional

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,
}

PERSISTENCE_FILE = Path(CONFIG_DIR) / ".active_profile"


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.
        """
        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 activate_profile(self, profile_name: Optional[str]) -> Optional[str]:
        """Activate a profile by name, or deactivate if None.

        Args:
            profile_name: Profile name to activate, or None to deactivate.

        Returns:
            None on success, or an error message string on failure.
        """
        if profile_name is not None:
            if profile_name not in self.config.profiles:
                return (
                    f"Profile '{profile_name}' is not defined in the profiles section"
                )

        # Track which camera/section pairs get changed for ZMQ publishing
        changed: dict[str, set[str]] = {}

        # Reset all cameras to base config
        self._reset_to_base(changed)

        # Apply new profile overrides if activating
        if profile_name is not None:
            err = self._apply_profile_overrides(profile_name, changed)
            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)
        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]]
    ) -> Optional[str]:
        """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,
                    )

    def _persist_active_profile(self, profile_name: Optional[str]) -> None:
        """Persist the active profile name to disk."""
        try:
            if profile_name is None:
                PERSISTENCE_FILE.unlink(missing_ok=True)
            else:
                PERSISTENCE_FILE.write_text(profile_name)
        except OSError:
            logger.exception("Failed to persist active profile")

    @staticmethod
    def load_persisted_profile() -> Optional[str]:
        """Load the persisted active profile name from disk."""
        try:
            if PERSISTENCE_FILE.exists():
                name = PERSISTENCE_FILE.read_text().strip()
                return name if name else None
        except OSError:
            logger.exception("Failed to load persisted profile")
        return 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."""
        return {
            "profiles": self.get_available_profiles(),
            "active_profile": self.config.active_profile,
        }