add enabled field to camera profiles for enabling/disabling cameras

refactor _publish_updates to only publish ZMQ updates for
sections that actually changed, not all sections on affected cameras.

frigate/api/app.py

@@ -31,7 +31,11 @@ from frigate.api.auth import (
     require_role,
 )
 from frigate.api.defs.query.app_query_parameters import AppTimelineHourlyQueryParameters
-from frigate.api.defs.request.app_body import AppConfigSetBody, MediaSyncBody
+from frigate.api.defs.request.app_body import (
+    AppConfigSetBody,
+    MediaSyncBody,
+    ProfileSetBody,
+)
 from frigate.api.defs.tags import Tags
 from frigate.config import FrigateConfig
 from frigate.config.camera.updater import (
@@ -201,6 +205,41 @@ def config(request: Request):
     return JSONResponse(content=config)
 
 
+@router.get("/profiles", dependencies=[Depends(allow_any_authenticated())])
+def get_profiles(request: Request):
+    """List all available profiles and the currently active profile."""
+    profile_manager = request.app.profile_manager
+    return JSONResponse(content=profile_manager.get_profile_info())
+
+
+@router.get("/profile/active", dependencies=[Depends(allow_any_authenticated())])
+def get_active_profile(request: Request):
+    """Get the currently active profile."""
+    config_obj: FrigateConfig = request.app.frigate_config
+    return JSONResponse(content={"active_profile": config_obj.active_profile})
+
+
+@router.put("/profile/set", dependencies=[Depends(require_role(["admin"]))])
+def set_profile(request: Request, body: ProfileSetBody):
+    """Activate or deactivate a profile."""
+    profile_manager = request.app.profile_manager
+    err = profile_manager.activate_profile(body.profile)
+    if err:
+        return JSONResponse(
+            content={"success": False, "message": err},
+            status_code=400,
+        )
+    request.app.dispatcher.publish(
+        "profile/state", body.profile or "none", retain=True
+    )
+    return JSONResponse(
+        content={
+            "success": True,
+            "active_profile": body.profile,
+        }
+    )
+
+
 @router.get("/ffmpeg/presets", dependencies=[Depends(allow_any_authenticated())])
 def ffmpeg_presets():
     """Return available ffmpeg preset keys for config UI usage."""

frigate/api/defs/request/app_body.py

@@ -30,6 +30,12 @@ class AppPutRoleBody(BaseModel):
     role: str
 
 
+class ProfileSetBody(BaseModel):
+    profile: Optional[str] = Field(
+        default=None, description="Profile name to activate, or null to deactivate"
+    )
+
+
 class MediaSyncBody(BaseModel):
     dry_run: bool = Field(
         default=True, description="If True, only report orphans without deleting them"

frigate/api/fastapi_app.py

@@ -68,6 +68,8 @@ def create_fastapi_app(
     event_metadata_updater: EventMetadataPublisher,
     config_publisher: CameraConfigUpdatePublisher,
     replay_manager: DebugReplayManager,
+    dispatcher=None,
+    profile_manager=None,
     enforce_default_admin: bool = True,
 ):
     logger.info("Starting FastAPI app")
@@ -149,6 +151,8 @@ def create_fastapi_app(
     app.event_metadata_updater = event_metadata_updater
     app.config_publisher = config_publisher
     app.replay_manager = replay_manager
+    app.dispatcher = dispatcher
+    app.profile_manager = profile_manager
 
     if frigate_config.auth.enabled:
         secret = get_jwt_secret()

frigate/app.py

@@ -30,6 +30,7 @@ from frigate.comms.ws import WebSocketClient
 from frigate.comms.zmq_proxy import ZmqProxy
 from frigate.config.camera.updater import CameraConfigUpdatePublisher
 from frigate.config.config import FrigateConfig
+from frigate.config.profile_manager import ProfileManager
 from frigate.const import (
     CACHE_DIR,
     CLIPS_DIR,
@@ -348,6 +349,19 @@ class FrigateApp:
             comms,
         )
 
+    def init_profile_manager(self) -> None:
+        self.profile_manager = ProfileManager(
+            self.config, self.inter_config_updater
+        )
+        self.dispatcher.profile_manager = self.profile_manager
+
+        persisted = ProfileManager.load_persisted_profile()
+        if persisted and any(
+            persisted in cam.profiles for cam in self.config.cameras.values()
+        ):
+            logger.info("Restoring persisted profile '%s'", persisted)
+            self.profile_manager.activate_profile(persisted)
+
     def start_detectors(self) -> None:
         for name in self.config.cameras.keys():
             try:
@@ -556,6 +570,7 @@ class FrigateApp:
         self.init_inter_process_communicator()
         self.start_detectors()
         self.init_dispatcher()
+        self.init_profile_manager()
         self.init_embeddings_client()
         self.start_video_output_processor()
         self.start_ptz_autotracker()
@@ -585,6 +600,8 @@ class FrigateApp:
                     self.event_metadata_updater,
                     self.inter_config_updater,
                     self.replay_manager,
+                    self.dispatcher,
+                    self.profile_manager,
                 ),
                 host="127.0.0.1",
                 port=5001,

frigate/comms/dispatcher.py

@@ -91,7 +91,9 @@ class Dispatcher:
         }
         self._global_settings_handlers: dict[str, Callable] = {
             "notifications": self._on_global_notification_command,
+            "profile": self._on_profile_command,
         }
+        self.profile_manager = None
 
         for comm in self.comms:
             comm.subscribe(self._receive)
@@ -298,6 +300,11 @@ class Dispatcher:
             )
             self.publish("birdseye_layout", json.dumps(self.birdseye_layout.copy()))
             self.publish("audio_detections", json.dumps(audio_detections))
