15 files

custom_components/album_slideshow/camera.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
 import asyncio
-from collections import OrderedDict
+from collections import OrderedDict, deque
+from dataclasses import dataclass
+from datetime import datetime, timezone
 import logging
 import random
 from pathlib import Path
+from typing import Any
 
 import async_timeout
 from PIL import Image
@@ -46,18 +49,44 @@ _ACCEPTED_IMAGE_PREFIX = ("image/",)
 _PAIR_SEARCH_LIMIT = 12
 _SKIP_SEARCH_LIMIT = 30
 
+_MAX_RENDER_ATTEMPTS = 10
+
+
+@dataclass(frozen=True, slots=True)
+class _NavigationCursor:
+    """All mutable ordering state needed to render the following slide."""
+
+    index: int
+    random_order: tuple[int, ...]
+    random_pos: int
+    recent_urls: tuple[str, ...]
+    rng_state: Any
+
+
+@dataclass(frozen=True, slots=True)
+class _RenderedFrame:
+    """A complete slide that can be displayed without any further work."""
+
+    data: bytes
+    cursor: _NavigationCursor
+    meta: dict
+
 
 def _ts_to_iso(ts_ms: int | None) -> str | None:
     """Convert epoch milliseconds to an ISO-8601 string in UTC, or None."""
     if not isinstance(ts_ms, int):
         return None
-    from datetime import datetime, timezone
     try:
         return datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).isoformat()
     except (OverflowError, OSError, ValueError):
         return None
 
 
+def _utc_now_iso() -> str:
+    """Return the current time as an ISO-8601 UTC string."""
+    return datetime.now(timezone.utc).isoformat()
+
+
 class _DownloadCache:
     """Byte-budget LRU cache for downloaded image data, O(1) per operation."""
 
@@ -155,22 +184,43 @@ class AlbumSlideshowCamera(Camera):
         self._frame_id: int = 0
 
         self._interrupt_event: asyncio.Event = asyncio.Event()
-        self._force_next: bool = False
+        # Navigation runs directly in the button/service coroutine. The lock
+        # serialises rapid presses while buffered swaps remain independent of
+        # the background timer loop.
+        self._navigation_lock: asyncio.Lock = asyncio.Lock()
+        self._navigation_pending: int = 0
+        # Rendered timeline around the current frame. Previous frames and future
+        # frames are final encoded JPEGs, so both navigation directions are an
+        # O(1) deque swap. The future deque is replenished in the background.
+        self._current_frame: _RenderedFrame | None = None
+        self._previous_frames: deque[_RenderedFrame] = deque()
+        self._next_frames: deque[_RenderedFrame] = deque()
+        self._timeline_generation: int = 0
+        self._timeline_dirty: bool = False
+        self._next_ready_event: asyncio.Event = asyncio.Event()
+        self._preload_task: asyncio.Task | None = None
+        # Visible navigation diagnostics. These are state attributes rather
+        # than debug-only log messages so they remain observable when Home
+        # Assistant's UI filters debug logs.
+        self._last_nav_direction: str | None = None
+        self._last_nav_requested_at: str | None = None
+        self._last_nav_started_at: str | None = None
+        self._last_nav_committed_at: str | None = None
+        self._last_nav_outcome: str | None = None
+        self._last_nav_error: str | None = None
         self._consecutive_failures: int = 0
         self._render_task: asyncio.Task | None = None
 
         def _on_coordinator_update() -> None:
             self._effective_cache = None
-            self._interrupt_event.set()
-            self.async_write_ha_state()
+            self._invalidate_timeline()
 
         coordinator.async_add_listener(_on_coordinator_update)
 
         def _on_store_change() -> None:
             self._download_cache.resize(self.store.image_cache_mb * 1024 * 1024)
             self._effective_cache = None
-            self._interrupt_event.set()
-            self.async_write_ha_state()
+            self._invalidate_timeline()
 
         store.add_listener(_on_store_change)
 
@@ -193,12 +243,19 @@ class AlbumSlideshowCamera(Camera):
         )
 
     async def async_will_remove_from_hass(self) -> None:
