music-assistant-server

1"""
2Async client for a local go-librespot daemon.
3
4go-librespot exposes a small HTTP+WebSocket API (enabled via its ``server`` config
5block). This client wraps the REST control endpoints (resume/pause/next/prev/seek/
6volume/play) and the ``/events`` WebSocket stream that pushes player state changes.
7See https://github.com/devgianlu/go-librespot/blob/master/API.md for the protocol.
8"""
9
10from __future__ import annotations
11
12import asyncio
13from collections.abc import Awaitable, Callable
14from http import HTTPStatus
15from typing import TYPE_CHECKING, Any
16
17from aiohttp import ClientError, ClientTimeout, WSMsgType
18
19if TYPE_CHECKING:
20    import logging
21
22    from music_assistant.mass import MusicAssistant
23
24# Called with (event_type, event_data) for every WebSocket event.
25EventCallback = Callable[[str, dict[str, Any]], Awaitable[None]]
26
27
28class GoLibrespotClient:
29    """Thin async wrapper around a go-librespot daemon's REST + WebSocket API."""
30
31    def __init__(self, mass: MusicAssistant, base_url: str, logger: logging.Logger) -> None:
32        """
33        Initialize the client.
34
35        :param mass: The MusicAssistant instance (for its shared HTTP session).
36        :param base_url: Base URL of the daemon's API server, e.g. ``http://127.0.0.1:3678``.
37        :param logger: Logger to use for diagnostics.
38        """
39        self.mass = mass
40        self.base_url = base_url.rstrip("/")
41        self.logger = logger
42
43    async def wait_until_ready(self, timeout: float = 30.0) -> bool:
44        """
45        Poll the daemon's root endpoint until the API server answers.
46
47        :param timeout: Maximum seconds to wait for the API to come up.
48        :return: True once the API responds, False if the timeout elapses.
49        """
50        deadline = self.mass.loop.time() + timeout
51        while self.mass.loop.time() < deadline:
52            try:
53                async with self.mass.http_session.get(
54                    f"{self.base_url}/", timeout=ClientTimeout(total=2)
55                ) as resp:
56                    if resp.status == HTTPStatus.OK:
57                        return True
58            except ClientError, TimeoutError, OSError:
59                pass
60            await asyncio.sleep(0.25)
61        return False
62
63    async def get_status(self) -> dict[str, Any] | None:
64        """
65        Fetch the current player status.
66
67        :return: The status payload, or None when there is no active session
68            (the daemon answers ``204 No Content`` in that case).
69        """
70        return await self._request("GET", "/status")
71
72    async def resume(self) -> None:
73        """Resume playback on the active session."""
74        await self._request("POST", "/player/resume")
75
76    async def pause(self) -> None:
77        """Pause playback on the active session."""
78        await self._request("POST", "/player/pause")
79
80    async def next(self) -> None:
81        """Skip to the next track."""
82        await self._request("POST", "/player/next")
83
84    async def prev(self) -> None:
85        """Skip to the previous track (or rewind the current one)."""
86        await self._request("POST", "/player/prev")
87
88    async def seek(self, position_ms: int) -> None:
89        """
90        Seek to an absolute position in the current track.
91
92        :param position_ms: Target position in milliseconds.
93        """
94        await self._request("POST", "/player/seek", {"position": max(0, position_ms)})
95
96    async def set_volume(self, volume: int) -> None:
97        """
98        Set the absolute playback volume.
99
100        :param volume: Volume on go-librespot's scale (0..volume_steps).
101        """
102        await self._request("POST", "/player/volume", {"volume": max(0, volume)})
103
104    async def play(self, uri: str, *, skip_to_uri: str | None = None, paused: bool = False) -> None:
105        """
106        Start playing a Spotify URI/context, making this device the active one.
107
108        go-librespot activates this device unconditionally for a play request, so
109        this doubles as a "take playback back to us" call when another device is
110        currently active.
111
112        :param uri: Spotify URI (track, album, playlist, ...) — typically a context.
113        :param skip_to_uri: Optional track URI within the context to start at.
114        :param paused: When True, load the content paused instead of playing.
115        """
116        body: dict[str, Any] = {"uri": uri, "paused": paused}
117        if skip_to_uri:
118            body["skip_to_uri"] = skip_to_uri
119        await self._request("POST", "/player/play", body)
120
121    async def listen_events(self, on_event: EventCallback) -> None:
122        """
123        Connect to the ``/events`` WebSocket and dispatch events until it closes.
124
125        Returns normally when the connection is closed by either side; raises on
126        connection errors so the caller can implement a reconnect loop.
127
128        :param on_event: Coroutine called with ``(event_type, event_data)`` per event.
129        """
130        ws_url = f"{self.base_url.replace('http', 'ws', 1)}/events"
131        async with self.mass.http_session.ws_connect(ws_url, heartbeat=30) as ws:
132            self.logger.debug("Connected to go-librespot events websocket")
133            async for msg in ws:
134                if msg.type == WSMsgType.TEXT:
135                    try:
136                        payload = msg.json()
137                    except ValueError:
138                        self.logger.debug("Ignoring non-JSON websocket message: %s", msg.data)
139                        continue
140                    if event_type := payload.get("type"):
141                        await on_event(event_type, payload.get("data") or {})
142                elif msg.type in (WSMsgType.CLOSE, WSMsgType.CLOSING, WSMsgType.CLOSED):
143                    break
144                elif msg.type == WSMsgType.ERROR:
145                    raise ws.exception() or ClientError("websocket error")
146
147    async def _request(
148        self, method: str, path: str, json_body: dict[str, Any] | None = None
149    ) -> dict[str, Any] | None:
150        """
151        Issue a request to the daemon's REST API.
152
153        :param method: HTTP method.
154        :param path: API path (e.g. ``/player/resume``).
155        :param json_body: Optional JSON request body.
156        :return: Parsed JSON response, or None when the daemon has no active session
157            (``204 No Content``) or returns no body.
158        :raises ClientError: When the daemon answers with an error status.
159        """
160        async with self.mass.http_session.request(
161            method, f"{self.base_url}{path}", json=json_body, timeout=ClientTimeout(total=10)
162        ) as resp:
163            # 204 = the daemon has no active Spotify session; surface as "no data"
164            # rather than an error so callers can treat it as a no-op.
165            if resp.status == HTTPStatus.NO_CONTENT:
166                self.logger.debug("go-librespot has no active session for %s %s", method, path)
167                return None
168            resp.raise_for_status()
169            if resp.content_type == "application/json":
170                data: dict[str, Any] = await resp.json()
171                return data
172            return None
173