music-assistant-server

1"""Various helpers/utilities for the AirPlay provider."""
2
3from __future__ import annotations
4
5import logging
6import os
7import platform
8import plistlib
9import re
10from fnmatch import fnmatchcase
11from typing import TYPE_CHECKING, Any
12
13from aiohttp import ClientError, ClientTimeout
14from music_assistant_models.enums import ContentType
15from music_assistant_models.media_items import AudioFormat
16
17from music_assistant.helpers.process import check_output
18from music_assistant.helpers.util import format_ip_for_url
19
20from .constants import AIRPLAY_BUFFER_DEPTH_DEFAULTS
21
22if TYPE_CHECKING:
23    from zeroconf.asyncio import AsyncServiceInfo
24
25    from music_assistant.mass import MusicAssistant
26
27_LOGGER = logging.getLogger(__name__)
28_COMPANION_PAIRING_DISABLED = 0x04
29_COMPANION_PAIRING_WITH_PIN = 0x4000
30# Bound the binary's `--check` probe. It normally answers instantly, but the first
31# execution of a freshly-fetched binary can stall (e.g. macOS Gatekeeper verification
32# of an unsigned download), and a wedged binary would otherwise block provider load or
33# a stream start indefinitely.
34_CLI_BINARY_CHECK_TIMEOUT = 15.0
35# Bound the /info capability probe: it runs in the discovery path, and a receiver
36# that is slow to answer must not hold up player registration.
37_INFO_PROBE_TIMEOUT = 5.0
38
39
40def convert_airplay_volume(value: float) -> int:
41    """Remap AirPlay dB volume (-30..0) to 0..100 scale."""
42    airplay_min = -30.0
43    airplay_max = 0.0
44    value = max(airplay_min, min(airplay_max, value))
45    portion = (value - airplay_min) * 100.0 / (airplay_max - airplay_min)
46    return max(0, min(100, round(portion)))
47
48
49def get_model_info(info: AsyncServiceInfo) -> tuple[str, str]:  # noqa: PLR0911
50    """Return Manufacturer and Model name from mdns info."""
51    manufacturer = info.decoded_properties.get("manufacturer")
52    model = info.decoded_properties.get("model")
53    if manufacturer and model:
54        return (manufacturer, model)
55    # try parse from am property
56    if am_property := info.decoded_properties.get("am"):
57        model = am_property
58
59    if not model:
60        model = "Unknown"
61
62    # parse apple model names
63    if model == "AudioAccessory6,1":
64        return ("Apple", "HomePod 2")
65    if model in ("AudioAccessory5,1", "AudioAccessorySingle5,1"):
66        return ("Apple", "HomePod Mini")
67    if model == "AppleTV1,1":
68        return ("Apple", "Apple TV Gen1")
69    if model == "AppleTV2,1":
70        return ("Apple", "Apple TV Gen2")
71    if model in ("AppleTV3,1", "AppleTV3,2"):
72        return ("Apple", "Apple TV Gen3")
73    if model == "AppleTV5,3":
74        return ("Apple", "Apple TV Gen4")
75    if model == "AppleTV6,2":
76        return ("Apple", "Apple TV 4K")
77    if model == "AppleTV11,1":
78        return ("Apple", "Apple TV 4K Gen2")
79    if model == "AppleTV14,1":
80        return ("Apple", "Apple TV 4K Gen3")
81    if model == "UPL-AMP":
82        return ("Ubiquiti Inc.", "UPL-AMP")
83    if "AirPort" in model:
84        return ("Apple", "AirPort Express")
85    if "AudioAccessory" in model:
86        return ("Apple", "HomePod")
87    if "AppleTV" in model:
88        model = "Apple TV"
89        manufacturer = "Apple"
90    # Detect Mac devices (Mac mini, MacBook, iMac, etc.)
91    # Model identifiers like: Mac16,11, MacBookPro18,3, iMac21,1
92    if model.startswith(("Mac", "iMac")):
93        # Parse Mac model to friendly name
94        if model.startswith("MacBookPro"):
95            return ("Apple", f"MacBook Pro ({model})")
96        if model.startswith("MacBookAir"):
97            return ("Apple", f"MacBook Air ({model})")
98        if model.startswith("MacBook"):
99            return ("Apple", f"MacBook ({model})")
100        if model.startswith("iMac"):
101            return ("Apple", f"iMac ({model})")
102        if model.startswith("Macmini"):
103            return ("Apple", f"Mac mini ({model})")
104        if model.startswith("MacPro"):
105            return ("Apple", f"Mac Pro ({model})")
106        if model.startswith("MacStudio"):
107            return ("Apple", f"Mac Studio ({model})")
108        # Generic Mac device (e.g. Mac16,11 for Mac mini M4)
109        return ("Apple", f"Mac ({model})")
110
111    return (manufacturer or "AirPlay", model)
112
113
114def parse_airplay_features(features_value: str | None) -> int:
115    """Return an AirPlay features bitmask, or zero for an invalid value."""
116    if not features_value:
117        return 0
118    try:
119        parts = features_value.split(",")
120        features = int(parts[0], 16)
121        if len(parts) > 1:
122            features |= int(parts[1], 16) << 32
123    except TypeError, ValueError:
124        return 0
125    return features
126
127
128def supports_airplay2(features_value: str | None) -> bool:
129    """
130    Check if a device advertises AirPlay 2 support in its features bitmask.
131
132    :param features_value: Raw features value from the mDNS TXT records
133        (``features`` on the _airplay service or ``ft`` on the _raop service),
134        formatted as ``0xLOW`` or ``0xLOW,0xHIGH``.
135    """
136    features = parse_airplay_features(features_value)
137    # SupportsUnifiedMediaControl (bit 38) / SupportsCoreUtilsPairingAndEncryption
138    # (bit 48): either one means the device speaks AirPlay 2. This mirrors the
139    # test the cliairplay binary uses for its automatic route selection.
140    return bool((features >> 38) & 1 or (features >> 48) & 1)
141
142
143def is_apple_device(manufacturer: str, model: str) -> bool:
144    """
145    Check if a device is a (standalone) Apple device with native AirPlay support.
146
147    Apple devices (HomePod, Apple TV) have native AirPlay support
148    and should be exposed as PlayerType.PLAYER.
149    We don't include MacBooks etc. here as they are not standalone devices
150    and may also be used for other protocols.
151    """
152    return manufacturer.lower().startswith("apple") and (
153        "homepod" in model.lower() or "apple tv" in model.lower()
154    )
155
156
157def is_macos_device(manufacturer: str, model: str) -> bool:
158    """Return whether an AirPlay device identifies as a Mac."""
159    return manufacturer.lower().startswith("apple") and model.lower().startswith(("mac", "imac"))
160
161
162def is_apple_tv(manufacturer: str, model: str) -> bool:
163    """
164    Check if a device identifies as an Apple TV (and not a HomePod).
165
166    Only Apple TVs run the tvOS dashboard app, so this narrows :func:`is_apple_device`
167    to the Apple TV family. The model strings come from :func:`get_model_info`
168    (e.g. "Apple TV 4K", "Apple TV Gen4").
169    """
170    return manufacturer.lower().startswith("apple") and "apple tv" in model.lower()
171
172
173def default_buffer_depth(manufacturer: str, model: str, fv: str | None) -> int:
174    """
175    Return the default receiver buffer depth in ms for a device, 0 for automatic.
176
177    :param manufacturer: Device manufacturer from discovery.
178    :param model: Device model from discovery.
179    :param fv: The device's _airplay fv (firmware) TXT record, when known.
180    """
181    for manufacturer_match, model_match, fv_match, depth_ms in AIRPLAY_BUFFER_DEPTH_DEFAULTS:
182        # fnmatchcase with both sides lowered: plain fnmatch only normalizes
183        # case on case-insensitive platforms, so a capitalized table row would
184        # match on macOS and silently fail on Linux.
185        if (
186            fnmatchcase(manufacturer.lower(), manufacturer_match.lower())
187            and fnmatchcase(model.lower(), model_match.lower())
188            and fnmatchcase((fv or "").lower(), fv_match.lower())
189        ):
190            return depth_ms
191    return 0
192
193
194def get_decoded_property(discovery_info: AsyncServiceInfo, key: str) -> str | None:
195    """
196    Return an mDNS TXT property value by case-insensitive key.
197
198    TXT record keys are case-insensitive (RFC 6763) and zeroconf preserves the
199    casing as advertised on the wire, which differs per device (e.g. Companion
200    services advertise ``rpFl``, MRP services ``SystemBuildVersion``).
201
202    :param discovery_info: The mDNS service info to read the property from.
203    :param key: The TXT record key to look up (any casing).
204    """
205    decoded_properties = discovery_info.decoded_properties
206    if (value := decoded_properties.get(key)) is not None:
207        return value
208    folded_key = key.casefold()
209    for prop_key, prop_value in decoded_properties.items():
210        if prop_key.casefold() == folded_key:
211            return prop_value
212    return None
213
214
215def supports_companion_pairing(discovery_info: AsyncServiceInfo | None) -> bool:
216    """Return whether a Companion service supports PIN pairing."""
217    if discovery_info is None:
218        return False
219    raw_flags = get_decoded_property(discovery_info, "rpFl")
220    if raw_flags is None:
221        return False
222    try:
223        flags = int(raw_flags, 16)
224    except TypeError, ValueError:
225        return False
226    return bool(flags & _COMPANION_PAIRING_WITH_PIN) and not bool(
227        flags & _COMPANION_PAIRING_DISABLED
228    )
229
230
231def supports_mrp_tunnel(discovery_info: AsyncServiceInfo | None) -> bool:
232    """Return whether an AirPlay service advertises tunneled MRP control."""
233    if discovery_info is None:
234        return False
235    features = parse_airplay_features(
236        discovery_info.decoded_properties.get("features")
237        or discovery_info.decoded_properties.get("ft")
238    )
239    return bool((features >> 58) & 1)
240
241
242def supports_transient_mrp(discovery_info: AsyncServiceInfo | None) -> bool:
243    """Return whether an AirPlay MRP tunnel supports transient authentication."""
244    if not supports_mrp_tunnel(discovery_info):
245        return False
246    assert discovery_info is not None
247    features = parse_airplay_features(
248        discovery_info.decoded_properties.get("features")
249        or discovery_info.decoded_properties.get("ft")
250    )
251    return bool((features >> 43) & 1 or (features >> 48) & 1)
252
253
254def supports_mrp_service(discovery_info: AsyncServiceInfo | None) -> bool:
255    """Return whether a native MRP service is usable."""
256    if discovery_info is None or discovery_info.port is None:
257        return False
258    build = get_decoded_property(discovery_info, "SystemBuildVersion") or ""
259    match = re.match(r"^(\d+)[A-Z]", build)
260    return match is None or int(match.group(1)) < 19
261
262
263async def probe_audio_formats(mass: MusicAssistant, host: str, port: int) -> int:
264    """
265    Return the audio formats an AirPlay 2 receiver advertises, as a bitmask.
266
267    Zero when the device is unreachable or publishes no format tables.
268
269    :param mass: The MusicAssistant instance.
270    :param host: Address of the receiver.
271    :param port: Port of the receiver's _airplay._tcp service.
272    """
273    # The tables live in the receiver's /info response, which is served
274    # unauthenticated, so this needs no pairing or credentials.
275    url = f"http://{format_ip_for_url(host)}:{port}/info"
276    try:
277        async with mass.http_session.get(
278            url, timeout=ClientTimeout(total=_INFO_PROBE_TIMEOUT)
279        ) as resp:
280            if resp.status != 200:
281                return 0
282            info = plistlib.loads(await resp.read())
283    except ClientError, TimeoutError, plistlib.InvalidFileException, ValueError:
284        return 0
285    return _parse_format_tables(info) if isinstance(info, dict) else 0
286
287
288async def get_cli_binary() -> str:
289    """
290    Find the cliairplay binary for the current platform.
291
292    :raises RuntimeError: If the binary cannot be found.
293    """
294    system = platform.system()
295    architecture = platform.machine()
296    binary_name = _get_cli_binary_name(system, architecture)
297    if binary_name is None:
298        msg = f"Unsupported cliairplay platform: {system.lower()}/{architecture.lower()}"
299        raise RuntimeError(msg)
300    base_path = os.path.join(os.path.dirname(__file__), "bin")
301    binary_path = os.path.join(base_path, binary_name)
302
303    try:
304        returncode, output = await check_output(
305            binary_path, "--check", timeout=_CLI_BINARY_CHECK_TIMEOUT
306        )
307        output_str = output.strip().decode()
308        if returncode == 0 and "cliairplay" in output_str and "check" in output_str:
309            return binary_path
310    except TimeoutError:
311        msg = (
312            f"{binary_name} did not respond to --check within "
313            f"{_CLI_BINARY_CHECK_TIMEOUT:.0f}s (first-run verification or a wedged binary)"
314        )
315        raise RuntimeError(msg) from None
316    except OSError:
317        pass
318
319    msg = f"Unable to locate {binary_name} for {system.lower()}/{architecture.lower()}"
320    raise RuntimeError(msg)
321
322
323def player_id_to_mac_address(player_id: str) -> str:
324    """Convert a player_id to a MAC address-like string."""
325    # the player_id is the mac address prefixed with "ap"
326    hex_str = player_id.replace("ap", "").upper()
327    return ":".join(hex_str[i : i + 2] for i in range(0, 12, 2))
328
329
330def generate_active_remote_id(mac_address: str) -> str:
331    """
332    Generate an Active-Remote ID for DACP communication.
333
334    The Active-Remote ID is used to match DACP callbacks from devices to the
335    correct stream. This function generates a consistent ID based on the
336    player_id (=macaddress, =device id), converted to uint32).
337
338    :return: Active-Remote ID as decimal string.
339    """
340    # Convert MAC address format to uint32
341    # Remove colons: "AA:BB:CC:DD:EE:FF" -> "AABBCCDDEEFF"
342    hex_str = mac_address.replace(":", "").upper()
343    # Parse as uint64 and truncate to uint32 (lower 32 bits)
344    device_id_u64 = int(hex_str, 16)
345    device_id_u32 = device_id_u64 & 0xFFFFFFFF
346    return str(device_id_u32)
347
348
349def serialize_txt_records(discovery_info: AsyncServiceInfo) -> str:
350    """
351    Serialize mDNS TXT records for cliairplay's --txt argument.
352
353    The binary receives the full _airplay._tcp TXT as a single
354    space-separated "key=value key=value ..." argument and uses it for
355    automatic route selection (RAOP vs AirPlay 2, native vs RAOP-compat,
356    PTP vs NTP). Pairs containing whitespace are skipped as the binary
357    splits the blob on spaces.
358
359    :param discovery_info: The _airplay._tcp discovery info of the device.
360    """
361    pairs: list[str] = []
362    for key, value in discovery_info.decoded_properties.items():
363        if value is None:
364            continue
365        if any(char.isspace() for char in key) or any(char.isspace() for char in value):
366            continue
367        pairs.append(f"{key}={value}")
368    return " ".join(pairs)
369
370
371def get_final_output_format(audio_format: AudioFormat) -> AudioFormat:
372    """
373    Determine the output format ffmpeg must encode to for the cliairplay binary.
374
375    The cliairplay binary always uses ALAC encoding internally.
376    """
377    return AudioFormat(
378        content_type=ContentType.ALAC,
379        sample_rate=audio_format.sample_rate,
380        bit_depth=audio_format.bit_depth,
381        channels=audio_format.channels,
382    )
383
384
385def _parse_format_tables(info: dict[str, Any]) -> int:
386    """Return the union of the format tables in a receiver's /info response."""
387    # Each stream advertises its formats either as a list of bit indices in
388    # supportedAudioFormatsExtended, or as a plain mask in the older
389    # supportedFormats. A device can use a different shape per stream.
390    extended = info.get("supportedAudioFormatsExtended")
391    legacy = info.get("supportedFormats")
392    formats = 0
393    for stream in ("audioStream", "bufferStream"):
394        if isinstance(extended, dict) and isinstance(bits := extended.get(stream), list):
395            for bit in bits:
396                if isinstance(bit, int) and 0 <= bit < 64:
397                    formats |= 1 << bit
398        elif isinstance(legacy, dict) and isinstance(mask := legacy.get(stream), int):
399            formats |= mask
400    return formats
401
402
403def _get_cli_binary_name(system: str, machine: str) -> str | None:
404    """Return the cliairplay release asset name for a platform."""
405    normalized_system = system.lower().replace("darwin", "macos")
406    normalized_machine = machine.lower()
407
408    if normalized_machine in ("amd64", "x86_64"):
409        architecture = "x86_64"
410    elif normalized_machine in ("aarch64", "arm64"):
411        architecture = "arm64" if normalized_system == "macos" else "aarch64"
412    else:
413        return None
414    if normalized_system not in ("linux", "macos"):
415        return None
416    return f"cliairplay-{normalized_system}-{architecture}"
417