+        preload_task = self._preload_task
+        self._cancel_preload()
         if self._render_task is not None:
             self._render_task.cancel()
             try:
                 await self._render_task
             except asyncio.CancelledError:
                 pass
+        if preload_task is not None:
+            try:
+                await preload_task
+            except asyncio.CancelledError:
+                pass
 
     @property
     def device_info(self):
@@ -258,6 +315,19 @@ class AlbumSlideshowCamera(Camera):
             "pair_divider_px": int(self.store.pair_divider_px),
             "pair_divider_color": self.store.pair_divider_color,
             "frame_id": self._frame_id,
+            "navigation_buffer_size": self._buffer_depth,
+            "previous_frames_cached": len(self._previous_frames),
+            "next_frames_preloaded": len(self._next_frames),
+            "navigation_preloading": bool(
+                self._preload_task is not None and not self._preload_task.done()
+            ),
+            "navigation_queue_size": self._navigation_pending,
+            "last_navigation_direction": self._last_nav_direction,
+            "last_navigation_requested_at": self._last_nav_requested_at,
+            "last_navigation_started_at": self._last_nav_started_at,
+            "last_navigation_committed_at": self._last_nav_committed_at,
+            "last_navigation_outcome": self._last_nav_outcome,
+            "last_navigation_error": self._last_nav_error,
             "pagination_debug": data.get("pagination_debug"),
         }
 
@@ -331,10 +401,53 @@ class AlbumSlideshowCamera(Camera):
         return ordered
 
     async def async_force_next(self) -> None:
-        self._force_next = True
+        """Display the next buffered slide immediately."""
+        await self._async_navigate(1)
+
+    async def async_force_prev(self) -> None:
+        """Display the previous retained slide immediately."""
+        await self._async_navigate(-1)
+
+    async def _async_navigate(self, direction: int) -> None:
+        """Serialise a manual navigation request and execute it directly."""
+        direction_name = "next" if direction > 0 else "previous"
+        self._navigation_pending += 1
+        self._last_nav_direction = direction_name
+        self._last_nav_requested_at = _utc_now_iso()
+        self._last_nav_outcome = "pending"
+        self._last_nav_error = None
+        # Wake the timer loop so the manual frame starts a fresh interval.
         self._interrupt_event.set()
         self.async_write_ha_state()
 
+        try:
+            async with self._navigation_lock:
+                self._last_nav_started_at = _utc_now_iso()
+                if self._timeline_dirty:
+                    await self._rebuild_current_frame()
+                changed = (
+                    await self._show_next_frame()
+                    if direction > 0
+                    else await self._show_previous_frame()
+                )
+                if changed:
+                    self._last_nav_committed_at = _utc_now_iso()
+                    self._last_nav_outcome = "displayed"
+                else:
+                    self._last_nav_outcome = "not_available"
+        except Exception as err:
+            self._last_nav_outcome = "error"
+            self._last_nav_error = str(err)
+            _LOGGER.warning(
+                "Album Slideshow %s: manual %s navigation failed: %s",
+                self.entry.title,
+                direction_name,
+                err,
+            )
+        finally:
+            self._navigation_pending -= 1
+            self.async_write_ha_state()
+
     async def async_force_refresh(self) -> None:
         await self.coordinator.async_request_refresh()
 
@@ -373,123 +486,387 @@ class AlbumSlideshowCamera(Camera):
     async def async_handle_async_mjpeg_stream(self, request):
         return await self.handle_async_mjpeg_stream(request)
 
