administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-08-07 03:11:15 +03:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Camera profile support (#22482)

Browse Source 
* add CameraProfileConfig model for named config overrides

* add profiles field to CameraConfig

* add active_profile field to FrigateConfig

Runtime-only field excluded from YAML serialization, tracks which
profile is currently active.

* add ProfileManager for profile activation and persistence

Handles snapshotting base configs, applying profile overrides via
deep_merge + apply_section_update, publishing ZMQ updates, and
persisting active profile to /config/.active_profile.

* add profile API endpoints (GET /profiles, GET/PUT /profile)

* add MQTT and dispatcher integration for profiles

- Subscribe to frigate/profile/set MQTT topic
- Publish profile/state and profiles/available on connect
- Add _on_profile_command handler to dispatcher
- Broadcast active profile state on WebSocket connect

* wire ProfileManager into app startup and FastAPI

- Create ProfileManager after dispatcher init
- Restore persisted profile on startup
- Pass dispatcher and profile_manager to FastAPI app

* add tests for invalid profile values and keys

Tests that Pydantic rejects: invalid field values (fps: "not_a_number"),
unknown section keys (ffmpeg in profile), invalid nested values, and
invalid profiles in full config parsing.

* formatting

* fix CameraLiveConfig JSON serialization error on profile activation

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

* consolidate

* add enabled field to camera profiles for enabling/disabling cameras

* add zones support to camera profiles

* add frontend profile types, color utility, and config save support

* add profile state management and save preview support

* add profileName prop to BaseSection for profile-aware config editing

* add profile section dropdown and wire into camera settings pages

* add per-profile camera enable/disable to Camera Management view

* add profiles summary page with card-based layout and fix backend zone comparison bug

* add active profile badge to settings toolbar

* i18n

* add red dot for any pending changes including profiles

* profile support for mask and zone editor

* fix hidden field validation errors caused by lodash wildcard and schema gaps

lodash unset does not support wildcard (*) segments, so hidden fields like
filters.*.mask were never stripped from form data, leaving null raw_coordinates
that fail RJSF anyOf validation. Add unsetWithWildcard helper and also strip
hidden fields from the JSON schema itself as defense-in-depth.

* add face_recognition and lpr to profile-eligible sections

* move profile dropdown from section panes to settings header

* add profiles enable toggle and improve empty state

* formatting

* tweaks

* tweak colors and switch

* fix profile save diff, masksAndZones delete, and config sync

* ui tweaks

* ensure profile manager gets updated config

* rename profile settings to ui settings

* refactor profilesview and add dots/border colors when overridden

* implement an update_config method for profile manager

* fix mask deletion

* more unique colors

* add top-level profiles config section with friendly names

* implement profile friendly names and improve profile UI

- Add ProfileDefinitionConfig type and profiles field to FrigateConfig
- Use ProfilesApiResponse type with friendly_name support throughout
- Replace Record<string, unknown> with proper JsonObject/JsonValue types
- Add profile creation form matching zone pattern (Zod + NameAndIdFields)
- Add pencil icon for renaming profile friendly names in ProfilesView
- Move Profiles menu item to first under Camera Configuration
- Add activity indicators on save/rename/delete buttons
- Display friendly names in CameraManagementView profile selector
- Fix duplicate colored dots in management profile dropdown
- Fix i18n namespace for overridden base config tooltips
- Move profile override deletion from dropdown trash icon to footer
  button with confirmation dialog, matching Reset to Global pattern
- Remove Add Profile from section header dropdown to prevent saving
  camera overrides before top-level profile definition exists
- Clean up newProfiles state after API profile deletion
- Refresh profiles SWR cache after saving profile definitions

* remove profile badge in settings and add profiles to main menu

* use icon only on mobile

* change color order

* docs

* show activity indicator on trash icon while deleting a profile

* tweak language

* immediately create profiles on backend instead of deferring to Save All

* hide restart-required fields when editing a profile section

fields that require a restart cannot take effect via profile switching,
so they are merged into hiddenFields when profileName is set

* show active profile indicator in desktop status bar

* fix profile config inheritance bug where Pydantic defaults override base values

The /config API was dumping profile overrides with model_dump() which included
all Pydantic defaults. When the frontend merged these over
the camera's base config, explicitly-set base values were
lost. Now profile overrides are re-dumped with exclude_unset=True so only
user-specified fields are returned.

Also fixes the Save All path generating spurious deletion markers for
restart-required fields that are hidden during profile
editing but not excluded from the raw data sanitization in
prepareSectionSavePayload.

* docs tweaks

* docs tweak

* formatting

* formatting

* fix typing

* fix test pollution

test_maintainer was injecting MagicMock() into sys.modules["frigate.config.camera.updater"] at module load time and never restoring it. When the profile tests later imported CameraConfigUpdateEnum and CameraConfigUpdateTopic from that module, they got mock objects instead of the real dataclass/enum, so equality comparisons always failed

* remove

* fix settings showing profile-merged values when editing base config

When a profile is active, the in-memory config contains effective
(profile-merged) values. The settings UI was displaying these merged
values even when the "Base Config" view was selected.

Backend: snapshot pre-profile base configs in ProfileManager and expose
them via a `base_config` key in the /api/config camera response when a
profile is active. The top-level sections continue to reflect the
effective running config.

Frontend: read from `base_config` when available in BaseSection,
useConfigOverride, useAllCameraOverrides, and prepareSectionSavePayload.
Include formData labels in Object/Audio switches widgets so that labels
added only by a profile override remain visible when editing that profile.

* use rasterized_mask as field

makes it easier to exclude from the schema with exclude=True
prevents leaking of the field when using model_dump for profiles

* fix zones

- Fix zone colors not matching across profiles by falling back to base zone color when profile zone data lacks a color field
- Use base_config for base-layer values in masks/zones view so profile-merged values don't pollute the base config editing view
- Handle zones separately in profile manager snapshot/restore since ZoneConfig requires special serialization (color as private attr, contour generation)
- Inherit base zone color and generate contours for profile zone overrides in profile manager

* formatting

* don't require restart for camera enabled change for profiles

* publish camera state when changing profiles

* formatting

* remove available profiles from mqtt

* improve typing
This commit is contained in:
Josh Hawkins
2026-03-19 09:47:57 -05:00
committed by GitHub
parent a05f35c747
commit c93dad9bd9
47 changed files with 4591 additions and 458 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/api/app.py
+66 -1
View File
@@ -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 (
@@ -154,6 +158,31 @@ def config(request: Request):
for zone_name, zone in config_obj.cameras[camera_name].zones.items():
camera_dict["zones"][zone_name]["color"] = zone.color
# Re-dump profile overrides with exclude_unset so that only
# explicitly-set fields are returned (not Pydantic defaults).
# Without this, the frontend merges defaults (e.g. threshold=30)
# over the camera's actual base values (e.g. threshold=20).
if camera.profiles:
for profile_name, profile_config in camera.profiles.items():
camera_dict.setdefault("profiles", {})[profile_name] = (
profile_config.model_dump(
mode="json", warnings="none", exclude_unset=True
)
)
# When a profile is active, the top-level camera sections contain
# profile-merged (effective) values. Include the original base
# configs so the frontend settings can display them separately.
if (
config_obj.active_profile is not None
and request.app.profile_manager is not None
):
base_sections = request.app.profile_manager.get_base_configs_for_api(
camera_name
)
if base_sections:
camera_dict["base_config"] = base_sections
# remove go2rtc stream passwords
go2rtc: dict[str, Any] = config_obj.go2rtc.model_dump(
mode="json", warnings="none", exclude_none=True
@@ -201,6 +230,39 @@ 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."""
@@ -589,6 +651,9 @@ def config_set(request: Request, body: AppConfigSetBody):
request.app.frigate_config = config
request.app.genai_manager.update_config(config)
if request.app.profile_manager is not None:
request.app.profile_manager.update_config(config)
if request.app.stats_emitter is not None:
request.app.stats_emitter.config = config
frigate/api/defs/request/app_body.py
+6
View File
@@ -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
+6
View File
@@ -29,11 +29,13 @@ from frigate.api import (
review,
)
from frigate.api.auth import get_jwt_secret, limiter, require_admin_by_default
from frigate.comms.dispatcher import Dispatcher
from frigate.comms.event_metadata_updater import (
EventMetadataPublisher,
)
from frigate.config import FrigateConfig
from frigate.config.camera.updater import CameraConfigUpdatePublisher
from frigate.config.profile_manager import ProfileManager
from frigate.debug_replay import DebugReplayManager
from frigate.embeddings import EmbeddingsContext
from frigate.genai import GenAIClientManager
@@ -69,6 +71,8 @@ def create_fastapi_app(
event_metadata_updater: EventMetadataPublisher,
config_publisher: CameraConfigUpdatePublisher,
replay_manager: DebugReplayManager,
dispatcher: Optional[Dispatcher] = None,
profile_manager: Optional[ProfileManager] = None,
enforce_default_admin: bool = True,
):
logger.info("Starting FastAPI app")
@@ -151,6 +155,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
+18
View File
@@ -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,
@@ -118,6 +119,7 @@ class FrigateApp:
self.ptz_metrics: dict[str, PTZMetrics] = {}
self.processes: dict[str, int] = {}
self.embeddings: Optional[EmbeddingsContext] = None
self.profile_manager: Optional[ProfileManager] = None
self.config = config
def ensure_dirs(self) -> None:
@@ -349,6 +351,19 @@ class FrigateApp:
comms,
)
def init_profile_manager(self) -> None:
self.profile_manager = ProfileManager(
self.config, self.inter_config_updater, self.dispatcher
)
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:
@@ -557,6 +572,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()
@@ -586,6 +602,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
+24
View File
@@ -16,6 +16,7 @@ from frigate.config.camera.updater import (
CameraConfigUpdateTopic,
)
from frigate.config.config import RuntimeFilterConfig, RuntimeMotionConfig
from frigate.config.profile_manager import ProfileManager
from frigate.const import (
CLEAR_ONGOING_REVIEW_SEGMENTS,
EXPIRE_AUDIO_ACTIVITY,
@@ -91,7 +92,9 @@ class Dispatcher:
}
self._global_settings_handlers: dict[str, Callable] = {
"notifications": self._on_global_notification_command,
"profile": self._on_profile_command,
}
self.profile_manager: Optional[ProfileManager] = None
for comm in self.comms:
comm.subscribe(self._receive)
@@ -298,6 +301,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 +564,22 @@ 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
+10
View File
@@ -163,6 +163,11 @@ class MqttClient(Communicator):
retain=True,
)
self.publish(
"profile/state",
self.config.active_profile or "none",
retain=True,
)
self.publish("available", "online", retain=True)
def on_mqtt_command(
@@ -289,6 +294,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
+7
View File
@@ -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
+44
View File
@@ -0,0 +1,44 @@
"""Camera profile configuration for named config overrides."""
from typing import Optional
from ..base import FrigateBaseModel
from ..classification import (
CameraFaceRecognitionConfig,
CameraLicensePlateRecognitionConfig,
)
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
from .zone import ZoneConfig
__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
face_recognition: Optional[CameraFaceRecognitionConfig] = None
lpr: Optional[CameraLicensePlateRecognitionConfig] = 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
zones: Optional[dict[str, ZoneConfig]] = None
frigate/config/camera/updater.py
+6
View File
@@ -27,6 +27,8 @@ class CameraConfigUpdateEnum(str, Enum):
review = "review"
review_genai = "review_genai"
semantic_search = "semantic_search" # for semantic search triggers
face_recognition = "face_recognition"
lpr = "lpr"
snapshots = "snapshots"
zones = "zones"
@@ -119,6 +121,10 @@ class CameraConfigUpdateSubscriber:
config.review.genai = updated_config
elif update_type == CameraConfigUpdateEnum.semantic_search:
config.semantic_search = updated_config
elif update_type == CameraConfigUpdateEnum.face_recognition:
config.face_recognition = updated_config
elif update_type == CameraConfigUpdateEnum.lpr:
config.lpr = updated_config
elif update_type == CameraConfigUpdateEnum.snapshots:
config.snapshots = updated_config
elif update_type == CameraConfigUpdateEnum.zones:
frigate/config/config.py
+25 -25
View File
@@ -12,7 +12,6 @@ from pydantic import (
Field,
TypeAdapter,
ValidationInfo,
field_serializer,
field_validator,
model_validator,
)
@@ -68,6 +67,7 @@ from .env import EnvVars
from .logger import LoggerConfig
from .mqtt import MqttConfig
from .network import NetworkingConfig
from .profile import ProfileDefinitionConfig
from .proxy import ProxyConfig
from .telemetry import TelemetryConfig
from .tls import TlsConfig
@@ -97,8 +97,7 @@ stream_info_retriever = StreamInfoRetriever()
class RuntimeMotionConfig(MotionConfig):
"""Runtime version of MotionConfig with rasterized masks."""
# The rasterized numpy mask (combination of all enabled masks)
rasterized_mask: np.ndarray = None
rasterized_mask: np.ndarray = Field(default=None, exclude=True)
def __init__(self, **config):
frame_shape = config.get("frame_shape", (1, 1))
@@ -144,24 +143,13 @@ class RuntimeMotionConfig(MotionConfig):
empty_mask[:] = 255
self.rasterized_mask = empty_mask
def dict(self, **kwargs):
ret = super().model_dump(**kwargs)
if "rasterized_mask" in ret:
ret.pop("rasterized_mask")
return ret
@field_serializer("rasterized_mask", when_used="json")
def serialize_rasterized_mask(self, value: Any, info):
return None
model_config = ConfigDict(arbitrary_types_allowed=True, extra="ignore")
class RuntimeFilterConfig(FilterConfig):
"""Runtime version of FilterConfig with rasterized masks."""
# The rasterized numpy mask (combination of all enabled masks)
rasterized_mask: Optional[np.ndarray] = None
rasterized_mask: Optional[np.ndarray] = Field(default=None, exclude=True)
def __init__(self, **config):
frame_shape = config.get("frame_shape", (1, 1))
@@ -225,16 +213,6 @@ class RuntimeFilterConfig(FilterConfig):
else:
self.rasterized_mask = None
def dict(self, **kwargs):
ret = super().model_dump(**kwargs)
if "rasterized_mask" in ret:
ret.pop("rasterized_mask")
return ret
@field_serializer("rasterized_mask", when_used="json")
def serialize_rasterized_mask(self, value: Any, info):
return None
model_config = ConfigDict(arbitrary_types_allowed=True, extra="ignore")
@@ -561,6 +539,19 @@ class FrigateConfig(FrigateBaseModel):
description="Configuration for named camera groups used to organize cameras in the UI.",
)
profiles: Dict[str, ProfileDefinitionConfig] = Field(
default_factory=dict,
title="Profiles",
description="Named profile definitions with friendly names. Camera profiles must reference names defined here.",
)
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
@@ -910,6 +901,15 @@ class FrigateConfig(FrigateBaseModel):
verify_objects_track(camera_config, labelmap_objects)
verify_lpr_and_face(self, camera_config)
# Validate camera profiles reference top-level profile definitions
for cam_name, cam_config in self.cameras.items():
for profile_name in cam_config.profiles:
if profile_name not in self.profiles:
raise ValueError(
f"Camera '{cam_name}' references profile '{profile_name}' "
f"which is not defined in the top-level 'profiles' section"
)
# set names on classification configs
for name, config in self.classification.custom.items():
config.name = name
frigate/config/profile.py
+20
View File
@@ -0,0 +1,20 @@
"""Top-level profile definition configuration."""
from pydantic import Field
from .base import FrigateBaseModel
__all__ = ["ProfileDefinitionConfig"]
class ProfileDefinitionConfig(FrigateBaseModel):
"""Defines a named profile with a human-readable display name.
The dict key is the machine name used internally; friendly_name
is the label shown in the UI and API responses.
"""
friendly_name: str = Field(
title="Friendly name",
description="Display name for this profile shown in the UI.",
)
frigate/config/profile_manager.py
+334
View File
@@ -0,0 +1,334 @@
"""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,
}
frigate/test/test_maintainer.py
+18 -5
View File
@@ -2,16 +2,29 @@ import sys
import unittest
from unittest.mock import MagicMock, patch
# Mock complex imports before importing maintainer
sys.modules["frigate.comms.inter_process"] = MagicMock()
sys.modules["frigate.comms.detections_updater"] = MagicMock()
sys.modules["frigate.comms.recordings_updater"] = MagicMock()
sys.modules["frigate.config.camera.updater"] = MagicMock()
# Mock complex imports before importing maintainer, saving originals so we can
# restore them after import and avoid polluting sys.modules for other tests.
_MOCKED_MODULES = [
"frigate.comms.inter_process",
"frigate.comms.detections_updater",
"frigate.comms.recordings_updater",
"frigate.config.camera.updater",
]
_originals = {name: sys.modules.get(name) for name in _MOCKED_MODULES}
for name in _MOCKED_MODULES:
sys.modules[name] = MagicMock()
# Now import the class under test
from frigate.config import FrigateConfig # noqa: E402
from frigate.record.maintainer import RecordingMaintainer # noqa: E402
# Restore original modules (or remove mock if there was no original)
for name, orig in _originals.items():
if orig is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = orig
class TestMaintainer(unittest.IsolatedAsyncioTestCase):
async def test_move_files_survives_bad_filename(self):
frigate/test/test_profiles.py
+629
View File
@@ -0,0 +1,629 @@
"""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 import ProfileDefinitionConfig
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_zones_field(self):
"""Profile with zones override."""
profile = CameraProfileConfig(
zones={
"driveway": {
"coordinates": "0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9",
"objects": ["car"],
}
}
)
assert profile.zones is not None
assert "driveway" in profile.zones
def test_zones_default_none(self):
"""Zones defaults to None when not set."""
profile = CameraProfileConfig()
assert profile.zones 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"},
"profiles": {
"armed": {"friendly_name": "Armed"},
},
"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)
def test_undefined_profile_reference_rejected(self):
"""Camera referencing a profile not defined in top-level profiles is rejected."""
from pydantic import ValidationError
config_data = {
"mqtt": {"host": "mqtt"},
"profiles": {
"armed": {"friendly_name": "Armed"},
},
"cameras": {
"front": {
"ffmpeg": {
"inputs": [
{
"path": "rtsp://10.0.0.1:554/video",
"roles": ["detect"],
}
]
},
"detect": {"height": 1080, "width": 1920, "fps": 5},
"profiles": {
"nonexistent": {
"detect": {"enabled": False},
},
},
},
},
}
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"},
"profiles": {
"armed": {"friendly_name": "Armed"},
"disarmed": {"friendly_name": "Disarmed"},
},
"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"},
"profiles": {
"armed": {"friendly_name": "Armed"},
"disarmed": {"friendly_name": "Disarmed"},
},
"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 come from top-level profile definitions."""
profiles = self.manager.get_available_profiles()
assert len(profiles) == 2
names = [p["name"] for p in profiles]
assert "armed" in names
assert "disarmed" in names
# Verify friendly_name is included
armed = next(p for p in profiles if p["name"] == "armed")
assert armed["friendly_name"] == "Armed"
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 defined" 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."""
self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away")
self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
enabled=False
)
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."""
self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away")
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_activate_profile_adds_zone(self, mock_persist):
"""Profile with zones adds/overrides zones on camera."""
from frigate.config.camera.zone import ZoneConfig
self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away")
self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
zones={
"driveway": ZoneConfig(
coordinates="0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9",
objects=["car"],
)
}
)
self.manager = ProfileManager(self.config, self.mock_updater)
assert "driveway" not in self.config.cameras["front"].zones
err = self.manager.activate_profile("away")
assert err is None
assert "driveway" in self.config.cameras["front"].zones
@patch.object(ProfileManager, "_persist_active_profile")
def test_deactivate_restores_zones(self, mock_persist):
"""Deactivating a profile restores base zones."""
from frigate.config.camera.zone import ZoneConfig
self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away")
self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
zones={
"driveway": ZoneConfig(
coordinates="0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9",
objects=["car"],
)
}
)
self.manager = ProfileManager(self.config, self.mock_updater)
self.manager.activate_profile("away")
assert "driveway" in self.config.cameras["front"].zones
self.manager.activate_profile(None)
assert "driveway" not in self.config.cameras["front"].zones
@patch.object(ProfileManager, "_persist_active_profile")
def test_zones_zmq_published(self, mock_persist):
"""ZMQ update is published for zones change."""
from frigate.config.camera.updater import (
CameraConfigUpdateEnum,
CameraConfigUpdateTopic,
)
from frigate.config.camera.zone import ZoneConfig
self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away")
self.config.cameras["front"].profiles["away"] = CameraProfileConfig(
zones={
"driveway": ZoneConfig(
coordinates="0.1,0.1,0.9,0.1,0.9,0.9,0.1,0.9",
objects=["car"],
)
}
)
self.manager = ProfileManager(self.config, self.mock_updater)
self.mock_updater.reset_mock()
self.manager.activate_profile("away")
zones_calls = [
call
for call in self.mock_updater.publish_update.call_args_list
if call[0][0]
== CameraConfigUpdateTopic(CameraConfigUpdateEnum.zones, "front")
]
assert len(zones_calls) == 1
@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.updater import (
CameraConfigUpdateEnum,
CameraConfigUpdateTopic,
)
self.config.profiles["away"] = ProfileDefinitionConfig(friendly_name="Away")
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 with friendly names."""
info = self.manager.get_profile_info()
assert "profiles" in info
assert "active_profile" in info
assert info["active_profile"] is None
names = [p["name"] for p in info["profiles"]]
assert "armed" in names
assert "disarmed" in names
@patch.object(ProfileManager, "_persist_active_profile")
def test_base_configs_for_api_unchanged_after_activation(self, mock_persist):
"""API base configs reflect pre-profile values after activation."""
base_track = self.config.cameras["front"].objects.track[:]
assert base_track == ["person"]
self.manager.activate_profile("armed")
# In-memory config has the profile-merged values
assert self.config.cameras["front"].objects.track == [
"person",
"car",
"package",
]
# But the API base configs still return the original base values
api_base = self.manager.get_base_configs_for_api("front")
assert "objects" in api_base
assert api_base["objects"]["track"] == ["person"]
def test_base_configs_for_api_are_json_serializable(self):
"""API base configs are JSON-serializable (mode='json')."""
import json
api_base = self.manager.get_base_configs_for_api("front")
# Should not raise
json.dumps(api_base)
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()
frigate/util/config.py
+2 -4
View File
@@ -717,7 +717,7 @@ def apply_section_update(camera_config, section: str, update: dict) -> Optional[
if section == "motion":
merged = deep_merge(
current.model_dump(exclude_unset=True, exclude={"rasterized_mask"}),
current.model_dump(exclude_unset=True),
update,
override=True,
)
@@ -727,9 +727,7 @@ def apply_section_update(camera_config, section: str, update: dict) -> Optional[
elif section == "objects":
merged = deep_merge(
current.model_dump(
exclude={"filters": {"__all__": {"rasterized_mask"}}}
),
current.model_dump(),
update,
override=True,
)