e-ink-dash/server/media.py

"""Windows Media playback tracking via GSMTC (Global System Media Transport Controls).

Uses the winsdk library to access the Windows Media API without polling.
"""

from __future__ import annotations

import asyncio
import base64
import io
import logging
from dataclasses import dataclass
from typing import Any

logger = logging.getLogger("paperdash.media")


# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------


@dataclass
class MediaState:
    """Current media playback state."""

    title: str = ""
    artist: str = ""
    album: str = ""
    artwork_base64: str = ""
    playback_position: float = 0.0  # seconds
    playback_duration: float = 0.0  # seconds
    playback_percentage: float = 0.0  # 0.0 - 1.0
    is_playing: bool = False
    is_paused: bool = False
    is_stopped: bool = True


# ---------------------------------------------------------------------------
# Media Manager
# ---------------------------------------------------------------------------


class MediaManager:
    """Tracks Windows media playback via GSMTC."""

    def __init__(self) -> None:
        self._state = MediaState()
        self._poll_interval: float = 2.0

    async def run(self) -> None:
        """Periodically poll media state from GSMTC."""
        while True:
            try:
                self._state = await self._poll_gsmtc()
            except Exception as e:
                logger.error(f"Media poll error: {e}")
            await asyncio.sleep(self._poll_interval)

    async def _poll_gsmtc(self) -> MediaState:
        """Read current media state from Windows GSMTC."""
        try:
            from winsdk.windows.media.control import (  # type: ignore
                GlobalSystemMediaTransportControlsSessionManager as Manager,
            )

            session_manager = await Manager.request_async()
            session = session_manager.get_current_session()

            if session is None:
                return MediaState()

            info = await session.try_get_media_properties_async()

            # Convert artwork to base64
            artwork_b64 = ""
            if info.thumbnail:
                try:
                    stream = await info.thumbnail.open_async()
                    reader = await asyncio.to_thread(
                        __import__("Windows.Storage.Streams").DataTypes.DataReader,
                        stream,
                    )
                    buffer = await asyncio.to_thread(reader.load_async, stream.size)
                    data = bytearray(buffer)
                    buf = io.BytesIO(data)
                    artwork_b64 = base64.b64encode(buf.read()).decode("ascii")
                except Exception:
                    artwork_b64 = ""

            # Compute percentage
            props = session.get_playback_info()
            pos = props.position.total_seconds() if props.position else 0.0
            duration = info.duration.total_seconds() if info.duration else 0.0
            percentage = (pos / duration * 100) if duration > 0 else 0.0

            return MediaState(
                title=info.title or "",
                artist=info.artist or "",
                album=info.album or "",
                artwork_base64=artwork_b64,
                playback_position=pos,
                playback_duration=duration,
                playback_percentage=percentage,
                is_playing=session.playback_status.value == 0,  # Playing
                is_paused=session.playback_status.value == 1,  # Paused
                is_stopped=session.playback_status.value == 2,  # Stopped
            )
        except ImportError:
            logger.warning("winsdk not installed — media tracking unavailable")
            return MediaState()
        except Exception as e:
            logger.error(f"GSMTC error: {e}")
            return MediaState()

    def get_state(self) -> dict[str, Any]:
        """Return current media state as a serializable dict."""
        return {
            "title": self._state.title,
            "artist": self._state.artist,
            "album": self._state.album,
            "artwork_base64": self._state.artwork_base64,
            "playback_position": self._state.playback_position,
            "playback_duration": self._state.playback_duration,
            "playback_percentage": self._state.playback_percentage,
            "is_playing": self._state.is_playing,
            "is_paused": self._state.is_paused,
            "is_stopped": self._state.is_stopped,
        }

    async def play(self) -> None:
        """Send play command to current media session."""
        await self._send_command("play")

    async def pause(self) -> None:
        """Send pause command to current media session."""
        await self._send_command("pause")

    async def skip(self) -> None:
        """Send next track command."""
        await self._send_command("next")

    async def previous(self) -> None:
        """Send previous track command."""
        await self._send_command("previous")

    async def _send_command(self, command: str) -> None:
        """Send a media transport command via GSMTC."""
        try:
            from winsdk.windows.media.control import (  # type: ignore
                GlobalSystemMediaTransportControlsSessionManager as Manager,
            )

            session_manager = await Manager.request_async()
            session = session_manager.get_current_session()
            if session:
                if command == "play":
                    await session.try_play_async()
                elif command == "pause":
                    await session.try_pause_async()
                elif command == "next":
                    await session.try_skip_next_async()
                elif command == "previous":
                    await session.try_skip_previous_async()
        except Exception as e:
            logger.error(f"Media command '{command}' failed: {e}")