Basic PTZ support

docker/Dockerfile.wheels

@@ -20,7 +20,7 @@ RUN apt-get -qq update \
 RUN wget -q https://bootstrap.pypa.io/get-pip.py -O get-pip.py \
     && python3 get-pip.py "pip==20.2.4"
 
-RUN pip3 install scikit-build
+RUN pip3 install scikit-build onvif_zeep
 
 RUN pip3 wheel --wheel-dir=/wheels \
     opencv-python-headless \

docs/docs/configuration/index.md

@@ -390,4 +390,22 @@ cameras:
       quality: 70
       # Optional: Restrict mqtt messages to objects that entered any of the listed zones (default: no required zones)
       required_zones: []
+
+    # Optional: Configuration for ONVIF support (used for PTZ)
+    onvif:
+      # Optional: The hostname of the ONVIF server for the camera (usually the IP address of the camera)
+      host: 192.168.1.2
+      # Optional: The port where the ONVIF server.
+      port: 8080
+      # Optional: The username needed to use the ONVIF server (usually the same as the one needed for RTSP, or logging into the admin portal)
+      username: admin
+      # Optional: The password needed to use the ONVIF server (same comment as for the username one)
+      password: password
+
+    # Optional: Configuration specific to PTZ support
+    ptz:
+      # Optional: How fast the camera should turn (default: shown below)
+      turn_speed: 0.5
+      # Optional: Whether the y-axis should be inverted (i.e. if the camera is mounted upside down) (default: shown below)
+      invert_y_axis: False
 ```

frigate/config.py

@@ -355,6 +355,20 @@ class CameraInput(FrigateBaseModel):
     )
 
 
+class CameraOnvifConfig(FrigateBaseModel):
+    host: Optional[str] = Field(title="ONVIF host.")
+    port: Optional[int] = Field(title="ONVIF port.")
+    username: Optional[str] = Field(title="ONVIF username.")
+    password: Optional[str] = Field(title="ONVIF password.")
+
+
+class CameraPtzConfig(FrigateBaseModel):
+    turn_speed: float = Field(default=0.1, title="How fast the camera should move.")
+    invert_y_axis: bool = Field(
+        default=False, title="Whether the y-axis of this camera should be inverted."
+    )
+
+
 class CameraFfmpegConfig(FfmpegConfig):
     inputs: List[CameraInput] = Field(title="Camera inputs.")
 
@@ -458,6 +472,12 @@ class CameraLiveConfig(FrigateBaseModel):
 class CameraConfig(FrigateBaseModel):
     name: Optional[str] = Field(title="Camera name.")
     ffmpeg: CameraFfmpegConfig = Field(title="FFmpeg configuration for the camera.")
+    ptz: CameraPtzConfig = Field(
+        default_factory=CameraPtzConfig, title="PTZ configuration for the camera."
+    )
+    onvif: CameraOnvifConfig = Field(
+        default_factory=CameraOnvifConfig, title="ONVIF configuration for the camera."
+    )
     best_image_timeout: int = Field(
         default=60,
         title="How long to wait for the image with the highest confidence score.",

frigate/http.py

@@ -31,6 +31,7 @@ from playhouse.shortcuts import model_to_dict
 
 from frigate.const import CLIPS_DIR, RECORD_DIR
 from frigate.models import Event, Recordings
+from frigate.ptz import Ptz
 from frigate.stats import stats_snapshot
 from frigate.util import calculate_region
 from frigate.version import VERSION
@@ -62,6 +63,8 @@ def create_app(
     app.stats_tracking = stats_tracking
     app.detected_frames_processor = detected_frames_processor
 
+    app.ptz_cameras = {}
+
     app.register_blueprint(bp)
 
     return app
@@ -382,6 +385,100 @@ def best(camera_name, label):
         return "Camera named {} not found".format(camera_name), 404
 
 
+@bp.route("/<camera_name>/ptz/move")
+def ptz_move(camera_name):
+    direction = request.args.get("direction", "up")
+    if (
+        camera_name in current_app.frigate_config.cameras
+        and current_app.frigate_config.cameras[camera_name].onvif.host is not None
+    ):
+        if current_app.ptz_cameras.get(camera_name) is None:
+            current_app.ptz_cameras[camera_name] = Ptz(
+                current_app.frigate_config.cameras[camera_name]
+            )
+
+        ptz = current_app.ptz_cameras[camera_name]
+
+        if direction == "up":
+            ptz.move_up()
+        elif direction == "down":
+            ptz.move_down()
+        elif direction == "left":
+            ptz.move_left()
+        elif direction == "right":
+            ptz.move_right()
+        else:
+            return "Bad direction {}".format(direction), 401
+
+        return "", 204
+    else:
+        return "Camera named {} not found".format(camera_name), 404
+
+
+@bp.route("/<camera_name>/ptz/zoom")
+def ptz_zoom(camera_name):
+    direction = bool(request.args.get("zoomIn", 0, type=int))
+    if (
+        camera_name in current_app.frigate_config.cameras
+        and current_app.frigate_config.cameras[camera_name].onvif.host is not None
+    ):
+        if current_app.ptz_cameras.get(camera_name) is None:
+            current_app.ptz_cameras[camera_name] = Ptz(
+                current_app.frigate_config.cameras[camera_name]
+            )
+
+        ptz = current_app.ptz_cameras[camera_name]
+
+        if direction:
+            ptz.zoom_in()
+        else:
+            ptz.zoom_out()
+
+        return "", 204
+    else:
+        return "Camera named {} not found".format(camera_name), 404
+
+
+@bp.route("/<camera_name>/ptz/sethome")
+def ptz_sethome(camera_name):
+    if (
+        camera_name in current_app.frigate_config.cameras
+        and current_app.frigate_config.cameras[camera_name].onvif.host is not None
+    ):
+        if current_app.ptz_cameras.get(camera_name) is None:
+            current_app.ptz_cameras[camera_name] = Ptz(
+                current_app.frigate_config.cameras[camera_name]
+            )
+
+        ptz = current_app.ptz_cameras[camera_name]
+
+        ptz.set_home()
+
+        return "", 204
+    else:
+        return "Camera named {} not found".format(camera_name), 404
+
+
+@bp.route("/<camera_name>/ptz/gotohome")
+def ptz_gotohome(camera_name):
+    if (
+        camera_name in current_app.frigate_config.cameras
+        and current_app.frigate_config.cameras[camera_name].onvif.host is not None
+    ):
+        if current_app.ptz_cameras.get(camera_name) is None:
+            current_app.ptz_cameras[camera_name] = Ptz(
+                current_app.frigate_config.cameras[camera_name]
+            )
+
+        ptz = current_app.ptz_cameras[camera_name]
+
+        ptz.goto_home()
+
+        return "", 204
+    else:
+        return "Camera named {} not found".format(camera_name), 404
+
+
 @bp.route("/<camera_name>")
 def mjpeg_feed(camera_name):
     fps = int(request.args.get("fps", "3"))

frigate/ptz.py

@@ -0,0 +1,100 @@
+from onvif import ONVIFCamera
+
+from frigate.config import CameraConfig
+
+
+class Ptz:
+    def __init__(self, config: CameraConfig):
+        if config.onvif.host is not None:
+            onvif_camera = ONVIFCamera(
+                config.onvif.host,
+                config.onvif.port,
+                config.onvif.username,
+                config.onvif.password,
+            )
+            media_service = onvif_camera.create_media_service()
+
+            self.ptz_service = onvif_camera.create_ptz_service()
+
+            profile = media_service.GetProfiles()[0]
+
+            request = self.ptz_service.create_type("GetConfigurationOptions")
+            request.ConfigurationToken = profile.PTZConfiguration.token
+            self.ptz_configuration_options = self.ptz_service.GetConfigurationOptions(
+                request
+            )
+
+            self.move_request = self.ptz_service.create_type("ContinuousMove")
+            self.move_request.ProfileToken = profile.token
+            if self.move_request.Velocity is None:
+                self.move_request.Velocity = self.ptz_service.GetStatus(
+                    {"ProfileToken": profile.token}
+                ).Position
+
+            self.set_preset_request = self.ptz_service.create_type("SetPreset")
+            self.set_preset_request.ProfileToken = profile.token
+
+            self.goto_preset_request = self.ptz_service.create_type("GotoPreset")
+            self.goto_preset_request.ProfileToken = profile.token
+            if self.goto_preset_request.Speed is None:
+                self.goto_preset_request.Speed = self.ptz_service.GetStatus(
+                    {"ProfileToken": profile.token}
+                ).Position
+
+            self.turn_speed = config.ptz.turn_speed
+            self.invert_y_axis = config.ptz.invert_y_axis
+
+            self.active = False
+
+    def move(self, request):
+        if self.active:
+            self.ptz_service.Stop({"ProfileToken": request.ProfileToken})
+
+        self.active = True
+        self.ptz_service.ContinuousMove(request)
+
+    def move_up(self):
+        request = self.move_request
+        request.Velocity.PanTilt.x = 0
+        request.Velocity.PanTilt.y = -1 if self.invert_y_axis else 1 * self.turn_speed
+        self.move(request)
+
+    def move_down(self):
+        request = self.move_request
+        request.Velocity.PanTilt.x = 0
+        request.Velocity.PanTilt.y = 1 if self.invert_y_axis else -1 * self.turn_speed
+        self.move(request)
+
+    def move_right(self):
+        request = self.move_request
+        request.Velocity.PanTilt.x = self.turn_speed
+        request.Velocity.PanTilt.y = 0
+        self.move(request)
+
+    def move_left(self):
+        request = self.move_request
+        request.Velocity.PanTilt.x = -1 * self.turn_speed
+        request.Velocity.PanTilt.y = 0
+        self.move(request)
+
+    def zoom_in(self):
+        request = self.move_request
+        request.Velocity.Zoom.x = 1
+        self.move(request)
+
+    def zoom_out(self):
+        request = self.move_request
+        request.Velocity.Zoom.x = -1
+        self.move(request)
+
+    def set_home(self):
+        request = self.set_preset_request
+        request.PresetToken = "1"
+        self.ptz_service.SetPreset(request)
+
+    def goto_home(self):
+        request = self.goto_preset_request
+        request.PresetToken = "1"
+        request.Speed.PanTilt.x = self.turn_speed
+        request.Speed.PanTilt.y = self.turn_speed
+        self.ptz_service.GotoPreset(request)