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

Submit GUI commands with fetch fallback

Browse Source
This commit is contained in:
michal
2026-08-03 03:05:34 +02:00
parent b6030860db
commit 789a9b349c
9 changed files with 454 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate_camera_control_bridge/app.py
+64 -22
View File
@@ -1636,6 +1636,22 @@ class CameraControlBridge:
payload["redirect_url"] = redirect_url
return payload
def action_payload(self, lang, action, command, result, redirect_url):
ok = bool(result.get("ok", True))
payload = {
"ok": ok,
"command": command,
"message": text_for(lang, "action.success" if ok else "action.failed_generic"),
"redirect_url": redirect_url,
"result": result,
}
if action is not None:
payload["action_id"] = action.id
payload["action"] = self.api_action(action, lang)
if action.kind == "all_off":
payload["actions"] = [self.api_action(item, lang) for item in self.actions_snapshot()]
return payload
def frigate_refresh_loop(self):
interval = int(self.config.get("defaults", {}).get("frigate_refresh_seconds", 300))
if interval <= 0:
@@ -1826,17 +1842,36 @@ class CameraControlBridge:
if len(parts) == 3 and parts[0] == "action":
action_id = urllib.parse.unquote(parts[1])
command = urllib.parse.unquote(parts[2]).lower()
token_param = urllib.parse.parse_qs(parsed.query).get("token", [None])[0]
token_query = f"&token={urllib.parse.quote(token_param)}" if token_param else ""
redirect_url = f"/?lang={urllib.parse.quote(lang)}{token_query}"
wants_json = "application/json" in self.headers.get("Accept", "").lower()
try:
result = bridge.execute_action(action_id, command, publish=True)
token_param = urllib.parse.parse_qs(parsed.query).get("token", [None])[0]
token_query = f"&token={urllib.parse.quote(token_param)}" if token_param else ""
self.send_response(303)
self.send_header("Location", f"/?lang={urllib.parse.quote(lang)}{token_query}")
self.end_headers()
action = bridge.action_by_id(action_id)
if wants_json:
payload = bridge.action_payload(lang, action, command, result, redirect_url)
self.send_json(payload, status=200 if payload.get("ok") else 500)
else:
self.send_response(303)
self.send_header("Location", redirect_url)
self.end_headers()
except Exception as exc:
self.send_response(500)
self.end_headers()
self.wfile.write(str(exc).encode("utf-8"))
if wants_json:
self.send_json(
{
"ok": False,
"action_id": action_id,
"command": command,
"message": text_for(lang, "action.failed", error=str(exc)),
"redirect_url": redirect_url,
},
status=500,
)
else:
self.send_response(500)
self.end_headers()
self.wfile.write(str(exc).encode("utf-8"))
return
self.send_response(404)
self.end_headers()
@@ -1874,20 +1909,27 @@ class CameraControlBridge:
LOG.info("HTTP fallback GUI listening on %s:%s", bind, port)
def api_actions(self):
return [
{
"id": action.id,
"name": action.display_name(self.default_language),
"platform": action.platform,
"commands": action.command_list(),
"state": action.state,
"snapshot": bool(action.cfg.get("snapshot")),
"auto_off_seconds": action.auto_off_seconds,
"auto_off_at": int(action.auto_off_at) if action.auto_off_at else None,
"requires_ack": bool(action.cfg.get("requires_ack")),
}
for action in self.actions_snapshot()
]
return [self.api_action(action, self.default_language) for action in self.actions_snapshot()]
def api_action(self, action, lang=None):
lang = lang or self.default_language
state_value = (action.state or "unknown").lower()
state_label = text_for(lang, f"state.{state_value}")
if state_label == f"state.{state_value}":
state_label = state_value.upper()
return {
"id": action.id,
"name": action.display_name(lang),
"platform": action.platform,
"commands": action.command_list(),
"state": action.state,
"state_label": state_label,
"snapshot": bool(action.cfg.get("snapshot")),
"snapshot_version": int(action.snapshot_version or 0),
"auto_off_seconds": action.auto_off_seconds,
"auto_off_at": int(action.auto_off_at) if action.auto_off_at else None,
"requires_ack": bool(action.cfg.get("requires_ack")),
}
def html_page(self, lang=None, token_param=None):
with self.action_lock: