Add support for Tasmota camera (#144067)

Co-authored-by: Erik Montnemery <erik@montnemery.com>

homeassistant/components/tasmota/camera.py

@@ -0,0 +1,110 @@
+"""Support for Tasmota Camera."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+import aiohttp
+from hatasmota import camera as tasmota_camera
+from hatasmota.entity import TasmotaEntity as HATasmotaEntity
+from hatasmota.models import DiscoveryHashType
+
+from homeassistant.components import camera
+from homeassistant.components.camera import Camera
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.aiohttp_client import (
+    async_aiohttp_proxy_web,
+    async_get_clientsession,
+)
+from homeassistant.helpers.dispatcher import async_dispatcher_connect
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+
+from .const import DATA_REMOVE_DISCOVER_COMPONENT
+from .discovery import TASMOTA_DISCOVERY_ENTITY_NEW
+from .entity import TasmotaAvailability, TasmotaDiscoveryUpdate, TasmotaEntity
+
+TIMEOUT = 10
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up Tasmota light dynamically through discovery."""
+
+    @callback
+    def async_discover(
+        tasmota_entity: HATasmotaEntity, discovery_hash: DiscoveryHashType
+    ) -> None:
+        """Discover and add a Tasmota camera."""
+        async_add_entities(
+            [
+                TasmotaCamera(
+                    tasmota_entity=tasmota_entity, discovery_hash=discovery_hash
+                )
+            ]
+        )
+
+    hass.data[DATA_REMOVE_DISCOVER_COMPONENT.format(camera.DOMAIN)] = (
+        async_dispatcher_connect(
+            hass,
+            TASMOTA_DISCOVERY_ENTITY_NEW.format(camera.DOMAIN),
+            async_discover,
+        )
+    )
+
+
+class TasmotaCamera(
+    TasmotaAvailability,
+    TasmotaDiscoveryUpdate,
+    TasmotaEntity,
+    Camera,
+):
+    """Representation of a Tasmota Camera."""
+
+    _tasmota_entity: tasmota_camera.TasmotaCamera
+
+    def __init__(self, **kwds: Any) -> None:
+        """Initialize."""
+        super().__init__(**kwds)
+        Camera.__init__(self)
+
+    async def async_camera_image(
+        self, width: int | None = None, height: int | None = None
+    ) -> bytes | None:
+        """Return a still image response from the camera."""
+
+        websession = async_get_clientsession(self.hass)
+        try:
+            async with asyncio.timeout(TIMEOUT):
+                response = await self._tasmota_entity.get_still_image_stream(websession)
+                return await response.read()
+
+        except TimeoutError as err:
+            raise HomeAssistantError(
+                f"Timeout getting camera image from {self.name}: {err}"
+            ) from err
+
+        except aiohttp.ClientError as err:
+            raise HomeAssistantError(
+                f"Error getting new camera image from {self.name}: {err}"
+            ) from err
+
+        return None
+
+    async def handle_async_mjpeg_stream(
+        self, request: aiohttp.web.Request
+    ) -> aiohttp.web.StreamResponse | None:
+        """Generate an HTTP MJPEG stream from the camera."""
+        # connect to stream
+        websession = async_get_clientsession(self.hass)
+        stream_coro = self._tasmota_entity.get_mjpeg_stream(websession)
+
+        return await async_aiohttp_proxy_web(self.hass, request, stream_coro)

homeassistant/components/tasmota/const.py

@@ -13,6 +13,7 @@ DOMAIN = "tasmota"
 
 PLATFORMS = [
     Platform.BINARY_SENSOR,
+    Platform.CAMERA,
     Platform.COVER,
     Platform.FAN,
     Platform.LIGHT,

tests/components/tasmota/test_camera.py

@@ -0,0 +1,308 @@
+"""The tests for the Tasmota camera platform."""
+
+from asyncio import Future
+import copy
+import json
+from unittest.mock import patch
+
+import pytest
+
+from homeassistant.components.camera import CameraState
+from homeassistant.components.tasmota.const import DEFAULT_PREFIX
+from homeassistant.const import ATTR_ASSUMED_STATE, Platform
+from homeassistant.core import HomeAssistant
+
+from .test_common import (
+    DEFAULT_CONFIG,
+    help_test_availability,
+    help_test_availability_discovery_update,
+    help_test_availability_poll_state,
+    help_test_availability_when_connection_lost,
+    help_test_deep_sleep_availability,
+    help_test_deep_sleep_availability_when_connection_lost,
+    help_test_discovery_device_remove,
+    help_test_discovery_removal,
+    help_test_discovery_update_unchanged,
+    help_test_entity_id_update_discovery_update,
+)
+
+from tests.common import async_fire_mqtt_message
+from tests.typing import ClientSessionGenerator, MqttMockHAClient, MqttMockPahoClient
+
+SMALLEST_VALID_JPEG = (
+    "ffd8ffe000104a46494600010101004800480000ffdb00430003020202020203020202030303030406040404040408060"
+    "6050609080a0a090809090a0c0f0c0a0b0e0b09090d110d0e0f101011100a0c12131210130f101010ffc9000b08000100"
+    "0101011100ffcc000600101005ffda0008010100003f00d2cf20ffd9"
+)
+SMALLEST_VALID_JPEG_BYTES = bytes.fromhex(SMALLEST_VALID_JPEG)
+
+
+async def test_controlling_state_via_mqtt(
+    hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_tasmota
+) -> None:
+    """Test state update via MQTT."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    mac = config["mac"]
+
+    async_fire_mqtt_message(
+        hass,
+        f"{DEFAULT_PREFIX}/{mac}/config",
+        json.dumps(config),
+    )
+    await hass.async_block_till_done()
+
+    state = hass.states.get("camera.tasmota")
+    assert state.state == "unavailable"
+    assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+    async_fire_mqtt_message(hass, "tasmota_49A3BC/tele/LWT", "Online")
+    await hass.async_block_till_done()
+    state = hass.states.get("camera.tasmota")
+    assert state.state == CameraState.IDLE
+    assert not state.attributes.get(ATTR_ASSUMED_STATE)
+
+
+async def test_availability_when_connection_lost(
+    hass: HomeAssistant,
+    mqtt_client_mock: MqttMockPahoClient,
+    mqtt_mock: MqttMockHAClient,
+    setup_tasmota,
+) -> None:
+    """Test availability after MQTT disconnection."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    await help_test_availability_when_connection_lost(
+        hass, mqtt_client_mock, mqtt_mock, Platform.CAMERA, config, object_id="tasmota"
+    )
+
+
+async def test_deep_sleep_availability_when_connection_lost(
+    hass: HomeAssistant,
+    mqtt_client_mock: MqttMockPahoClient,
+    mqtt_mock: MqttMockHAClient,
+    setup_tasmota,
+) -> None:
+    """Test availability after MQTT disconnection."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    await help_test_deep_sleep_availability_when_connection_lost(
+        hass, mqtt_client_mock, mqtt_mock, Platform.CAMERA, config, object_id="tasmota"
+    )
+
+
+async def test_availability(
+    hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_tasmota
+) -> None:
+    """Test availability."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    await help_test_availability(
+        hass, mqtt_mock, Platform.CAMERA, config, object_id="tasmota"
+    )
+
+
+async def test_deep_sleep_availability(
+    hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_tasmota
+) -> None:
+    """Test availability."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    await help_test_deep_sleep_availability(
+        hass, mqtt_mock, Platform.CAMERA, config, object_id="tasmota"
+    )
+
+
+async def test_availability_discovery_update(
+    hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_tasmota
+) -> None:
+    """Test availability discovery update."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    await help_test_availability_discovery_update(
+        hass, mqtt_mock, Platform.CAMERA, config, object_id="tasmota"
+    )
+
+
+async def test_availability_poll_state(
+    hass: HomeAssistant,
+    mqtt_client_mock: MqttMockPahoClient,
+    mqtt_mock: MqttMockHAClient,
+    setup_tasmota,
+) -> None:
+    """Test polling after MQTT connection (re)established."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    poll_topic = "tasmota_49A3BC/cmnd/STATE"
+    await help_test_availability_poll_state(
+        hass, mqtt_client_mock, mqtt_mock, Platform.CAMERA, config, poll_topic, ""
+    )
+
+
+async def test_discovery_removal_camera(
+    hass: HomeAssistant,
+    mqtt_mock: MqttMockHAClient,
+    caplog: pytest.LogCaptureFixture,
+    setup_tasmota,
+) -> None:
+    """Test removal of discovered camera."""
+    config1 = copy.deepcopy(DEFAULT_CONFIG)
+    config1["cam"] = 1
+    config2 = copy.deepcopy(DEFAULT_CONFIG)
+    config2["cam"] = 0
+
+    await help_test_discovery_removal(
+        hass,
+        mqtt_mock,
+        caplog,
+        Platform.CAMERA,
+        config1,
+        config2,
+        object_id="tasmota",
+        name="Tasmota",
+    )
+
+
+async def test_discovery_update_unchanged_camera(
+    hass: HomeAssistant,
+    mqtt_mock: MqttMockHAClient,
+    caplog: pytest.LogCaptureFixture,
+    setup_tasmota,
+) -> None:
+    """Test update of discovered camera."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    with patch(
+        "homeassistant.components.tasmota.camera.TasmotaCamera.discovery_update"
+    ) as discovery_update:
+        await help_test_discovery_update_unchanged(
+            hass,
+            mqtt_mock,
+            caplog,
+            Platform.CAMERA,
+            config,
+            discovery_update,
+            object_id="tasmota",
+            name="Tasmota",
+        )
+
+
+async def test_discovery_device_remove(
+    hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_tasmota
+) -> None:
+    """Test device registry remove."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    unique_id = f"{DEFAULT_CONFIG['mac']}_camera_camera_0"
+    await help_test_discovery_device_remove(
+        hass, mqtt_mock, Platform.CAMERA, unique_id, config
+    )
+
+
+async def test_entity_id_update_discovery_update(
+    hass: HomeAssistant, mqtt_mock: MqttMockHAClient, setup_tasmota
+) -> None:
+    """Test MQTT discovery update when entity_id is updated."""
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+    await help_test_entity_id_update_discovery_update(
+        hass, mqtt_mock, Platform.CAMERA, config, object_id="tasmota"
+    )
+
+
+async def test_camera_single_frame(
+    hass: HomeAssistant,
+    mqtt_mock: MqttMockHAClient,
+    setup_tasmota,
+    hass_client: ClientSessionGenerator,
+) -> None:
+    """Test single frame capture."""
+
+    class MockClientResponse:
+        def __init__(self, text) -> None:
+            self._text = text
+
+        async def read(self):
+            return self._text
+
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+
+    mac = config["mac"]
+    async_fire_mqtt_message(
+        hass,
+        f"{DEFAULT_PREFIX}/{mac}/config",
+        json.dumps(config),
+    )
+
+    mock_single_image_stream = Future()
+    mock_single_image_stream.set_result(MockClientResponse(SMALLEST_VALID_JPEG_BYTES))
+
+    with patch(
+        "hatasmota.camera.TasmotaCamera.get_still_image_stream",
+        return_value=mock_single_image_stream,
+    ):
+        client = await hass_client()
+        resp = await client.get("/api/camera_proxy/camera.tasmota")
+        await hass.async_block_till_done()
+
+    assert resp.status == 200
+    assert resp.content_type == "image/jpeg"
+    assert resp.content_length == len(SMALLEST_VALID_JPEG_BYTES)
+    assert await resp.read() == SMALLEST_VALID_JPEG_BYTES
+
+
+async def test_camera_stream(
+    hass: HomeAssistant,
+    mqtt_mock: MqttMockHAClient,
+    setup_tasmota,
+    hass_client: ClientSessionGenerator,
+) -> None:
+    """Test mjpeg stream capture."""
+
+    class MockClientResponse:
+        def __init__(self, text) -> None:
+            self._text = text
+            self._frame_available = True
+
+        async def read(self, buffer_size):
+            if self._frame_available:
+                self._frame_available = False
+                return self._text
+            return None
+
+        def close(self):
+            pass
+
+        @property
+        def headers(self):
+            return {"Content-Type": "multipart/x-mixed-replace"}
+
+        @property
+        def content(self):
+            return self
+
+    config = copy.deepcopy(DEFAULT_CONFIG)
+    config["cam"] = 1
+
+    mac = config["mac"]
+    async_fire_mqtt_message(
+        hass,
+        f"{DEFAULT_PREFIX}/{mac}/config",
+        json.dumps(config),
+    )
+
+    mock_mjpeg_stream = Future()
+    mock_mjpeg_stream.set_result(MockClientResponse(SMALLEST_VALID_JPEG_BYTES))
+
+    with patch(
+        "hatasmota.camera.TasmotaCamera.get_mjpeg_stream",
+        return_value=mock_mjpeg_stream,
+    ):
+        client = await hass_client()
+        resp = await client.get("/api/camera_proxy_stream/camera.tasmota")
+        await hass.async_block_till_done()
+
+    assert resp.status == 200
+    assert resp.content_type == "multipart/x-mixed-replace"
+    assert await resp.read() == SMALLEST_VALID_JPEG_BYTES

tests/components/tasmota/test_common.py

@@ -36,6 +36,7 @@ DEFAULT_CONFIG = {
     "fn": ["Test", "Beer", "Milk", "Four", None],
     "hn": "tasmota_49A3BC-0956",
     "if": 0,  # iFan
+    "cam": 0,  # webcam
     "lk": 1,  # RGB + white channels linked to a single light
     "mac": "00000049A3BC",
     "md": "Sonoff Basic",