michal/frigate-camera-control-bridge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Refresh snapshots after light actions

Browse Source
This commit is contained in:
michal
2026-08-03 01:28:56 +02:00
parent 127d07dc6b
commit 4ea737024b
14 changed files with 177 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2
View File
@@ -36,6 +36,8 @@ through Home Assistant MQTT Discovery plus an emergency HTTP fallback GUI.
| --- | --- |
| `FRIGATE_NAME` | Instance name, for example `frigate-main`. |
| `FRIGATE_URL` | Frigate API URL, for example `http://frigate-main.example.test:5000`. |
| `FRIGATE_PUBLIC_URL` | Optional public GUI URL used by camera stream links, for example a reverse proxy URL. If omitted, stream links use `FRIGATE_URL`. |
| `FRIGATE_CAMERA_PATH_TEMPLATE` | Optional camera page path or full URL template, defaults to `/#{camera_quoted}`. Available placeholders: `{camera}`, `{camera_quoted}`, `{frigate_url}`, `{frigate_public_url}`. |
| `MQTT_HOST`, `MQTT_PORT` | MQTT broker. |
| `MQTT_USER`, `MQTT_PASSWORD` | MQTT credentials. |
| `DEFAULT_CAMERA_USER`, `DEFAULT_CAMERA_PASSWORD` | Default credentials for cameras discovered from Frigate. |
examples/docker-compose/camera-control.env.example
+2
View File
@@ -8,6 +8,8 @@ DEFAULT_CAMERA_PASSWORD=change-me
HTTP_PORT=5011
UI_LANGUAGE=en
FRIGATE_PUBLIC_URL=
FRIGATE_CAMERA_PATH_TEMPLATE=/#{camera_quoted}
WAIT_FOR_FRIGATE=true
FRIGATE_RETRY_SECONDS=5
FRIGATE_STARTUP_WAIT_SECONDS=0
examples/docker-compose/compose.sidecar-only.yaml
+2
View File
@@ -11,6 +11,8 @@ services:
FRIGATE_NAME: frigate-main
# Set this to the Frigate container DNS name or another reachable URL.
FRIGATE_URL: http://frigate:5000
# Optional public Frigate GUI URL for stream links shown in the bridge GUI.
FRIGATE_PUBLIC_URL: ""
BASE_TOPIC: frigate_camera_control/frigate-main
ports:
- "5011:5011"
examples/docker-compose/compose.yaml
+2
View File
@@ -26,6 +26,8 @@ services:
# Because both containers share the same Compose network, the sidecar can
# reach Frigate by service name.
FRIGATE_URL: http://frigate:5000
# Optional public Frigate GUI URL for stream links shown in the bridge GUI.
FRIGATE_PUBLIC_URL: ""
BASE_TOPIC: frigate_camera_control/frigate-main
ports:
- "5011:5011"
examples/kubernetes/same-pod.yaml
+4
View File
@@ -25,6 +25,10 @@ spec:
value: frigate-main
- name: FRIGATE_URL
value: http://127.0.0.1:5000
- name: FRIGATE_PUBLIC_URL
value: ""
- name: FRIGATE_CAMERA_PATH_TEMPLATE
value: /#{camera_quoted}
- name: MQTT_HOST
value: mqtt.default.svc.cluster.local
- name: BASE_TOPIC
examples/kubernetes/two-pods.yaml
+4
View File
@@ -53,6 +53,10 @@ spec:
value: frigate-main
- name: FRIGATE_URL
value: http://frigate:5000
- name: FRIGATE_PUBLIC_URL
value: ""
- name: FRIGATE_CAMERA_PATH_TEMPLATE
value: /#{camera_quoted}
- name: MQTT_HOST
value: mqtt.default.svc.cluster.local
- name: BASE_TOPIC
examples/podman-quadlet/frigate-camera-control-bridge.env.example
+2
View File
@@ -12,6 +12,8 @@ DEFAULT_CAMERA_PASSWORD=change-me
HTTP_PORT=5011
UI_LANGUAGE=en
FRIGATE_PUBLIC_URL=
FRIGATE_CAMERA_PATH_TEMPLATE=/#{camera_quoted}
WAIT_FOR_FRIGATE=true
FRIGATE_RETRY_SECONDS=5
FRIGATE_STARTUP_WAIT_SECONDS=0
frigate_camera_control_bridge/app.py
+34 -3
View File
@@ -235,6 +235,11 @@ def config_from_env():
"snapshot_retry_seconds": int(os.environ.get("SNAPSHOT_RETRY_SECONDS", "600")),
"snapshot_refresh_seconds": int(os.environ.get("SNAPSHOT_REFRESH_SECONDS", "3600")),
"snapshot_live_on_gui": env_bool("SNAPSHOT_LIVE_ON_GUI", True),
"frigate_public_url": os.environ.get("FRIGATE_PUBLIC_URL"),
"frigate_camera_path_template": os.environ.get(
"FRIGATE_CAMERA_PATH_TEMPLATE",
"/#{camera_quoted}",
),
"frigate_refresh_seconds": int(os.environ.get("FRIGATE_REFRESH_SECONDS", "300")),
"auto_off_siren_seconds": int(os.environ.get("AUTO_OFF_SIREN_SECONDS", "180")),
"auto_off_red_blue_seconds": int(os.environ.get("AUTO_OFF_RED_BLUE_SECONDS", "300")),
@@ -333,6 +338,7 @@ class Action:
self.state = None
self.auto_off_seconds = int(cfg.get("auto_off_seconds") or 0)
self.auto_off_at = None
self.snapshot_version = 0
def display_name(self, lang):
names = self.cfg.get("name_i18n") or {}
@@ -714,6 +720,7 @@ class CameraControlBridge:
continue
action.state = old.state
action.auto_off_at = old.auto_off_at
action.snapshot_version = old.snapshot_version
def initialize_action_states(self):
for action in self.actions.values():
@@ -1181,12 +1188,17 @@ class CameraControlBridge:
"configuration_url": item["frigate_url"],
}
@staticmethod
def snapshot_config(item):
return {
def snapshot_config(self, item):
config = {
"frigate_url": item["frigate_url"],
"camera": item["camera"],
}
defaults = self.config.get("defaults", {})
if defaults.get("frigate_public_url"):
config["public_url"] = defaults["frigate_public_url"]
if defaults.get("frigate_camera_path_template"):
config["camera_path_template"] = defaults["frigate_camera_path_template"]
return config
@staticmethod
def xml_text(xml, tag):
@@ -1243,6 +1255,23 @@ class CameraControlBridge:
LOG.info("snapshot unavailable for %s: %s", action.id, exc)
return ok
@staticmethod
def should_refresh_snapshot_after_command(action, command):
if command.lower() not in ("on", "off", "safe_off"):
return False
return action.cfg.get("name_key") in ("action.manual_white", "action.red_blue")
def invalidate_action_snapshot(self, action):
url = self.snapshot_url(action)
if not url:
return
with self.action_lock:
self.snapshot_cache.pop(url, None)
version = int(time.time() * 1000)
for item in self.actions.values():
if self.snapshot_url(item) == url:
item.snapshot_version = version
def snapshot_url(self, action):
snapshot = action.cfg.get("snapshot")
if not snapshot:
@@ -1390,6 +1419,8 @@ class CameraControlBridge:
self.schedule_auto_off(action)
elif command in ("off", "safe_off"):
self.cancel_auto_off(action.id)
if self.should_refresh_snapshot_after_command(action, command):
self.invalidate_action_snapshot(action)
if publish:
try:
self.publish_result(action, command, result)
frigate_camera_control_bridge/static/app.css
+39
View File
@@ -302,6 +302,13 @@ select {
min-width: 0;
}
.card-title-row {
display: grid;
grid-template-columns: minmax(0, 1fr) 34px;
gap: 8px;
align-items: start;
}
h2 {
margin: 0 0 8px;
font-size: 1.02rem;
@@ -310,6 +317,38 @@ h2 {
overflow-wrap: anywhere;
}
.stream-link {
display: grid;
width: 34px;
height: 34px;
place-items: center;
border: 1px solid var(--line);
border-radius: 6px;
background: #ffffff;
color: #1f3a5f;
text-decoration: none;
}
.stream-link:hover {
border-color: var(--blue);
color: var(--blue-dark);
}
.stream-link:focus-visible {
outline: 3px solid rgba(29, 78, 216, 0.25);
outline-offset: 2px;
}
.stream-link svg {
width: 18px;
height: 18px;
fill: none;
stroke: currentColor;
stroke-linecap: round;
stroke-linejoin: round;
stroke-width: 2;
}
.camera-count {
display: inline-flex;
color: var(--muted);
frigate_camera_control_bridge/ui.py
+47 -1
View File
@@ -149,7 +149,8 @@ def snapshot_markup_for_group(group, lang, token_query):
if action.cfg.get("snapshot"):
snapshot_url = (
f"/snapshot/{urllib.parse.quote(action.id)}.jpg"
f"?lang={urllib.parse.quote(lang)}{token_query}"
f"?lang={urllib.parse.quote(lang)}"
f"&v={int(getattr(action, 'snapshot_version', 0) or 0)}{token_query}"
)
snapshot_markup = (
'<div class="media">'
@@ -162,6 +163,48 @@ def snapshot_markup_for_group(group, lang, token_query):
return snapshot_markup
def camera_stream_url(action):
snapshot = action.cfg.get("snapshot") or {}
base_url = snapshot.get("public_url") or snapshot.get("frigate_url")
camera = snapshot.get("camera")
if not base_url or not camera:
return None
template = snapshot.get("camera_path_template") or "/#{camera_quoted}"
values = {
"camera": camera,
"camera_quoted": urllib.parse.quote(camera, safe=""),
"frigate_url": snapshot.get("frigate_url", "").rstrip("/"),
"frigate_public_url": base_url.rstrip("/"),
}
try:
link = template.format(**values)
except KeyError:
link = template
if urllib.parse.urlsplit(link).scheme:
return link
return f"{base_url.rstrip('/')}/{link.lstrip('/')}"
def camera_stream_link(group, lang):
action = next((item for item in group["actions"] if item.cfg.get("snapshot")), None)
if not action:
return ""
url = camera_stream_url(action)
if not url:
return ""
label = text_for(lang, "link.open_stream")
return (
f'<a class="stream-link" href="{html.escape(url)}" target="_blank" rel="noreferrer" '
f'title="{html.escape(label)}" aria-label="{html.escape(label)}">'
'<svg viewBox="0 0 24 24" aria-hidden="true">'
'<path d="M15 3h6v6"></path>'
'<path d="M10 14 21 3"></path>'
'<path d="M21 13v5a3 3 0 0 1-3 3H6a3 3 0 0 1-3-3V6a3 3 0 0 1 3-3h5"></path>'
"</svg>"
"</a>"
)
def render_action_control(action, lang, token_query):
display_name = action_label(action, lang)
@@ -213,7 +256,10 @@ def render_group_card(group, lang, token_query):
f"{snapshot_markup_for_group(group, lang, token_query)}"
'<div class="card-body">'
'<div class="card-main">'
'<div class="card-title-row">'
f"<h2>{html.escape(group['name'])}</h2>"
f"{camera_stream_link(group, lang)}"
"</div>"
f'<span class="camera-count">{html.escape(text_for(lang, "camera.controls_count", count=len(group["actions"])))}</span>'
"</div>"
f'<div class="control-stack">{controls}</div>'
locale/en/LC_MESSAGES/frigate_camera_control_bridge.po
+3
View File
@@ -95,6 +95,9 @@ msgstr "PRESS"
msgid "button.refresh_cameras"
msgstr "Refresh cameras"
msgid "link.open_stream"
msgstr "Open Frigate stream"
msgid "refresh.working"
msgstr "Refreshing cameras..."
locale/pl/LC_MESSAGES/frigate_camera_control_bridge.po
+3
View File
@@ -95,6 +95,9 @@ msgstr "URUCHOM"
msgid "button.refresh_cameras"
msgstr "Odswiez kamery"
msgid "link.open_stream"
msgstr "Otworz stream we Frigate"
msgid "refresh.working"
msgstr "Odswiezanie kamer..."
tests/test_refresh_summary.py
+13 -1
View File
@@ -4,10 +4,12 @@ from frigate_camera_control_bridge.app import CameraControlBridge
class DummyAction:
def __init__(self, action_id, camera=None, kind="hikvision_colorvu"):
def __init__(self, action_id, camera=None, kind="hikvision_colorvu", name_key=None):
self.id = action_id
self.kind = kind
self.cfg = {}
if name_key:
self.cfg["name_key"] = name_key
if camera:
self.cfg["snapshot"] = {
"frigate_url": "http://frigate.example.test:5000",
@@ -52,6 +54,16 @@ class RefreshSummaryTest(unittest.TestCase):
self.assertEqual(message, "No camera changes found.")
def test_light_actions_refresh_snapshot_after_on_off(self):
white = DummyAction("white", "one", name_key="action.manual_white")
red_blue = DummyAction("red_blue", "one", name_key="action.red_blue")
siren = DummyAction("siren", "one", name_key="action.siren")
self.assertTrue(CameraControlBridge.should_refresh_snapshot_after_command(white, "on"))
self.assertTrue(CameraControlBridge.should_refresh_snapshot_after_command(red_blue, "off"))
self.assertFalse(CameraControlBridge.should_refresh_snapshot_after_command(siren, "on"))
self.assertFalse(CameraControlBridge.should_refresh_snapshot_after_command(white, "press"))
if __name__ == "__main__":
unittest.main()
tests/test_ui_render.py
+20 -1
View File
@@ -1,6 +1,6 @@
import unittest
from frigate_camera_control_bridge.ui import render_page
from frigate_camera_control_bridge.ui import camera_stream_url, render_page
class DummyAction:
@@ -11,6 +11,7 @@ class DummyAction:
self.state = state
self.auto_off_seconds = 180
self.auto_off_at = 1234567890
self.snapshot_version = 42
self.cfg = {
"name_key": name_key,
"camera_name": "Example camera",
@@ -35,6 +36,7 @@ class UiRenderTest(unittest.TestCase):
self.assertIn('data-auto-off-at="1234567890"', html)
self.assertIn('data-requires-ack="true"', html)
self.assertIn("Example camera", html)
self.assertIn("v=42", html)
def test_controls_from_same_camera_share_one_card(self):
actions = [
@@ -73,6 +75,23 @@ class UiRenderTest(unittest.TestCase):
self.assertIn('data-working-label="Refreshing cameras..."', html)
self.assertIn("<svg", html)
def test_camera_stream_link_uses_default_frigate_url(self):
action = DummyAction()
self.assertEqual(
camera_stream_url(action),
"http://frigate.example.test:5000/#example",
)
def test_camera_stream_link_uses_public_proxy_url_from_env_config(self):
action = DummyAction()
action.cfg["snapshot"]["public_url"] = "https://frigate.example.test"
self.assertEqual(
camera_stream_url(action),
"https://frigate.example.test/#example",
)
if __name__ == "__main__":
unittest.main()