forked from Github/frigate
Basic PTZ object autotracking functionality (#6913)
* Basic functionality
* Threaded motion estimator
* Revert "Threaded motion estimator"
This reverts commit 3171801607.
* Don't detect motion when ptz is moving
* fix motion logic
* fix mypy error
* Add threaded queue for movement for slower ptzs
* Move queues per camera
* Move autotracker start to app.py
* iou value for tracked object
* mqtt callback
* tracked object should be initially motionless
* only draw thicker box if autotracking is enabled
* Init if enabled when initially disabled in config
* Fix init
* Thread names
* Always use motion estimator
* docs
* clarify fov support
* remove size ratio
* use mp event instead of value for ptz status
* update autotrack at half fps
* fix merge conflict
* fix event type for mypy
* clean up
* Clean up
* remove unused code
* merge conflict fix
* docs: update link to object_detectors page
* Update docs/docs/configuration/autotracking.md
Co-authored-by: Nicolas Mowen <nickmowen213@gmail.com>
* clarify wording
* pass actual instances directly
* default return preset
* fix type
* Error message when onvif init fails
* disable autotracking if onvif init fails
* disable autotracking if onvif init fails
* ptz module
* verify required_zones in config
* update util after dev merge
---------
Co-authored-by: Nicolas Mowen <nickmowen213@gmail.com>
This commit is contained in:
Josh Hawkins
362
frigate/ptz/autotrack.py
Normal file
362
frigate/ptz/autotrack.py
Normal file
@@ -0,0 +1,362 @@
|
||||
"""Automatically pan, tilt, and zoom on detected objects via onvif."""
|
||||
|
||||
import copy
|
||||
import logging
|
||||
import queue
|
||||
import threading
|
||||
import time
|
||||
from functools import partial
|
||||
from multiprocessing.synchronize import Event as MpEvent
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
from norfair.camera_motion import MotionEstimator, TranslationTransformationGetter
|
||||
|
||||
from frigate.config import CameraConfig, FrigateConfig
|
||||
from frigate.ptz.onvif import OnvifController
|
||||
from frigate.types import CameraMetricsTypes
|
||||
from frigate.util.image import SharedMemoryFrameManager, intersection_over_union
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class PtzMotionEstimator:
|
||||
def __init__(self, config: CameraConfig, ptz_stopped) -> None:
|
||||
self.frame_manager = SharedMemoryFrameManager()
|
||||
# homography is nice (zooming) but slow, translation is pan/tilt only but fast.
|
||||
self.norfair_motion_estimator = MotionEstimator(
|
||||
transformations_getter=TranslationTransformationGetter(),
|
||||
min_distance=30,
|
||||
max_points=500,
|
||||
)
|
||||
self.camera_config = config
|
||||
self.coord_transformations = None
|
||||
self.ptz_stopped = ptz_stopped
|
||||
logger.debug(f"Motion estimator init for cam: {config.name}")
|
||||
|
||||
def motion_estimator(self, detections, frame_time, camera_name):
|
||||
if (
|
||||
self.camera_config.onvif.autotracking.enabled
|
||||
and not self.ptz_stopped.is_set()
|
||||
):
|
||||
logger.debug(
|
||||
f"Motion estimator running for {camera_name} - frame time: {frame_time}"
|
||||
)
|
||||
|
||||
frame_id = f"{camera_name}{frame_time}"
|
||||
yuv_frame = self.frame_manager.get(
|
||||
frame_id, self.camera_config.frame_shape_yuv
|
||||
)
|
||||
|
||||
frame = cv2.cvtColor(yuv_frame, cv2.COLOR_YUV2GRAY_I420)
|
||||
|
||||
# mask out detections for better motion estimation
|
||||
mask = np.ones(frame.shape[:2], frame.dtype)
|
||||
|
||||
detection_boxes = [x[2] for x in detections]
|
||||
for detection in detection_boxes:
|
||||
x1, y1, x2, y2 = detection
|
||||
mask[y1:y2, x1:x2] = 0
|
||||
|
||||
# merge camera config motion mask with detections. Norfair function needs 0,1 mask
|
||||
mask = np.bitwise_and(mask, self.camera_config.motion.mask).clip(max=1)
|
||||
|
||||
# Norfair estimator function needs color so it can convert it right back to gray
|
||||
frame = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGRA)
|
||||
|
||||
self.coord_transformations = self.norfair_motion_estimator.update(
|
||||
frame, mask
|
||||
)
|
||||
|
||||
self.frame_manager.close(frame_id)
|
||||
|
||||
logger.debug(
|
||||
f"Motion estimator transformation: {self.coord_transformations.rel_to_abs((0,0))}"
|
||||
)
|
||||
|
||||
return self.coord_transformations
|
||||
|
||||
return None
|
||||
|
||||
|
||||
class PtzAutoTrackerThread(threading.Thread):
|
||||
def __init__(
|
||||
self,
|
||||
config: FrigateConfig,
|
||||
onvif: OnvifController,
|
||||
camera_metrics: dict[str, CameraMetricsTypes],
|
||||
stop_event: MpEvent,
|
||||
) -> None:
|
||||
threading.Thread.__init__(self)
|
||||
self.name = "ptz_autotracker"
|
||||
self.ptz_autotracker = PtzAutoTracker(config, onvif, camera_metrics)
|
||||
self.stop_event = stop_event
|
||||
self.config = config
|
||||
|
||||
def run(self):
|
||||
while not self.stop_event.is_set():
|
||||
for camera_name, cam in self.config.cameras.items():
|
||||
if cam.onvif.autotracking.enabled:
|
||||
self.ptz_autotracker.camera_maintenance(camera_name)
|
||||
time.sleep(1)
|
||||
else:
|
||||
# disabled dynamically by mqtt
|
||||
if self.ptz_autotracker.tracked_object.get(camera_name):
|
||||
self.ptz_autotracker.tracked_object[camera_name] = None
|
||||
self.ptz_autotracker.tracked_object_previous[camera_name] = None
|
||||
time.sleep(0.1)
|
||||
logger.info("Exiting autotracker...")
|
||||
|
||||
|
||||
class PtzAutoTracker:
|
||||
def __init__(
|
||||
self,
|
||||
config: FrigateConfig,
|
||||
onvif: OnvifController,
|
||||
camera_metrics: CameraMetricsTypes,
|
||||
) -> None:
|
||||
self.config = config
|
||||
self.onvif = onvif
|
||||
self.camera_metrics = camera_metrics
|
||||
self.tracked_object: dict[str, object] = {}
|
||||
self.tracked_object_previous: dict[str, object] = {}
|
||||
self.object_types = {}
|
||||
self.required_zones = {}
|
||||
self.move_queues = {}
|
||||
self.move_threads = {}
|
||||
self.autotracker_init = {}
|
||||
|
||||
# if cam is set to autotrack, onvif should be set up
|
||||
for camera_name, cam in self.config.cameras.items():
|
||||
self.autotracker_init[camera_name] = False
|
||||
if cam.onvif.autotracking.enabled:
|
||||
self._autotracker_setup(cam, camera_name)
|
||||
|
||||
def _autotracker_setup(self, cam, camera_name):
|
||||
logger.debug(f"Autotracker init for cam: {camera_name}")
|
||||
|
||||
self.object_types[camera_name] = cam.onvif.autotracking.track
|
||||
self.required_zones[camera_name] = cam.onvif.autotracking.required_zones
|
||||
|
||||
self.tracked_object[camera_name] = None
|
||||
self.tracked_object_previous[camera_name] = None
|
||||
|
||||
self.move_queues[camera_name] = queue.Queue()
|
||||
|
||||
if not self.onvif.cams[camera_name]["init"]:
|
||||
if not self.onvif._init_onvif(camera_name):
|
||||
logger.warning(f"Unable to initialize onvif for {camera_name}")
|
||||
cam.onvif.autotracking.enabled = False
|
||||
self.camera_metrics[camera_name][
|
||||
"ptz_autotracker_enabled"
|
||||
].value = False
|
||||
|
||||
return
|
||||
|
||||
if not self.onvif.cams[camera_name]["relative_fov_supported"]:
|
||||
cam.onvif.autotracking.enabled = False
|
||||
self.camera_metrics[camera_name][
|
||||
"ptz_autotracker_enabled"
|
||||
].value = False
|
||||
logger.warning(
|
||||
f"Disabling autotracking for {camera_name}: FOV relative movement not supported"
|
||||
)
|
||||
|
||||
return
|
||||
|
||||
# movement thread per camera
|
||||
if not self.move_threads or not self.move_threads[camera_name]:
|
||||
self.move_threads[camera_name] = threading.Thread(
|
||||
name=f"move_thread_{camera_name}",
|
||||
target=partial(self._process_move_queue, camera_name),
|
||||
)
|
||||
self.move_threads[camera_name].daemon = True
|
||||
self.move_threads[camera_name].start()
|
||||
|
||||
self.autotracker_init[camera_name] = True
|
||||
|
||||
def _process_move_queue(self, camera):
|
||||
while True:
|
||||
try:
|
||||
if self.move_queues[camera].qsize() > 1:
|
||||
# Accumulate values since last moved
|
||||
pan = 0
|
||||
tilt = 0
|
||||
|
||||
while not self.move_queues[camera].empty():
|
||||
queued_pan, queued_tilt = self.move_queues[camera].queue[0]
|
||||
|
||||
# If exceeding the movement range, keep it in the queue and move now
|
||||
if abs(pan + queued_pan) > 1.0 or abs(tilt + queued_tilt) > 1.0:
|
||||
logger.debug("Pan or tilt value exceeds 1.0")
|
||||
break
|
||||
|
||||
queued_pan, queued_tilt = self.move_queues[camera].get()
|
||||
|
||||
pan += queued_pan
|
||||
tilt += queued_tilt
|
||||
else:
|
||||
move_data = self.move_queues[camera].get()
|
||||
pan, tilt = move_data
|
||||
|
||||
self.onvif._move_relative(camera, pan, tilt, 1)
|
||||
|
||||
# Wait until the camera finishes moving
|
||||
self.camera_metrics[camera]["ptz_stopped"].wait()
|
||||
|
||||
except queue.Empty:
|
||||
time.sleep(0.1)
|
||||
|
||||
def _enqueue_move(self, camera, pan, tilt):
|
||||
move_data = (pan, tilt)
|
||||
logger.debug(f"enqueue pan: {pan}, enqueue tilt: {tilt}")
|
||||
self.move_queues[camera].put(move_data)
|
||||
|
||||
def _autotrack_move_ptz(self, camera, obj):
|
||||
camera_config = self.config.cameras[camera]
|
||||
|
||||
# # frame width and height
|
||||
camera_width = camera_config.frame_shape[1]
|
||||
camera_height = camera_config.frame_shape[0]
|
||||
|
||||
# Normalize coordinates. top right of the fov is (1,1).
|
||||
pan = 0.5 - (obj.obj_data["centroid"][0] / camera_width)
|
||||
tilt = 0.5 - (obj.obj_data["centroid"][1] / camera_height)
|
||||
|
||||
# ideas: check object velocity for camera speed?
|
||||
self._enqueue_move(camera, -pan, tilt)
|
||||
|
||||
def autotrack_object(self, camera, obj):
|
||||
camera_config = self.config.cameras[camera]
|
||||
|
||||
# check if ptz is moving
|
||||
self.onvif.get_camera_status(camera)
|
||||
|
||||
if camera_config.onvif.autotracking.enabled:
|
||||
# either this is a brand new object that's on our camera, has our label, entered the zone, is not a false positive,
|
||||
# and is not initially motionless - or one we're already tracking, which assumes all those things are already true
|
||||
if (
|
||||
# new object
|
||||
self.tracked_object[camera] is None
|
||||
and obj.camera == camera
|
||||
and obj.obj_data["label"] in self.object_types[camera]
|
||||
and set(obj.entered_zones) & set(self.required_zones[camera])
|
||||
and not obj.previous["false_positive"]
|
||||
and not obj.false_positive
|
||||
and self.tracked_object_previous[camera] is None
|
||||
and obj.obj_data["motionless_count"] == 0
|
||||
):
|
||||
logger.debug(
|
||||
f"Autotrack: New object: {obj.obj_data['id']} {obj.obj_data['box']} {obj.obj_data['frame_time']}"
|
||||
)
|
||||
self.tracked_object[camera] = obj
|
||||
self.tracked_object_previous[camera] = copy.deepcopy(obj)
|
||||
self._autotrack_move_ptz(camera, obj)
|
||||
|
||||
return
|
||||
|
||||
if (
|
||||
# already tracking an object
|
||||
self.tracked_object[camera] is not None
|
||||
and self.tracked_object_previous[camera] is not None
|
||||
and obj.obj_data["id"] == self.tracked_object[camera].obj_data["id"]
|
||||
and obj.obj_data["frame_time"]
|
||||
!= self.tracked_object_previous[camera].obj_data["frame_time"]
|
||||
):
|
||||
# don't move the ptz if we're relatively close to the existing box
|
||||
# should we use iou or euclidean distance or both?
|
||||
# distance = math.sqrt((obj.obj_data["centroid"][0] - camera_width/2)**2 + (obj.obj_data["centroid"][1] - obj.camera_height/2)**2)
|
||||
# if distance <= (self.camera_width * .15) or distance <= (self.camera_height * .15)
|
||||
if (
|
||||
intersection_over_union(
|
||||
self.tracked_object_previous[camera].obj_data["box"],
|
||||
obj.obj_data["box"],
|
||||
)
|
||||
> 0.5
|
||||
):
|
||||
logger.debug(
|
||||
f"Autotrack: Existing object (do NOT move ptz): {obj.obj_data['id']} {obj.obj_data['box']} {obj.obj_data['frame_time']}"
|
||||
)
|
||||
self.tracked_object_previous[camera] = copy.deepcopy(obj)
|
||||
return
|
||||
|
||||
logger.debug(
|
||||
f"Autotrack: Existing object (move ptz): {obj.obj_data['id']} {obj.obj_data['box']} {obj.obj_data['frame_time']}"
|
||||
)
|
||||
self.tracked_object_previous[camera] = copy.deepcopy(obj)
|
||||
self._autotrack_move_ptz(camera, obj)
|
||||
|
||||
return
|
||||
|
||||
if (
|
||||
# The tracker lost an object, so let's check the previous object's region and compare it with the incoming object
|
||||
# If it's within bounds, start tracking that object.
|
||||
# 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 == camera
|
||||
and obj.obj_data["label"] in self.object_types[camera]
|
||||
and not obj.previous["false_positive"]
|
||||
and not obj.false_positive
|
||||
and obj.obj_data["motionless_count"] == 0
|
||||
and self.tracked_object_previous[camera] is not None
|
||||
):
|
||||
if (
|
||||
intersection_over_union(
|
||||
self.tracked_object_previous[camera].obj_data["region"],
|
||||
obj.obj_data["box"],
|
||||
)
|
||||
< 0.2
|
||||
):
|
||||
logger.debug(
|
||||
f"Autotrack: Reacquired object: {obj.obj_data['id']} {obj.obj_data['box']} {obj.obj_data['frame_time']}"
|
||||
)
|
||||
self.tracked_object[camera] = obj
|
||||
self.tracked_object_previous[camera] = copy.deepcopy(obj)
|
||||
self._autotrack_move_ptz(camera, obj)
|
||||
|
||||
return
|
||||
|
||||
def end_object(self, camera, obj):
|
||||
if self.config.cameras[camera].onvif.autotracking.enabled:
|
||||
if (
|
||||
self.tracked_object[camera] is not None
|
||||
and obj.obj_data["id"] == self.tracked_object[camera].obj_data["id"]
|
||||
):
|
||||
logger.debug(
|
||||
f"Autotrack: End object: {obj.obj_data['id']} {obj.obj_data['box']}"
|
||||
)
|
||||
self.tracked_object[camera] = None
|
||||
self.onvif.get_camera_status(camera)
|
||||
|
||||
def camera_maintenance(self, camera):
|
||||
# calls get_camera_status to check/update ptz movement
|
||||
# returns camera to preset after timeout when tracking is over
|
||||
autotracker_config = self.config.cameras[camera].onvif.autotracking
|
||||
|
||||
if not self.autotracker_init[camera]:
|
||||
self._autotracker_setup(self.config.cameras[camera], camera)
|
||||
# regularly update camera status
|
||||
if not self.camera_metrics[camera]["ptz_stopped"].is_set():
|
||||
self.onvif.get_camera_status(camera)
|
||||
|
||||
# return to preset if tracking is over
|
||||
if (
|
||||
self.tracked_object[camera] is None
|
||||
and self.tracked_object_previous[camera] is not None
|
||||
and (
|
||||
# might want to use a different timestamp here?
|
||||
time.time()
|
||||
- self.tracked_object_previous[camera].obj_data["frame_time"]
|
||||
> autotracker_config.timeout
|
||||
)
|
||||
and autotracker_config.return_preset
|
||||
):
|
||||
self.camera_metrics[camera]["ptz_stopped"].wait()
|
||||
logger.debug(
|
||||
f"Autotrack: Time is {time.time()}, returning to preset: {autotracker_config.return_preset}"
|
||||
)
|
||||
self.onvif._move_to_preset(
|
||||
camera,
|
||||
autotracker_config.return_preset.lower(),
|
||||
)
|
||||
self.tracked_object_previous[camera] = None
|
||||
366
frigate/ptz/onvif.py
Normal file
366
frigate/ptz/onvif.py
Normal file
@@ -0,0 +1,366 @@
|
||||
"""Configure and control camera via onvif."""
|
||||
|
||||
import logging
|
||||
import site
|
||||
from enum import Enum
|
||||
|
||||
import numpy
|
||||
from onvif import ONVIFCamera, ONVIFError
|
||||
|
||||
from frigate.config import FrigateConfig
|
||||
from frigate.types import CameraMetricsTypes
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class OnvifCommandEnum(str, Enum):
|
||||
"""Holds all possible move commands"""
|
||||
|
||||
init = "init"
|
||||
move_down = "move_down"
|
||||
move_left = "move_left"
|
||||
move_right = "move_right"
|
||||
move_up = "move_up"
|
||||
preset = "preset"
|
||||
stop = "stop"
|
||||
zoom_in = "zoom_in"
|
||||
zoom_out = "zoom_out"
|
||||
|
||||
|
||||
class OnvifController:
|
||||
def __init__(
|
||||
self, config: FrigateConfig, camera_metrics: dict[str, CameraMetricsTypes]
|
||||
) -> None:
|
||||
self.cams: dict[str, ONVIFCamera] = {}
|
||||
self.camera_metrics = camera_metrics
|
||||
|
||||
for cam_name, cam in config.cameras.items():
|
||||
if not cam.enabled:
|
||||
continue
|
||||
|
||||
if cam.onvif.host:
|
||||
try:
|
||||
self.cams[cam_name] = {
|
||||
"onvif": ONVIFCamera(
|
||||
cam.onvif.host,
|
||||
cam.onvif.port,
|
||||
cam.onvif.user,
|
||||
cam.onvif.password,
|
||||
wsdl_dir=site.getsitepackages()[0].replace(
|
||||
"dist-packages", "site-packages"
|
||||
)
|
||||
+ "/wsdl",
|
||||
),
|
||||
"init": False,
|
||||
"active": False,
|
||||
"presets": {},
|
||||
}
|
||||
except ONVIFError as e:
|
||||
logger.error(f"Onvif connection to {cam.name} failed: {e}")
|
||||
|
||||
def _init_onvif(self, camera_name: str) -> bool:
|
||||
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
|
||||
|
||||
# create init services
|
||||
media = onvif.create_media_service()
|
||||
|
||||
try:
|
||||
profile = media.GetProfiles()[0]
|
||||
except ONVIFError as e:
|
||||
logger.error(f"Unable to connect to camera: {camera_name}: {e}")
|
||||
return False
|
||||
|
||||
ptz = onvif.create_ptz_service()
|
||||
request = ptz.create_type("GetConfigurationOptions")
|
||||
request.ConfigurationToken = profile.PTZConfiguration.token
|
||||
ptz_config = ptz.GetConfigurationOptions(request)
|
||||
|
||||
fov_space_id = next(
|
||||
(
|
||||
i
|
||||
for i, space in enumerate(
|
||||
ptz_config.Spaces.RelativePanTiltTranslationSpace
|
||||
)
|
||||
if "TranslationSpaceFov" in space["URI"]
|
||||
),
|
||||
None,
|
||||
)
|
||||
|
||||
# setup continuous moving request
|
||||
move_request = ptz.create_type("ContinuousMove")
|
||||
move_request.ProfileToken = profile.token
|
||||
self.cams[camera_name]["move_request"] = move_request
|
||||
|
||||
# setup relative moving request for autotracking
|
||||
move_request = ptz.create_type("RelativeMove")
|
||||
move_request.ProfileToken = profile.token
|
||||
if move_request.Translation is None and fov_space_id is not None:
|
||||
move_request.Translation = ptz.GetStatus(
|
||||
{"ProfileToken": profile.token}
|
||||
).Position
|
||||
move_request.Translation.PanTilt.space = ptz_config["Spaces"][
|
||||
"RelativePanTiltTranslationSpace"
|
||||
][fov_space_id]["URI"]
|
||||
move_request.Translation.Zoom.space = ptz_config["Spaces"][
|
||||
"RelativeZoomTranslationSpace"
|
||||
][0]["URI"]
|
||||
if move_request.Speed is None:
|
||||
move_request.Speed = ptz.GetStatus({"ProfileToken": profile.token}).Position
|
||||
self.cams[camera_name]["relative_move_request"] = move_request
|
||||
|
||||
# setup relative moving request for autotracking
|
||||
move_request = ptz.create_type("AbsoluteMove")
|
||||
move_request.ProfileToken = profile.token
|
||||
self.cams[camera_name]["absolute_move_request"] = move_request
|
||||
|
||||
# status request for autotracking
|
||||
status_request = ptz.create_type("GetStatus")
|
||||
status_request.ProfileToken = profile.token
|
||||
self.cams[camera_name]["status_request"] = status_request
|
||||
|
||||
# setup existing presets
|
||||
try:
|
||||
presets: list[dict] = ptz.GetPresets({"ProfileToken": profile.token})
|
||||
except ONVIFError as e:
|
||||
logger.warning(f"Unable to get presets from camera: {camera_name}: {e}")
|
||||
presets = []
|
||||
|
||||
for preset in presets:
|
||||
self.cams[camera_name]["presets"][preset["Name"].lower()] = preset["token"]
|
||||
|
||||
# get list of supported features
|
||||
ptz_config = ptz.GetConfigurationOptions(request)
|
||||
supported_features = []
|
||||
|
||||
if ptz_config.Spaces and ptz_config.Spaces.ContinuousPanTiltVelocitySpace:
|
||||
supported_features.append("pt")
|
||||
|
||||
if ptz_config.Spaces and ptz_config.Spaces.ContinuousZoomVelocitySpace:
|
||||
supported_features.append("zoom")
|
||||
|
||||
if ptz_config.Spaces and ptz_config.Spaces.RelativePanTiltTranslationSpace:
|
||||
supported_features.append("pt-r")
|
||||
|
||||
if ptz_config.Spaces and ptz_config.Spaces.RelativeZoomTranslationSpace:
|
||||
supported_features.append("zoom-r")
|
||||
|
||||
if fov_space_id is not None:
|
||||
supported_features.append("pt-r-fov")
|
||||
self.cams[camera_name][
|
||||
"relative_fov_range"
|
||||
] = ptz_config.Spaces.RelativePanTiltTranslationSpace[fov_space_id]
|
||||
|
||||
self.cams[camera_name]["relative_fov_supported"] = fov_space_id is not None
|
||||
|
||||
self.cams[camera_name]["features"] = supported_features
|
||||
|
||||
self.cams[camera_name]["init"] = True
|
||||
return True
|
||||
|
||||
def _stop(self, camera_name: str) -> None:
|
||||
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
|
||||
move_request = self.cams[camera_name]["move_request"]
|
||||
onvif.get_service("ptz").Stop(
|
||||
{
|
||||
"ProfileToken": move_request.ProfileToken,
|
||||
"PanTilt": True,
|
||||
"Zoom": True,
|
||||
}
|
||||
)
|
||||
self.cams[camera_name]["active"] = False
|
||||
|
||||
def _move(self, camera_name: str, command: OnvifCommandEnum) -> None:
|
||||
if self.cams[camera_name]["active"]:
|
||||
logger.warning(
|
||||
f"{camera_name} is already performing an action, stopping..."
|
||||
)
|
||||
self._stop(camera_name)
|
||||
|
||||
self.cams[camera_name]["active"] = True
|
||||
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
|
||||
move_request = self.cams[camera_name]["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,
|
||||
}
|
||||
}
|
||||
|
||||
onvif.get_service("ptz").ContinuousMove(move_request)
|
||||
|
||||
def _move_relative(self, camera_name: str, pan, tilt, speed) -> None:
|
||||
if not self.cams[camera_name]["relative_fov_supported"]:
|
||||
logger.error(f"{camera_name} does not support ONVIF RelativeMove (FOV).")
|
||||
return
|
||||
|
||||
logger.debug(f"{camera_name} called RelativeMove: pan: {pan} tilt: {tilt}")
|
||||
self.get_camera_status(camera_name)
|
||||
|
||||
if self.cams[camera_name]["active"]:
|
||||
logger.warning(
|
||||
f"{camera_name} is already performing an action, not moving..."
|
||||
)
|
||||
return
|
||||
|
||||
self.cams[camera_name]["active"] = True
|
||||
self.camera_metrics[camera_name]["ptz_stopped"].clear()
|
||||
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
|
||||
move_request = self.cams[camera_name]["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
|
||||
pan = numpy.interp(
|
||||
pan,
|
||||
[-1, 1],
|
||||
[
|
||||
self.cams[camera_name]["relative_fov_range"]["XRange"]["Min"],
|
||||
self.cams[camera_name]["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"],
|
||||
],
|
||||
)
|
||||
|
||||
move_request.Speed = {
|
||||
"PanTilt": {
|
||||
"x": speed,
|
||||
"y": speed,
|
||||
},
|
||||
"Zoom": 0,
|
||||
}
|
||||
|
||||
# move pan and tilt separately
|
||||
move_request.Translation.PanTilt.x = pan
|
||||
move_request.Translation.PanTilt.y = 0
|
||||
move_request.Translation.Zoom.x = 0
|
||||
|
||||
onvif.get_service("ptz").RelativeMove(move_request)
|
||||
|
||||
move_request.Translation.PanTilt.x = 0
|
||||
move_request.Translation.PanTilt.y = tilt
|
||||
move_request.Translation.Zoom.x = 0
|
||||
|
||||
onvif.get_service("ptz").RelativeMove(move_request)
|
||||
|
||||
self.cams[camera_name]["active"] = False
|
||||
|
||||
def _move_to_preset(self, camera_name: str, preset: str) -> None:
|
||||
if preset not in self.cams[camera_name]["presets"]:
|
||||
logger.error(f"{preset} is not a valid preset for {camera_name}")
|
||||
return
|
||||
|
||||
self.cams[camera_name]["active"] = True
|
||||
self.camera_metrics[camera_name]["ptz_stopped"].clear()
|
||||
move_request = self.cams[camera_name]["move_request"]
|
||||
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
|
||||
preset_token = self.cams[camera_name]["presets"][preset]
|
||||
onvif.get_service("ptz").GotoPreset(
|
||||
{
|
||||
"ProfileToken": move_request.ProfileToken,
|
||||
"PresetToken": preset_token,
|
||||
}
|
||||
)
|
||||
self.camera_metrics[camera_name]["ptz_stopped"].set()
|
||||
self.cams[camera_name]["active"] = False
|
||||
|
||||
def _zoom(self, camera_name: str, command: OnvifCommandEnum) -> None:
|
||||
if self.cams[camera_name]["active"]:
|
||||
logger.warning(
|
||||
f"{camera_name} is already performing an action, stopping..."
|
||||
)
|
||||
self._stop(camera_name)
|
||||
|
||||
self.cams[camera_name]["active"] = True
|
||||
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
|
||||
move_request = self.cams[camera_name]["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}}
|
||||
|
||||
onvif.get_service("ptz").ContinuousMove(move_request)
|
||||
|
||||
def handle_command(
|
||||
self, camera_name: str, command: OnvifCommandEnum, param: str = ""
|
||||
) -> None:
|
||||
if camera_name not in self.cams.keys():
|
||||
logger.error(f"Onvif is not setup for {camera_name}")
|
||||
return
|
||||
|
||||
if not self.cams[camera_name]["init"]:
|
||||
if not self._init_onvif(camera_name):
|
||||
return
|
||||
|
||||
if command == OnvifCommandEnum.init:
|
||||
# already init
|
||||
return
|
||||
elif command == OnvifCommandEnum.stop:
|
||||
self._stop(camera_name)
|
||||
elif command == OnvifCommandEnum.preset:
|
||||
self._move_to_preset(camera_name, param)
|
||||
elif (
|
||||
command == OnvifCommandEnum.zoom_in or command == OnvifCommandEnum.zoom_out
|
||||
):
|
||||
self._zoom(camera_name, command)
|
||||
else:
|
||||
self._move(camera_name, command)
|
||||
|
||||
def get_camera_info(self, camera_name: str) -> dict[str, any]:
|
||||
if camera_name not in self.cams.keys():
|
||||
logger.error(f"Onvif is not setup for {camera_name}")
|
||||
return {}
|
||||
|
||||
if not self.cams[camera_name]["init"]:
|
||||
self._init_onvif(camera_name)
|
||||
|
||||
return {
|
||||
"name": camera_name,
|
||||
"features": self.cams[camera_name]["features"],
|
||||
"presets": list(self.cams[camera_name]["presets"].keys()),
|
||||
}
|
||||
|
||||
def get_camera_status(self, camera_name: str) -> dict[str, any]:
|
||||
if camera_name not in self.cams.keys():
|
||||
logger.error(f"Onvif is not setup for {camera_name}")
|
||||
return {}
|
||||
|
||||
if not self.cams[camera_name]["init"]:
|
||||
self._init_onvif(camera_name)
|
||||
|
||||
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
|
||||
status_request = self.cams[camera_name]["status_request"]
|
||||
status = onvif.get_service("ptz").GetStatus(status_request)
|
||||
|
||||
if status.MoveStatus.PanTilt == "IDLE" or status.MoveStatus.Zoom == "IDLE":
|
||||
self.cams[camera_name]["active"] = False
|
||||
self.camera_metrics[camera_name]["ptz_stopped"].set()
|
||||
else:
|
||||
self.cams[camera_name]["active"] = True
|
||||
self.camera_metrics[camera_name]["ptz_stopped"].clear()
|
||||
|
||||
return {
|
||||
"pan": status.Position.PanTilt.x,
|
||||
"tilt": status.Position.PanTilt.y,
|
||||
"zoom": status.Position.Zoom.x,
|
||||
"pantilt_moving": status.MoveStatus.PanTilt,
|
||||
"zoom_moving": status.MoveStatus.Zoom,
|
||||
}
|
||||
Reference in New Issue
Block a user