Refactor ESPHome camera to avoid creating tasks

homeassistant/components/esphome/camera.py

@@ -2,10 +2,13 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Callable, Coroutine
+from functools import partial
 from typing import Any
 
 from aioesphomeapi import CameraInfo, CameraState
 from aiohttp import web
+import async_timeout
 
 from homeassistant.components import camera
 from homeassistant.components.camera import Camera
@@ -18,6 +21,9 @@
     platform_async_setup_entry,
 )
 
+# ESPHome camera timeout is 5 seconds, but we add some buffer for wifi latency
+CAMERA_TIMEOUT = 10
+
 
 async def async_setup_entry(
     hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
@@ -40,48 +46,43 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         """Initialize."""
         Camera.__init__(self)
         EsphomeEntity.__init__(self, *args, **kwargs)
-        self._image_cond = asyncio.Condition()
+        self._image_event = asyncio.Event()
 
     @callback
     def _on_state_update(self) -> None:
         """Notify listeners of new image when update arrives."""
         super()._on_state_update()
-        self.hass.async_create_task(self._on_state_update_coro())
-
-    async def _on_state_update_coro(self) -> None:
-        async with self._image_cond:
-            self._image_cond.notify_all()
+        image_event = self._image_event
+        image_event.set()
+        image_event.clear()
 
     async def async_camera_image(
         self, width: int | None = None, height: int | None = None
     ) -> bytes | None:
         """Return single camera image bytes."""
-        if not self.available:
-            return None
-        await self._client.request_single_image()
-        async with self._image_cond:
-            await self._image_cond.wait()
-            if not self.available:
-                # Availability can change while waiting for 'self._image.cond'
-                return None  # type: ignore[unreachable]
-            return self._state.data[:]
+        return await self._async_request_image(self._client.request_single_image)
 
-    async def _async_camera_stream_image(self) -> bytes | None:
-        """Return a single camera image in a stream."""
+    async def _async_request_image(
+        self, request_method: Callable[[], Coroutine[Any, Any, None]]
+    ) -> bytes | None:
+        """Wait for an image to be available and return it."""
         if not self.available:
             return None
-        await self._client.request_image_stream()
-        async with self._image_cond:
-            await self._image_cond.wait()
-            if not self.available:
-                # Availability can change while waiting for 'self._image.cond'
-                return None  # type: ignore[unreachable]
-            return self._state.data[:]
+        await request_method()
+        async with async_timeout.timeout(CAMERA_TIMEOUT):
+            await self._image_event.wait()
+        if not self.available:
+            # Availability can change while waiting for 'image_event'
+            return None  # type: ignore[unreachable]
+        return self._state.data
 
     async def handle_async_mjpeg_stream(
         self, request: web.Request
     ) -> web.StreamResponse:
         """Serve an HTTP MJPEG stream from the camera."""
+        stream_request = partial(
+            self._async_request_image, self._client.request_image_stream
+        )
         return await camera.async_get_still_stream(
-            request, self._async_camera_stream_image, camera.DEFAULT_CONTENT_TYPE, 0.0
+            request, stream_request, camera.DEFAULT_CONTENT_TYPE, 0.0
         )