blakeblackshear.frigate/frigate/ptz/onvif.py
Nicolas Mowen cb30450060
Migrate pydantic to V2 (#10142)
* Run pydantic migration tool

* Finish removing deprecated functions

* Formatting

* Fix movement weights type

* Fix movement weight test

* Fix config checks

* formatting

* fix typing

* formatting

* Fix

* Fix serialization issues

* Formatting

* fix model namespace warnings

* Update formatting

* Format go2rtc file

* Cleanup migrations

* Fix warnings

* Don't include null values in config json

* Formatting

* Fix test

---------

Co-authored-by: Josh Hawkins <32435876+hawkeye217@users.noreply.github.com>
2024-02-29 16:10:13 -07:00

663 lines
25 KiB
Python

"""Configure and control camera via onvif."""
import logging
import site
from enum import Enum
import numpy
from onvif import ONVIFCamera, ONVIFError
from zeep.exceptions import Fault, TransportError
from frigate.config import FrigateConfig, ZoomingModeEnum
from frigate.types import PTZMetricsTypes
from frigate.util.builtin import find_by_key
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, ptz_metrics: dict[str, PTZMetricsTypes]
) -> None:
self.cams: dict[str, ONVIFCamera] = {}
self.config = config
self.ptz_metrics = ptz_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,
"features": [],
"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()
logger.debug(f"Onvif media xaddr for {camera_name}: {media.xaddr}")
try:
# 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 (ONVIFError, Fault, TransportError) as e:
logger.error(
f"Unable to get Onvif capabilities for camera: {camera_name}: {e}"
)
return False
try:
profiles = media.GetProfiles()
except (ONVIFError, Fault, TransportError) as e:
logger.error(
f"Unable to get Onvif media profiles for camera: {camera_name}: {e}"
)
return False
profile = None
for key, onvif_profile in enumerate(profiles):
if (
onvif_profile.VideoEncoderConfiguration
and onvif_profile.VideoEncoderConfiguration.Encoding == "H264"
and onvif_profile.PTZConfiguration
and onvif_profile.PTZConfiguration.DefaultContinuousPanTiltVelocitySpace
is not None
):
profile = onvif_profile
logger.debug(f"Selected Onvif profile for {camera_name}: {profile}")
break
if profile is None:
logger.error(
f"No appropriate Onvif profiles found for camera: {camera_name}."
)
return False
# 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
ptz = onvif.create_ptz_service()
request = ptz.create_type("GetConfigurationOptions")
request.ConfigurationToken = profile.PTZConfiguration.token
ptz_config = ptz.GetConfigurationOptions(request)
logger.debug(f"Onvif config for {camera_name}: {ptz_config}")
service_capabilities_request = ptz.create_type("GetServiceCapabilities")
self.cams[camera_name]["service_capabilities_request"] = (
service_capabilities_request
)
fov_space_id = next(
(
i
for i, space in enumerate(
ptz_config.Spaces.RelativePanTiltTranslationSpace
)
if "TranslationSpaceFov" in space["URI"]
),
None,
)
# status request for autotracking and filling ptz-parameters
status_request = ptz.create_type("GetStatus")
status_request.ProfileToken = profile.token
self.cams[camera_name]["status_request"] = status_request
try:
status = ptz.GetStatus(status_request)
logger.debug(f"Onvif status config for {camera_name}: {status}")
except Exception as e:
logger.warning(f"Unable to get status from camera: {camera_name}: {e}")
status = None
# autoracking relative panning/tilting needs a relative zoom value set to 0
# if camera supports relative movement
if (
self.config.cameras[camera_name].onvif.autotracking.zooming
!= ZoomingModeEnum.disabled
):
zoom_space_id = next(
(
i
for i, space in enumerate(
ptz_config.Spaces.RelativeZoomTranslationSpace
)
if "TranslationGenericSpace" 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
logger.debug(f"{camera_name}: Relative move request: {move_request}")
if move_request.Translation is None and fov_space_id is not None:
move_request.Translation = status.Position
move_request.Translation.PanTilt.space = ptz_config["Spaces"][
"RelativePanTiltTranslationSpace"
][fov_space_id]["URI"]
# try setting relative zoom translation space
try:
if (
self.config.cameras[camera_name].onvif.autotracking.zooming
!= ZoomingModeEnum.disabled
):
if zoom_space_id is not None:
move_request.Translation.Zoom.space = ptz_config["Spaces"][
"RelativeZoomTranslationSpace"
][zoom_space_id]["URI"]
else:
move_request.Translation.Zoom = []
except Exception:
self.config.cameras[
camera_name
].onvif.autotracking.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: Relative zoom not supported"
)
if move_request.Speed is None:
move_request.Speed = configs.DefaultPTZSpeed if configs else None
logger.debug(
f"{camera_name}: Relative move request after setup: {move_request}"
)
self.cams[camera_name]["relative_move_request"] = move_request
# setup absolute moving request for autotracking zooming
move_request = ptz.create_type("AbsoluteMove")
move_request.ProfileToken = profile.token
self.cams[camera_name]["absolute_move_request"] = move_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"][
(getattr(preset, "Name") or f"preset {preset['token']}").lower()
] = preset["token"]
# get list of supported features
supported_features = []
if configs.DefaultContinuousPanTiltVelocitySpace:
supported_features.append("pt")
if configs.DefaultContinuousZoomVelocitySpace:
supported_features.append("zoom")
if configs.DefaultRelativePanTiltTranslationSpace:
supported_features.append("pt-r")
if configs.DefaultRelativeZoomTranslationSpace:
supported_features.append("zoom-r")
try:
# get camera's zoom limits from onvif config
self.cams[camera_name]["relative_zoom_range"] = (
ptz_config.Spaces.RelativeZoomTranslationSpace[0]
)
except Exception:
if (
self.config.cameras[camera_name].onvif.autotracking.zooming
== ZoomingModeEnum.relative
):
self.config.cameras[
camera_name
].onvif.autotracking.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: Relative zoom not supported"
)
if configs.DefaultAbsoluteZoomPositionSpace:
supported_features.append("zoom-a")
try:
# get camera's zoom limits from onvif config
self.cams[camera_name]["absolute_zoom_range"] = (
ptz_config.Spaces.AbsoluteZoomPositionSpace[0]
)
self.cams[camera_name]["zoom_limits"] = configs.ZoomLimits
except Exception:
if self.config.cameras[camera_name].onvif.autotracking.zooming:
self.config.cameras[
camera_name
].onvif.autotracking.zooming = ZoomingModeEnum.disabled
logger.warning(
f"Disabling autotracking zooming for {camera_name}: Absolute zoom not supported"
)
# set relative pan/tilt space for autotracker
if (
fov_space_id is not None
and configs.DefaultRelativePanTiltTranslationSpace 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]["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, zoom, speed) -> None:
if "pt-r-fov" not in self.cams[camera_name]["features"]:
logger.error(f"{camera_name} does not support ONVIF RelativeMove (FOV).")
return
logger.debug(
f"{camera_name} called RelativeMove: pan: {pan} tilt: {tilt} zoom: {zoom}"
)
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.ptz_metrics[camera_name]["ptz_motor_stopped"].clear()
logger.debug(
f"{camera_name}: PTZ start time: {self.ptz_metrics[camera_name]['ptz_frame_time'].value}"
)
self.ptz_metrics[camera_name]["ptz_start_time"].value = self.ptz_metrics[
camera_name
]["ptz_frame_time"].value
self.ptz_metrics[camera_name]["ptz_stop_time"].value = 0
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,
},
}
move_request.Translation.PanTilt.x = pan
move_request.Translation.PanTilt.y = tilt
if (
"zoom-r" in self.cams[camera_name]["features"]
and self.config.cameras[camera_name].onvif.autotracking.zooming
== ZoomingModeEnum.relative
):
move_request.Speed = {
"PanTilt": {
"x": speed,
"y": speed,
},
"Zoom": {"x": speed},
}
move_request.Translation.Zoom.x = zoom
onvif.get_service("ptz").RelativeMove(move_request)
# reset after the move request
move_request.Translation.PanTilt.x = 0
move_request.Translation.PanTilt.y = 0
if (
"zoom-r" in self.cams[camera_name]["features"]
and self.config.cameras[camera_name].onvif.autotracking.zooming
== ZoomingModeEnum.relative
):
move_request.Translation.Zoom.x = 0
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.ptz_metrics[camera_name]["ptz_motor_stopped"].clear()
self.ptz_metrics[camera_name]["ptz_start_time"].value = 0
self.ptz_metrics[camera_name]["ptz_stop_time"].value = 0
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.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 _zoom_absolute(self, camera_name: str, zoom, speed) -> None:
if "zoom-a" not in self.cams[camera_name]["features"]:
logger.error(f"{camera_name} does not support ONVIF AbsoluteMove zooming.")
return
logger.debug(f"{camera_name} called AbsoluteMove: zoom: {zoom}")
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.ptz_metrics[camera_name]["ptz_motor_stopped"].clear()
logger.debug(
f"{camera_name}: PTZ start time: {self.ptz_metrics[camera_name]['ptz_frame_time'].value}"
)
self.ptz_metrics[camera_name]["ptz_start_time"].value = self.ptz_metrics[
camera_name
]["ptz_frame_time"].value
self.ptz_metrics[camera_name]["ptz_stop_time"].value = 0
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
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"],
],
)
move_request.Speed = {"Zoom": speed}
move_request.Position = {"Zoom": zoom}
logger.debug(f"{camera_name}: Absolute zoom: {zoom}")
onvif.get_service("ptz").AbsoluteMove(move_request)
self.cams[camera_name]["active"] = False
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_service_capabilities(self, camera_name: 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"]:
self._init_onvif(camera_name)
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
service_capabilities_request = self.cams[camera_name][
"service_capabilities_request"
]
service_capabilities = onvif.get_service("ptz").GetServiceCapabilities(
service_capabilities_request
)
logger.debug(
f"Onvif service capabilities for {camera_name}: {service_capabilities}"
)
# MoveStatus is required for autotracking - should return "true" if supported
return find_by_key(vars(service_capabilities), "MoveStatus")
def get_camera_status(self, camera_name: 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"]:
self._init_onvif(camera_name)
onvif: ONVIFCamera = self.cams[camera_name]["onvif"]
status_request = self.cams[camera_name]["status_request"]
try:
status = onvif.get_service("ptz").GetStatus(status_request)
except Exception:
pass # We're unsupported, that'll be reported in the next check.
try:
pan_tilt_status = getattr(status.MoveStatus, "PanTilt", None)
zoom_status = getattr(status.MoveStatus, "Zoom", None)
# if it's not an attribute, see if MoveStatus even exists in the status result
if pan_tilt_status is None:
pan_tilt_status = getattr(status, "MoveStatus", None)
# we're unsupported
if pan_tilt_status is None or pan_tilt_status not in [
"IDLE",
"MOVING",
]:
raise Exception
except Exception:
logger.warning(
f"Camera {camera_name} does not support the ONVIF GetStatus method. Autotracking will not function correctly and must be disabled in your config."
)
return
if pan_tilt_status == "IDLE" and (zoom_status is None or zoom_status == "IDLE"):
self.cams[camera_name]["active"] = False
if not self.ptz_metrics[camera_name]["ptz_motor_stopped"].is_set():
self.ptz_metrics[camera_name]["ptz_motor_stopped"].set()
logger.debug(
f"{camera_name}: PTZ stop time: {self.ptz_metrics[camera_name]['ptz_frame_time'].value}"
)
self.ptz_metrics[camera_name]["ptz_stop_time"].value = self.ptz_metrics[
camera_name
]["ptz_frame_time"].value
else:
self.cams[camera_name]["active"] = True
if self.ptz_metrics[camera_name]["ptz_motor_stopped"].is_set():
self.ptz_metrics[camera_name]["ptz_motor_stopped"].clear()
logger.debug(
f"{camera_name}: PTZ start time: {self.ptz_metrics[camera_name]['ptz_frame_time'].value}"
)
self.ptz_metrics[camera_name][
"ptz_start_time"
].value = self.ptz_metrics[camera_name]["ptz_frame_time"].value
self.ptz_metrics[camera_name]["ptz_stop_time"].value = 0
if (
self.config.cameras[camera_name].onvif.autotracking.zooming
!= ZoomingModeEnum.disabled
):
# store absolute zoom level as 0 to 1 interpolated from the values of the camera
self.ptz_metrics[camera_name]["ptz_zoom_level"].value = numpy.interp(
round(status.Position.Zoom.x, 2),
[0, 1],
[
self.cams[camera_name]["absolute_zoom_range"]["XRange"]["Min"],
self.cams[camera_name]["absolute_zoom_range"]["XRange"]["Max"],
],
)
logger.debug(
f'{camera_name}: Camera zoom level: {self.ptz_metrics[camera_name]["ptz_zoom_level"].value}'
)
# some hikvision cams won't update MoveStatus, so warn if it hasn't changed
if (
not self.ptz_metrics[camera_name]["ptz_motor_stopped"].is_set()
and not self.ptz_metrics[camera_name]["ptz_reset"].is_set()
and self.ptz_metrics[camera_name]["ptz_start_time"].value != 0
and self.ptz_metrics[camera_name]["ptz_frame_time"].value
> (self.ptz_metrics[camera_name]["ptz_start_time"].value + 10)
and self.ptz_metrics[camera_name]["ptz_stop_time"].value == 0
):
logger.debug(
f'Start time: {self.ptz_metrics[camera_name]["ptz_start_time"].value}, Stop time: {self.ptz_metrics[camera_name]["ptz_stop_time"].value}, Frame time: {self.ptz_metrics[camera_name]["ptz_frame_time"].value}'
)
# set the stop time so we don't come back into this again and spam the logs
self.ptz_metrics[camera_name]["ptz_stop_time"].value = self.ptz_metrics[
camera_name
]["ptz_frame_time"].value
logger.warning(f"Camera {camera_name} is still in ONVIF 'MOVING' status.")