/
/
/
1"""Streaming functionality using librespot for Spotify provider."""
2
3from __future__ import annotations
4
5import asyncio
6from collections import deque
7from collections.abc import AsyncGenerator
8from typing import TYPE_CHECKING
9
10from music_assistant_models.enums import MediaType
11from music_assistant_models.errors import AudioError
12
13from music_assistant.constants import VERBOSE_LOG_LEVEL
14from music_assistant.helpers.process import AsyncProcess
15
16if TYPE_CHECKING:
17 from music_assistant_models.streamdetails import StreamDetails
18
19 from .provider import SpotifyProvider
20
21
22class LibrespotStreamer:
23 """Handles streaming functionality using librespot."""
24
25 def __init__(self, provider: SpotifyProvider) -> None:
26 """Initialize the LibrespotStreamer."""
27 self.provider = provider
28
29 async def get_audio_stream(
30 self, streamdetails: StreamDetails, seek_position: int = 0
31 ) -> AsyncGenerator[bytes]:
32 """Return the audio stream for the provider item."""
33 # Regular track/episode streaming - audiobooks are handled in the provider
34 media_type = "episode" if streamdetails.media_type == MediaType.PODCAST_EPISODE else "track"
35 spotify_uri = f"spotify://{media_type}:{streamdetails.item_id}"
36 async for chunk in self.stream_spotify_uri(spotify_uri, seek_position):
37 yield chunk
38
39 async def stream_spotify_uri(
40 self, spotify_uri: str, seek_position: int = 0
41 ) -> AsyncGenerator[bytes]:
42 """Return the audio stream for the Spotify URI."""
43 self.provider.logger.log(
44 VERBOSE_LOG_LEVEL, f"Start streaming {spotify_uri} using librespot"
45 )
46 # Validate that librespot binary is available
47 if not self.provider._librespot_bin:
48 raise AudioError("Librespot binary not available")
49
50 args = [
51 self.provider._librespot_bin,
52 "--cache",
53 self.provider.cache_dir,
54 "--disable-audio-cache",
55 "--passthrough",
56 "--bitrate",
57 "320",
58 "--backend",
59 "pipe",
60 "--single-track",
61 spotify_uri,
62 "--disable-discovery",
63 "--dither",
64 "none",
65 ]
66 if seek_position:
67 args += ["--start-position", str(int(seek_position))]
68
69 async with AsyncProcess(
70 args,
71 stdout=True,
72 stderr=True,
73 name="librespot",
74 ) as librespot_proc:
75 log_history: deque[str] = deque(maxlen=10)
76 logger = self.provider.logger
77
78 async def log_librespot_output() -> None:
79 """Log librespot output if verbose logging is enabled."""
80 async for line in librespot_proc.iter_stderr():
81 log_history.append(line)
82 if "ERROR" in line or "WARNING" in line:
83 logger.warning("[librespot] %s", line)
84 if "unable to" in line.lower() or "skipping" in line.lower():
85 # if librespot reports a fatal error (e.g. unable to load
86 # or read audio), terminate the process to avoid hanging
87 # indefinitely as it won't produce any audio output.
88 # NOTE: we terminate the underlying process directly instead
89 # of calling close() because this task IS the stderr reader
90 # and close() would try to await itself.
91 if librespot_proc.proc and librespot_proc.proc.returncode is None:
92 librespot_proc.proc.terminate()
93 return
94 else:
95 logger.log(VERBOSE_LOG_LEVEL, "[librespot] %s", line)
96
97 librespot_proc.attach_stderr_reader(asyncio.create_task(log_librespot_output()))
98 # yield from librespot's stdout
99 async for chunk in librespot_proc.iter_chunked():
100 yield chunk
101
102 if librespot_proc.returncode != 0:
103 raise AudioError(
104 f"Librespot exited with code {librespot_proc.returncode} for {spotify_uri}"
105 )
106