+            self.publish(
+                "profile/state",
+                self.config.active_profile or "none",
+                retain=True,
+            )
 
         def handle_notification_test() -> None:
             self.publish("notification_test", "Test notification")
@@ -556,6 +563,20 @@ class Dispatcher:
         )
         self.publish("notifications/state", payload, retain=True)
 
+    def _on_profile_command(self, payload: str) -> None:
+        """Callback for profile/set topic."""
+        if self.profile_manager is None:
+            logger.error("Profile manager not initialized")
+            return
+
+        profile_name = payload.strip() if payload.strip() not in ("", "none", "None") else None
+        err = self.profile_manager.activate_profile(profile_name)
+        if err:
+            logger.error("Failed to activate profile: %s", err)
+            return
+
+        self.publish("profile/state", payload.strip() or "none", retain=True)
+
     def _on_audio_command(self, camera_name: str, payload: str) -> None:
         """Callback for audio topic."""
         audio_settings = self.config.cameras[camera_name].audio

frigate/comms/mqtt.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import threading
 from typing import Any, Callable
@@ -163,6 +164,22 @@ class MqttClient(Communicator):
                 retain=True,
             )
 
+        self.publish(
+            "profile/state",
+            self.config.active_profile or "none",
+            retain=True,
+        )
+        available_profiles: list[str] = []
+        for camera in self.config.cameras.values():
+            for profile_name in camera.profiles:
+                if profile_name not in available_profiles:
+                    available_profiles.append(profile_name)
+        self.publish(
+            "profiles/available",
+            json.dumps(sorted(available_profiles)),
+            retain=True,
+        )
+
         self.publish("available", "online", retain=True)
 
     def on_mqtt_command(
@@ -289,6 +306,11 @@ class MqttClient(Communicator):
                 self.on_mqtt_command,
             )
 
+        self.client.message_callback_add(
+            f"{self.mqtt_config.topic_prefix}/profile/set",
+            self.on_mqtt_command,
+        )
+
         self.client.message_callback_add(
             f"{self.mqtt_config.topic_prefix}/onConnect", self.on_mqtt_command
         )

frigate/config/camera/camera.py

@@ -34,6 +34,7 @@ from .mqtt import CameraMqttConfig
 from .notification import NotificationConfig
 from .objects import ObjectConfig
 from .onvif import OnvifConfig
