Improve profile information

This commit is contained in:
Nicolas Mowen 2026-03-31 08:58:52 -06:00
parent b821420dee
commit 028238bd71
3 changed files with 163 additions and 38 deletions

View File

@ -1,7 +1,9 @@
"""Profile manager for activating/deactivating named config profiles.""" """Profile manager for activating/deactivating named config profiles."""
import copy import copy
import json
import logging import logging
from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Optional from typing import Optional
@ -32,7 +34,7 @@ PROFILE_SECTION_UPDATES: dict[str, CameraConfigUpdateEnum] = {
"zones": CameraConfigUpdateEnum.zones, "zones": CameraConfigUpdateEnum.zones,
} }
PERSISTENCE_FILE = Path(CONFIG_DIR) / ".active_profile" PERSISTENCE_FILE = Path(CONFIG_DIR) / ".profiles"
class ProfileManager: class ProfileManager:
@ -291,25 +293,36 @@ class ProfileManager:
) )
def _persist_active_profile(self, profile_name: Optional[str]) -> None: def _persist_active_profile(self, profile_name: Optional[str]) -> None:
"""Persist the active profile name to disk.""" """Persist the active profile state to disk as JSON."""
try: try:
if profile_name is None: data = self._load_persisted_data()
PERSISTENCE_FILE.unlink(missing_ok=True) data["active"] = profile_name
else: if profile_name is not None:
PERSISTENCE_FILE.write_text(profile_name) data.setdefault("last_activated", {})[profile_name] = datetime.now(
timezone.utc
).timestamp()
PERSISTENCE_FILE.write_text(json.dumps(data))
except OSError: except OSError:
logger.exception("Failed to persist active profile") logger.exception("Failed to persist active profile")
@staticmethod @staticmethod
def load_persisted_profile() -> Optional[str]: def _load_persisted_data() -> dict:
"""Load the persisted active profile name from disk.""" """Load the full persisted profile data from disk."""
try: try:
if PERSISTENCE_FILE.exists(): if PERSISTENCE_FILE.exists():
name = PERSISTENCE_FILE.read_text().strip() raw = PERSISTENCE_FILE.read_text().strip()
return name if name else None if raw:
except OSError: return json.loads(raw)
logger.exception("Failed to load persisted profile") except (OSError, json.JSONDecodeError):
return None logger.exception("Failed to load persisted profile data")
return {"active": None, "last_activated": {}}
@staticmethod
def load_persisted_profile() -> Optional[str]:
"""Load the persisted active profile name from disk."""
data = ProfileManager._load_persisted_data()
name = data.get("active")
return name if name else None
def get_base_configs_for_api(self, camera_name: str) -> dict[str, dict]: def get_base_configs_for_api(self, camera_name: str) -> dict[str, dict]:
"""Return base (pre-profile) section configs for a camera. """Return base (pre-profile) section configs for a camera.
@ -328,7 +341,9 @@ class ProfileManager:
def get_profile_info(self) -> dict: def get_profile_info(self) -> dict:
"""Get profile state info for API responses.""" """Get profile state info for API responses."""
data = self._load_persisted_data()
return { return {
"profiles": self.get_available_profiles(), "profiles": self.get_available_profiles(),
"active_profile": self.config.active_profile, "active_profile": self.config.active_profile,
"last_activated": data.get("last_activated", {}),
} }

View File

