administrator/frigate
1
0
Fork 0
mirror of https://github.com/blakeblackshear/frigate.git synced 2026-10-05 22:32:50 +03:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

simplify onvif and autotracking code

Browse Source 
Removes about 230 lines from the ONVIF controller and autotracker without changing how PTZ moves are calculated. `OnvifController` no longer keeps its own `camera_configs` copy of the camera config, the cached `GetStatus`, `GetServiceCapabilities`, and `AbsoluteMove` request objects are gone in favor of plain dicts at the call site, and `PtzAutoTrackerThread` is merged into `PtzAutoTracker`. Repeated blocks in the autotracker (waiting for the motor to stop, disabling autotracking on a failed setup step) are now single helpers.
This commit is contained in:
Josh Hawkins
2026-10-03 15:05:56 -05:00
parent 9b2839f4fb
commit cc1ff63b53
7 changed files with 427 additions and 655 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frigate/app.py
+2 -2
View File
@@ -77,7 +77,7 @@ from frigate.notices.registry import NoticeRegistry
from frigate.object_detection.base import ObjectDetectProcess
from frigate.object_detection.util import detection_frame_size
from frigate.output.output import OutputProcess
from frigate.ptz.autotrack import PtzAutoTrackerThread
from frigate.ptz.autotrack import PtzAutoTracker
from frigate.ptz.onvif import OnvifController
from frigate.record.cleanup import RecordingCleanup
from frigate.record.export import migrate_exports
@@ -443,7 +443,7 @@ class FrigateApp:
)
def start_ptz_autotracker(self) -> None:
self.ptz_autotracker_thread = PtzAutoTrackerThread(
self.ptz_autotracker_thread = PtzAutoTracker(
self.config,
self.onvif_controller,
self.ptz_metrics,
frigate/camera/state.py
+11 -13
View File
@@ -16,7 +16,7 @@ from frigate.config import (
ZoomingModeEnum,
)
from frigate.const import CLIPS_DIR, THUMB_DIR
from frigate.ptz.autotrack import PtzAutoTrackerThread
from frigate.ptz.autotrack import PtzAutoTracker
from frigate.track.tracked_object import TrackedObject
from frigate.util.image import (
SharedMemoryFrameManager,
@@ -35,7 +35,7 @@ class CameraState:
name: str,
config: FrigateConfig,
frame_manager: SharedMemoryFrameManager,
ptz_autotracker_thread: PtzAutoTrackerThread,
ptz_autotracker_thread: PtzAutoTracker,
) -> None:
self.name = name
self.config = config
@@ -115,17 +115,13 @@ class CameraState:
# draw thicker box around ptz autotracked object
if (
self.camera_config.onvif.autotracking.enabled
and self.ptz_autotracker_thread.ptz_autotracker.autotracker_init.get(
self.name
)
and self.ptz_autotracker_thread.ptz_autotracker.tracked_object[
self.name
]
and self.ptz_autotracker_thread.autotracker_init.get(self.name)
and self.ptz_autotracker_thread.tracked_object[self.name]
is not None
and obj["id"]
== self.ptz_autotracker_thread.ptz_autotracker.tracked_object[
== self.ptz_autotracker_thread.tracked_object[ # type: ignore[union-attr]
self.name
].obj_data["id"] # type: ignore[attr-defined]
].obj_data["id"]
and obj["frame_time"] == frame_time
):
thickness = 5
@@ -138,9 +134,11 @@ class CameraState:
and self.camera_config.detect.width is not None
and self.camera_config.detect.height is not None
):
max_target_box = self.ptz_autotracker_thread.ptz_autotracker.tracked_object_metrics[
self.name
]["max_target_box"] # type: ignore[index]
max_target_box = (
self.ptz_autotracker_thread.tracked_object_metrics[
self.name
]["max_target_box"]
)
side_length = max_target_box * (
max(
self.camera_config.detect.width,
frigate/ptz/autotrack.py
+170 -298
View File
@@ -180,7 +180,7 @@ class PtzMotionEstimator:
return self.coord_transformations
class PtzAutoTrackerThread(threading.Thread):
class PtzAutoTracker(threading.Thread):
def __init__(
self,
config: FrigateConfig,
@@ -190,56 +190,14 @@ class PtzAutoTrackerThread(threading.Thread):
stop_event: MpEvent,
) -> None:
super().__init__(name="ptz_autotracker")
self.ptz_autotracker = PtzAutoTracker(
config, onvif, ptz_metrics, dispatcher, stop_event
)
self.stop_event = stop_event
self.config = config
def run(self):
while not self.stop_event.wait(1):
self.ptz_autotracker.check_for_updates()
for camera, camera_config in list(self.config.cameras.items()):
if not camera_config.enabled:
continue
if camera_config.onvif.autotracking.enabled:
future = asyncio.run_coroutine_threadsafe(
self.ptz_autotracker.camera_maintenance(camera),
self.ptz_autotracker.onvif.loop,
)
# Wait for the coroutine to complete
future.result()
else:
# disabled dynamically by mqtt
if self.ptz_autotracker.tracked_object.get(camera):
self.ptz_autotracker.tracked_object[camera] = None
self.ptz_autotracker.tracked_object_history[camera].clear()
self.ptz_autotracker.config_subscriber.stop()
logger.info("Exiting autotracker...")
class PtzAutoTracker:
def __init__(
self,
config: FrigateConfig,
onvif: OnvifController,
ptz_metrics: PTZMetrics,
dispatcher: Dispatcher,
stop_event: MpEvent,
) -> None:
self.config = config
self.onvif = onvif
self.ptz_metrics = ptz_metrics
self.dispatcher = dispatcher
self.stop_event = stop_event
self.tracked_object: dict[str, object] = {}
self.tracked_object: dict[str, TrackedObject | None] = {}
self.tracked_object_history: dict[str, object] = {}
self.tracked_object_metrics: dict[str, object] = {}
self.object_types: dict[str, object] = {}
self.required_zones: dict[str, object] = {}
self.tracked_object_metrics: dict[str, dict[str, Any]] = {}
self.move_queues: dict[str, object] = {}
self.move_queue_locks: dict[str, object] = {}
self.move_threads: dict[str, object] = {}
@@ -249,7 +207,6 @@ class PtzAutoTracker:
self.intercept: dict[str, object] = {}
self.move_coefficients: dict[str, object] = {}
self.zoom_time: dict[str, float] = {}
self.zoom_factor: dict[str, object] = {}
self.config_subscriber = CameraConfigUpdateSubscriber(
self.config,
@@ -277,6 +234,29 @@ class PtzAutoTracker:
# Wait for the coroutine to complete
future.result()
def run(self) -> None:
while not self.stop_event.wait(1):
self.check_for_updates()
for camera, camera_config in list(self.config.cameras.items()):
if not camera_config.enabled:
continue
if camera_config.onvif.autotracking.enabled:
future = asyncio.run_coroutine_threadsafe(
self.camera_maintenance(camera), self.onvif.loop
)
# Wait for the coroutine to complete
future.result()
else:
# disabled dynamically by mqtt
if self.tracked_object.get(camera):
self.tracked_object[camera] = None
self.tracked_object_history[camera].clear()
self.config_subscriber.stop()
logger.info("Exiting autotracker...")
def check_for_updates(self) -> None:
"""Apply camera config updates and mirror autotracking state to ptz metrics.
@@ -303,18 +283,11 @@ class PtzAutoTracker:
async def _autotracker_setup(self, camera_config: CameraConfig, camera: str):
logger.debug(f"{camera}: Autotracker init")
self.object_types[camera] = camera_config.onvif.autotracking.track
self.required_zones[camera] = camera_config.onvif.autotracking.required_zones
self.zoom_factor[camera] = camera_config.onvif.autotracking.zoom_factor
self.tracked_object[camera] = None
self.tracked_object_history[camera] = deque(
maxlen=round(camera_config.detect.fps * 1.5)
)
self.tracked_object_metrics[camera] = {
"max_target_box": AUTOTRACKING_MAX_AREA_RATIO
** (1 / self.zoom_factor[camera])
}
self._reset_tracked_object_metrics(camera)
self.calibrating[camera] = False
self.move_metrics[camera] = []
@@ -326,43 +299,22 @@ class PtzAutoTracker:
# handle onvif constructor failing due to no connection
if camera not in self.onvif.cams:
logger.warning(
f"Disabling autotracking for {camera}: onvif connection failed"
)
camera_config.onvif.autotracking.enabled = False
self.ptz_metrics[camera].autotracker_enabled.value = False
self._disable(camera, "onvif connection failed")
return
if not self.onvif.cams[camera]["init"]:
if not await self.onvif._init_onvif(camera):
logger.warning(
f"Disabling autotracking for {camera}: Unable to initialize onvif"
)
camera_config.onvif.autotracking.enabled = False
self.ptz_metrics[camera].autotracker_enabled.value = False
self._disable(camera, "Unable to initialize onvif")
return
if "pt-r-fov" not in self.onvif.cams[camera]["features"]:
logger.warning(
f"Disabling autotracking for {camera}: FOV relative movement not supported"
)
camera_config.onvif.autotracking.enabled = False
self.ptz_metrics[camera].autotracker_enabled.value = False
self._disable(camera, "FOV relative movement not supported")
return
move_status_supported = await self.onvif.get_service_capabilities(camera)
if not (
isinstance(move_status_supported, bool) and move_status_supported
) and not (
isinstance(move_status_supported, str)
and move_status_supported.lower() == "true"
):
logger.warning(
f"Disabling autotracking for {camera}: ONVIF MoveStatus not supported"
)
camera_config.onvif.autotracking.enabled = False
self.ptz_metrics[camera].autotracker_enabled.value = False
if str(move_status_supported).lower() != "true":
self._disable(camera, "ONVIF MoveStatus not supported")
return
if self.onvif.cams[camera]["init"]:
@@ -374,36 +326,13 @@ class PtzAutoTracker:
)
if camera_config.onvif.autotracking.movement_weights:
if len(camera_config.onvif.autotracking.movement_weights) == 6:
camera_config.onvif.autotracking.movement_weights = [
float(val)
for val in camera_config.onvif.autotracking.movement_weights
]
self.ptz_metrics[
camera
].min_zoom.value = (
camera_config.onvif.autotracking.movement_weights[0]
)
self.ptz_metrics[
camera
].max_zoom.value = (
camera_config.onvif.autotracking.movement_weights[1]
)
self.intercept[camera] = (
camera_config.onvif.autotracking.movement_weights[2]
)
self.move_coefficients[camera] = (
camera_config.onvif.autotracking.movement_weights[3:5]
)
self.zoom_time[camera] = (
camera_config.onvif.autotracking.movement_weights[5]
)
else:
camera_config.onvif.autotracking.enabled = False
self.ptz_metrics[camera].autotracker_enabled.value = False
logger.warning(
f"Autotracker recalibration is required for {camera}. Disabling autotracking."
)
(
self.ptz_metrics[camera].min_zoom.value,
self.ptz_metrics[camera].max_zoom.value,
self.intercept[camera],
*self.move_coefficients[camera],
self.zoom_time[camera],
) = map(float, camera_config.onvif.autotracking.movement_weights)
if camera_config.onvif.autotracking.calibrate_on_startup:
await self._calibrate_camera(camera)
@@ -412,21 +341,24 @@ class PtzAutoTracker:
self.dispatcher.publish(f"{camera}/ptz_autotracker/active", "OFF", retain=False)
self.autotracker_init[camera] = True
def _write_config(self, camera):
config_file = find_config_file()
def _disable(self, camera: str, reason: str) -> None:
logger.warning(f"Disabling autotracking for {camera}: {reason}")
self.config.cameras[camera].onvif.autotracking.enabled = False
self.ptz_metrics[camera].autotracker_enabled.value = False
logger.debug(
f"{camera}: Writing new config with autotracker motion coefficients: {self.config.cameras[camera].onvif.autotracking.movement_weights}"
)
def _reset_tracked_object_metrics(self, camera: str) -> None:
zoom_factor = self.config.cameras[camera].onvif.autotracking.zoom_factor
self.tracked_object_metrics[camera] = {
"max_target_box": AUTOTRACKING_MAX_AREA_RATIO ** (1 / zoom_factor)
}
update_yaml_file_bulk(
config_file,
{
f"cameras.{camera}.onvif.autotracking.movement_weights": self.config.cameras[
camera
].onvif.autotracking.movement_weights
},
)
async def _wait_until_stopped(
self, camera: str, metrics: PTZMetrics | None = None
) -> None:
metrics = metrics or self.ptz_metrics[camera]
while not metrics.motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
async def _calibrate_camera(self, camera):
# move the camera from the preset in steps and measure the time it takes to move that amount
@@ -459,8 +391,7 @@ class PtzAutoTracker:
1,
)
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
zoom_out_values.append(self.ptz_metrics[camera].zoom_level.value)
@@ -470,8 +401,7 @@ class PtzAutoTracker:
1,
)
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
zoom_in_values.append(self.ptz_metrics[camera].zoom_level.value)
@@ -488,8 +418,7 @@ class PtzAutoTracker:
1,
)
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
zoom_out_values.append(self.ptz_metrics[camera].zoom_level.value)
@@ -503,8 +432,7 @@ class PtzAutoTracker:
1,
)
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
zoom_stop_time = time.time()
@@ -518,8 +446,7 @@ class PtzAutoTracker:
1,
)
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
full_relative_stop_time = time.time()
@@ -531,8 +458,7 @@ class PtzAutoTracker:
1,
)
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
self.zoom_time[camera] = (
full_relative_stop_time - full_relative_start_time
@@ -558,9 +484,7 @@ class PtzAutoTracker:
self.ptz_metrics[camera].reset.set()
self.ptz_metrics[camera].motor_stopped.clear()
# Wait until the camera finishes moving
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
for step in range(num_steps):
pan = step_sizes[step]
@@ -569,9 +493,7 @@ class PtzAutoTracker:
start_time = time.time()
await self.onvif._move_relative(camera, pan, tilt, 0, 1)
# Wait until the camera finishes moving
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
stop_time = time.time()
self.move_metrics[camera].append(
@@ -590,9 +512,7 @@ class PtzAutoTracker:
self.ptz_metrics[camera].reset.set()
self.ptz_metrics[camera].motor_stopped.clear()
# Wait until the camera finishes moving
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
logger.info(
f"Calibration for {camera} in progress: {round((step / num_steps) * 100)}% complete"
@@ -668,7 +588,14 @@ class PtzAutoTracker:
f"{camera}: New regression parameters - intercept: {self.intercept[camera]}, coefficients: {self.move_coefficients[camera]}"
)
self._write_config(camera)
update_yaml_file_bulk(
find_config_file(),
{
f"cameras.{camera}.onvif.autotracking.movement_weights": self.config.cameras[
camera
].onvif.autotracking.movement_weights
},
)
def _predict_movement_time(self, camera, pan, tilt):
combined_movement = abs(pan) + abs(tilt)
@@ -682,6 +609,18 @@ class PtzAutoTracker:
[self.tracked_object_history[camera][-1]["frame_time"] + time],
)
def _predict_target_box(self, camera, predicted_time):
target_box = self.tracked_object_metrics[camera]["target_box"]
if not predicted_time:
return target_box
frame_shape = self.config.cameras[camera].frame_shape
return target_box + self._predict_area_after_time(camera, predicted_time) / (
frame_shape[0] * frame_shape[1]
)
def _calculate_tracked_object_metrics(self, camera, obj):
def remove_outliers(data):
areas = [item["area"] for item in data]
@@ -703,19 +642,20 @@ class PtzAutoTracker:
return filtered_data
camera_config = self.config.cameras[camera]
tom = self.tracked_object_metrics[camera]
zoom_factor = camera_config.onvif.autotracking.zoom_factor
camera_width = camera_config.frame_shape[1]
camera_height = camera_config.frame_shape[0]
# Extract areas and calculate weighted average
# grab the largest dimension of the bounding box and create a square from that
# Filter out the initial frame and use a recent time window
# Use a recent time window
current_time = obj.obj_data["frame_time"]
time_window = 1.5 # seconds
history = [
entry
for entry in self.tracked_object_history[camera]
if not entry.get("is_initial_frame", False)
and current_time - entry["frame_time"] <= time_window
if current_time - entry["frame_time"] <= time_window
]
if not history: # Fallback to latest if no recent entries
history = [self.tracked_object_history[camera][-1]]
@@ -748,33 +688,29 @@ class PtzAutoTracker:
)
y = np.array([item["area"] for item in filtered_areas_not_touching_edge])
self.tracked_object_metrics[camera]["area_coefficients"] = np.linalg.lstsq(
X.reshape(-1, 1), y, rcond=None
)[0]
tom["area_coefficients"] = np.linalg.lstsq(X.reshape(-1, 1), y, rcond=None)[
0
]
else:
self.tracked_object_metrics[camera]["area_coefficients"] = np.array([0])
tom["area_coefficients"] = np.array([0])
weights = np.arange(1, len(filtered_areas) + 1)
weighted_area = np.average(
[item["area"] for item in filtered_areas], weights=weights
)
self.tracked_object_metrics[camera]["target_box"] = (
tom["target_box"] = (
weighted_area / (camera_width * camera_height)
) ** self.zoom_factor[camera]
) ** zoom_factor
if "original_target_box" not in self.tracked_object_metrics[camera]:
self.tracked_object_metrics[camera]["original_target_box"] = (
self.tracked_object_metrics[camera]["target_box"]
)
if "original_target_box" not in tom:
tom["original_target_box"] = tom["target_box"]
(
self.tracked_object_metrics[camera]["valid_velocity"],
self.tracked_object_metrics[camera]["velocity"],
tom["valid_velocity"],
tom["velocity"],
) = self._get_valid_velocity(camera, obj)
self.tracked_object_metrics[camera]["distance"] = self._get_distance_threshold(
camera, obj
)
tom["distance"] = self._get_distance_threshold(camera, obj)
centroid_distance = np.linalg.norm(
[
@@ -785,9 +721,7 @@ class PtzAutoTracker:
logger.debug(f"{camera}: Centroid distance: {centroid_distance}")
self.tracked_object_metrics[camera]["below_distance_threshold"] = (
centroid_distance < self.tracked_object_metrics[camera]["distance"]
)
tom["below_distance_threshold"] = centroid_distance < tom["distance"]
async def _process_move_queue(self, camera):
move_queue = self.move_queues[camera]
@@ -833,16 +767,12 @@ class PtzAutoTracker:
if pan != 0 or tilt != 0:
await self.onvif._move_relative(camera, pan, tilt, 0, 1)
# Wait until the camera finishes moving
while not metrics.motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera, metrics)
if zoom > 0 and metrics.zoom_level.value != zoom:
await self.onvif._zoom_absolute(camera, zoom, 1)
# Wait until the camera finishes moving
while not metrics.motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera, metrics)
if camera_config.onvif.autotracking.movement_weights:
logger.debug(
@@ -878,41 +808,20 @@ class PtzAutoTracker:
await move_queue.get()
def _enqueue_move(self, camera, frame_time, pan, tilt, zoom):
def split_value(value, suppress_diff=True):
clipped = np.clip(value, -1, 1)
# don't make small movements
if -0.05 < clipped < 0.05 and suppress_diff:
diff = 0.0
else:
diff = value - clipped
return clipped, diff
pan, tilt, zoom = (np.clip(value, -1, 1) for value in (pan, tilt, zoom))
if (
frame_time > self.ptz_metrics[camera].start_time.value
(pan != 0 or tilt != 0 or zoom != 0)
and frame_time > self.ptz_metrics[camera].start_time.value
and frame_time > self.ptz_metrics[camera].stop_time.value
and not self.move_queue_locks[camera].locked()
):
# we can split up any large moves caused by velocity estimated movements if necessary
# get an excess amount and assign it instead of 0 below
while pan != 0 or tilt != 0 or zoom != 0:
pan, _ = split_value(pan)
tilt, _ = split_value(tilt)
zoom, _ = split_value(zoom, False)
logger.debug(
f"{camera}: Enqueue movement for frame time: {frame_time} pan: {pan}, tilt: {tilt}, zoom: {zoom}"
)
move_data = (frame_time, pan, tilt, zoom)
self.onvif.loop.call_soon_threadsafe(
self.move_queues[camera].put_nowait, move_data
)
# reset values to not split up large movements
pan = 0
tilt = 0
zoom = 0
logger.debug(
f"{camera}: Enqueue movement for frame time: {frame_time} pan: {pan}, tilt: {tilt}, zoom: {zoom}"
)
self.onvif.loop.call_soon_threadsafe(
self.move_queues[camera].put_nowait, (frame_time, pan, tilt, zoom)
)
def _touching_frame_edges(self, camera, box):
camera_config = self.config.cameras[camera]
@@ -1046,16 +955,16 @@ class PtzAutoTracker:
return distance_threshold
def _should_zoom_in(
self, camera: str, obj: TrackedObject, box, predicted_time, debug_zooming=False
):
def _should_zoom_in(self, camera: str, box, predicted_time):
# returns True if we should zoom in, False if we should zoom out, None to do nothing
camera_config = self.config.cameras[camera]
tom = self.tracked_object_metrics[camera]
zoom_factor = camera_config.onvif.autotracking.zoom_factor
camera_width = camera_config.frame_shape[1]
camera_height = camera_config.frame_shape[0]
camera_fps = camera_config.detect.fps
average_velocity = self.tracked_object_metrics[camera]["velocity"]
average_velocity = tom["velocity"]
bb_left, bb_top, bb_right, bb_bottom = box
@@ -1073,13 +982,11 @@ class PtzAutoTracker:
touching_frame_edges = self._touching_frame_edges(camera, box)
# make sure object is centered in the frame
below_distance_threshold = self.tracked_object_metrics[camera][
"below_distance_threshold"
]
below_distance_threshold = tom["below_distance_threshold"]
below_dimension_threshold = (bb_right - bb_left) <= camera_width * (
self.zoom_factor[camera] + 0.1
) and (bb_bottom - bb_top) <= camera_height * (self.zoom_factor[camera] + 0.1)
zoom_factor + 0.1
) and (bb_bottom - bb_top) <= camera_height * (zoom_factor + 0.1)
# ensure object is not moving quickly
below_velocity_threshold = np.all(
@@ -1087,30 +994,18 @@ class PtzAutoTracker:
< np.tile([velocity_threshold_x, velocity_threshold_y], 2)
) or np.all(average_velocity == 0)
if not predicted_time:
calculated_target_box = self.tracked_object_metrics[camera]["target_box"]
else:
calculated_target_box = self.tracked_object_metrics[camera][
"target_box"
] + self._predict_area_after_time(camera, predicted_time) / (
camera_width * camera_height
)
calculated_target_box = self._predict_target_box(camera, predicted_time)
below_area_threshold = (
calculated_target_box
< self.tracked_object_metrics[camera]["max_target_box"]
)
below_area_threshold = calculated_target_box < tom["max_target_box"]
# introduce some hysteresis to prevent a yo-yo zooming effect
zoom_out_hysteresis = (
calculated_target_box
> self.tracked_object_metrics[camera]["max_target_box"]
* AUTOTRACKING_ZOOM_OUT_HYSTERESIS
> tom["max_target_box"] * AUTOTRACKING_ZOOM_OUT_HYSTERESIS
)
zoom_in_hysteresis = (
calculated_target_box
< self.tracked_object_metrics[camera]["max_target_box"]
* AUTOTRACKING_ZOOM_IN_HYSTERESIS
< tom["max_target_box"] * AUTOTRACKING_ZOOM_IN_HYSTERESIS
)
at_max_zoom = (
@@ -1122,31 +1017,29 @@ class PtzAutoTracker:
== self.ptz_metrics[camera].min_zoom.value
)
# debug zooming
if debug_zooming:
logger.debug(
f"{camera}: Zoom test: touching edges: count: {touching_frame_edges} left: {bb_left < AUTOTRACKING_ZOOM_EDGE_THRESHOLD * camera_width}, right: {bb_right > (1 - AUTOTRACKING_ZOOM_EDGE_THRESHOLD) * camera_width}, top: {bb_top < AUTOTRACKING_ZOOM_EDGE_THRESHOLD * camera_height}, bottom: {bb_bottom > (1 - AUTOTRACKING_ZOOM_EDGE_THRESHOLD) * camera_height}"
)
logger.debug(
f"{camera}: Zoom test: below distance threshold: {(below_distance_threshold)}"
)
logger.debug(
f"{camera}: Zoom test: below area threshold: {(below_area_threshold)} target: {self.tracked_object_metrics[camera]['target_box']}, calculated: {calculated_target_box}, max: {self.tracked_object_metrics[camera]['max_target_box']}"
)
logger.debug(
f"{camera}: Zoom test: below dimension threshold: {below_dimension_threshold} width: {bb_right - bb_left}, max width: {camera_width * (self.zoom_factor[camera] + 0.1)}, height: {bb_bottom - bb_top}, max height: {camera_height * (self.zoom_factor[camera] + 0.1)}"
)
logger.debug(
f"{camera}: Zoom test: below velocity threshold: {below_velocity_threshold} velocity x: {abs(average_velocity[0])}, x threshold: {velocity_threshold_x}, velocity y: {abs(average_velocity[1])}, y threshold: {velocity_threshold_y}"
)
logger.debug(f"{camera}: Zoom test: at max zoom: {at_max_zoom}")
logger.debug(f"{camera}: Zoom test: at min zoom: {at_min_zoom}")
logger.debug(
f"{camera}: Zoom test: zoom in hysteresis limit: {zoom_in_hysteresis} value: {AUTOTRACKING_ZOOM_IN_HYSTERESIS} original: {self.tracked_object_metrics[camera]['original_target_box']} max: {self.tracked_object_metrics[camera]['max_target_box']} target: {calculated_target_box if calculated_target_box else self.tracked_object_metrics[camera]['target_box']}"
)
logger.debug(
f"{camera}: Zoom test: zoom out hysteresis limit: {zoom_out_hysteresis} value: {AUTOTRACKING_ZOOM_OUT_HYSTERESIS} original: {self.tracked_object_metrics[camera]['original_target_box']} max: {self.tracked_object_metrics[camera]['max_target_box']} target: {calculated_target_box if calculated_target_box else self.tracked_object_metrics[camera]['target_box']}"
)
logger.debug(
f"{camera}: Zoom test: touching edges: count: {touching_frame_edges} left: {bb_left < AUTOTRACKING_ZOOM_EDGE_THRESHOLD * camera_width}, right: {bb_right > (1 - AUTOTRACKING_ZOOM_EDGE_THRESHOLD) * camera_width}, top: {bb_top < AUTOTRACKING_ZOOM_EDGE_THRESHOLD * camera_height}, bottom: {bb_bottom > (1 - AUTOTRACKING_ZOOM_EDGE_THRESHOLD) * camera_height}"
)
logger.debug(
f"{camera}: Zoom test: below distance threshold: {(below_distance_threshold)}"
)
logger.debug(
f"{camera}: Zoom test: below area threshold: {(below_area_threshold)} target: {tom['target_box']}, calculated: {calculated_target_box}, max: {tom['max_target_box']}"
)
logger.debug(
f"{camera}: Zoom test: below dimension threshold: {below_dimension_threshold} width: {bb_right - bb_left}, max width: {camera_width * (zoom_factor + 0.1)}, height: {bb_bottom - bb_top}, max height: {camera_height * (zoom_factor + 0.1)}"
)
logger.debug(
f"{camera}: Zoom test: below velocity threshold: {below_velocity_threshold} velocity x: {abs(average_velocity[0])}, x threshold: {velocity_threshold_x}, velocity y: {abs(average_velocity[1])}, y threshold: {velocity_threshold_y}"
)
logger.debug(f"{camera}: Zoom test: at max zoom: {at_max_zoom}")
logger.debug(f"{camera}: Zoom test: at min zoom: {at_min_zoom}")
logger.debug(
f"{camera}: Zoom test: zoom in hysteresis limit: {zoom_in_hysteresis} value: {AUTOTRACKING_ZOOM_IN_HYSTERESIS} original: {tom['original_target_box']} max: {tom['max_target_box']} target: {calculated_target_box if calculated_target_box else tom['target_box']}"
)
logger.debug(
f"{camera}: Zoom test: zoom out hysteresis limit: {zoom_out_hysteresis} value: {AUTOTRACKING_ZOOM_OUT_HYSTERESIS} original: {tom['original_target_box']} max: {tom['max_target_box']} target: {calculated_target_box if calculated_target_box else tom['target_box']}"
)
# Zoom in conditions (and)
if (
@@ -1237,7 +1130,7 @@ class PtzAutoTracker:
)
zoom = self._get_zoom_amount(
camera, obj, predicted_box, predicted_movement_time, debug_zoom=True
camera, obj, predicted_box, predicted_movement_time
)
if (
@@ -1298,9 +1191,10 @@ class PtzAutoTracker:
obj: TrackedObject,
predicted_box,
predicted_movement_time,
debug_zoom=True,
):
camera_config = self.config.cameras[camera]
tom = self.tracked_object_metrics[camera]
zoom_factor = camera_config.onvif.autotracking.zoom_factor
# frame width and height
camera_width = camera_config.frame_shape[1]
@@ -1317,16 +1211,12 @@ class PtzAutoTracker:
# absolute zooming separately from pan/tilt
if camera_config.onvif.autotracking.zooming == ZoomingModeEnum.absolute:
# don't zoom on initial move
if "target_box" not in self.tracked_object_metrics[camera]:
if "target_box" not in tom:
zoom = current_zoom_level
else:
if (
result := self._should_zoom_in(
camera,
obj,
obj.obj_data["box"],
predicted_movement_time,
debug_zoom,
camera, obj.obj_data["box"], predicted_movement_time
)
) is not None:
# divide zoom in 10 increments and always zoom out more than in
@@ -1342,46 +1232,35 @@ class PtzAutoTracker:
# relative zooming concurrently with pan/tilt
if camera_config.onvif.autotracking.zooming == ZoomingModeEnum.relative:
# this is our initial zoom in on a new object
if "target_box" not in self.tracked_object_metrics[camera]:
zoom = target_box ** self.zoom_factor[camera]
if zoom > self.tracked_object_metrics[camera]["max_target_box"]:
if "target_box" not in tom:
zoom = target_box**zoom_factor
if zoom > tom["max_target_box"]:
zoom = -(1 - zoom)
logger.debug(
f"{camera}: target box: {target_box}, max: {self.tracked_object_metrics[camera]['max_target_box']}, calc zoom: {zoom}"
f"{camera}: target box: {target_box}, max: {tom['max_target_box']}, calc zoom: {zoom}"
)
else:
if (
result := self._should_zoom_in(
camera,
obj,
predicted_box
if camera_config.onvif.autotracking.movement_weights
else obj.obj_data["box"],
predicted_movement_time,
debug_zoom,
)
) is not None:
if predicted_movement_time:
calculated_target_box = self.tracked_object_metrics[camera][
"target_box"
] + self._predict_area_after_time(
camera, predicted_movement_time
) / (camera_width * camera_height)
logger.debug(
f"{camera}: Zooming prediction: predicted movement time: {predicted_movement_time}, original box: {self.tracked_object_metrics[camera]['target_box']}, calculated box: {calculated_target_box}"
)
else:
calculated_target_box = self.tracked_object_metrics[camera][
"target_box"
]
# zoom value
ratio = (
self.tracked_object_metrics[camera]["max_target_box"]
/ calculated_target_box
calculated_target_box = self._predict_target_box(
camera, predicted_movement_time
)
if predicted_movement_time:
logger.debug(
f"{camera}: Zooming prediction: predicted movement time: {predicted_movement_time}, original box: {tom['target_box']}, calculated box: {calculated_target_box}"
)
# zoom value
ratio = tom["max_target_box"] / calculated_target_box
zoom = (ratio - 1) / (ratio + 1)
logger.debug(
f"{camera}: limit: {self.tracked_object_metrics[camera]['max_target_box']}, ratio: {ratio} zoom calculation: {zoom}"
f"{camera}: limit: {tom['max_target_box']}, ratio: {ratio} zoom calculation: {zoom}"
)
if not result:
# zoom out with special condition if zooming out because of velocity, edges, etc.
@@ -1394,9 +1273,6 @@ class PtzAutoTracker:
return zoom
def is_autotracking(self, camera: str):
return self.tracked_object[camera] is not None
def autotrack_object(self, camera: str, obj: TrackedObject):
if camera not in self.config.cameras:
return
@@ -1420,8 +1296,9 @@ class PtzAutoTracker:
# new object
self.tracked_object[camera] is None
and obj.camera_config.name == camera
and obj.obj_data["label"] in self.object_types[camera]
and set(obj.entered_zones) & set(self.required_zones[camera])
and obj.obj_data["label"] in camera_config.onvif.autotracking.track
and set(obj.entered_zones)
& set(camera_config.onvif.autotracking.required_zones)
and not obj.previous["false_positive"]
and not obj.false_positive
and not self.tracked_object_history[camera]
@@ -1479,7 +1356,7 @@ class PtzAutoTracker:
# Should we check region (maybe too broad) or expand the previous object's box a bit and check that?
self.tracked_object[camera] is None
and obj.camera_config.name == camera
and obj.obj_data["label"] in self.object_types[camera]
and obj.obj_data["label"] in camera_config.onvif.autotracking.track
and not obj.previous["false_positive"]
and not obj.false_positive
and self.tracked_object_history[camera]
@@ -1515,10 +1392,7 @@ class PtzAutoTracker:
f"{camera}: End object: {obj.obj_data['id']} {obj.obj_data['box']}"
)
self.tracked_object[camera] = None
self.tracked_object_metrics[camera] = {
"max_target_box": AUTOTRACKING_MAX_AREA_RATIO
** (1 / self.zoom_factor[camera])
}
self._reset_tracked_object_metrics(camera)
async def camera_maintenance(self, camera):
# bail and don't check anything if we're not set up yet, calibrating, or
@@ -1555,8 +1429,7 @@ class PtzAutoTracker:
self.tracked_object[camera] = None
self.tracked_object_history[camera].clear()
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
logger.debug(
f"{camera}: Time is {self.ptz_metrics[camera].frame_time.value}, returning to preset: {autotracker_config.return_preset}"
)
@@ -1566,8 +1439,7 @@ class PtzAutoTracker:
)
# update stored zoom level from preset
while not self.ptz_metrics[camera].motor_stopped.is_set():
await self.onvif.get_camera_status(camera)
await self._wait_until_stopped(camera)
self.ptz_metrics[camera].tracking_active.clear()
self.dispatcher.publish(
frigate/ptz/onvif.py
+202 -302
View File
@@ -10,8 +10,7 @@ from pathlib import Path
from typing import Any
import numpy
from onvif import ONVIFCamera, ONVIFError, ONVIFService
from zeep.exceptions import Fault, TransportError
from onvif import ONVIFCamera, ONVIFService
from frigate.camera import PTZMetrics
from frigate.config import FrigateConfig, ZoomingModeEnum
@@ -41,6 +40,14 @@ class OnvifCommandEnum(str, Enum):
focus_out = "focus_out"
PAN_TILT_VELOCITY = {
OnvifCommandEnum.move_left: (-0.5, 0),
OnvifCommandEnum.move_right: (0.5, 0),
OnvifCommandEnum.move_up: (0, 0.5),
OnvifCommandEnum.move_down: (0, -0.5),
}
class OnvifController:
ptz_metrics: dict[str, PTZMetrics]
@@ -61,14 +68,6 @@ class OnvifController:
self.loop_thread = threading.Thread(target=self._run_event_loop, daemon=True)
self.loop_thread.start()
self.camera_configs = {}
for cam_name, cam in config.cameras.items():
if not cam.enabled:
continue
if cam.onvif.host:
self.camera_configs[cam_name] = cam
self.status_locks[cam_name] = asyncio.Lock()
self.config_subscriber = CameraConfigUpdateSubscriber(
self.config,
self.config.cameras,
@@ -92,8 +91,9 @@ class OnvifController:
async def _init_cameras(self) -> None:
"""Initialize all configured cameras."""
for cam_name in self.camera_configs:
await self._init_single_camera(cam_name)
for cam_name, cam in list(self.config.cameras.items()):
if cam.enabled and cam.onvif.host:
await self._init_single_camera(cam_name)
async def _poll_config_updates(self) -> None:
"""Poll for ONVIF config updates and re-initialize cameras as needed."""
@@ -129,13 +129,9 @@ class OnvifController:
async def _remove_camera(self, cam_name: str) -> None:
"""Tear down the ONVIF session for a camera removed at runtime."""
if cam_name not in self.cams and cam_name not in self.camera_configs:
return
logger.debug(f"Tearing down ONVIF for {cam_name} after camera removal")
await self._close_camera(cam_name)
self.cams.pop(cam_name, None)
self.camera_configs.pop(cam_name, None)
self.failed_cams.pop(cam_name, None)
self.status_locks.pop(cam_name, None)
@@ -143,25 +139,14 @@ class OnvifController:
"""Re-initialize a camera after config change."""
logger.info(f"Re-initializing ONVIF for {cam_name} due to config change")
# close existing session before re-init
# close existing session and reset state before re-init
await self._close_camera(cam_name)
cam = self.config.cameras.get(cam_name)
if not cam or not cam.onvif.host:
# ONVIF removed from config, clean up
self.cams.pop(cam_name, None)
self.camera_configs.pop(cam_name, None)
self.failed_cams.pop(cam_name, None)
return
# update stored config and reset state
self.camera_configs[cam_name] = cam
if cam_name not in self.status_locks:
self.status_locks[cam_name] = asyncio.Lock()
self.cams.pop(cam_name, None)
self.failed_cams.pop(cam_name, None)
await self._init_single_camera(cam_name)
cam = self.config.cameras.get(cam_name)
if cam and cam.onvif.host:
await self._init_single_camera(cam_name)
async def _init_single_camera(self, cam_name: str) -> bool:
"""Initialize a single camera by name.
@@ -172,11 +157,12 @@ class OnvifController:
Returns:
bool: True if initialization succeeded, False otherwise
"""
if cam_name not in self.camera_configs:
cam = self.config.cameras.get(cam_name)
if cam is None:
logger.error(f"No configuration found for camera {cam_name}")
return False
cam = self.camera_configs[cam_name]
self.status_locks.setdefault(cam_name, asyncio.Lock())
try:
self.cams[cam_name] = {
"onvif": ONVIFCamera(
@@ -195,12 +181,11 @@ class OnvifController:
"profiles": [],
}
return True
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.error(f"Failed to create ONVIF camera instance for {cam_name}: {e}")
# track initial failures
self.failed_cams[cam_name] = {
"retry_attempts": 0,
"last_error": str(e),
"last_attempt": time.time(),
}
return False
@@ -211,7 +196,8 @@ class OnvifController:
if camera_config is None:
return False
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
cam = self.cams[camera_name]
onvif: ONVIFCamera = cam["onvif"]
try:
await onvif.update_xaddrs()
except Exception as e:
@@ -226,7 +212,7 @@ class OnvifController:
# this will fire an exception if camera is not a ptz
capabilities = onvif.get_definition("ptz")
logger.debug(f"Onvif capabilities for {camera_name}: {capabilities}")
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.error(
f"Unable to get Onvif capabilities for camera: {camera_name}: {e}"
)
@@ -235,7 +221,7 @@ class OnvifController:
try:
profiles = await media.GetProfiles()
logger.debug(f"Onvif profiles for {camera_name}: {profiles}")
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.error(
f"Unable to get Onvif media profiles for camera: {camera_name}: {e}"
)
@@ -254,7 +240,7 @@ class OnvifController:
]
# store available profiles for API response and log for debugging
self.cams[camera_name]["profiles"] = [
cam["profiles"] = [
{"name": getattr(p, "Name", None) or p.token, "token": p.token}
for p in valid_profiles
]
@@ -267,19 +253,18 @@ class OnvifController:
)
configured_profile = camera_config.onvif.profile
profile = None
if configured_profile is not None:
# match by exact token first, then by name
for p in valid_profiles:
if p.token == configured_profile:
profile = p
break
if profile is None:
for p in valid_profiles:
if getattr(p, "Name", None) == configured_profile:
profile = p
break
profile = next(
(
p
for key in ("token", "Name")
for p in valid_profiles
if getattr(p, key, None) == configured_profile
),
None,
)
if profile is None:
available = [
f"name='{getattr(p, 'Name', None)}', token='{p.token}'"
@@ -304,39 +289,30 @@ class OnvifController:
logger.debug(f"Selected Onvif profile for {camera_name}: {profile}")
# get the PTZ config for the profile
try:
configs = profile.PTZConfiguration
logger.debug(
f"Onvif ptz config for media profile in {camera_name}: {configs}"
)
except Exception as e:
logger.error(
f"Invalid Onvif PTZ configuration for camera: {camera_name}: {e}"
)
return False
configs = profile.PTZConfiguration
logger.debug(f"Onvif ptz config for media profile in {camera_name}: {configs}")
ptz: ONVIFService = await onvif.create_ptz_service()
self.cams[camera_name]["ptz"] = ptz
cam["ptz"] = ptz
try:
imaging: ONVIFService = await onvif.create_imaging_service()
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.debug(f"Imaging service not supported for {camera_name}: {e}")
imaging = None
self.cams[camera_name]["imaging"] = imaging
cam["imaging"] = imaging
try:
video_sources = await media.GetVideoSources()
if video_sources and len(video_sources) > 0:
self.cams[camera_name]["video_source_token"] = video_sources[0].token
except (Fault, ONVIFError, TransportError, Exception) as e:
cam["video_source_token"] = video_sources[0].token
except Exception as e:
logger.debug(f"Unable to get video sources for {camera_name}: {e}")
self.cams[camera_name]["video_source_token"] = None
cam["video_source_token"] = None
# setup continuous moving request
move_request = ptz.create_type("ContinuousMove")
move_request.ProfileToken = profile.token
self.cams[camera_name]["move_request"] = move_request
cam["move_request"] = move_request
# get PTZ configuration options for feature detection and relative movement
ptz_config = None
@@ -349,7 +325,7 @@ class OnvifController:
logger.debug(
f"Onvif PTZ configuration options for {camera_name}: {ptz_config}"
)
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.debug(
f"Unable to get PTZ configuration options for {camera_name}: {e}"
)
@@ -375,18 +351,6 @@ class OnvifController:
autotracking_config.enabled_in_config and autotracking_config.enabled
)
# these are local and cost nothing to build, and autotracking can be enabled
# after a camera is initialized, so always create them rather than baking the
# current config value into init state
status_request = ptz.create_type("GetStatus")
status_request.ProfileToken = profile.token
self.cams[camera_name]["status_request"] = status_request
service_capabilities_request = ptz.create_type("GetServiceCapabilities")
self.cams[camera_name]["service_capabilities_request"] = (
service_capabilities_request
)
# setup relative move request when FOV relative movement is supported
if (
fov_space_id is not None
@@ -395,9 +359,7 @@ class OnvifController:
# one-off GetStatus to seed Translation field
status = None
try:
one_off_status_request = ptz.create_type("GetStatus")
one_off_status_request.ProfileToken = profile.token
status = await ptz.GetStatus(one_off_status_request)
status = await ptz.GetStatus({"ProfileToken": profile.token})
logger.debug(f"Onvif status for {camera_name}: {status}")
except Exception as e:
logger.warning(f"Unable to get status from camera {camera_name}: {e}")
@@ -424,7 +386,7 @@ class OnvifController:
# configure zoom on relative move request
if (
autotracking_enabled
and autotracking_config.zooming != ZoomingModeEnum.disabled
and autotracking_config.zooming == ZoomingModeEnum.relative
):
zoom_space_id = next(
(
@@ -463,21 +425,16 @@ class OnvifController:
)
if rel_move_request.Speed is None:
rel_move_request.Speed = configs.DefaultPTZSpeed if configs else None
rel_move_request.Speed = configs.DefaultPTZSpeed
logger.debug(
f"{camera_name}: Relative move request after setup: {rel_move_request}"
)
self.cams[camera_name]["relative_move_request"] = rel_move_request
# setup absolute move request
abs_move_request = ptz.create_type("AbsoluteMove")
abs_move_request.ProfileToken = profile.token
self.cams[camera_name]["absolute_move_request"] = abs_move_request
cam["relative_move_request"] = rel_move_request
# setup existing presets
try:
presets: list[dict] = await ptz.GetPresets({"ProfileToken": profile.token})
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.warning(f"Unable to get presets from camera: {camera_name}: {e}")
presets = []
@@ -490,7 +447,7 @@ class OnvifController:
preset_name = preset_name.encode("latin-1").decode("utf-8")
except (UnicodeEncodeError, UnicodeDecodeError):
pass
self.cams[camera_name]["presets"][preset_name.lower()] = preset["token"]
cam["presets"][preset_name.lower()] = preset["token"]
# get list of supported features
supported_features = []
@@ -504,64 +461,40 @@ class OnvifController:
if configs.DefaultRelativePanTiltTranslationSpace:
supported_features.append("pt-r")
spaces = getattr(ptz_config, "Spaces", None)
if configs.DefaultRelativeZoomTranslationSpace:
supported_features.append("zoom-r")
if ptz_config is not None:
try:
self.cams[camera_name]["relative_zoom_range"] = (
ptz_config.Spaces.RelativeZoomTranslationSpace[0]
)
except Exception as e:
if autotracking_config.zooming == ZoomingModeEnum.relative:
autotracking_config.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: Relative zoom not supported. Exception: {e}"
)
if getattr(spaces, "RelativeZoomTranslationSpace", None):
cam["relative_zoom_range"] = spaces.RelativeZoomTranslationSpace[0]
if configs.DefaultAbsoluteZoomPositionSpace:
supported_features.append("zoom-a")
if ptz_config is not None:
try:
self.cams[camera_name]["absolute_zoom_range"] = (
ptz_config.Spaces.AbsoluteZoomPositionSpace[0]
)
self.cams[camera_name]["zoom_limits"] = configs.ZoomLimits
except Exception as e:
if autotracking_config.zooming != ZoomingModeEnum.disabled:
autotracking_config.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: Absolute zoom not supported. Exception: {e}"
)
if getattr(spaces, "AbsoluteZoomPositionSpace", None):
cam["absolute_zoom_range"] = spaces.AbsoluteZoomPositionSpace[0]
# disable autotracking zoom if required ranges are unavailable
if autotracking_config.zooming != ZoomingModeEnum.disabled:
if autotracking_config.zooming == ZoomingModeEnum.relative:
if "relative_zoom_range" not in self.cams[camera_name]:
autotracking_config.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: Relative zoom range unavailable"
)
if autotracking_config.zooming == ZoomingModeEnum.absolute:
if "absolute_zoom_range" not in self.cams[camera_name]:
autotracking_config.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: Absolute zoom range unavailable"
)
if (
self.cams[camera_name]["video_source_token"] is not None
and imaging is not None
# autotracking zoom needs the range for its mode, and get_camera_status
# reads the absolute range in both modes
zooming = autotracking_config.zooming
if zooming != ZoomingModeEnum.disabled and (
"absolute_zoom_range" not in cam or f"{zooming.value}_zoom_range" not in cam
):
autotracking_config.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: {zooming.value} zoom range unavailable"
)
if cam["video_source_token"] is not None and imaging is not None:
try:
imaging_capabilities = await imaging.GetImagingSettings(
{"VideoSourceToken": self.cams[camera_name]["video_source_token"]}
{"VideoSourceToken": cam["video_source_token"]}
)
if (
hasattr(imaging_capabilities, "Focus")
and imaging_capabilities.Focus
):
supported_features.append("focus")
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.debug(f"Focus not supported for {camera_name}: {e}")
# detect FOV relative movement support
@@ -570,17 +503,18 @@ class OnvifController:
and configs.DefaultRelativePanTiltTranslationSpace is not None
):
supported_features.append("pt-r-fov")
self.cams[camera_name]["relative_fov_range"] = (
cam["relative_fov_range"] = (
ptz_config.Spaces.RelativePanTiltTranslationSpace[fov_space_id]
)
self.cams[camera_name]["features"] = supported_features
self.cams[camera_name]["init"] = True
cam["features"] = supported_features
cam["init"] = True
return True
async def _stop(self, camera_name: str) -> None:
move_request = self.cams[camera_name]["move_request"]
await self.cams[camera_name]["ptz"].Stop(
cam = self.cams[camera_name]
move_request = cam["move_request"]
await cam["ptz"].Stop(
{
"ProfileToken": move_request.ProfileToken,
"PanTilt": True,
@@ -588,60 +522,46 @@ class OnvifController:
}
)
if (
"focus" in self.cams[camera_name]["features"]
and self.cams[camera_name]["video_source_token"]
and self.cams[camera_name]["imaging"] is not None
"focus" in cam["features"]
and cam["video_source_token"]
and cam["imaging"] is not None
):
try:
stop_request = self.cams[camera_name]["imaging"].create_type("Stop")
stop_request.VideoSourceToken = self.cams[camera_name][
"video_source_token"
]
await self.cams[camera_name]["imaging"].Stop(stop_request)
except (Fault, ONVIFError, TransportError, Exception) as e:
stop_request = cam["imaging"].create_type("Stop")
stop_request.VideoSourceToken = cam["video_source_token"]
await cam["imaging"].Stop(stop_request)
except Exception as e:
logger.warning(f"Failed to stop focus for {camera_name}: {e}")
self.cams[camera_name]["active"] = False
cam["active"] = False
async def _move(self, camera_name: str, command: OnvifCommandEnum) -> None:
if self.cams[camera_name]["active"]:
cam = self.cams[camera_name]
if cam["active"]:
logger.warning(
f"{camera_name} is already performing an action, stopping..."
)
await self._stop(camera_name)
if "pt" not in self.cams[camera_name]["features"]:
if "pt" not in cam["features"]:
logger.error(f"{camera_name} does not support ONVIF pan/tilt movement.")
return
self.cams[camera_name]["active"] = True
move_request = self.cams[camera_name]["move_request"]
cam["active"] = True
move_request = cam["move_request"]
if command == OnvifCommandEnum.move_left:
move_request.Velocity = {"PanTilt": {"x": -0.5, "y": 0}}
elif command == OnvifCommandEnum.move_right:
move_request.Velocity = {"PanTilt": {"x": 0.5, "y": 0}}
elif command == OnvifCommandEnum.move_up:
move_request.Velocity = {
"PanTilt": {
"x": 0,
"y": 0.5,
}
}
elif command == OnvifCommandEnum.move_down:
move_request.Velocity = {
"PanTilt": {
"x": 0,
"y": -0.5,
}
}
x, y = PAN_TILT_VELOCITY[command]
move_request.Velocity = {"PanTilt": {"x": x, "y": y}}
try:
await self.cams[camera_name]["ptz"].ContinuousMove(move_request)
except (Fault, ONVIFError, TransportError, Exception) as e:
await cam["ptz"].ContinuousMove(move_request)
except Exception as e:
logger.warning(f"Onvif sending move request to {camera_name} failed: {e}")
async def _move_relative(self, camera_name: str, pan, tilt, zoom, speed) -> None:
if "pt-r-fov" not in self.cams[camera_name]["features"]:
cam = self.cams[camera_name]
if "pt-r-fov" not in cam["features"]:
logger.error(f"{camera_name} does not support ONVIF RelativeMove (FOV).")
return
@@ -654,13 +574,13 @@ class OnvifController:
f"{camera_name} called RelativeMove: pan: {pan} tilt: {tilt} zoom: {zoom}"
)
if self.cams[camera_name]["active"]:
if cam["active"]:
logger.warning(
f"{camera_name} is already performing an action, not moving..."
)
return
self.cams[camera_name]["active"] = True
cam["active"] = True
# only track start_time for autotracking
if metrics.autotracker_enabled.value:
@@ -669,7 +589,7 @@ class OnvifController:
metrics.start_time.value = metrics.frame_time.value
metrics.stop_time.value = 0
move_request = self.cams[camera_name]["relative_move_request"]
move_request = cam["relative_move_request"]
# function takes in -1 to 1 for pan and tilt, interpolate to the values of the camera.
# The onvif spec says this can report as +INF and -INF, so this may need to be modified
@@ -677,55 +597,49 @@ class OnvifController:
pan,
[-1, 1],
[
self.cams[camera_name]["relative_fov_range"]["XRange"]["Min"],
self.cams[camera_name]["relative_fov_range"]["XRange"]["Max"],
cam["relative_fov_range"]["XRange"]["Min"],
cam["relative_fov_range"]["XRange"]["Max"],
],
)
tilt = numpy.interp(
tilt,
[-1, 1],
[
self.cams[camera_name]["relative_fov_range"]["YRange"]["Min"],
self.cams[camera_name]["relative_fov_range"]["YRange"]["Max"],
cam["relative_fov_range"]["YRange"]["Min"],
cam["relative_fov_range"]["YRange"]["Max"],
],
)
move_request.Speed = {
"PanTilt": {
"x": speed,
"y": speed,
},
}
move_speed = {"PanTilt": {"x": speed, "y": speed}}
move_request.Translation.PanTilt.x = pan
move_request.Translation.PanTilt.y = tilt
# include zoom if requested and camera supports relative zoom
if zoom != 0 and "zoom-r" in self.cams[camera_name]["features"]:
move_request.Speed = {
"PanTilt": {
"x": speed,
"y": speed,
},
"Zoom": {"x": speed},
}
include_zoom = zoom != 0 and "zoom-r" in cam["features"]
if include_zoom:
move_speed["Zoom"] = {"x": speed}
move_request["Translation"]["Zoom"] = {"x": zoom}
await self.cams[camera_name]["ptz"].RelativeMove(move_request)
move_request.Speed = move_speed
await cam["ptz"].RelativeMove(move_request)
# reset after the move request
move_request.Translation.PanTilt.x = 0
move_request.Translation.PanTilt.y = 0
if zoom != 0 and "zoom-r" in self.cams[camera_name]["features"]:
if include_zoom:
del move_request["Translation"]["Zoom"]
self.cams[camera_name]["active"] = False
cam["active"] = False
async def _move_to_preset(self, camera_name: str, preset: str) -> None:
cam = self.cams[camera_name]
preset = preset.lower()
if preset not in self.cams[camera_name]["presets"]:
if preset not in cam["presets"]:
logger.error(f"{preset} is not a valid preset for {camera_name}")
return
@@ -734,44 +648,48 @@ class OnvifController:
if metrics is None:
return
self.cams[camera_name]["active"] = True
cam["active"] = True
metrics.start_time.value = 0
metrics.stop_time.value = 0
move_request = self.cams[camera_name]["move_request"]
preset_token = self.cams[camera_name]["presets"][preset]
move_request = cam["move_request"]
preset_token = cam["presets"][preset]
await self.cams[camera_name]["ptz"].GotoPreset(
await cam["ptz"].GotoPreset(
{
"ProfileToken": move_request.ProfileToken,
"PresetToken": preset_token,
}
)
self.cams[camera_name]["active"] = False
cam["active"] = False
async def _zoom(self, camera_name: str, command: OnvifCommandEnum) -> None:
if self.cams[camera_name]["active"]:
cam = self.cams[camera_name]
if cam["active"]:
logger.warning(
f"{camera_name} is already performing an action, stopping..."
)
await self._stop(camera_name)
if "zoom" not in self.cams[camera_name]["features"]:
if "zoom" not in cam["features"]:
logger.error(f"{camera_name} does not support ONVIF zooming.")
return
self.cams[camera_name]["active"] = True
move_request = self.cams[camera_name]["move_request"]
cam["active"] = True
move_request = cam["move_request"]
if command == OnvifCommandEnum.zoom_in:
move_request.Velocity = {"Zoom": {"x": 0.5}}
elif command == OnvifCommandEnum.zoom_out:
move_request.Velocity = {"Zoom": {"x": -0.5}}
await self.cams[camera_name]["ptz"].ContinuousMove(move_request)
await cam["ptz"].ContinuousMove(move_request)
async def _zoom_absolute(self, camera_name: str, zoom, speed) -> None:
if "zoom-a" not in self.cams[camera_name]["features"]:
cam = self.cams[camera_name]
if "zoom-a" not in cam["features"]:
logger.error(f"{camera_name} does not support ONVIF AbsoluteMove zooming.")
return
@@ -782,56 +700,59 @@ class OnvifController:
logger.debug(f"{camera_name} called AbsoluteMove: zoom: {zoom}")
if self.cams[camera_name]["active"]:
if cam["active"]:
logger.warning(
f"{camera_name} is already performing an action, not moving..."
)
return
self.cams[camera_name]["active"] = True
cam["active"] = True
metrics.motor_stopped.clear()
logger.debug(f"{camera_name}: PTZ start time: {metrics.frame_time.value}")
metrics.start_time.value = metrics.frame_time.value
metrics.stop_time.value = 0
move_request = self.cams[camera_name]["absolute_move_request"]
# function takes in 0 to 1 for zoom, interpolate to the values of the camera.
zoom = numpy.interp(
zoom,
[0, 1],
[
self.cams[camera_name]["absolute_zoom_range"]["XRange"]["Min"],
self.cams[camera_name]["absolute_zoom_range"]["XRange"]["Max"],
cam["absolute_zoom_range"]["XRange"]["Min"],
cam["absolute_zoom_range"]["XRange"]["Max"],
],
)
move_request.Speed = {"Zoom": speed}
move_request.Position = {"Zoom": zoom}
logger.debug(f"{camera_name}: Absolute zoom: {zoom}")
await self.cams[camera_name]["ptz"].AbsoluteMove(move_request)
await cam["ptz"].AbsoluteMove(
{
"ProfileToken": cam["move_request"].ProfileToken,
"Position": {"Zoom": zoom},
"Speed": {"Zoom": speed},
}
)
self.cams[camera_name]["active"] = False
cam["active"] = False
async def _focus(self, camera_name: str, command: OnvifCommandEnum) -> None:
if self.cams[camera_name]["active"]:
cam = self.cams[camera_name]
if cam["active"]:
logger.warning(
f"{camera_name} is already performing an action, not moving..."
)
await self._stop(camera_name)
if (
"focus" not in self.cams[camera_name]["features"]
or not self.cams[camera_name]["video_source_token"]
or self.cams[camera_name]["imaging"] is None
"focus" not in cam["features"]
or not cam["video_source_token"]
or cam["imaging"] is None
):
logger.error(f"{camera_name} does not support ONVIF continuous focus.")
return
self.cams[camera_name]["active"] = True
move_request = self.cams[camera_name]["imaging"].create_type("Move")
move_request.VideoSourceToken = self.cams[camera_name]["video_source_token"]
cam["active"] = True
move_request = cam["imaging"].create_type("Move")
move_request.VideoSourceToken = cam["video_source_token"]
move_request.Focus = {
"Continuous": {
"Speed": 0.5 if command == OnvifCommandEnum.focus_in else -0.5
@@ -839,10 +760,10 @@ class OnvifController:
}
try:
await self.cams[camera_name]["imaging"].Move(move_request)
except (Fault, ONVIFError, TransportError, Exception) as e:
await cam["imaging"].Move(move_request)
except Exception as e:
logger.warning(f"Onvif sending focus request to {camera_name} failed: {e}")
self.cams[camera_name]["active"] = False
cam["active"] = False
async def handle_command_async(
self, camera_name: str, command: OnvifCommandEnum, param: str = ""
@@ -883,7 +804,7 @@ class OnvifController:
await self._focus(camera_name, command)
else:
await self._move(camera_name, command)
except (Fault, ONVIFError, TransportError, Exception) as e:
except Exception as e:
logger.error(f"Unable to handle onvif command: {e}")
def handle_command(
@@ -906,6 +827,15 @@ class OnvifController:
f"Error executing command {command} for camera {camera_name}: {e}"
)
def _camera_info(self, camera_name: str) -> dict[str, Any]:
cam = self.cams[camera_name]
return {
"name": camera_name,
"features": cam["features"],
"presets": list(cam["presets"]),
"profiles": cam["profiles"],
}
async def get_camera_info(self, camera_name: str) -> dict[str, Any]:
"""
Get ptz capabilities and presets, attempting to reconnect if ONVIF is configured
@@ -929,60 +859,21 @@ class OnvifController:
return {}
if camera_name in self.cams.keys() and self.cams[camera_name]["init"]:
return {
"name": camera_name,
"features": self.cams[camera_name]["features"],
"presets": list(self.cams[camera_name]["presets"].keys()),
"profiles": self.cams[camera_name].get("profiles", []),
}
return self._camera_info(camera_name)
if camera_name not in self.cams.keys() and camera_name in self.config.cameras:
success = await self._init_single_camera(camera_name)
if not success:
return {}
# Reset retry count after timeout
attempts = self.failed_cams.get(camera_name, {}).get("retry_attempts", 0)
last_attempt = self.failed_cams.get(camera_name, {}).get("last_attempt", 0)
failed = self.failed_cams.get(camera_name, {})
attempts = failed.get("retry_attempts", 0)
last_attempt = failed.get("last_attempt", 0)
# Reset retry count after timeout
if last_attempt and (time.time() - last_attempt) > self.reset_timeout:
logger.debug(f"Resetting retry count for {camera_name} after timeout")
attempts = 0
self.failed_cams[camera_name]["retry_attempts"] = 0
# Attempt initialization/reconnection
if attempts < self.max_retries:
logger.info(
f"Attempting ONVIF initialization for {camera_name} (retry {attempts + 1}/{self.max_retries})"
)
try:
if await self._init_onvif(camera_name):
if camera_name in self.failed_cams:
del self.failed_cams[camera_name]
return {
"name": camera_name,
"features": self.cams[camera_name]["features"],
"presets": list(self.cams[camera_name]["presets"].keys()),
}
else:
logger.warning(f"ONVIF initialization failed for {camera_name}")
self.failed_cams[camera_name] = {
"retry_attempts": attempts + 1,
"last_attempt": time.time(),
}
except Exception as e:
logger.error(
f"Error during ONVIF initialization for {camera_name}: {e}"
)
if camera_name not in self.failed_cams:
self.failed_cams[camera_name] = {"retry_attempts": 0}
self.failed_cams[camera_name].update(
{
"retry_attempts": attempts + 1,
"last_error": str(e),
"last_attempt": time.time(),
}
)
if attempts >= self.max_retries:
remaining_time = max(
@@ -991,8 +882,24 @@ class OnvifController:
logger.error(
f"Too many ONVIF initialization attempts for {camera_name}, retry in {remaining_time} minute{'s' if remaining_time != 1 else ''}"
)
return {}
logger.debug(f"Could not initialize ONVIF for {camera_name}")
logger.info(
f"Attempting ONVIF initialization for {camera_name} (retry {attempts + 1}/{self.max_retries})"
)
try:
if await self._init_onvif(camera_name):
self.failed_cams.pop(camera_name, None)
return self._camera_info(camera_name)
logger.warning(f"ONVIF initialization failed for {camera_name}")
except Exception as e:
logger.error(f"Error during ONVIF initialization for {camera_name}: {e}")
self.failed_cams[camera_name] = {
"retry_attempts": attempts + 1,
"last_attempt": time.time(),
}
return {}
async def get_service_capabilities(self, camera_name: str) -> None:
@@ -1000,16 +907,13 @@ class OnvifController:
logger.error(f"ONVIF is not configured for {camera_name}")
return {}
if not self.cams[camera_name]["init"]:
cam = self.cams[camera_name]
if not cam["init"]:
await self._init_onvif(camera_name)
service_capabilities_request = self.cams[camera_name][
"service_capabilities_request"
]
try:
service_capabilities = await self.cams[camera_name][
"ptz"
].GetServiceCapabilities(service_capabilities_request)
service_capabilities = await cam["ptz"].GetServiceCapabilities()
logger.debug(
f"Onvif service capabilities for {camera_name}: {service_capabilities}"
@@ -1035,13 +939,16 @@ class OnvifController:
if metrics is None or camera_config is None:
return
if not self.cams[camera_name]["init"]:
cam = self.cams[camera_name]
if not cam["init"]:
if not await self._init_onvif(camera_name):
return
status_request = self.cams[camera_name]["status_request"]
try:
status = await self.cams[camera_name]["ptz"].GetStatus(status_request)
status = await cam["ptz"].GetStatus(
{"ProfileToken": cam["move_request"].ProfileToken}
)
except Exception:
pass # We're unsupported, that'll be reported in the next check.
@@ -1072,7 +979,7 @@ class OnvifController:
if pan_tilt_status == "IDLE" and (
zoom_status is None or zoom_status == "IDLE"
):
self.cams[camera_name]["active"] = False
cam["active"] = False
if not metrics.motor_stopped.is_set():
metrics.motor_stopped.set()
@@ -1082,7 +989,7 @@ class OnvifController:
metrics.stop_time.value = metrics.frame_time.value
else:
self.cams[camera_name]["active"] = True
cam["active"] = True
if metrics.motor_stopped.is_set():
metrics.motor_stopped.clear()
@@ -1098,8 +1005,8 @@ class OnvifController:
metrics.zoom_level.value = numpy.interp(
round(status.Position.Zoom.x, 2),
[
self.cams[camera_name]["absolute_zoom_range"]["XRange"]["Min"],
self.cams[camera_name]["absolute_zoom_range"]["XRange"]["Max"],
cam["absolute_zoom_range"]["XRange"]["Min"],
cam["absolute_zoom_range"]["XRange"]["Max"],
],
[0, 1],
)
@@ -1138,7 +1045,7 @@ class OnvifController:
def close(self) -> None:
"""Gracefully shut down the ONVIF controller."""
if not hasattr(self, "loop") or self.loop.is_closed():
if self.loop.is_closed():
logger.debug("ONVIF controller already closed")
return
@@ -1155,13 +1062,6 @@ class OnvifController:
self.config_subscriber.stop()
def stop_and_cleanup():
try:
self.loop.stop()
except Exception as e:
logger.error(f"Error during loop cleanup: {e}")
# Schedule stop and cleanup in the loop thread
self.loop.call_soon_threadsafe(stop_and_cleanup)
self.loop.call_soon_threadsafe(self.loop.stop)
self.loop_thread.join()
frigate/test/test_ptz_autotrack.py
+22
View File
@@ -83,6 +83,28 @@ class TestAutotrackerInitGuards(unittest.IsolatedAsyncioTestCase):
tracker.onvif.get_camera_status.assert_not_called()
class TestAutotrackerEnqueueMove(unittest.TestCase):
def _enqueue(self, pan: float, tilt: float, zoom: float) -> MagicMock:
tracker = _make_tracker()
tracker.move_queues = {CAMERA: MagicMock()}
tracker.move_queue_locks = {CAMERA: MagicMock()}
tracker.move_queue_locks[CAMERA].locked.return_value = False
tracker._enqueue_move(CAMERA, 1000.0, pan, tilt, zoom)
return tracker.onvif.loop.call_soon_threadsafe
def test_move_is_clipped_to_the_onvif_range(self) -> None:
# velocity estimates can push the predicted centroid outside the frame
call_soon = self._enqueue(1.7, -2.5, 0.4)
call_soon.assert_called_once()
self.assertEqual(call_soon.call_args.args[1], (1000.0, 1.0, -1.0, 0.4))
def test_empty_move_is_not_enqueued(self) -> None:
self._enqueue(0, 0, 0).assert_not_called()
class TestAutotrackerMetricSync(unittest.TestCase):
def test_metric_follows_config_when_enabled_by_update(self) -> None:
# autotracking enabled via a config save: the metric was seeded False when
frigate/test/test_ptz_onvif.py
+16 -36
View File
@@ -2,14 +2,9 @@
Regression coverage for a camera that is initialized while autotracking is off and
has it enabled later, which is the normal wizard flow: set the camera up first,
configure autotracking afterwards. The autotracking-only request objects used to
be created only when autotracking was enabled at init time, so the camera was left
with init=True but no status_request. get_camera_status skips its re-init branch
when init is True, so it went straight to the missing key and raised KeyError on
the tracking thread.
The request objects are built from the locally parsed WSDL and cost no network, so
they are always created and init=True now implies they exist.
configure autotracking afterwards. get_camera_status skips its re-init branch when
init is True, so everything it reads must exist whether or not autotracking was
enabled at init time.
Also covers the inverse direction: the ptz movement timestamps must not be written
for a camera that has autotracking off, because nothing clears them back out.
@@ -99,7 +94,6 @@ def _make_controller(autotracking_enabled: bool) -> OnvifController:
controller.config = config
controller.cams = {CAMERA: {"onvif": _make_onvif_camera(), "init": False}}
controller.failed_cams = {}
controller.camera_configs = {CAMERA: config.cameras[CAMERA]}
controller.ptz_metrics = {CAMERA: MagicMock()}
return controller
@@ -110,7 +104,6 @@ def _make_move_controller(autotracking_enabled: bool) -> OnvifController:
config = _config(autotracking_enabled)
controller = OnvifController.__new__(OnvifController)
controller.config = config
controller.camera_configs = {CAMERA: config.cameras[CAMERA]}
controller.failed_cams = {}
ptz = MagicMock()
@@ -135,38 +128,25 @@ def _make_move_controller(autotracking_enabled: bool) -> OnvifController:
class TestOnvifInitRequests(unittest.IsolatedAsyncioTestCase):
async def test_status_request_created_when_autotracking_disabled(self) -> None:
async def test_camera_status_independent_of_autotracking_at_init(self) -> None:
# the wizard flow: onvif configured first, autotracking enabled later
controller = _make_controller(autotracking_enabled=False)
self.assertTrue(await controller._init_onvif(CAMERA))
cam = controller.cams[CAMERA]
self.assertTrue(cam["init"])
self.assertIn("status_request", cam)
self.assertIn("service_capabilities_request", cam)
async def test_status_request_created_when_autotracking_enabled(self) -> None:
controller = _make_controller(autotracking_enabled=True)
self.assertTrue(await controller._init_onvif(CAMERA))
cam = controller.cams[CAMERA]
self.assertIn("status_request", cam)
self.assertIn("service_capabilities_request", cam)
async def test_init_implies_status_request_exists(self) -> None:
# the invariant get_camera_status relies on: it skips re-init when init is
# True and then reads status_request without guarding
for autotracking_enabled in (True, False):
with self.subTest(autotracking_enabled=autotracking_enabled):
controller = _make_controller(autotracking_enabled)
controller.status_locks = {CAMERA: asyncio.Lock()}
await controller._init_onvif(CAMERA)
self.assertTrue(await controller._init_onvif(CAMERA))
cam = controller.cams[CAMERA]
if cam["init"]:
self.assertEqual(cam["status_request"].request_type, "GetStatus")
status = MagicMock()
status.MoveStatus.PanTilt = "IDLE"
status.MoveStatus.Zoom = "IDLE"
ptz = controller.cams[CAMERA]["ptz"]
ptz.GetStatus = AsyncMock(return_value=status)
await controller.get_camera_status(CAMERA)
ptz.GetStatus.assert_awaited_once_with({"ProfileToken": "profile_1"})
self.assertFalse(controller.cams[CAMERA]["active"])
async def test_requests_built_without_contacting_camera(self) -> None:
# create_type is a local WSDL lookup; cameras that do not implement
frigate/track/object_processing.py
+4 -4
View File
@@ -41,7 +41,7 @@ from frigate.const import (
)
from frigate.events.types import EventStateEnum, EventTypeEnum
from frigate.models import Event, ReviewSegment, Timeline
from frigate.ptz.autotrack import PtzAutoTrackerThread
from frigate.ptz.autotrack import PtzAutoTracker
from frigate.track.tracked_object import TrackedObject
from frigate.util.image import SharedMemoryFrameManager
@@ -60,7 +60,7 @@ class TrackedObjectProcessor(threading.Thread):
config: FrigateConfig,
dispatcher: Dispatcher,
tracked_objects_queue: MpQueue,
ptz_autotracker_thread: PtzAutoTrackerThread,
ptz_autotracker_thread: PtzAutoTracker,
stop_event: MpEvent,
) -> None:
super().__init__(name="detected_frames_processor")
@@ -153,7 +153,7 @@ class TrackedObjectProcessor(threading.Thread):
)
def autotrack(camera: str, obj: TrackedObject, frame_name: str) -> None:
self.ptz_autotracker_thread.ptz_autotracker.autotrack_object(camera, obj)
self.ptz_autotracker_thread.autotrack_object(camera, obj)
def end(camera: str, obj: TrackedObject, frame_name: str) -> None:
# populate has_snapshot
@@ -177,7 +177,7 @@ class TrackedObjectProcessor(threading.Thread):
"type": "end",
}
self.dispatcher.publish("events", json.dumps(message), retain=False)
self.ptz_autotracker_thread.ptz_autotracker.end_object(camera, obj)
self.ptz_autotracker_thread.end_object(camera, obj)
self.event_sender.publish(
(