+from .profile import CameraProfileConfig
 from .record import RecordConfig
 from .review import ReviewConfig
 from .snapshots import SnapshotsConfig
@@ -184,6 +185,12 @@ class CameraConfig(FrigateBaseModel):
         title="Camera URL",
         description="URL to visit the camera directly from system page",
     )
+
+    profiles: dict[str, CameraProfileConfig] = Field(
+        default_factory=dict,
+        title="Profiles",
+        description="Named config profiles with partial overrides that can be activated at runtime.",
+    )
     zones: dict[str, ZoneConfig] = Field(
         default_factory=dict,
         title="Zones",

frigate/config/camera/profile.py

@@ -0,0 +1,36 @@
+"""Camera profile configuration for named config overrides."""
+
+from typing import Optional
+
+from ..base import FrigateBaseModel
+from .audio import AudioConfig
+from .birdseye import BirdseyeCameraConfig
+from .detect import DetectConfig
+from .motion import MotionConfig
+from .notification import NotificationConfig
+from .objects import ObjectConfig
+from .record import RecordConfig
+from .review import ReviewConfig
+from .snapshots import SnapshotsConfig
+
+__all__ = ["CameraProfileConfig"]
+
+
+class CameraProfileConfig(FrigateBaseModel):
+    """A named profile containing partial camera config overrides.
+
+    Sections set to None inherit from the camera's base config.
+    Sections that are defined get Pydantic-validated, then only
+    explicitly-set fields are used as overrides via exclude_unset.
+    """
+
+    enabled: Optional[bool] = None
+    audio: Optional[AudioConfig] = None
+    birdseye: Optional[BirdseyeCameraConfig] = None
+    detect: Optional[DetectConfig] = None
+    motion: Optional[MotionConfig] = None
+    notifications: Optional[NotificationConfig] = None
+    objects: Optional[ObjectConfig] = None
+    record: Optional[RecordConfig] = None
+    review: Optional[ReviewConfig] = None
+    snapshots: Optional[SnapshotsConfig] = None

frigate/config/config.py

@@ -560,6 +560,13 @@ class FrigateConfig(FrigateBaseModel):
         description="Configuration for named camera groups used to organize cameras in the UI.",
     )
 
+    active_profile: Optional[str] = Field(
+        default=None,
+        title="Active profile",
+        description="Currently active profile name. Runtime-only, not persisted in YAML.",
+        exclude=True,
+    )
+
     _plus_api: PlusApi
 
     @property

frigate/config/profile_manager.py

@@ -0,0 +1,221 @@
+"""Profile manager for activating/deactivating named config profiles."""
+
+import logging
+from pathlib import Path
+from typing import Optional
+
+from frigate.config.camera.updater import (
+    CameraConfigUpdateEnum,
+    CameraConfigUpdatePublisher,
+    CameraConfigUpdateTopic,
+)
+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,
+    "motion": CameraConfigUpdateEnum.motion,
+    "notifications": CameraConfigUpdateEnum.notifications,
+    "objects": CameraConfigUpdateEnum.objects,
+    "record": CameraConfigUpdateEnum.record,
+    "review": CameraConfigUpdateEnum.review,
+    "snapshots": CameraConfigUpdateEnum.snapshots,
+}
+
+PERSISTENCE_FILE = Path(CONFIG_DIR) / ".active_profile"
+
+
+class ProfileManager:
+    """Manages profile activation, persistence, and config application."""
+
+    def __init__(
+        self,
+        config,
+        config_updater: CameraConfigUpdatePublisher,
+    ):
+        from frigate.config.config import FrigateConfig
+
+        self.config: FrigateConfig = config
+        self.config_updater = config_updater
+        self._base_configs: dict[str, dict[str, dict]] = {}
+        self._base_enabled: dict[str, bool] = {}
+        self._snapshot_base_configs()
+
+    def _snapshot_base_configs(self) -> None:
+        """Snapshot each camera's current section configs and enabled state."""
+        for cam_name, cam_config in self.config.cameras.items():
+            self._base_configs[cam_name] = {}
+            self._base_enabled[cam_name] = cam_config.enabled
+            for section in PROFILE_SECTION_UPDATES:
+                section_config = getattr(cam_config, section, None)
+                if section_config is not None:
+                    self._base_configs[cam_name][section] = section_config.model_dump()
+
+    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:
+            has_profile = any(
+                profile_name in cam.profiles
+                for cam in self.config.cameras.values()
+            )
+            if not has_profile:
+                return f"Profile '{profile_name}' not found on any camera"
+
+        # 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 section configs
+            base = self._base_configs.get(cam_name, {})
+            for section in PROFILE_SECTION_UPDATES:
+                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")
+
+            base = self._base_configs.get(cam_name, {})
+
+            for section in PROFILE_SECTION_UPDATES:
+                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,
+                    )
+                    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_available_profiles(self) -> list[str]:
+        """Get a deduplicated list of all profile names across cameras."""
+        profiles: set[str] = set()
+        for cam_config in self.config.cameras.values():
+            profiles.update(cam_config.profiles.keys())
+        return sorted(profiles)
+
+    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,
+        }

