Add support for PTZ service

custom_components/frigate/camera.py

@@ -35,12 +35,15 @@
     ATTR_CONFIG,
     ATTR_EVENT_ID,
     ATTR_FAVORITE,
+    ATTR_PTZ_ACTION,
+    ATTR_PTZ_ARGUMENT,
     CONF_RTMP_URL_TEMPLATE,
     CONF_RTSP_URL_TEMPLATE,
     DEVICE_CLASS_CAMERA,
     DOMAIN,
     NAME,
     SERVICE_FAVORITE_EVENT,
+    SERVICE_PTZ,
     STATE_DETECTED,
     STATE_IDLE,
 )
@@ -84,6 +87,14 @@ async def async_setup_entry(
         },
         SERVICE_FAVORITE_EVENT,
     )
+    platform.async_register_entity_service(
+        SERVICE_PTZ,
+        {
+            vol.Required(ATTR_PTZ_ACTION): str,
+            vol.Optional(ATTR_PTZ_ARGUMENT, default=""): str,
+        },
+        SERVICE_PTZ,
+    )
 
 
 class FrigateCamera(FrigateMQTTEntity, Camera):  # type: ignore[misc]
@@ -145,6 +156,9 @@ def __init__(
         self._attr_motion_detection_enabled = self._camera_config.get("motion", {}).get(
             "enabled"
         )
+        self._ptz_topic = (
+            f"{frigate_config['mqtt']['topic_prefix']}" f"/{self._cam_name}/ptz"
+        )
         self._set_motion_topic = (
             f"{frigate_config['mqtt']['topic_prefix']}" f"/{self._cam_name}/motion/set"
         )
@@ -288,6 +302,16 @@ async def favorite_event(self, event_id: str, favorite: bool) -> None:
         """Favorite an event."""
         await self._client.async_retain(event_id, favorite)
 
+    async def ptz(self, action: str, argument: str) -> None:
+        """Run PTZ command."""
+        await async_publish(
+            self.hass,
+            self._ptz_topic,
+            f"{action}{f'_{argument}' if argument else ''}",
+            0,
+            False,
+        )
+
 
 class BirdseyeCamera(FrigateEntity, Camera):  # type: ignore[misc]
     """Representation of the Frigate birdseye camera."""

custom_components/frigate/const.py

@@ -32,6 +32,8 @@
 ATTR_EVENT_ID = "event_id"
 ATTR_FAVORITE = "favorite"
 ATTR_MQTT = "mqtt"
+ATTR_PTZ_ACTION = "action"
+ATTR_PTZ_ARGUMENT = "argument"
 
 # Configuration and options
 CONF_CAMERA_STATIC_IMAGE_HEIGHT = "camera_image_height"
@@ -77,3 +79,4 @@
 
 # Frigate Services
 SERVICE_FAVORITE_EVENT = "favorite_event"
+SERVICE_PTZ = "ptz"

custom_components/frigate/services.yaml

@@ -30,3 +30,38 @@ favorite_event:
       default: true
       selector:
         boolean:
+
+ptz:
+  name: Control camera via PTZ
+  description: >
+    Pan / Tilt, Zoom, or move a camera to a preset
+  target:
+    entity:
+      integration: frigate
+      domain: camera
+      device_class: camera
+  fields:
+    action:
+      name: PTZ Service
+      description: Type of PTZ action
+      required: true
+      advanced: false
+      example: move
+      default: move
+      selector:
+        select:
+          options:
+            - "move"
+            - "preset"
+            - "stop"
+            - "zoom"
+    argument:
+      name: PTZ Action
+      description: >
+        left, right, up, down for move; in, out for zoom; name of preset
+      required: false
+      advanced: false
+      example: down
+      default: ""
+      selector:
+        text:

tests/test_camera.py

@@ -12,11 +12,14 @@
 from custom_components.frigate.const import (
     ATTR_EVENT_ID,
     ATTR_FAVORITE,
+    ATTR_PTZ_ACTION,
+    ATTR_PTZ_ARGUMENT,
     CONF_RTMP_URL_TEMPLATE,
     CONF_RTSP_URL_TEMPLATE,
     DOMAIN,
     NAME,
     SERVICE_FAVORITE_EVENT,
+    SERVICE_PTZ,
 )
 from homeassistant.components.camera import (
     DOMAIN as CAMERA_DOMAIN,
@@ -491,3 +494,71 @@ async def test_retain_service_call(
         blocking=True,
     )
     client.async_retain.assert_called_with(event_id, True)
+
+
+async def test_ptz_move_service_call(
+    hass: HomeAssistant,
+    mqtt_mock: Any,
+) -> None:
+    """Test ptz service call."""
+    client = create_mock_frigate_client()
+    await setup_mock_frigate_config_entry(hass, client=client)
+
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_PTZ,
+        {
+            ATTR_ENTITY_ID: TEST_CAMERA_FRONT_DOOR_ENTITY_ID,
+            ATTR_PTZ_ACTION: "move",
+            ATTR_PTZ_ARGUMENT: "up",
+        },
+        blocking=True,
+    )
+    mqtt_mock.async_publish.assert_called_once_with(
+        "frigate/front_door/ptz", "move_up", 0, False
+    )
+
+
+async def test_ptz_preset_service_call(
+    hass: HomeAssistant,
+    mqtt_mock: Any,
+) -> None:
+    """Test ptz service call."""
+    client = create_mock_frigate_client()
+    await setup_mock_frigate_config_entry(hass, client=client)
+
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_PTZ,
+        {
+            ATTR_ENTITY_ID: TEST_CAMERA_FRONT_DOOR_ENTITY_ID,
+            ATTR_PTZ_ACTION: "preset",
+            ATTR_PTZ_ARGUMENT: "main",
+        },
+        blocking=True,
+    )
+    mqtt_mock.async_publish.assert_called_once_with(
+        "frigate/front_door/ptz", "preset_main", 0, False
+    )
+
+
+async def test_ptz_stop_service_call(
+    hass: HomeAssistant,
+    mqtt_mock: Any,
+) -> None:
+    """Test ptz service call."""
+    client = create_mock_frigate_client()
+    await setup_mock_frigate_config_entry(hass, client=client)
+
+    await hass.services.async_call(
+        DOMAIN,
+        SERVICE_PTZ,
+        {
+            ATTR_ENTITY_ID: TEST_CAMERA_FRONT_DOOR_ENTITY_ID,
+            ATTR_PTZ_ACTION: "stop",
+        },
+        blocking=True,
+    )
+    mqtt_mock.async_publish.assert_called_once_with(
+        "frigate/front_door/ptz", "stop", 0, False
+    )