"""Configure and control camera via onvif.""" import logging from enum import Enum from importlib.util import find_spec from pathlib import Path 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_relative = "move_relative" 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=Path(find_spec("onvif").origin).parent / "../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() # setup continuous moving request move_request = ptz.create_type("ContinuousMove") move_request.ProfileToken = profile.token self.cams[camera_name]["move_request"] = move_request # extra setup for autotracking cameras if ( self.config.cameras[camera_name].onvif.autotracking.enabled_in_config and self.config.cameras[camera_name].onvif.autotracking.enabled ): 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 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") if ( self.config.cameras[camera_name].onvif.autotracking.enabled_in_config and self.config.cameras[camera_name].onvif.autotracking.enabled ): 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") if ( self.config.cameras[camera_name].onvif.autotracking.enabled_in_config and self.config.cameras[camera_name].onvif.autotracking.enabled ): 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 ( self.config.cameras[camera_name].onvif.autotracking.enabled_in_config and self.config.cameras[camera_name].onvif.autotracking.enabled and 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, } } try: onvif.get_service("ptz").ContinuousMove(move_request) except ONVIFError as e: logger.warning(f"Onvif sending move request to {camera_name} failed: {e}") 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.move_relative: _, pan, tilt = param.split("_") self._move_relative(camera_name, float(pan), float(tilt), 0, 1) 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" ] try: 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") except Exception: logger.warning( f"Camera {camera_name} does not support the ONVIF GetServiceCapabilities method. Autotracking will not function correctly and must be disabled in your config." ) return False 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.")