/
/
/
1"""
2go-librespot backend for the Spotify Connect provider.
3
4go-librespot is driven entirely over its local HTTP+WebSocket API: the WebSocket
5``/events`` stream feeds player/session/metadata/volume state into Music Assistant,
6and transport + volume commands are issued via the REST endpoints. Playback control
7works without a configured Spotify *music* provider or the Spotify Web API.
8"""
9
10from __future__ import annotations
11
12import asyncio
13import json
14import os
15from contextlib import suppress
16from functools import partial
17from pathlib import Path
18from typing import TYPE_CHECKING, Any
19
20from music_assistant_models.enums import ContentType, StreamType
21from music_assistant_models.media_items import AudioFormat
22
23from music_assistant.helpers.process import AsyncProcess
24from music_assistant.helpers.util import (
25 interface_name_for_ip,
26 is_port_in_use,
27 select_free_port,
28)
29from music_assistant.providers.spotify_connect.base import (
30 AUDIO_QUALITY_LOSSLESS,
31 LOSSY_BIT_RATES,
32 MAX_LOSSY_BIT_RATE,
33 SpotifyConnectBackend,
34 spotify_source_audio_format,
35)
36from music_assistant.providers.spotify_connect.helpers import (
37 generate_device_id,
38 get_go_librespot_binary,
39)
40from music_assistant.providers.spotify_connect.models import (
41 BackendEvent,
42 BackendEventType,
43 BackendStreamSource,
44 BackendTrackMetadata,
45)
46
47from .client import GoLibrespotClient
48
49if TYPE_CHECKING:
50 import logging
51
52 from music_assistant.mass import MusicAssistant
53 from music_assistant.providers.spotify_connect.models import (
54 AudioChunkReader,
55 BackendEventCallback,
56 )
57
58# go-librespot volume scale; we pin volume_steps to this so the daemon's 0..max
59# volume maps 1:1 to a 0-100 percentage.
60VOLUME_STEPS = 100
61
62# Read size for pulling PCM off the daemon's stdout.
63STREAM_READ_CHUNK = 16384
64
65# Port range the go-librespot API server binds to (loopback only, one per instance).
66API_PORT_RANGE_START = 38800
67API_PORT_RANGE_END = 38900
68
69
70class GoLibrespotBackend(SpotifyConnectBackend):
71 """Spotify Connect backend wrapping a supervised go-librespot daemon."""
72
73 def __init__(
74 self,
75 mass: MusicAssistant,
76 *,
77 identity_key: str,
78 publish_name: str,
79 name: str,
80 logger: logging.Logger,
81 event_callback: BackendEventCallback,
82 crossfade_ms: int = 0,
83 loudness_normalization: bool = True,
84 audio_quality: str = AUDIO_QUALITY_LOSSLESS,
85 ) -> None:
86 """
87 Initialize the backend (cheap; the daemon is launched in ``start``).
88
89 :param mass: The MusicAssistant instance.
90 :param identity_key: Unique identity of this daemon (one per connected
91 player); keys the credential/cache dir and the stable Spotify
92 device id.
93 :param publish_name: Device name advertised to the Spotify app.
94 :param name: Display name of the owning provider instance (log messages).
95 :param logger: Logger to use for diagnostics.
96 :param event_callback: Awaited with a normalized BackendEvent for every
97 state change the daemon reports.
98 :param crossfade_ms: Crossfade duration between tracks in milliseconds
99 (0 disables crossfade).
100 :param loudness_normalization: Whether Spotify's loudness normalization
101 should be applied to the audio.
102 :param audio_quality: Ceiling for the streaming quality Spotify is asked
103 to deliver (one of the AUDIO_QUALITY_* tiers).
104 """
105 self.mass = mass
106 self.logger = logger
107 self.name = name
108 self._identity_key = identity_key
109 self._publish_name = publish_name
110 self._event_callback = event_callback
111 self._crossfade_ms = crossfade_ms
112 self._loudness_normalization = loudness_normalization
113 self._audio_quality = audio_quality
114 self.cache_dir = os.path.join(self.mass.cache_path, identity_key)
115 self._binary: str | None = None
116 self._api_port: int = 0
117 self._client: GoLibrespotClient | None = None
118 self._stop_called: bool = False
119 self._daemon_task: asyncio.Task[None] | None = None
120 self._events_task: asyncio.Task[None] | None = None
121 self._proc: AsyncProcess | None = None
122 self._restart_error_count = 0
123 # _audio_format is the original Spotify source codec (Ogg Vorbis at the
124 # configured tier's bitrate), advertised to clients for display.
125 # _decoded_audio_format is the raw PCM go-librespot actually writes to its
126 # stdout after decoding â what the audio reader yields and what the streams
127 # controller hands ffmpeg as the input format. We always emit the source's
128 # own format here; MA is responsible for converting it to whatever each
129 # player needs.
130 self._audio_format = spotify_source_audio_format(audio_quality, lossless=False)
131 self._decoded_audio_format = AudioFormat(
132 content_type=ContentType.PCM_S16LE,
133 codec_type=ContentType.PCM_S16LE,
134 sample_rate=44100,
135 bit_depth=16,
136 channels=2,
137 )
138
139 @property
140 def audio_format(self) -> AudioFormat:
141 """Return the source audio format (advertised to clients for display)."""
142 return self._audio_format
143
144 @property
145 def decoded_audio_format(self) -> AudioFormat:
146 """Return the decoded PCM format the audio reader actually delivers."""
147 return self._decoded_audio_format
148
149 async def start(self) -> None:
150 """Start the backend and its supervised go-librespot daemon."""
151 self._binary = get_go_librespot_binary()
152 self._api_port = await select_free_port(
153 API_PORT_RANGE_START, API_PORT_RANGE_END, host="127.0.0.1"
154 )
155 self._client = GoLibrespotClient(
156 self.mass, f"http://127.0.0.1:{self._api_port}", self.logger
157 )
158 # Two self-healing supervisors: one keeps the daemon process alive, the
159 # other keeps the events websocket connected (reconnecting across daemon
160 # restarts). The events runner resets the daemon's restart backoff once
161 # the websocket is healthy again.
162 self._daemon_task = self.mass.create_task(self._daemon_runner())
163 self._events_task = self.mass.create_task(self._events_runner())
164
165 async def stop(self) -> None:
166 """Stop the daemon and all supervisor tasks."""
167 self._stop_called = True
168 for task in (self._events_task, self._daemon_task):
169 if task and not task.done():
170 task.cancel()
171 with suppress(asyncio.CancelledError):
172 await task
173
174 async def get_stream_source(self) -> BackendStreamSource:
175 """Return the CUSTOM stream source, consumed through the audio reader."""
176 # CUSTOM: the core pulls PCM through get_audio_reader. `-fflags nobuffer`
177 # keeps ffmpeg's own input buffering low so the controller's realtime
178 # pacer owns the (small, bounded) read-ahead.
179 return BackendStreamSource(
180 stream_type=StreamType.CUSTOM,
181 extra_input_args=["-fflags", "nobuffer"],
182 )
183
184 def get_audio_reader(self) -> AudioChunkReader | None:
185 """
186 Return a PCM chunk reader bound to the currently running daemon.
187
188 The reader stays bound to this daemon process: once it exits, the
189 reader returns b"" (clean EOF) even if a restarted daemon is already
190 up. None is returned when no daemon is running.
191 """
192 if (proc := self._proc) is None:
193 return None
194 return partial(proc.read, STREAM_READ_CHUNK)
195
196 async def play(self, uri: str, *, skip_to_uri: str | None = None) -> None:
197 """
198 Start playing a Spotify URI/context, making this device the active one.
199
200 :param uri: Spotify URI (track, album, playlist, ...) â typically a context.
201 :param skip_to_uri: Optional track URI within the context to start at.
202 """
203 assert self._client is not None
204 await self._client.play(uri, skip_to_uri=skip_to_uri)
205
206 async def resume(self) -> None:
207 """Resume playback on the active session."""
208 assert self._client is not None
209 await self._client.resume()
210
211 async def pause(self) -> None:
212 """Pause playback on the active session."""
213 assert self._client is not None
214 await self._client.pause()
215
216 async def deactivate(self) -> None:
217 """Release this device as the active Spotify Connect device."""
218 assert self._client is not None
219 await self._client.stop()
220
221 async def next(self) -> None:
222 """Skip to the next track."""
223 assert self._client is not None
224 await self._client.next()
225
226 async def previous(self) -> None:
227 """Skip to the previous track (or rewind the current one)."""
228 assert self._client is not None
229 await self._client.prev()
230
231 async def seek(self, position_ms: int) -> None:
232 """
233 Seek to an absolute position in the current track.
234
235 :param position_ms: Target position in milliseconds.
236 """
237 assert self._client is not None
238 await self._client.seek(position_ms)
239
240 async def set_volume(self, volume: int) -> None:
241 """
242 Set the daemon's playback volume.
243
244 :param volume: Absolute volume as a 0-100 percentage (translated to
245 go-librespot's own volume scale).
246 """
247 assert self._client is not None
248 await self._client.set_volume(round(volume / 100 * VOLUME_STEPS))
249
250 def _write_config(self, source_ip: str | None) -> None:
251 """
252 Write the go-librespot ``config.yml`` for this instance.
253
254 go-librespot reads a YAML config; JSON is valid YAML, so we emit JSON to
255 sidestep an extra dependency and any string-quoting pitfalls (the device
256 name is user-provided). The config dir doubles as the credential/device
257 cache so the Spotify Connect device stays paired across restarts.
258
259 :param source_ip: Local address of the player-facing interface, or None to
260 advertise the Spotify Connect device on all interfaces.
261 """
262 Path(self.cache_dir).mkdir(parents=True, exist_ok=True)
263 config: dict[str, Any] = {
264 "device_name": self._publish_name,
265 "device_type": "speaker",
266 "device_id": generate_device_id(self._identity_key),
267 "bitrate": LOSSY_BIT_RATES.get(self._audio_quality, MAX_LOSSY_BIT_RATE),
268 "audio_backend": "pipe",
269 # write decoded PCM to the daemon's stdout, which we capture and
270 # forward (the process pipe is always attached, so the daemon's
271 # non-blocking pipe open never fails for lack of a reader). s16le is
272 # the Spotify source representation; MA converts it per player.
273 "audio_output_pipe": "/dev/stdout",
274 "audio_output_pipe_format": "s16le",
275 # external_volume: don't let go-librespot attenuate the PCM â MA / the
276 # target player owns the actual volume. We still receive 'volume'
277 # events and push volume back so the Spotify app slider stays in sync.
278 # No initial_volume: go-librespot ignores it with external_volume set;
279 # the provider pushes the player's volume instead.
280 "external_volume": True,
281 "volume_steps": VOLUME_STEPS,
282 # normalisation is applied by go-librespot itself (-14 LUFS target);
283 # crossfade_duration is in milliseconds, 0 disables it. The crossfade
284 # key needs go-librespot >= 0.8.0; older daemons ignore unknown keys.
285 "normalisation_disabled": not self._loudness_normalization,
286 "crossfade_duration": self._crossfade_ms,
287 "zeroconf_enabled": True,
288 "credentials": {"type": "zeroconf", "zeroconf": {"persist_credentials": True}},
289 "server": {"enabled": True, "address": "127.0.0.1", "port": self._api_port},
290 }
291 # Advertise the Spotify Connect device only on the interface the streams
292 # server binds to, so it lands on the right network on multi-homed hosts.
293 # go-librespot selects advertise interfaces by name, so map the IP to one.
294 if source_ip:
295 if iface_name := interface_name_for_ip(source_ip):
296 config["zeroconf_interfaces_to_advertise"] = [iface_name]
297 else:
298 self.logger.debug(
299 "No interface found for stream bind IP %s; advertising on all interfaces",
300 source_ip,
301 )
302 config_file = os.path.join(self.cache_dir, "config.yml")
303 with open(config_file, "w", encoding="utf-8") as fileobj:
304 json.dump(config, fileobj, indent=2)
305
306 async def _daemon_runner(self) -> None:
307 """Run and supervise the go-librespot daemon, restarting it if it exits."""
308 assert self._binary
309 assert self._client
310 # Loop forever; stop() cancels this task and the explicit stop-check below
311 # handles a graceful exit without a restart.
312 while True:
313 # If the API port was taken while the daemon was down, move to a
314 # fresh port instead of crash-looping on a bind error.
315 if await is_port_in_use(self._api_port, host="127.0.0.1"):
316 self._api_port = await select_free_port(
317 API_PORT_RANGE_START, API_PORT_RANGE_END, host="127.0.0.1"
318 )
319 self._client.base_url = f"http://127.0.0.1:{self._api_port}"
320 self.logger.warning(
321 "API port in use by another process; switching to port %s", self._api_port
322 )
323 self._write_config(await self.mass.streams.get_source_ip())
324 proc: AsyncProcess | None = None
325 try:
326 # stdout carries the decoded PCM (audio_output_pipe=/dev/stdout) and
327 # is consumed through get_audio_reader; stderr carries the daemon's
328 # logs, read here. Because the process pipe owns stdout from spawn
329 # there is always a reader, so go-librespot's non-blocking pipe open
330 # never fails for lack of a consumer.
331 self._proc = proc = AsyncProcess(
332 [self._binary, "--config_dir", self.cache_dir],
333 stdout=True,
334 stderr=True,
335 name=f"go-librespot[{self.name}]",
336 )
337 await proc.start()
338 self.logger.info("Started Spotify Connect background daemon [%s]", self.name)
339 async for line in proc.iter_stderr():
340 self.logger.debug("[%s] %s", self.name, line)
341 except asyncio.CancelledError:
342 raise
343 except Exception as err:
344 self.logger.warning("go-librespot daemon error [%s]: %s", self.name, err)
345 finally:
346 if proc:
347 await proc.close()
348 # The daemon â and thus the Spotify session â is gone. Tell the
349 # provider so a dead/restarting daemon isn't treated as active and
350 # controllable; a fresh 'active' event re-establishes it on reconnect.
351 self._proc = None
352 try:
353 await self._event_callback(BackendEvent(BackendEventType.CONNECTION_LOST))
354 except Exception:
355 # never let a callback error replace a propagating
356 # cancellation or kill the daemon supervisor
357 self.logger.exception("Error while handling daemon exit")
358 if self._stop_called:
359 break
360 self.logger.info("Spotify Connect background daemon stopped for %s", self.name)
361 self._restart_error_count += 1
362 if self._restart_error_count >= 5:
363 await self._event_callback(
364 BackendEvent(
365 BackendEventType.FATAL_ERROR,
366 error="go-librespot daemon failed to start multiple times.",
367 )
368 )
369 return
370 await asyncio.sleep(2)
371
372 async def _events_runner(self) -> None:
373 """Keep the go-librespot events websocket connected, reconnecting as needed."""
374 assert self._client is not None
375 while not self._stop_called:
376 try:
377 if not await self._client.wait_until_ready():
378 await asyncio.sleep(2)
379 continue
380 # A live websocket means the daemon is healthy: reset the restart
381 # backoff counter the daemon supervisor uses.
382 self._restart_error_count = 0
383 await self._client.listen_events(self._handle_event)
384 except asyncio.CancelledError:
385 raise
386 except Exception as err:
387 self.logger.debug("go-librespot events websocket dropped: %s", err)
388 if not self._stop_called:
389 await asyncio.sleep(2)
390
391 async def _handle_event(self, event_type: str, data: dict[str, Any]) -> None:
392 """Translate a single go-librespot websocket event and emit it normalized."""
393 self.logger.debug("Received event [%s]: %s %s", self.name, event_type, data)
394 await self._event_callback(self._translate_event(event_type, data))
395
396 def _translate_event(self, event_type: str, data: dict[str, Any]) -> BackendEvent:
397 """Map a raw go-librespot event onto the normalized BackendEvent model."""
398 # Every event carries the latest context/track uris when present, so the
399 # provider can take playback back when the user moves the active device
400 # away in the Spotify app.
401 context_uri: str | None = data.get("context_uri") or None
402 track_uri: str | None = data.get("uri") or None
403 if event_type == "active":
404 return BackendEvent(
405 BackendEventType.SESSION_ACTIVE, context_uri=context_uri, track_uri=track_uri
406 )
407 if event_type == "inactive":
408 return BackendEvent(
409 BackendEventType.SESSION_INACTIVE, context_uri=context_uri, track_uri=track_uri
410 )
411 if event_type == "playing":
412 return BackendEvent(
413 BackendEventType.PLAYING, context_uri=context_uri, track_uri=track_uri
414 )
415 if event_type == "paused":
416 return BackendEvent(
417 BackendEventType.PAUSED, context_uri=context_uri, track_uri=track_uri
418 )
419 if event_type == "stopped":
420 return BackendEvent(
421 BackendEventType.STOPPED, context_uri=context_uri, track_uri=track_uri
422 )
423 if event_type == "metadata":
424 artists = data.get("artist_names") or []
425 duration_ms = data.get("duration")
426 return BackendEvent(
427 BackendEventType.METADATA,
428 context_uri=context_uri,
429 track_uri=track_uri,
430 metadata=BackendTrackMetadata(
431 track_uri=data.get("uri"),
432 title=data.get("name") or None,
433 artist=artists[0] if artists else None,
434 album=data.get("album_name"),
435 image_url=data.get("album_cover_url"),
436 duration=duration_ms // 1000 if duration_ms else None,
437 position=int(data.get("position", 0)) // 1000,
438 ),
439 )
440 if event_type == "seek":
441 return BackendEvent(
442 BackendEventType.POSITION,
443 context_uri=context_uri,
444 track_uri=track_uri,
445 position=int(data.get("position", 0)) // 1000,
446 )
447 if event_type == "volume" and data.get("value") is not None:
448 # translate the daemon's 0..max scale to a 0-100 percentage
449 max_value = data.get("max") or VOLUME_STEPS
450 return BackendEvent(
451 BackendEventType.VOLUME,
452 context_uri=context_uri,
453 track_uri=track_uri,
454 volume=int(int(data["value"]) / int(max_value) * 100),
455 )
456 return BackendEvent(BackendEventType.OTHER, context_uri=context_uri, track_uri=track_uri)
457