camera.py

Go to the documentation of this file.

"""Support for Tuya cameras."""
 
from __future__ import annotations
 
from tuya_sharing import CustomerDevice, Manager
 
from homeassistant.components import ffmpeg
from homeassistant.components.camera import Camera as CameraEntity, CameraEntityFeature
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
from . import TuyaConfigEntry
from .const import TUYA_DISCOVERY_NEW, DPCode
from .entity import TuyaEntity
 
# All descriptions can be found here:
# https://developer.tuya.com/en/docs/iot/standarddescription?id=K9i5ql6waswzq
CAMERAS: tuple[str, ...] = (
    # Smart Camera (including doorbells)
    # https://developer.tuya.com/en/docs/iot/categorysgbj?id=Kaiuz37tlpbnu
    "sp",
)
 
 
async def async_setup_entry(
    hass: HomeAssistant, entry: TuyaConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
    """Set up Tuya cameras dynamically through Tuya discovery."""
    hass_data = entry.runtime_data
 
    @callback
    def async_discover_device(device_ids: list[str]) -> None:
        """Discover and add a discovered Tuya camera."""
        entities: list[TuyaCameraEntity] = []
        for device_id in device_ids:
            device = hass_data.manager.device_map[device_id]
            if device.category in CAMERAS:
                entities.append(TuyaCameraEntity(device, hass_data.manager))
 
        async_add_entities(entities)
 
    async_discover_device([*hass_data.manager.device_map])
 
    entry.async_on_unload(
        async_dispatcher_connect(hass, TUYA_DISCOVERY_NEW, async_discover_device)
    )
 
 
class TuyaCameraEntity(TuyaEntity, CameraEntity):
    """Tuya Camera Entity."""
 
    _attr_supported_features = CameraEntityFeature.STREAM
    _attr_brand = "Tuya"
    _attr_name = None
 
    def __init__(
        self,
        device: CustomerDevice,
        device_manager: Manager,
    ) -> None:
        """Init Tuya Camera."""
        super().__init__(device, device_manager)
        CameraEntity.__init__(self)
        self._attr_model_attr_model = device.product_name
 
    @property
    def is_recording(self) -> bool:
        """Return true if the device is recording."""
        return self.devicedevice.status.get(DPCode.RECORD_SWITCH, False)
 
    @property
    def motion_detection_enabled(self) -> bool:
        """Return the camera motion detection status."""
        return self.devicedevice.status.get(DPCode.MOTION_SWITCH, False)
 
    async def stream_source(self) -> str | None:
        """Return the source of the stream."""
        return await self.hasshass.async_add_executor_job(
            self.device_managerdevice_manager.get_device_stream_allocate,
            self.devicedevice.id,
            "rtsp",
        )
 
    async def async_camera_image(
        self, width: int | None = None, height: int | None = None
    ) -> bytes | None:
        """Return a still image response from the camera."""
        stream_source = await self.stream_sourcestream_source()
        if not stream_source:
            return None
        return await ffmpeg.async_get_image(
            self.hasshass,
            stream_source,
            width=width,
            height=height,
        )
 
    def enable_motion_detection(self) -> None:
        """Enable motion detection in the camera."""
        self._send_command_send_command([{"code": DPCode.MOTION_SWITCH, "value": True}])
 
    def disable_motion_detection(self) -> None:
        """Disable motion detection in camera."""
        self._send_command_send_command([{"code": DPCode.MOTION_SWITCH, "value": False}])