Migrate Tuya camera to use wrapper class (home-assistant#156542)

Author: epenet

homeassistant/components/tuya/camera.py

@@ -13,6 +13,7 @@
 from . import TuyaConfigEntry
 from .const import TUYA_DISCOVERY_NEW, DeviceCategory, DPCode
 from .entity import TuyaEntity
+from .models import DPCodeBooleanWrapper
 
 CAMERAS: tuple[DeviceCategory, ...] = (
     DeviceCategory.DGHSXJ,
@@ -35,7 +36,18 @@ def async_discover_device(device_ids: list[str]) -> None:
         for device_id in device_ids:
             device = manager.device_map[device_id]
             if device.category in CAMERAS:
-                entities.append(TuyaCameraEntity(device, manager))
+                entities.append(
+                    TuyaCameraEntity(
+                        device,
+                        manager,
+                        motion_detection_switch=DPCodeBooleanWrapper.find_dpcode(
+                            device, DPCode.MOTION_SWITCH, prefer_function=True
+                        ),
+                        recording_status=DPCodeBooleanWrapper.find_dpcode(
+                            device, DPCode.RECORD_SWITCH
+                        ),
+                    )
+                )
 
         async_add_entities(entities)
 
@@ -57,21 +69,30 @@ def __init__(
         self,
         device: CustomerDevice,
         device_manager: Manager,
+        *,
+        motion_detection_switch: DPCodeBooleanWrapper | None = None,
+        recording_status: DPCodeBooleanWrapper | None = None,
     ) -> None:
         """Init Tuya Camera."""
         super().__init__(device, device_manager)
         CameraEntity.__init__(self)
         self._attr_model = device.product_name
+        self._motion_detection_switch = motion_detection_switch
+        self._recording_status = recording_status
 
     @property
     def is_recording(self) -> bool:
         """Return true if the device is recording."""
-        return self.device.status.get(DPCode.RECORD_SWITCH, False)
+        if (status := self._read_wrapper(self._recording_status)) is not None:
+            return status
+        return False
 
     @property
     def motion_detection_enabled(self) -> bool:
         """Return the camera motion detection status."""
-        return self.device.status.get(DPCode.MOTION_SWITCH, False)
+        if (status := self._read_wrapper(self._motion_detection_switch)) is not None:
+            return status
+        return False
 
     async def stream_source(self) -> str | None:
         """Return the source of the stream."""
@@ -95,10 +116,10 @@ async def async_camera_image(
             height=height,
         )
 
-    def enable_motion_detection(self) -> None:
+    async def async_enable_motion_detection(self) -> None:
         """Enable motion detection in the camera."""
-        self._send_command([{"code": DPCode.MOTION_SWITCH, "value": True}])
+        await self._async_send_dpcode_update(self._motion_detection_switch, True)
 
-    def disable_motion_detection(self) -> None:
+    async def async_disable_motion_detection(self) -> None:
         """Disable motion detection in camera."""
-        self._send_command([{"code": DPCode.MOTION_SWITCH, "value": False}])
+        await self._async_send_dpcode_update(self._motion_detection_switch, False)

homeassistant/components/tuya/entity.py

@@ -66,10 +66,18 @@ def _send_command(self, commands: list[dict[str, Any]]) -> None:
         LOGGER.debug("Sending commands for device %s: %s", self.device.id, commands)
         self.device_manager.send_commands(self.device.id, commands)
 
+    def _read_wrapper(self, dpcode_wrapper: DPCodeWrapper | None) -> Any | None:
+        """Read the wrapper device status."""
+        if dpcode_wrapper is None:
+            return None
+        return dpcode_wrapper.read_device_status(self.device)
+
     async def _async_send_dpcode_update(
-        self, dpcode_wrapper: DPCodeWrapper, value: Any
+        self, dpcode_wrapper: DPCodeWrapper | None, value: Any
     ) -> None:
         """Send command to the device."""
+        if dpcode_wrapper is None:
+            return
         await self.hass.async_add_executor_job(
             self._send_command,
             [dpcode_wrapper.get_update_command(self.device, value)],

tests/components/tuya/test_camera.py

@@ -2,13 +2,19 @@
 
 from __future__ import annotations
 
+from typing import Any
 from unittest.mock import patch
 
 import pytest
 from syrupy.assertion import SnapshotAssertion
 from tuya_sharing import CustomerDevice, Manager
 
-from homeassistant.const import Platform
+from homeassistant.components.camera import (
+    DOMAIN as CAMERA_DOMAIN,
+    SERVICE_DISABLE_MOTION,
+    SERVICE_ENABLE_MOTION,
+)
+from homeassistant.const import ATTR_ENTITY_ID, Platform
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import entity_registry as er
 
@@ -46,3 +52,48 @@ async def test_platform_setup_and_discovery(
         snapshot,
         mock_config_entry.entry_id,
     )
+
+
+@patch("homeassistant.components.tuya.PLATFORMS", [Platform.CAMERA])
+@pytest.mark.parametrize(
+    "mock_device_code",
+    ["sp_rudejjigkywujjvs"],
+)
+@pytest.mark.parametrize(
+    ("service", "expected_command"),
+    [
+        (
+            SERVICE_DISABLE_MOTION,
+            {"code": "motion_switch", "value": False},
+        ),
+        (
+            SERVICE_ENABLE_MOTION,
+            {"code": "motion_switch", "value": True},
+        ),
+    ],
+)
+async def test_motion_detection(
+    hass: HomeAssistant,
+    mock_manager: Manager,
+    mock_config_entry: MockConfigEntry,
+    mock_device: CustomerDevice,
+    service: str,
+    expected_command: dict[str, Any],
+) -> None:
+    """Test turning off a switch."""
+    entity_id = "camera.burocam"
+    await initialize_entry(hass, mock_manager, mock_config_entry, mock_device)
+
+    state = hass.states.get(entity_id)
+    assert state is not None, f"{entity_id} does not exist"
+    await hass.services.async_call(
+        CAMERA_DOMAIN,
+        service,
+        {
+            ATTR_ENTITY_ID: entity_id,
+        },
+        blocking=True,
+    )
+    mock_manager.send_commands.assert_called_once_with(
+        mock_device.id, [expected_command]
+    )