@ -1,5 +1,6 @@
"""Tests for the profiles system.""" """Tests for the profiles system."""
import json
import os import os
import tempfile import tempfile
import unittest import unittest
@ -549,10 +550,17 @@ class TestProfileManager(unittest.TestCase):
def test_get_profile_info(self): def test_get_profile_info(self):
"""Profile info returns correct structure with friendly names.""" """Profile info returns correct structure with friendly names."""
info = self.manager.get_profile_info() with patch.object(
ProfileManager,
"_load_persisted_data",
return_value={"active": None, "last_activated": {}},
):
info = self.manager.get_profile_info()
assert "profiles" in info assert "profiles" in info
assert "active_profile" in info assert "active_profile" in info
assert "last_activated" in info
assert info["active_profile"] is None assert info["active_profile"] is None
assert info["last_activated"] == {}
names = [p["name"] for p in info["profiles"]] names = [p["name"] for p in info["profiles"]]
assert "armed" in names assert "armed" in names
assert "disarmed" in names assert "disarmed" in names
@ -590,33 +598,22 @@ class TestProfilePersistence(unittest.TestCase):
"""Test profile persistence to disk.""" """Test profile persistence to disk."""
def test_persist_and_load(self): def test_persist_and_load(self):
"""Active profile name can be persisted and loaded.""" """Active profile name can be persisted and loaded via JSON."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: data = {"active": "armed", "last_activated": {"armed": 1700000000.0}}
temp_path = f.name with patch.object(
ProfileManager,
try: "_load_persisted_data",
from pathlib import Path return_value=data,
):
path = Path(temp_path) result = ProfileManager.load_persisted_profile()
path.write_text("armed") assert result == "armed"
loaded = path.read_text().strip()
assert loaded == "armed"
finally:
os.unlink(temp_path)
def test_load_empty_file(self): def test_load_empty_file(self):
"""Empty persistence file returns None.""" """Empty persistence file returns None."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: with patch.object(type(PERSISTENCE_FILE), "exists", return_value=True):
f.write("") with patch.object(type(PERSISTENCE_FILE), "read_text", return_value=""):
temp_path = f.name result = ProfileManager.load_persisted_profile()
assert result is None
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): def test_load_missing_file(self):
"""Missing persistence file returns None.""" """Missing persistence file returns None."""
@ -624,6 +621,118 @@ class TestProfilePersistence(unittest.TestCase):
result = ProfileManager.load_persisted_profile() result = ProfileManager.load_persisted_profile()
assert result is None assert result is None
def test_load_persisted_data_valid_json(self):
"""Valid JSON file is loaded correctly."""
data = {"active": "home", "last_activated": {"home": 1700000000.0}}
with patch.object(type(PERSISTENCE_FILE), "exists", return_value=True):
with patch.object(
type(PERSISTENCE_FILE),
"read_text",
return_value=json.dumps(data),
):
result = ProfileManager._load_persisted_data()
assert result == data
def test_load_persisted_data_invalid_json(self):
"""Invalid JSON returns default structure."""
with patch.object(type(PERSISTENCE_FILE), "exists", return_value=True):
with patch.object(
type(PERSISTENCE_FILE), "read_text", return_value="not json"
):
result = ProfileManager._load_persisted_data()
assert result == {"active": None, "last_activated": {}}
def test_load_persisted_data_missing_file(self):
"""Missing file returns default structure."""
with patch.object(type(PERSISTENCE_FILE), "exists", return_value=False):
result = ProfileManager._load_persisted_data()
assert result == {"active": None, "last_activated": {}}
def test_persist_records_timestamp(self):
"""Persisting a profile records the activation timestamp."""
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": {"enabled": True}}},
},
},
}
if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR):
os.makedirs(MODEL_CACHE_DIR)
config = FrigateConfig(**config_data)
manager = ProfileManager(config, MagicMock())
written_data = {}
def mock_write(content):
written_data.update(json.loads(content))
with patch.object(
ProfileManager,
"_load_persisted_data",
return_value={"active": None, "last_activated": {}},
):
with patch.object(type(PERSISTENCE_FILE), "write_text", mock_write):
manager._persist_active_profile("armed")
assert written_data["active"] == "armed"
assert "armed" in written_data["last_activated"]
assert isinstance(written_data["last_activated"]["armed"], float)
def test_persist_deactivate_keeps_timestamps(self):
"""Deactivating sets active to None but preserves last_activated."""
existing = {
"active": "armed",
"last_activated": {"armed": 1700000000.0},
}
written_data = {}
def mock_write(content):
written_data.update(json.loads(content))
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": {"enabled": True}}},
},
},
}
if not os.path.exists(MODEL_CACHE_DIR) and not os.path.islink(MODEL_CACHE_DIR):
os.makedirs(MODEL_CACHE_DIR)
config = FrigateConfig(**config_data)
manager = ProfileManager(config, MagicMock())
with patch.object(
ProfileManager, "_load_persisted_data", return_value=existing
):
with patch.object(type(PERSISTENCE_FILE), "write_text", mock_write):
manager._persist_active_profile(None)
assert written_data["active"] is None
assert written_data["last_activated"]["armed"] == 1700000000.0
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()

View File

@ -14,6 +14,7 @@ export type ProfileInfo = {
export type ProfilesApiResponse = { export type ProfilesApiResponse = {
profiles: ProfileInfo[]; profiles: ProfileInfo[];
active_profile: string | null; active_profile: string | null;
last_activated: Record<string, number>;
}; };
export type ProfileState = { export type ProfileState = {