add zones support to camera profiles

This commit is contained in:
Josh Hawkins 2026-03-06 16:09:35 -06:00
parent d15fc4e58e
commit 60930e50c2
3 changed files with 123 additions and 1 deletions

View File

@ -12,6 +12,7 @@ from .objects import ObjectConfig
from .record import RecordConfig from .record import RecordConfig
from .review import ReviewConfig from .review import ReviewConfig
from .snapshots import SnapshotsConfig from .snapshots import SnapshotsConfig
from .zone import ZoneConfig
__all__ = ["CameraProfileConfig"] __all__ = ["CameraProfileConfig"]
@ -34,3 +35,4 @@ class CameraProfileConfig(FrigateBaseModel):
record: Optional[RecordConfig] = None record: Optional[RecordConfig] = None
review: Optional[ReviewConfig] = None review: Optional[ReviewConfig] = None
snapshots: Optional[SnapshotsConfig] = None snapshots: Optional[SnapshotsConfig] = None
zones: Optional[dict[str, ZoneConfig]] = None

View File

@ -1,5 +1,6 @@
"""Profile manager for activating/deactivating named config profiles.""" """Profile manager for activating/deactivating named config profiles."""
import copy
import logging import logging
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
@ -9,6 +10,7 @@ from frigate.config.camera.updater import (
CameraConfigUpdatePublisher, CameraConfigUpdatePublisher,
CameraConfigUpdateTopic, CameraConfigUpdateTopic,
) )
from frigate.config.camera.zone import ZoneConfig
from frigate.const import CONFIG_DIR from frigate.const import CONFIG_DIR
from frigate.util.builtin import deep_merge from frigate.util.builtin import deep_merge
from frigate.util.config import apply_section_update from frigate.util.config import apply_section_update
@ -44,13 +46,15 @@ class ProfileManager:
self.config_updater = config_updater self.config_updater = config_updater
self._base_configs: dict[str, dict[str, dict]] = {} self._base_configs: dict[str, dict[str, dict]] = {}
self._base_enabled: dict[str, bool] = {} self._base_enabled: dict[str, bool] = {}
self._base_zones: dict[str, dict[str, ZoneConfig]] = {}
self._snapshot_base_configs() self._snapshot_base_configs()
def _snapshot_base_configs(self) -> None: def _snapshot_base_configs(self) -> None:
"""Snapshot each camera's current section configs and enabled state.""" """Snapshot each camera's current section configs, enabled, and zones."""
for cam_name, cam_config in self.config.cameras.items(): for cam_name, cam_config in self.config.cameras.items():
self._base_configs[cam_name] = {} self._base_configs[cam_name] = {}
self._base_enabled[cam_name] = cam_config.enabled self._base_enabled[cam_name] = cam_config.enabled
self._base_zones[cam_name] = copy.deepcopy(cam_config.zones)
for section in PROFILE_SECTION_UPDATES: for section in PROFILE_SECTION_UPDATES:
section_config = getattr(cam_config, section, None) section_config = getattr(cam_config, section, None)
if section_config is not None: if section_config is not None:
@ -105,6 +109,12 @@ class ProfileManager:
cam_config.enabled = base_enabled cam_config.enabled = base_enabled
changed.setdefault(cam_name, set()).add("enabled") changed.setdefault(cam_name, set()).add("enabled")
# Restore zones
base_zones = self._base_zones.get(cam_name)
if base_zones is not None and cam_config.zones != base_zones:
cam_config.zones = copy.deepcopy(base_zones)
changed.setdefault(cam_name, set()).add("zones")
# Restore section configs # Restore section configs
base = self._base_configs.get(cam_name, {}) base = self._base_configs.get(cam_name, {})
for section in PROFILE_SECTION_UPDATES: for section in PROFILE_SECTION_UPDATES:
@ -136,6 +146,14 @@ class ProfileManager:
cam_config.enabled = profile.enabled cam_config.enabled = profile.enabled
changed.setdefault(cam_name, set()).add("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)
cam_config.zones = merged_zones
changed.setdefault(cam_name, set()).add("zones")
base = self._base_configs.get(cam_name, {}) base = self._base_configs.get(cam_name, {})
for section in PROFILE_SECTION_UPDATES: for section in PROFILE_SECTION_UPDATES:
@ -175,6 +193,15 @@ class ProfileManager:
) )
continue 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) update_enum = PROFILE_SECTION_UPDATES.get(section)
if update_enum is None: if update_enum is None:
continue continue

View File

@ -70,6 +70,24 @@ class TestCameraProfileConfig(unittest.TestCase):
profile = CameraProfileConfig() profile = CameraProfileConfig()
assert profile.enabled is None 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): def test_none_sections_not_in_dump(self):
"""Sections left as None should not appear in exclude_unset dump.""" """Sections left as None should not appear in exclude_unset dump."""
profile = CameraProfileConfig(detect={"enabled": False}) profile = CameraProfileConfig(detect={"enabled": False})
@ -380,6 +398,81 @@ class TestProfileManager(unittest.TestCase):
self.manager.activate_profile(None) self.manager.activate_profile(None)
assert self.config.cameras["front"].enabled is True 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.profile import CameraProfileConfig
from frigate.config.camera.zone import ZoneConfig
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.profile import CameraProfileConfig
from frigate.config.camera.zone import ZoneConfig
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.profile import CameraProfileConfig
from frigate.config.camera.updater import (
CameraConfigUpdateEnum,
CameraConfigUpdateTopic,
)
from frigate.config.camera.zone import ZoneConfig
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") @patch.object(ProfileManager, "_persist_active_profile")
def test_enabled_zmq_published(self, mock_persist): def test_enabled_zmq_published(self, mock_persist):
"""ZMQ update is published for enabled state change.""" """ZMQ update is published for enabled state change."""