-    async def _wait_or_interrupt(self, timeout: float) -> bool:
-        """Wait up to ``timeout`` seconds, returning True if interrupted.
+    @property
+    def _buffer_depth(self) -> int:
+        """Configured number of rendered frames retained in each direction."""
+        return min(10, max(0, int(self.store.navigation_buffer_size)))
 
-        Does NOT clear ``_interrupt_event`` - the render loop clears it once
-        per cycle, before rendering, so a signal that arrives while we're
-        rendering (a "next slide" press, a coordinator/store change) survives
-        until we get here instead of being wiped.
+    def _capture_cursor(self, source=None) -> _NavigationCursor:
+        """Snapshot ordering state from this camera or a private renderer."""
+        source = source or self
+        return _NavigationCursor(
+            index=int(source._index),
+            random_order=tuple(source._random_order),
+            random_pos=int(source._random_pos),
+            recent_urls=tuple(source._recent_urls),
+            rng_state=source._rng.getstate(),
+        )
 
-        A force-next that's already pending is honored immediately without
-        sleeping, so a press that landed during the last render (or right at
-        the loop boundary) isn't held until the full slide interval elapses.
+    def _make_renderer(self, cursor: _NavigationCursor):
+        """Create an isolated render context sharing only immutable services/cache.
+
+        Composition helpers historically operate on ``self._index`` and random
+        ordering fields. A private camera-shaped context lets background
+        preloading reuse those mature helpers without ever mutating the live
+        entity's current frame or navigation cursor.
         """
-        if self._force_next:
+        renderer = AlbumSlideshowCamera.__new__(AlbumSlideshowCamera)
+        renderer.hass = self.hass
+        renderer.entry = self.entry
+        renderer.coordinator = self.coordinator
+        renderer.store = self.store
+        renderer._download_cache = self._download_cache
+        renderer._index = cursor.index
+        renderer._random_order = list(cursor.random_order)
+        renderer._random_pos = cursor.random_pos
+        renderer._recent_urls = list(cursor.recent_urls)
+        renderer._rng = random.Random()
+        renderer._rng.setstate(cursor.rng_state)
+        return renderer
+
+    async def _render_available_frame(
+        self,
+        cursor: _NavigationCursor,
+        items: list[MediaItem],
+        *,
+        advance: bool,
+    ) -> _RenderedFrame:
+        """Render the current or next usable slide from ``cursor``.
+
+        Broken candidates are skipped, matching the old loop's retry behavior.
+        All state mutations occur on a private renderer. The returned JPEG and
+        cursor are therefore safe to place in the future buffer.
+        """
+        if not items:
+            raise RuntimeError("No media available")
+
+        attempts = min(len(items), _MAX_RENDER_ATTEMPTS)
+        last_error: Exception | None = None
+        should_advance = advance
+
+        for _ in range(attempts):
+            renderer = self._make_renderer(cursor)
+            renderer._index %= len(items)
+            if should_advance:
+                renderer._do_advance(len(items), items)
+
+            composed: Image.Image | None = None
+            try:
+                async with self._compose_semaphore:
+                    composed, meta = await renderer._compose_for_index(items)
+                    cursor = self._capture_cursor(renderer)
+                    if composed is None:
+                        raise RuntimeError("Image composition returned no frame")
+                    encoded = await self.hass.async_add_executor_job(
+                        ip.encode_image, composed
+                    )
+                return _RenderedFrame(encoded, cursor, meta or {})
+            except asyncio.CancelledError:
+                raise
+            except Exception as err:
+                last_error = err
+                cursor = self._capture_cursor(renderer)
+                should_advance = True
+                _LOGGER.debug(
+                    "Album Slideshow %s: skipping unrenderable buffered slide: %s",
+                    self.entry.title,
+                    err,
+                )
+            finally:
+                ip.safe_close(composed)
+
+        raise RuntimeError(
+            f"Could not render a usable slide after {attempts} attempts: {last_error}"
+        )
+
+    def _apply_frame(self, frame: _RenderedFrame) -> None:
+        """Make a rendered frame current and publish it to Home Assistant."""
+        self._current_frame = frame
+        self._framebuffer = frame.data
+        self.store.last_frame = frame.data
+        self._index = frame.cursor.index
+        self._random_order = list(frame.cursor.random_order)
+        self._random_pos = frame.cursor.random_pos
+        self._recent_urls = list(frame.cursor.recent_urls)
+        self._rng.setstate(frame.cursor.rng_state)
+
+        meta = frame.meta
+        self._last_is_portrait = meta.get("is_portrait")
+        self._last_captured_at_pair = meta.get("captured_at_pair")
+        self._last_pair_frames = meta.get("pair_frames")
+        self._last_pair_orientation = meta.get("pair_orientation")
+        self._frame_id += 1
+
+        _LOGGER.debug(
+            "Album Slideshow %s: displayed buffered frame_id=%d index=%d "
+            "previous=%d next=%d",
+            self.entry.title,
+            self._frame_id,
+            self._index,
+            len(self._previous_frames),
+            len(self._next_frames),
+        )
+        self.async_write_ha_state()
+
+    def _trim_timeline(self) -> None:
+        """Enforce the configured frame count on both sides of current."""
+        depth = self._buffer_depth
+        while len(self._previous_frames) > depth:
+            self._previous_frames.popleft()
+        while len(self._next_frames) > depth:
+            self._next_frames.pop()
+
+    def _cancel_preload(self) -> None:
+        task = self._preload_task
+        self._preload_task = None
+        if task is not None and not task.done():
+            task.cancel()
+        self._next_ready_event.set()
+
+    def _invalidate_timeline(self) -> None:
+        """Drop frames rendered from stale media/settings and wake the loop."""
+        self._timeline_generation += 1
+        self._timeline_dirty = True
+        self._previous_frames.clear()
+        self._next_frames.clear()
+        self._cancel_preload()
+        self._interrupt_event.set()
+        self.async_write_ha_state()
+
+    def _schedule_preload(self) -> None:
+        """Start the single per-camera worker that fills future frames."""
+        self._trim_timeline()
+        if (
+            self._buffer_depth <= 0
+            or self._current_frame is None
+            or len(self._next_frames) >= self._buffer_depth
+        ):
+            return
+        if self._preload_task is not None and not self._preload_task.done():
+            return
+
+        generation = self._timeline_generation
+        self._preload_task = self.hass.async_create_background_task(
+            self._preload_loop(generation),
+            name="album_slideshow_preload",
+        )
+
+    async def _preload_loop(self, generation: int) -> None:
+        """Render future slides sequentially until the configured buffer is full."""
+        this_task = asyncio.current_task()
+        try:
+            while generation == self._timeline_generation:
+                self._trim_timeline()
+                if len(self._next_frames) >= self._buffer_depth:
+                    return
+
+                base = self._next_frames[-1] if self._next_frames else self._current_frame
+                if base is None:
+                    return
+                items = self._effective_items()
+                if not items:
+                    return
+
+                frame = await self._render_available_frame(
+                    base.cursor,
+                    items,
+                    advance=True,
+                )
+                if generation != self._timeline_generation:
+                    return
+
+                # Navigation may have moved the base from future to current or
+                # vice versa while rendering. It is still valid if it remains
+                # the last frame in the known timeline.
+                current_tail = (
+                    self._next_frames[-1]
+                    if self._next_frames
+                    else self._current_frame
+                )
+                if current_tail is not base:
+                    continue
+                if len(self._next_frames) < self._buffer_depth:
+                    self._next_frames.append(frame)
+                    self._next_ready_event.set()
+                    self.async_write_ha_state()
+                await asyncio.sleep(0)
+        except asyncio.CancelledError:
+            raise
+        except Exception as err:
+            _LOGGER.warning(
+                "Album Slideshow %s: could not fill navigation buffer: %s",
+                self.entry.title,
+                err,
+            )
+        finally:
+            if self._preload_task is this_task:
+                self._preload_task = None
+            self._next_ready_event.set()
+            self.async_write_ha_state()
+
+    async def _await_preloaded_frame(self) -> bool:
+        """Wait for the in-flight worker's first frame, without waiting for all X."""
+        if self._next_frames:
+            return True
+        if self._buffer_depth <= 0:
+            return False
+
+        self._schedule_preload()
+        while not self._next_frames:
+            task = self._preload_task
+            if task is None or task.done():
+                return False
+            self._next_ready_event.clear()
+            if self._next_frames:
+                return True
+            waiter = asyncio.create_task(self._next_ready_event.wait())
+            try:
+                await asyncio.wait(
+                    {task, waiter},
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+            finally:
+                waiter.cancel()
+                try:
+                    await waiter
+                except asyncio.CancelledError:
+                    pass
+        return True
+
+    async def _show_next_frame(self) -> bool:
+        """Display the next buffered frame, rendering on demand only if empty."""
+        if self._current_frame is None:
+            return await self._rebuild_current_frame()
+
+        await self._await_preloaded_frame()
+        if self._next_frames:
+            frame = self._next_frames.popleft()
+        else:
+            items = self._effective_items()
+            frame = await self._render_available_frame(
+                self._current_frame.cursor,
+                items,
+                advance=True,
+            )
+
+        if self._buffer_depth > 0:
+            self._previous_frames.append(self._current_frame)
+        self._trim_timeline()
+        self._apply_frame(frame)
+        self._schedule_preload()
+        return True
+
+    async def _show_previous_frame(self) -> bool:
+        """Restore the most recent retained frame without I/O or composition."""
+        if not self._previous_frames:
+            return False
+
+        frame = self._previous_frames.pop()
+        if self._current_frame is not None and self._buffer_depth > 0:
+            self._next_frames.appendleft(self._current_frame)
+        self._trim_timeline()
+        self._apply_frame(frame)
+        self._schedule_preload()
+        return True
+
+    async def _rebuild_current_frame(self) -> bool:
+        """Render the current cursor after startup or playlist/settings changes."""
+        generation = self._timeline_generation
+        self._timeline_dirty = False
+        items = self._effective_items()
+        if not items:
+            return False
+
+        cursor = (
+            self._current_frame.cursor
+            if self._current_frame is not None
+            else self._capture_cursor()
+        )
+        frame = await self._render_available_frame(cursor, items, advance=False)
+        if generation != self._timeline_generation:
+            return False
+        self._previous_frames.clear()
+        self._next_frames.clear()
+        self._apply_frame(frame)
+        self._schedule_preload()
+        return True
+
+    async def _wait_or_interrupt(self, timeout: float) -> bool:
+        """Wait for configuration/navigation or for the slide timer to expire."""
+        if self._timeline_dirty:
+            return True
+        self._interrupt_event.clear()
+        # Close the clear/wait race: callbacks cannot run between these two
+        # synchronous statements without setting the event again.
+        if self._timeline_dirty:
             return True
         try:
-            await asyncio.wait_for(self._interrupt_event.wait(), timeout=timeout)
+            if self.store.paused:
+                await self._interrupt_event.wait()
+            else:
+                await asyncio.wait_for(
+                    self._interrupt_event.wait(),
+                    timeout=timeout,
+                )
             return True
         except asyncio.TimeoutError:
             return False
 
     async def _render_loop(self, initial_delay: float = 0.0) -> None:
-        """Background task: render slides into _framebuffer, advance on timer or interrupt."""
+        """Display buffered frames on command/timer and refill them in the background."""
         if initial_delay > 0:
+            await asyncio.sleep(initial_delay)
+
+        while self._current_frame is None:
             try:
-                await asyncio.sleep(initial_delay)
-            except asyncio.CancelledError:
-                raise
-        should_advance = False  # Don't advance on the very first render
-        while True:
-            # Clear the wake signal before rendering. Anything that happens
-            # from here on (a "next slide" press, a coordinator/store change)
-            # re-sets it and is picked up after this frame commits, so no
-            # wake is lost while we're mid-render.
-            self._interrupt_event.clear()
-            try:
-                await self._render_cycle(advance=should_advance)
-                self._consecutive_failures = 0
+                async with self._navigation_lock:
+                    if self._current_frame is None:
+                        await self._rebuild_current_frame()
+                if self._current_frame is not None:
+                    self._consecutive_failures = 0
+                    break
             except asyncio.CancelledError:
                 raise
             except Exception as err:
                 self._consecutive_failures += 1
                 backoff = min(2 ** self._consecutive_failures, 60)
                 _LOGGER.warning(
-                    "Album Slideshow: render cycle failed (attempt %d), retrying in %ds: %s",
-                    self._consecutive_failures, backoff, err,
+                    "Album Slideshow: initial render failed (attempt %d), retrying in %ds: %s",
+                    self._consecutive_failures,
+                    backoff,
+                    err,
                 )
-                try:
-                    await asyncio.sleep(backoff)
-                except asyncio.CancelledError:
-                    raise
-                should_advance = True  # Skip the broken image on retry
-                continue
+                await asyncio.sleep(backoff)
+            if not self._effective_items():
+                self._interrupt_event.clear()
+                if not self._effective_items():
+                    await self._interrupt_event.wait()
 
-            interrupted = await self._wait_or_interrupt(float(int(self.store.slide_interval)))
-            if self._force_next:
-                # Explicit "next slide" request: always advance.
-                should_advance = True
-                self._force_next = False
-            elif interrupted:
-                # A coordinator/store change woke us: re-render the current
-                # frame (new data or settings) without skipping ahead.
-                should_advance = False
-            else:
-                # Paused slideshows hold the current frame until the user
-                # un-pauses or hits "next slide" explicitly.
-                should_advance = not bool(self.store.paused)
-
-    async def _render_cycle(self, advance: bool) -> None:
-        """Render one frame.
-
-        The slideshow is just "advance index, compose, encode, broadcast".
-        Visible transitions are handled by the Lovelace card client-side,
-        so this path stays minimal: at most one PIL decode + encode per
-        slide change.
-
-        Compose work is serialised across all albums via a domain-wide
-        semaphore so 4 cameras don't all decode + encode at once.
-        """
-        items: list[MediaItem] = self._effective_items()
-        if not items:
-            return
-
-        count = len(items)
-        if advance:
-            self._do_advance(count, items)
-
-        async with self._compose_semaphore:
-            composed, meta = await self._compose_for_index(items)
-            if composed is None:
-                return
+        while True:
             try:
-                await self._commit_composed(composed, meta)
-            finally:
-                ip.safe_close(composed)
+                if self._timeline_dirty:
+                    async with self._navigation_lock:
+                        if self._timeline_dirty:
+                            await self._rebuild_current_frame()
+                    continue
 
-    async def _commit_composed(self, composed: Image.Image, meta: dict) -> None:
-        """Encode the composed slide into the framebuffer and broadcast.
-
-        Encodes off the loop so the JPEG encode (30-80 ms at 1080p, more
-        at 4K) doesn't block HA.
-        """
-        encoded = await self.hass.async_add_executor_job(
-            ip.encode_image, composed
-        )
-
-        self._framebuffer = encoded
-        self.store.last_frame = encoded
-        self._frame_id += 1
-        if meta:
-            self._last_is_portrait = meta.get("is_portrait")
-        else:
-            self._last_is_portrait = None
-        self._last_captured_at_pair = meta.get("captured_at_pair") if meta else None
-        self._last_pair_frames = meta.get("pair_frames") if meta else None
-        self._last_pair_orientation = meta.get("pair_orientation") if meta else None
-
-        self.async_write_ha_state()
+                interrupted = await self._wait_or_interrupt(
+                    float(int(self.store.slide_interval))
+                )
+                if not interrupted and not self.store.paused:
+                    async with self._navigation_lock:
+                        if not self._timeline_dirty:
+                            await self._show_next_frame()
+                self._consecutive_failures = 0
+            except asyncio.CancelledError:
+                raise
+            except Exception as err:
+                self._consecutive_failures += 1
+                _LOGGER.warning(
+                    "Album Slideshow: buffered navigation/render failed (attempt %d): %s",
+                    self._consecutive_failures,
+                    err,
+                )
 
     @property
     def _compose_semaphore(self) -> asyncio.Semaphore:
@@ -849,6 +1226,7 @@ class AlbumSlideshowCamera(Camera):
         return None
 
     async def _http_get(self, url: str) -> bytes | None:
+        """Fetch one remote image with validation and a hard timeout."""
         session = async_get_clientsession(self.hass)
         try:
             async with async_timeout.timeout(30):