frigate/test/test_profiles.py

@@ -0,0 +1,467 @@
+"""Tests for the profiles system."""
+
+import os
+import tempfile
+import unittest
+from unittest.mock import MagicMock, patch
+
+from frigate.config import FrigateConfig
+from frigate.config.camera.profile import CameraProfileConfig
+from frigate.config.profile_manager import PERSISTENCE_FILE, ProfileManager
+from frigate.const import MODEL_CACHE_DIR
+
+
+class TestCameraProfileConfig(unittest.TestCase):
+    """Test the CameraProfileConfig Pydantic model."""
+
+    def test_empty_profile(self):
+        """All sections default to None."""
+        profile = CameraProfileConfig()
+        assert profile.detect is None
+        assert profile.motion is None
+        assert profile.objects is None
+        assert profile.review is None
+        assert profile.notifications is None
+
+    def test_partial_detect(self):
+        """Profile with only detect.enabled set."""
+        profile = CameraProfileConfig(detect={"enabled": False})
+        assert profile.detect is not None
+        assert profile.detect.enabled is False
+        dumped = profile.detect.model_dump(exclude_unset=True)
+        assert dumped == {"enabled": False}
+
+    def test_partial_notifications(self):
+        """Profile with only notifications.enabled set."""
+        profile = CameraProfileConfig(notifications={"enabled": True})
+        assert profile.notifications is not None
+        assert profile.notifications.enabled is True
+        dumped = profile.notifications.model_dump(exclude_unset=True)
+        assert dumped == {"enabled": True}
+
+    def test_partial_objects(self):
+        """Profile with objects.track set."""
+        profile = CameraProfileConfig(objects={"track": ["car", "package"]})
+        assert profile.objects is not None
+        assert profile.objects.track == ["car", "package"]
+
+    def test_partial_review(self):
+        """Profile with nested review.alerts.labels."""
+        profile = CameraProfileConfig(
+            review={"alerts": {"labels": ["person", "car"]}}
+        )
+        assert profile.review is not None
+        assert profile.review.alerts.labels == ["person", "car"]
+
+    def test_enabled_field(self):
+        """Profile with enabled set to False."""
+        profile = CameraProfileConfig(enabled=False)
+        assert profile.enabled is False
+        dumped = profile.model_dump(exclude_unset=True)
+        assert dumped == {"enabled": False}
+
+    def test_enabled_field_true(self):
+        """Profile with enabled set to True."""
+        profile = CameraProfileConfig(enabled=True)
+        assert profile.enabled is True
+
+    def test_enabled_default_none(self):
+        """Enabled defaults to None when not set."""
+        profile = CameraProfileConfig()
+        assert profile.enabled is None
+
+    def test_none_sections_not_in_dump(self):
+        """Sections left as None should not appear in exclude_unset dump."""
+        profile = CameraProfileConfig(detect={"enabled": False})
+        dumped = profile.model_dump(exclude_unset=True)
+        assert "detect" in dumped
+        assert "motion" not in dumped
+        assert "objects" not in dumped
+
+    def test_invalid_field_value_rejected(self):
+        """Invalid field values are caught by Pydantic."""
+        from pydantic import ValidationError
+
+        with self.assertRaises(ValidationError):
+            CameraProfileConfig(detect={"fps": "not_a_number"})
+
+    def test_invalid_section_key_rejected(self):
+        """Unknown section keys are rejected (extra=forbid from FrigateBaseModel)."""
+        from pydantic import ValidationError
+
+        with self.assertRaises(ValidationError):
+            CameraProfileConfig(ffmpeg={"inputs": []})
+
+    def test_invalid_nested_field_rejected(self):
+        """Invalid nested field values are caught."""
+        from pydantic import ValidationError
+
+        with self.assertRaises(ValidationError):
+            CameraProfileConfig(
+                review={"alerts": {"labels": "not_a_list"}}
+            )
+
+    def test_invalid_profile_in_camera_config(self):
+        """Invalid profile section in full config is caught at parse time."""
+        from pydantic import ValidationError
+
+        config_data = {
+            "mqtt": {"host": "mqtt"},
+            "cameras": {
+                "front": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {
+                                "path": "rtsp://10.0.0.1:554/video",
+                                "roles": ["detect"],
+                            }
+                        ]
+                    },
+                    "detect": {"height": 1080, "width": 1920, "fps": 5},
+                    "profiles": {
+                        "armed": {
+                            "detect": {"fps": "invalid"},
+                        },
+                    },
+                },
+            },
+        }
+        with self.assertRaises(ValidationError):
+            FrigateConfig(**config_data)
+
+
+class TestProfileInConfig(unittest.TestCase):
+    """Test that profiles parse correctly in FrigateConfig."""
+
+    def setUp(self):
+        self.base_config = {
+            "mqtt": {"host": "mqtt"},
+            "cameras": {
+                "front": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {
+                                "path": "rtsp://10.0.0.1:554/video",
+                                "roles": ["detect"],
+                            }
+                        ]
+                    },
+                    "detect": {"height": 1080, "width": 1920, "fps": 5},
+                    "profiles": {
+                        "armed": {
+                            "notifications": {"enabled": True},
+                            "objects": {"track": ["person", "car", "package"]},
+                        },
+                        "disarmed": {
+                            "notifications": {"enabled": False},
+                            "objects": {"track": ["package"]},
+                        },
+                    },
+                },
+                "back": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {
+                                "path": "rtsp://10.0.0.2:554/video",
+                                "roles": ["detect"],
+                            }
+                        ]
+                    },
+                    "detect": {"height": 1080, "width": 1920, "fps": 5},
+                    "profiles": {
+                        "armed": {
+                            "detect": {"enabled": True},
+                        },
+                    },
+                },
+            },
+        }
+
+        if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR):
+            os.makedirs(MODEL_CACHE_DIR)
+
+    def test_profiles_parse(self):
+        """Profiles are parsed into Dict[str, CameraProfileConfig]."""
+        config = FrigateConfig(**self.base_config)
+        front = config.cameras["front"]
+        assert "armed" in front.profiles
+        assert "disarmed" in front.profiles
+        assert isinstance(front.profiles["armed"], CameraProfileConfig)
+
+    def test_profile_sections_parsed(self):
+        """Profile sections are properly typed."""
+        config = FrigateConfig(**self.base_config)
+        armed = config.cameras["front"].profiles["armed"]
+        assert armed.notifications is not None
+        assert armed.notifications.enabled is True
+        assert armed.objects is not None
+        assert armed.objects.track == ["person", "car", "package"]
+        assert armed.detect is None  # not set in this profile
+
+    def test_camera_without_profiles(self):
+        """Camera with no profiles has empty dict."""
+        config_data = {
+            "mqtt": {"host": "mqtt"},
+            "cameras": {
+                "front": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {
+                                "path": "rtsp://10.0.0.1:554/video",
+                                "roles": ["detect"],
+                            }
+                        ]
+                    },
+                    "detect": {"height": 1080, "width": 1920, "fps": 5},
+                },
+            },
+        }
+        config = FrigateConfig(**config_data)
+        assert config.cameras["front"].profiles == {}
+
+
+class TestProfileManager(unittest.TestCase):
+    """Test ProfileManager activation, deactivation, and switching."""
+
+    def setUp(self):
+        self.config_data = {
+            "mqtt": {"host": "mqtt"},
+            "cameras": {
+                "front": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {
+                                "path": "rtsp://10.0.0.1:554/video",
+                                "roles": ["detect"],
+                            }
+                        ]
+                    },
+                    "detect": {"height": 1080, "width": 1920, "fps": 5},
+                    "notifications": {"enabled": False},
+                    "objects": {"track": ["person"]},
+                    "profiles": {
+                        "armed": {
+                            "notifications": {"enabled": True},
+                            "objects": {"track": ["person", "car", "package"]},
+                        },
+                        "disarmed": {
+                            "notifications": {"enabled": False},
+                            "objects": {"track": ["package"]},
+                        },
+                    },
+                },
+                "back": {
+                    "ffmpeg": {
+                        "inputs": [
+                            {
+                                "path": "rtsp://10.0.0.2:554/video",
+                                "roles": ["detect"],
+                            }
+                        ]
+                    },
+                    "detect": {"height": 1080, "width": 1920, "fps": 5},
+                    "profiles": {
+                        "armed": {
+                            "notifications": {"enabled": True},
+                        },
+                    },
+                },
+            },
+        }
+
+        if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR):
+            os.makedirs(MODEL_CACHE_DIR)
+
+        self.config = FrigateConfig(**self.config_data)
+        self.mock_updater = MagicMock()
+        self.manager = ProfileManager(self.config, self.mock_updater)
+
+    def test_get_available_profiles(self):
+        """Available profiles are collected from all cameras."""
+        profiles = self.manager.get_available_profiles()
+        assert "armed" in profiles
+        assert "disarmed" in profiles
+        assert len(profiles) == 2
+
+    def test_activate_invalid_profile(self):
+        """Activating non-existent profile returns error."""
+        err = self.manager.activate_profile("nonexistent")
+        assert err is not None
+        assert "not found" in err
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_activate_profile(self, mock_persist):
+        """Activating a profile applies overrides."""
+        err = self.manager.activate_profile("armed")
+        assert err is None
+        assert self.config.active_profile == "armed"
+
+        # Front camera should have armed overrides
+        front = self.config.cameras["front"]
+        assert front.notifications.enabled is True
+        assert front.objects.track == ["person", "car", "package"]
+
+        # Back camera should have armed overrides
+        back = self.config.cameras["back"]
+        assert back.notifications.enabled is True
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_deactivate_profile(self, mock_persist):
+        """Deactivating a profile restores base config."""
+        # Activate first
+        self.manager.activate_profile("armed")
+        assert self.config.cameras["front"].notifications.enabled is True
+
+        # Deactivate
+        err = self.manager.activate_profile(None)
+        assert err is None
+        assert self.config.active_profile is None
+
+        # Should be back to base
+        front = self.config.cameras["front"]
+        assert front.notifications.enabled is False
+        assert front.objects.track == ["person"]
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_switch_profiles(self, mock_persist):
+        """Switching from one profile to another works."""
+        self.manager.activate_profile("armed")
+        assert self.config.cameras["front"].objects.track == [
+            "person",
+            "car",
+            "package",
+        ]
+
+        self.manager.activate_profile("disarmed")
+        assert self.config.active_profile == "disarmed"
+        assert self.config.cameras["front"].objects.track == ["package"]
+        assert self.config.cameras["front"].notifications.enabled is False
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_unaffected_camera(self, mock_persist):
+        """Camera without the activated profile is unaffected."""
+        back_base_notifications = self.config.cameras["back"].notifications.enabled
+
+        self.manager.activate_profile("disarmed")
+
+        # Back camera has no "disarmed" profile, should be unchanged
+        assert self.config.cameras["back"].notifications.enabled == back_base_notifications
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_activate_profile_disables_camera(self, mock_persist):
+        """Profile with enabled=false disables the camera."""
+        # Add a profile that disables the front camera
+        from frigate.config.camera.profile import CameraProfileConfig
+
+        self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
+            enabled=False
+        )
+        # Re-create manager to pick up new profile
+        self.manager = ProfileManager(self.config, self.mock_updater)
+
+        assert self.config.cameras["front"].enabled is True
+        err = self.manager.activate_profile("away")
+        assert err is None
+        assert self.config.cameras["front"].enabled is False
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_deactivate_restores_enabled(self, mock_persist):
+        """Deactivating a profile restores the camera's base enabled state."""
+        from frigate.config.camera.profile import CameraProfileConfig
+
+        self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
+            enabled=False
+        )
+        self.manager = ProfileManager(self.config, self.mock_updater)
+
+        self.manager.activate_profile("away")
+        assert self.config.cameras["front"].enabled is False
+
+        self.manager.activate_profile(None)
+        assert self.config.cameras["front"].enabled is True
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_enabled_zmq_published(self, mock_persist):
+        """ZMQ update is published for enabled state change."""
+        from frigate.config.camera.profile import CameraProfileConfig
+        from frigate.config.camera.updater import (
+            CameraConfigUpdateEnum,
+            CameraConfigUpdateTopic,
+        )
+
+        self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
+            enabled=False
+        )
+        self.manager = ProfileManager(self.config, self.mock_updater)
+        self.mock_updater.reset_mock()
+
+        self.manager.activate_profile("away")
+
+        # Find the enabled update call
+        enabled_calls = [
+            call
+            for call in self.mock_updater.publish_update.call_args_list
+            if call[0][0]
+            == CameraConfigUpdateTopic(CameraConfigUpdateEnum.enabled, "front")
+        ]
+        assert len(enabled_calls) == 1
+        assert enabled_calls[0][0][1] is False
+
+    @patch.object(ProfileManager, "_persist_active_profile")
+    def test_zmq_updates_published(self, mock_persist):
+        """ZMQ updates are published when a profile is activated."""
+        self.manager.activate_profile("armed")
+        assert self.mock_updater.publish_update.called
+
+    def test_get_profile_info(self):
+        """Profile info returns correct structure."""
+        info = self.manager.get_profile_info()
+        assert "profiles" in info
+        assert "active_profile" in info
+        assert info["active_profile"] is None
+        assert "armed" in info["profiles"]
+
+
+class TestProfilePersistence(unittest.TestCase):
+    """Test profile persistence to disk."""
+
+    def test_persist_and_load(self):
+        """Active profile name can be persisted and loaded."""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
+            temp_path = f.name
+
+        try:
+            from pathlib import Path
+
+            path = Path(temp_path)
+            path.write_text("armed")
+            loaded = path.read_text().strip()
+            assert loaded == "armed"
+        finally:
+            os.unlink(temp_path)
+
+    def test_load_empty_file(self):
+        """Empty persistence file returns None."""
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f:
+            f.write("")
+            temp_path = f.name
+
+        try:
+            with patch.object(type(PERSISTENCE_FILE), "exists", return_value=True):
+                with patch.object(
+                    type(PERSISTENCE_FILE), "read_text", return_value=""
+                ):
+                    result = ProfileManager.load_persisted_profile()
+                    assert result is None
+        finally:
+            os.unlink(temp_path)
+
+    def test_load_missing_file(self):
+        """Missing persistence file returns None."""
+        with patch.object(type(PERSISTENCE_FILE), "exists", return_value=False):
+            result = ProfileManager.load_persisted_profile()
+            assert result is None
+
+
+if __name__ == "__main__":
+    unittest.main()