music-assistant-server
15.6 KB•PY
metadata.py
15.6 KB • 375 lines • python
1"""Metadata reader for shairport-sync metadata pipe."""
2
3from __future__ import annotations
4
5import asyncio
6import base64
7import os
8import re
9import struct
10import time
11from contextlib import suppress
12from typing import TYPE_CHECKING, Any
13
14from music_assistant.constants import VERBOSE_LOG_LEVEL
15
16if TYPE_CHECKING:
17 from collections.abc import Callable
18 from logging import Logger
19
20
21class MetadataReader:
22 """Read and parse metadata from shairport-sync metadata pipe."""
23
24 def __init__(
25 self,
26 metadata_pipe: str,
27 logger: Logger,
28 on_metadata: Callable[[dict[str, Any]], None] | None = None,
29 ) -> None:
30 """Initialize metadata reader.
31
32 :param metadata_pipe: Path to the metadata pipe.
33 :param logger: Logger instance.
34 :param on_metadata: Callback function for metadata updates.
35 """
36 self.metadata_pipe = metadata_pipe
37 self.logger = logger
38 self.on_metadata = on_metadata
39 self._reader_task: asyncio.Task[None] | None = None
40 self._stop = False
41 self._current_metadata: dict[str, Any] = {}
42 self._fd: int | None = None
43 self._buffer = ""
44 self.cover_art_bytes: bytes | None = None
45
46 async def start(self) -> None:
47 """Start reading metadata from the pipe."""
48 self._stop = False
49 self._reader_task = asyncio.create_task(self._read_metadata())
50
51 async def stop(self) -> None:
52 """Stop reading metadata."""
53 self._stop = True
54 if self._reader_task and not self._reader_task.done():
55 self._reader_task.cancel()
56 with suppress(asyncio.CancelledError):
57 await self._reader_task
58
59 async def _read_metadata(self) -> None:
60 """Read metadata from the pipe using async file descriptor."""
61 loop = asyncio.get_event_loop()
62 try:
63 # Open the metadata pipe in non-blocking mode
64 # Use O_RDONLY | O_NONBLOCK to avoid blocking on open
65 self._fd = await loop.run_in_executor(
66 None, os.open, self.metadata_pipe, os.O_RDONLY | os.O_NONBLOCK
67 )
68
69 # Create an asyncio.Event to signal when data is available
70 data_available = asyncio.Event()
71
72 def on_readable() -> None:
73 """Set data available flag when file descriptor is readable."""
74 data_available.set()
75
76 # Register the file descriptor with the event loop
77 loop.add_reader(self._fd, on_readable)
78
79 try:
80 while not self._stop:
81 # Wait for data to be available
82 await data_available.wait()
83 data_available.clear()
84
85 # Read available data from the pipe
86 try:
87 chunk = os.read(self._fd, 4096)
88 if chunk:
89 # Decode as text and add to buffer
90 self._buffer += chunk.decode("utf-8", errors="ignore")
91 # Process all complete metadata items in the buffer
92 self._process_buffer()
93 except BlockingIOError:
94 # No data available right now, wait for next notification
95 continue
96 except OSError as err:
97 self.logger.debug("Error reading from pipe: %s", err)
98 await asyncio.sleep(0.1)
99
100 finally:
101 # Remove the reader callback
102 loop.remove_reader(self._fd)
103
104 except Exception as err:
105 self.logger.error("Error reading metadata pipe: %s", err)
106 finally:
107 if self._fd is not None:
108 with suppress(OSError):
109 os.close(self._fd)
110 self._fd = None
111
112 def _process_buffer(self) -> None:
113 """Process all complete metadata items in the buffer (XML format or plain text markers)."""
114 # First, check for plain text markers from sessioncontrol hooks
115 while "\n" in self._buffer:
116 # Check if we have a complete line before any XML
117 line_end = self._buffer.index("\n")
118 if "<item>" not in self._buffer or self._buffer.index("<item>") > line_end:
119 # We have a plain text line before any XML
120 line = self._buffer[:line_end].strip()
121 self._buffer = self._buffer[line_end + 1 :]
122
123 # Handle our custom markers
124 if line == "MA_PLAY_BEGIN":
125 self.logger.info("Playback started (via sessioncontrol hook)")
126 if self.on_metadata:
127 self.on_metadata({"play_state": "playing"})
128 elif line == "MA_PLAY_END":
129 self.logger.info("Playback ended (via sessioncontrol hook)")
130 if self.on_metadata:
131 self.on_metadata({"play_state": "stopped"})
132 # Ignore other plain text lines
133 else:
134 # XML item comes first, stop looking for lines
135 break
136
137 # Look for complete <item>...</item> blocks
138 while "<item>" in self._buffer and "</item>" in self._buffer:
139 try:
140 # Find the boundaries of the next item
141 start_idx = self._buffer.index("<item>")
142 end_idx = self._buffer.index("</item>") + len("</item>")
143
144 # Extract the item
145 item_xml = self._buffer[start_idx:end_idx]
146
147 # Remove processed item from buffer
148 self._buffer = self._buffer[end_idx:]
149
150 # Parse the item
151 self._parse_xml_item(item_xml)
152
153 except (ValueError, IndexError) as err:
154 self.logger.debug("Error processing buffer: %s", err)
155 # Clear malformed data
156 if "</item>" in self._buffer:
157 # Skip to after the next </item>
158 self._buffer = self._buffer[self._buffer.index("</item>") + len("</item>") :]
159 else:
160 # Wait for more data
161 break
162 except Exception as err:
163 self.logger.error("Unexpected error processing buffer: %s", err)
164 # Clear the buffer on unexpected error
165 self._buffer = ""
166 break
167
168 def _parse_xml_item(self, item_xml: str) -> None:
169 """Parse a single XML metadata item.
170
171 :param item_xml: XML string containing a metadata item.
172 """
173 try:
174 # Extract type (hex format)
175 type_match = re.search(r"<type>([0-9a-fA-F]{8})</type>", item_xml)
176 code_match = re.search(r"<code>([0-9a-fA-F]{8})</code>", item_xml)
177 length_match = re.search(r"<length>(\d+)</length>", item_xml)
178
179 if not type_match or not code_match or not length_match:
180 return
181
182 # Convert hex type and code to ASCII strings
183 type_hex = int(type_match.group(1), 16)
184 code_hex = int(code_match.group(1), 16)
185 length = int(length_match.group(1))
186
187 # Convert hex to 4-character ASCII codes
188 type_str = type_hex.to_bytes(4, "big").decode("ascii", errors="ignore")
189 code_str = code_hex.to_bytes(4, "big").decode("ascii", errors="ignore")
190
191 # Extract data if present
192 data: str | bytes | None = None
193 if length > 0:
194 data_match = re.search(r"<data encoding=\"base64\">([^<]+)</data>", item_xml)
195 if data_match:
196 try:
197 # Decode base64 data
198 data_b64 = data_match.group(1).strip()
199 decoded_data = base64.b64decode(data_b64)
200
201 # For binary fields (PICT, astm), keep as raw bytes
202 # For text fields, decode to UTF-8
203 if code_str in ("PICT", "astm"):
204 # Cover art and duration: keep as raw bytes
205 data = decoded_data
206 else:
207 # Text metadata: decode to UTF-8
208 data = decoded_data.decode("utf-8", errors="ignore")
209 except Exception as err:
210 self.logger.debug("Error decoding base64 data: %s", err)
211
212 # Process the metadata item
213 asyncio.create_task(self._process_metadata_item(type_str, code_str, data))
214
215 except Exception as err:
216 self.logger.debug("Error parsing XML item: %s", err)
217
218 async def _process_metadata_item(
219 self, item_type: str, code: str, data: str | bytes | None
220 ) -> None:
221 """Process a metadata item and update current metadata.
222
223 :param item_type: Type of metadata (e.g., 'core' or 'ssnc').
224 :param code: Metadata code identifier.
225 :param data: Optional metadata data (string, bytes, or None).
226 """
227 # Don't log binary data (like cover art)
228 if code == "PICT":
229 self.logger.log(
230 VERBOSE_LOG_LEVEL,
231 "Metadata: type=%s, code=%s, data=<binary image data>",
232 item_type,
233 code,
234 )
235 else:
236 self.logger.log(
237 VERBOSE_LOG_LEVEL, "Metadata: type=%s, code=%s, data=%s", item_type, code, data
238 )
239
240 # Handle metadata start/end markers
241 if item_type == "ssnc" and code == "mdst":
242 self._current_metadata = {}
243 # Note: We don't clear cover_art_bytes here because:
244 # 1. Cover art may arrive before mdst (at playback start)
245 # 2. New cover art will overwrite old bytes when it arrives
246 # 3. Cache-busting timestamp ensures browser gets correct image
247 if self.on_metadata:
248 self.on_metadata({"metadata_start": True})
249 return
250
251 if item_type == "ssnc" and code == "mden":
252 if self.on_metadata and self._current_metadata:
253 self.on_metadata(dict(self._current_metadata))
254 return
255
256 # Parse core metadata (from iTunes/iOS)
257 if item_type == "core" and data is not None:
258 self._parse_core_metadata(code, data)
259
260 # Parse shairport-sync metadata
261 if item_type == "ssnc" and data is not None:
262 self._parse_ssnc_metadata(code, data)
263
264 def _parse_core_metadata(self, code: str, data: str | bytes) -> None:
265 """Parse core metadata from iTunes/iOS.
266
267 :param code: Metadata code identifier.
268 :param data: Metadata data.
269 """
270 # Text metadata fields - expect string data
271 if isinstance(data, str):
272 if code == "asar": # Artist
273 self._current_metadata["artist"] = data
274 elif code == "asal": # Album
275 self._current_metadata["album"] = data
276 elif code == "minm": # Title
277 self._current_metadata["title"] = data
278
279 # Binary metadata fields - expect bytes data
280 elif isinstance(data, bytes):
281 if code == "PICT": # Cover art (raw bytes)
282 # Store raw bytes for later retrieval via resolve_image
283 self.cover_art_bytes = data
284 self.logger.debug("Stored cover art: %d bytes", len(data))
285 # Signal that cover art is available with timestamp for cache-busting
286 timestamp = str(int(time.time() * 1000))
287 self._current_metadata["cover_art_timestamp"] = timestamp
288 # Send cover art update immediately (cover art often arrives in separate block)
289 if self.on_metadata:
290 self.on_metadata({"cover_art_timestamp": timestamp})
291 elif code == "astm": # Track duration in milliseconds (stored as 32-bit big-endian int)
292 try:
293 # Duration is sent as 4-byte big-endian integer
294 if len(data) >= 4:
295 duration_ms = struct.unpack(">I", data[:4])[0]
296 self._current_metadata["duration"] = duration_ms // 1000
297 except (ValueError, TypeError, struct.error) as err:
298 self.logger.debug("Error parsing duration: %s", err)
299
300 def _parse_ssnc_metadata(self, code: str, data: str | bytes) -> None:
301 """Parse shairport-sync metadata.
302
303 :param code: Metadata code identifier.
304 :param data: Metadata data.
305 """
306 # Handle binary data (cover art can come as ssnc type)
307 if isinstance(data, bytes):
308 if code == "PICT": # Cover art (raw bytes)
309 # Store raw bytes for later retrieval via resolve_image
310 self.cover_art_bytes = data
311 self.logger.debug("Stored cover art: %d bytes", len(data))
312 # Signal that cover art is available with timestamp for cache-busting
313 timestamp = str(int(time.time() * 1000))
314 self._current_metadata["cover_art_timestamp"] = timestamp
315 # Send cover art update immediately (cover art often arrives in separate block)
316 if self.on_metadata:
317 self.on_metadata({"cover_art_timestamp": timestamp})
318 return
319
320 # Process string data for ssnc metadata (volume/progress are text-based)
321 if code == "pvol": # Volume
322 self._parse_volume(data)
323 # Send volume updates immediately (not batched with mden)
324 if self.on_metadata and "volume" in self._current_metadata:
325 self.on_metadata({"volume": self._current_metadata["volume"]})
326 elif code == "prgr": # Progress
327 self._parse_progress(data)
328 # Send progress updates immediately (not batched with mden)
329 if self.on_metadata and "elapsed_time" in self._current_metadata:
330 self.on_metadata({"elapsed_time": self._current_metadata["elapsed_time"]})
331 elif code == "paus": # Paused
332 self._current_metadata["paused"] = True
333 elif code == "prsm": # Playing/resumed
334 self._current_metadata["paused"] = False
335
336 def _parse_volume(self, data: str) -> None:
337 """Parse volume metadata from shairport-sync.
338
339 Format: airplay_volume,min_volume,max_volume,mute
340 AirPlay volume is in dB, typically ranging from -30.0 (silent) to 0.0 (max).
341 Special value -144.0 means muted.
342
343 :param data: Volume data string (e.g., "-21.88,0.00,0.00,0.00").
344 """
345 try:
346 parts = data.split(",")
347 if len(parts) >= 1:
348 airplay_volume = float(parts[0])
349 # -144.0 means muted
350 if airplay_volume <= -144.0:
351 volume_percent = 0
352 else:
353 # Convert dB to percentage: -30dB = 0%, 0dB = 100%
354 volume_percent = int(((airplay_volume + 30.0) / 30.0) * 100)
355 volume_percent = max(0, min(100, volume_percent))
356 self._current_metadata["volume"] = volume_percent
357 except (ValueError, IndexError) as err:
358 self.logger.debug("Error parsing volume: %s", err)
359
360 def _parse_progress(self, data: str) -> None:
361 """Parse progress metadata.
362
363 :param data: Progress data string.
364 """
365 try:
366 parts = data.split("/")
367 if len(parts) >= 3:
368 start_rtp = int(parts[0])
369 current_rtp = int(parts[1])
370 elapsed_frames = current_rtp - start_rtp
371 elapsed_seconds = elapsed_frames / 44100
372 self._current_metadata["elapsed_time"] = int(elapsed_seconds)
373 except (ValueError, IndexError) as err:
374 self.logger.debug("Error parsing progress: %s", err)
375