Save initial camera state to update when websocket connects (#11174)

* Send camera state to dispatcher

* Fix logic

* Cleanup

* Send camera activitiy in on connect

* Support reading initial camera state

* Fix key

* Formatting

* Sorting
This commit is contained in:
Nicolas Mowen
2024-04-30 07:09:50 -06:00
committed by GitHub
parent 0080c28c4d
commit 461442b399
9 changed files with 147 additions and 43 deletions

View File

@@ -1,6 +1,7 @@
"""Handle communication between Frigate and other applications."""
import datetime
import json
import logging
from abc import ABC, abstractmethod
from typing import Any, Callable, Optional
@@ -12,6 +13,7 @@ from frigate.const import (
INSERT_MANY_RECORDINGS,
INSERT_PREVIEW,
REQUEST_REGION_GRID,
UPDATE_CAMERA_ACTIVITY,
UPSERT_REVIEW_SEGMENT,
)
from frigate.models import Previews, Recordings, ReviewSegment
@@ -76,6 +78,8 @@ class Dispatcher:
for comm in self.comms:
comm.subscribe(self._receive)
self.camera_activity = {}
def _receive(self, topic: str, payload: str) -> Optional[Any]:
"""Handle receiving of payload from communicators."""
if topic.endswith("set"):
@@ -122,6 +126,10 @@ class Dispatcher:
ReviewSegment.update(end_time=datetime.datetime.now().timestamp()).where(
ReviewSegment.end_time == None
).execute()
elif topic == UPDATE_CAMERA_ACTIVITY:
self.camera_activity = payload
elif topic == "onConnect":
self.publish("camera_activity", json.dumps(self.camera_activity))
else:
self.publish(topic, payload, retain=False)

View File

@@ -50,6 +50,10 @@ class WebSocketClient(Communicator): # type: ignore[misc]
class _WebSocketHandler(WebSocket): # type: ignore[misc]
receiver = self._dispatcher
def opened(self) -> None:
"""A new websocket is opened, we need to send an update message"""
threading.Timer(1.0, self.receiver, ("onConnect", "")).start()
def received_message(self, message: WebSocket.received_message) -> None:
try:
json_message = json.loads(message.data.decode("utf-8"))

View File

@@ -80,6 +80,7 @@ INSERT_PREVIEW = "insert_preview"
REQUEST_REGION_GRID = "request_region_grid"
UPSERT_REVIEW_SEGMENT = "upsert_review_segment"
CLEAR_ONGOING_REVIEW_SEGMENTS = "clear_ongoing_review_segments"
UPDATE_CAMERA_ACTIVITY = "update_camera_activity"
# Autotracking

View File

@@ -16,6 +16,7 @@ import numpy as np
from frigate.comms.detections_updater import DetectionPublisher, DetectionTypeEnum
from frigate.comms.dispatcher import Dispatcher
from frigate.comms.events_updater import EventEndSubscriber, EventUpdatePublisher
from frigate.comms.inter_process import InterProcessRequestor
from frigate.config import (
CameraConfig,
FrigateConfig,
@@ -24,7 +25,7 @@ from frigate.config import (
SnapshotsConfig,
ZoomingModeEnum,
)
from frigate.const import CLIPS_DIR
from frigate.const import ALL_ATTRIBUTE_LABELS, CLIPS_DIR, UPDATE_CAMERA_ACTIVITY
from frigate.events.types import EventStateEnum, EventTypeEnum
from frigate.ptz.autotrack import PtzAutoTrackerThread
from frigate.util.image import (
@@ -724,8 +725,31 @@ class CameraState:
# TODO: can i switch to looking this up and only changing when an event ends?
# maintain best objects
camera_activity: dict[str, list[any]] = {
"motion": len(motion_boxes) > 0,
"objects": [],
}
for obj in tracked_objects.values():
object_type = obj.obj_data["label"]
active = (
obj.obj_data["motionless_count"]
< self.camera_config.detect.stationary.threshold
)
if not obj.false_positive:
label = object_type
if (
obj.obj_data.get("sub_label")
and obj.obj_data.get("sub_label")[0] in ALL_ATTRIBUTE_LABELS
):
label = obj.obj_data["sub_label"]
camera_activity["objects"].append(
{"id": obj.obj_data["id"], "label": label, "stationary": not active}
)
# if the object's thumbnail is not from the current frame
if obj.false_positive or obj.thumbnail_data["frame_time"] != frame_time:
continue
@@ -752,6 +776,9 @@ class CameraState:
for c in self.callbacks["snapshot"]:
c(self.name, self.best_objects[object_type], frame_time)
for c in self.callbacks["camera_activity"]:
c(self.name, camera_activity)
# update overall camera state for each object type
obj_counter = Counter(
obj.obj_data["label"]
@@ -841,10 +868,14 @@ class TrackedObjectProcessor(threading.Thread):
self.frame_manager = SharedMemoryFrameManager()
self.last_motion_detected: dict[str, float] = {}
self.ptz_autotracker_thread = ptz_autotracker_thread
self.requestor = InterProcessRequestor()
self.detection_publisher = DetectionPublisher(DetectionTypeEnum.video)
self.event_sender = EventUpdatePublisher()
self.event_end_subscriber = EventEndSubscriber()
self.camera_activity: dict[str, dict[str, any]] = {}
def start(camera, obj: TrackedObject, current_frame_time):
self.event_sender.publish(
(
@@ -962,6 +993,13 @@ class TrackedObjectProcessor(threading.Thread):
def object_status(camera, object_name, status):
self.dispatcher.publish(f"{camera}/{object_name}", status, retain=False)
def camera_activity(camera, activity):
last_activity = self.camera_activity.get(camera)
if not last_activity or activity != last_activity:
self.camera_activity[camera] = activity
self.requestor.send_data(UPDATE_CAMERA_ACTIVITY, self.camera_activity)
for camera in self.config.cameras.keys():
camera_state = CameraState(
camera, self.config, self.frame_manager, self.ptz_autotracker_thread
@@ -972,6 +1010,7 @@ class TrackedObjectProcessor(threading.Thread):
camera_state.on("end", end)
camera_state.on("snapshot", snapshot)
camera_state.on("object_status", object_status)
camera_state.on("camera_activity", camera_activity)
self.camera_states[camera] = camera_state
# {
@@ -1228,6 +1267,7 @@ class TrackedObjectProcessor(threading.Thread):
event_id, camera = update
self.camera_states[camera].finished(event_id)
self.requestor.stop()
self.detection_publisher.stop()
self.event_sender.stop()
self.event_end_subscriber.stop()