music-assistant-server

1"""Various (server-only) tools and helpers."""
2
3from __future__ import annotations
4
5import asyncio
6import codecs
7import functools
8import html
9import importlib
10import inspect
11import logging
12import os
13import platform
14import re
15import shutil
16import signal
17import socket
18import sys
19import time
20import unicodedata
21import urllib.error
22import urllib.request
23import weakref
24from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Coroutine
25from concurrent.futures import ThreadPoolExecutor
26from contextlib import suppress
27from importlib.metadata import PackageNotFoundError
28from importlib.metadata import version as pkg_version
29from ipaddress import IPv4Address, IPv6Address, ip_address
30from itertools import islice
31from pathlib import Path
32from types import ModuleType, TracebackType
33from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, Protocol, Self, TypeVar, cast
34from urllib.parse import urlparse
35
36import ifaddr
37from markdownify import markdownify
38from music_assistant_models.enums import AlbumType, IdentifierType
39from music_assistant_models.errors import UnsupportedSystemError
40from zeroconf import InterfaceChoice, IPVersion
41
42from music_assistant.constants import (
43    ANNOUNCE_ALERT_FILE,
44    LIVE_INDICATORS,
45    SOUNDTRACK_INDICATORS,
46    VERBOSE_LOG_LEVEL,
47    WILDCARD_BIND_IPS,
48)
49from music_assistant.helpers.process import check_output
50
51if TYPE_CHECKING:
52    from collections.abc import Iterator
53
54    from music_assistant_models.player import DeviceInfo
55    from zeroconf.asyncio import AsyncServiceInfo
56
57    from music_assistant.mass import MusicAssistant
58    from music_assistant.models import ProviderModuleType
59
60
61LOGGER = logging.getLogger(__name__)
62
63CALLBACK_TYPE = Callable[[], None]
64
65
66async def warn_if_missing_x86_64_v2(logger: logging.Logger) -> None:
67    """
68    Log a deprecation warning if the CPU lacks x86-64-v2 support.
69
70    :param logger: Logger instance to write the warning to.
71    """
72    if platform.machine() not in ("x86_64", "AMD64"):
73        return
74
75    def _check() -> bool | None:
76        try:
77            cpuinfo = Path("/proc/cpuinfo").read_text()
78        except FileNotFoundError, PermissionError:
79            return None
80
81        flags: set[str] = set()
82        for line in cpuinfo.splitlines():
83            if line.startswith("flags"):
84                flags.update(line.split())
85                break
86
87        if not flags:
88            return None
89
90        # x86-64-v2 requires: CMPXCHG16B, LAHF/SAHF, POPCNT, SSE3, SSSE3, SSE4.1, SSE4.2
91        # SSE3 may appear as "pni" (Prescott New Instructions) on older kernels
92        required = {"cx16", "lahf_lm", "popcnt", "sse4_1", "sse4_2", "ssse3"}
93        has_sse3 = bool({"sse3", "pni"} & flags)
94        return required.issubset(flags) and has_sse3
95
96    if await asyncio.to_thread(_check) is False:
97        logger.warning(
98            "\n\n"
99            "########################################################"
100            "########################\n"
101            "###               CPU DEPRECATION WARNING"
102            "                                    ###\n"
103            "########################################################"
104            "########################\n"
105            "\n"
106            "Your CPU does not support the x86-64-v2 instruction "
107            "set, which will be\n"
108            "required starting with Music Assistant 2.9.\n"
109            "\n"
110            "If you are running in a virtual machine (e.g. Proxmox),"
111            " change the CPU type\n"
112            "to 'host' or select a more modern CPU type preset "
113            "(e.g. x86-64-v2 or newer).\n"
114            "\n"
115            "If your physical CPU predates 2009, you will likely "
116            "need to upgrade\n"
117            "your hardware before updating Music Assistant to 2.9.\n"
118            "\n"
119            "########################################################"
120            "########################\n"
121        )
122
123
124def get_total_system_memory() -> float:
125    """
126    Return the memory available to this process in GB (0.0 when unknown).
127
128    On Linux this is min(physical RAM, cgroup memory limit), so a container's
129    --memory limit is honored when sizing buffers and gating heavy features.
130    Returns 0.0 when the platform cannot report memory (e.g. Windows), which
131    callers treat as "unknown" and fail open.
132    """
133    host_gb = _get_host_memory_gb()
134    if host_gb <= 0.0:
135        return 0.0
136    if sys.platform != "linux":
137        return host_gb
138    cgroup_gb = _get_cgroup_memory_limit_gb()
139    if cgroup_gb is None or cgroup_gb <= 0.0:
140        return host_gb
141    return min(host_gb, cgroup_gb)
142
143
144def _get_host_memory_gb() -> float:
145    """Return host physical RAM in GB via sysconf, or 0.0 when unavailable."""
146    try:
147        total_memory_bytes = os.sysconf("SC_PAGE_SIZE") * os.sysconf("SC_PHYS_PAGES")
148        return total_memory_bytes / (1024**3)
149    except AttributeError, ValueError, OSError:
150        # sysconf is unavailable on some platforms (e.g. Windows); treat as unknown.
151        return 0.0
152
153
154def _get_cgroup_memory_limit_gb(
155    cgroup_root: str = "/sys/fs/cgroup", proc_cgroup: str = "/proc/self/cgroup"
156) -> float | None:
157    """
158    Return this process's cgroup memory limit in GB, or None if unlimited/unavailable.
159
160    cgroup v2 (memory.max) is tried first, then v1 (memory/memory.limit_in_bytes).
161
162    :param cgroup_root: Mount point of the cgroup filesystem (overridable for tests).
163    :param proc_cgroup: Path to the process cgroup file (overridable for tests).
164    """
165    limit = _read_cgroup_v2_limit(cgroup_root, proc_cgroup)
166    if limit is not None:
167        return limit
168    return _read_cgroup_v1_limit(cgroup_root, proc_cgroup)
169
170
171def _read_cgroup_v2_limit(cgroup_root: str, proc_cgroup: str) -> float | None:
172    """Read the effective cgroup v2 memory limit in GB, or None."""
173    rel = _read_self_cgroup_path(proc_cgroup, controller=None)
174    return _min_hierarchical_limit(cgroup_root, rel, "memory.max")
175
176
177def _read_cgroup_v1_limit(cgroup_root: str, proc_cgroup: str) -> float | None:
178    """Read the effective cgroup v1 memory limit in GB, or None."""
179    # On v1 the memory controller is conventionally mounted at <root>/memory.
180    rel = _read_self_cgroup_path(proc_cgroup, controller="memory")
181    return _min_hierarchical_limit(
182        os.path.join(cgroup_root, "memory"), rel, "memory.limit_in_bytes"
183    )
184
185
186def _min_hierarchical_limit(base: str, rel: str | None, filename: str) -> float | None:
187    """
188    Return the smallest bounded memory limit (GB) across the cgroup and its ancestors, or None.
189
190    The effective limit is the minimum imposed anywhere from the process's own cgroup
191    up to the mount root, since a parent slice can cap memory even when the leaf cgroup
192    itself is unlimited (e.g. systemd slices or nested k8s cgroups).
193
194    :param base: Base of the hierarchy (the cgroup mount, or <mount>/memory on v1).
195    :param rel: The process's cgroup path relative to base (from /proc/self/cgroup).
196    :param filename: Limit file to read at each level (memory.max or memory.limit_in_bytes).
197    """
198    parts = [p for p in (rel or "").split("/") if p]
199    limits: list[float] = []
200    # Walk from the process's own cgroup up to the mount root.
201    while True:
202        directory = os.path.join(base, *parts) if parts else base
203        limit = _read_cgroup_limit_file(os.path.join(directory, filename))
204        if limit is not None:
205            limits.append(limit)
206        if not parts:
207            break
208        parts.pop()
209    return min(limits) if limits else None
210
211
212def _read_cgroup_limit_file(path: str) -> float | None:
213    """
214    Parse a cgroup memory-limit file into GB, or None if missing/unlimited/invalid.
215
216    :param path: Path to a cgroup memory.max (v2) or memory.limit_in_bytes (v1) file.
217    """
218    try:
219        with open(path) as fh:
220            raw = fh.read().strip()
221    except OSError:
222        # File absent or unreadable (covers FileNotFoundError/PermissionError/etc.).
223        return None
224    # "max" (v2) or a near-INT64_MAX sentinel (v1) both mean "no limit set".
225    if not raw or raw == "max":
226        return None
227    try:
228        limit_bytes = int(raw)
229    except ValueError:
230        return None
231    if limit_bytes <= 0 or limit_bytes >= _CGROUP_UNLIMITED_THRESHOLD:
232        return None
233    return limit_bytes / (1024**3)
234
235
236def _read_self_cgroup_path(proc_cgroup: str, *, controller: str | None) -> str | None:
237    """
238    Return the process's cgroup path from /proc/self/cgroup, or None.
239
240    :param proc_cgroup: Path to the process cgroup file.
241    :param controller: For cgroup v1, the controller name (e.g. "memory") whose path
242        to return. None selects the cgroup v2 unified hierarchy line ("0::<path>").
243    """
244    try:
245        with open(proc_cgroup) as fh:
246            for line in fh:
247                parts = line.strip().split(":", 2)
248                if len(parts) != 3:
249                    continue
250                hierarchy_id, controllers, path = parts
251                if controller is None:
252                    if hierarchy_id == "0" and controllers == "":
253                        return path or "/"
254                elif controller in controllers.split(","):
255                    return path or "/"
256    except OSError:
257        return None
258    return None
259
260
261# cgroup v1 writes a near-INT64_MAX value (PAGE_SIZE * LONG_MAX on most kernels) to
262# memory.limit_in_bytes when no limit is set; treat anything this large as unlimited.
263_CGROUP_UNLIMITED_THRESHOLD: int = 1 << 62
264
265
266def is_arm() -> bool:
267    """Return whether the host CPU is ARM-based (32- or 64-bit)."""
268    return platform.machine().lower() in ("arm64", "aarch64", "armv8l", "armv7l")
269
270
271def inference_thread_budget() -> int:
272    """
273    Return the native thread budget for on-device inference.
274
275    Defaults to ~25% of the available cores, or to an operator-supplied OMP_NUM_THREADS
276    when that is set, so the torch and native pool budgets stay in agreement.
277    """
278    override = os.environ.get("OMP_NUM_THREADS", "")
279    if override.isdigit() and (threads := int(override)) > 0:
280        return threads
281    return max(1, (os.process_cpu_count() or os.cpu_count() or 4) // 4)
282
283
284def cap_native_thread_pools() -> int:
285    """
286    Cap the native BLAS/OpenMP thread pools process-wide and return the applied budget.
287
288    Left uncapped, every one of these pools sizes itself to the full core count per worker
289    and, across concurrent analysis sessions, saturates the box and starves playback.
290
291    Must be called before any native math library is loaded, because these pools read their
292    size from the environment once, at library load time. Capping them after the fact means
293    walking the dynamic linker's loaded-library list (what threadpoolctl does), which
294    deadlocks against a concurrent import: the walk holds the loader lock while it needs the
295    GIL back for each callback, and an importing thread holds the GIL while it waits in
296    dlopen() for that same loader lock.
297    """
298    budget = inference_thread_budget()
299    for env_var in (
300        "OMP_NUM_THREADS",
301        "OPENBLAS_NUM_THREADS",
302        "MKL_NUM_THREADS",
303        "NUMEXPR_NUM_THREADS",
304        "VECLIB_MAXIMUM_THREADS",
305    ):
306        # setdefault so an operator-supplied value always wins
307        os.environ.setdefault(env_var, str(budget))
308    return budget
309
310
311async def verify_system_meets_requirements(
312    *,
313    feature_name: str,
314    min_memory_gb: float = 0.0,
315    min_cpu_cores: int = 0,
316    require_ml_inference: bool = False,
317) -> None:
318    """
319    Verify the host meets the minimum CPU/RAM requirements for a heavy provider.
320
321    :param feature_name: Human-readable provider name used in the error message.
322    :param min_memory_gb: Minimum total system RAM in GB (0 disables the check).
323    :param min_cpu_cores: Minimum CPU core count (0 disables the check).
324    :param require_ml_inference: When True, also verify the CPU can run on-device
325        torch inference. Checked last, as it spawns a probe subprocess.
326    :raises UnsupportedSystemError: If the system does not meet the requirements.
327    """
328    if shortfall := _resource_shortfall(min_memory_gb=min_memory_gb, min_cpu_cores=min_cpu_cores):
329        message, translation_key, translation_args = shortfall
330        raise UnsupportedSystemError(
331            f"This system does not meet the minimal requirements for {feature_name}: {message}",
332            translation_key=translation_key,
333            translation_args=[feature_name, *translation_args],
334        )
335    if require_ml_inference:
336        await verify_cpu_supports_ml_inference()
337
338
339def system_meets_requirements(
340    *,
341    min_memory_gb: float = 0.0,
342    min_cpu_cores: int = 0,
343) -> bool:
344    """
345    Return whether the host meets the given RAM/CPU thresholds.
346
347    A non-raising companion to verify_system_meets_requirements for soft UI hints
348    (e.g. hiding a recommended-hardware notice) rather than gating setup. The
349    ML-inference capability is not considered here.
350
351    :param min_memory_gb: Minimum total system RAM in GB (0 disables the check).
352    :param min_cpu_cores: Minimum CPU core count (0 disables the check).
353    """
354    return _resource_shortfall(min_memory_gb=min_memory_gb, min_cpu_cores=min_cpu_cores) is None
355
356
357# The kernel reports MemTotal — installed RAM minus firmware/reserved pages — so a host
358# always shows a little under its nominal size (a "4GB" box reports ~3.8GB). Allow this
359# fraction of slack when checking a RAM target, in one place rather than per call site, so
360# nominal requirements (4, 8 GB) match the hardware they describe without ad-hoc thresholds.
361MEMORY_REPORTING_TOLERANCE: float = 0.08
362
363
364def meets_memory_target(total_memory_gb: float, target_gb: float) -> bool:
365    """
366    Return whether reported RAM satisfies a nominal target within the reporting tolerance.
367
368    Fails open (True) when the target is 0 (no requirement) or memory is unknown
369    (0.0, e.g. Windows), so callers never block on a guess.
370
371    :param total_memory_gb: RAM reported by get_total_system_memory() in GB.
372    :param target_gb: Nominal RAM target in GB (e.g. 4 or 8).
373    """
374    if not target_gb or not total_memory_gb:
375        return True
376    return total_memory_gb >= target_gb * (1.0 - MEMORY_REPORTING_TOLERANCE)
377
378
379def _resource_shortfall(
380    *, min_memory_gb: float, min_cpu_cores: int
381) -> tuple[str, str, list[Any]] | None:
382    """
383    Return an unmet RAM/CPU threshold as (message, translation_key, translation_args), or None.
384
385    translation_args exclude the feature name, which the caller prepends.
386
387    :param min_memory_gb: Minimum total system RAM in GB (0 disables the check).
388    :param min_cpu_cores: Minimum CPU core count (0 disables the check).
389    """
390    cpu_cores = os.process_cpu_count() or os.cpu_count() or 1
391    if min_cpu_cores and cpu_cores < min_cpu_cores:
392        return (
393            f"at least {min_cpu_cores} CPU cores are required ({cpu_cores} detected).",
394            "unsupported_system_cpu_cores",
395            [min_cpu_cores, cpu_cores],
396        )
397    total_memory_gb = get_total_system_memory()
398    # meets_memory_target() fails open on unknown memory (0.0, e.g. Windows) and absorbs
399    # the kernel's MemTotal under-report, so min_memory_gb stays a clean nominal figure.
400    if min_memory_gb and not meets_memory_target(total_memory_gb, min_memory_gb):
401        return (
402            f"at least {min_memory_gb:.0f}GB of RAM is required ({total_memory_gb:.1f}GB detected).",
403            "unsupported_system_memory",
404            [f"{min_memory_gb:.0f}", f"{total_memory_gb:.1f}"],
405        )
406    return None
407
408
409# How long to wait for the out-of-process inference probe before treating it as
410# inconclusive. The probe only imports torch and runs a few tiny tensors, but a cold,
411# heavily loaded VM can be slow to start the interpreter, so keep this generous.
412_ML_INFERENCE_PROBE_TIMEOUT = 60.0
413# POSIX signals that mean the CPU could not execute the inference (the probe exits with the
414# negated signal number). Any of these disables the feature; other exits fail open.
415_ML_INFERENCE_FAULT_SIGNALS = frozenset(
416    {signal.SIGILL, signal.SIGSEGV, signal.SIGABRT, signal.SIGFPE}
417)
418
419
420async def verify_cpu_supports_ml_inference() -> None:
421    """
422    Verify the CPU can actually execute on-device ML (torch) inference.
423
424    Runs a representative inference in a throwaway subprocess, so a CPU that reports a
425    capability it cannot actually execute (common on virtual machines without host CPU
426    passthrough) crashes the probe instead of the server. Inconclusive probe results fail
427    open, so a probe malfunction never blocks a capable host.
428
429    :raises UnsupportedSystemError: If the CPU lacks AVX2, or reports it but cannot execute
430        the required instructions.
431    """
432    if platform.machine().lower() not in ("x86_64", "amd64", "i386", "i686", "x86"):
433        # non-x86 (ARM) machines run quantized inference via QNNPACK instead of FBGEMM
434        return
435    from music_assistant.helpers import _ml_inference_probe  # noqa: PLC0415
436
437    returncode = await _run_ml_inference_probe()
438    if returncode == _ml_inference_probe.PROBE_CAPABLE:
439        return
440    if returncode == _ml_inference_probe.PROBE_NO_AVX2:
441        raise UnsupportedSystemError(
442            "On-device audio analysis requires a CPU with AVX2 support "
443            "(Intel Haswell / AMD Zen or newer). This CPU does not support AVX2. "
444            "If you are running in a virtual machine (e.g. Proxmox), changing the "
445            "CPU type to 'host' may expose AVX2 to the guest.",
446            translation_key="unsupported_system_avx2",
447        )
448    if returncode is not None and returncode < 0 and -returncode in _ML_INFERENCE_FAULT_SIGNALS:
449        raise UnsupportedSystemError(
450            "On-device audio analysis cannot run on this CPU: it reports AVX2 support but "
451            "fails to execute the required instructions. This is common on virtual machines "
452            "without host CPU passthrough -- if you are running in a VM (e.g. Proxmox or "
453            "TrueNAS), set the CPU type to 'host'.",
454            translation_key="unsupported_system_ml_inference_failed",
455        )
456    # Inconclusive: the probe could not be spawned, timed out, was OOM-killed, or exited for
457    # an unexpected reason. Assume the host is capable rather than block a working setup.
458    LOGGER.warning(
459        "On-device ML inference capability probe was inconclusive (exit code %s); "
460        "assuming this CPU is capable",
461        returncode,
462    )
463
464
465async def _run_ml_inference_probe() -> int | None:
466    """
467    Run the inference probe subprocess and return its exit code.
468
469    Returns None when the probe could not be started or did not finish in time; otherwise
470    the process return code (negative if a signal killed it).
471    """
472    from music_assistant.helpers import _ml_inference_probe  # noqa: PLC0415
473
474    try:
475        # Run with -m, not by file path: a path run puts the probe's own directory on
476        # sys.path, which would shadow the stdlib (e.g. helpers/logging.py over logging).
477        proc = await asyncio.create_subprocess_exec(
478            sys.executable,
479            "-m",
480            _ml_inference_probe.__name__,
481            stdout=asyncio.subprocess.DEVNULL,
482            stderr=asyncio.subprocess.DEVNULL,
483        )
484    except OSError as err:
485        LOGGER.warning("Could not start the ML inference capability probe: %s", err)
486        return None
487    try:
488        await asyncio.wait_for(proc.wait(), timeout=_ML_INFERENCE_PROBE_TIMEOUT)
489    except TimeoutError:
490        proc.kill()
491        with suppress(ProcessLookupError):
492            await proc.wait()
493        LOGGER.warning("The ML inference capability probe timed out")
494        return None
495    return proc.returncode
496
497
498keyword_pattern = re.compile("title=|artist=")
499title_pattern = re.compile(r"title=\"(?P<title>.*?)\"")
500artist_pattern = re.compile(r"artist=\"(?P<artist>.*?)\"")
501dot_com_pattern = re.compile(r"(?P<netloc>\(?\w+\.(?:\w+\.)?(\w{2,3})\)?)")
502ad_pattern = re.compile(r"((ad|advertisement)_)|^AD\s\d+$|ADBREAK", flags=re.IGNORECASE)
503title_artist_order_pattern = re.compile(r"(?P<title>.+)\sBy:\s(?P<artist>.+)", flags=re.IGNORECASE)
504# German format used by some stations: "Track" von Artist
505german_von_pattern = re.compile(r'^"(?P<title>[^"]+)"\s+von\s+(?P<artist>.+)$', flags=re.IGNORECASE)
506# English format used by some stations: "Track" by Artist from "Album" (album optional).
507# Title and album are quote-delimited, so the non-greedy artist plus the anchored,
508# quoted album group keep "by"/"from" inside the artist name from being mis-split.
509english_by_pattern = re.compile(
510    r'^"(?P<title>[^"]+)"\s+by\s+(?P<artist>.+?)(?:\s+from\s+"(?P<album>[^"]*)")?$',
511    flags=re.IGNORECASE,
512)
513multi_space_pattern = re.compile(r"\s{2,}")
514end_junk_pattern = re.compile(r"(.+?)(\s\W+)$")
515
516# HTML tags worth preserving as markdown; any other tag is stripped (text kept)
517MARKDOWN_SAFE_TAGS = [
518    "a",
519    "b",
520    "blockquote",
521    "br",
522    "em",
523    "h1",
524    "h2",
525    "h3",
526    "h4",
527    "h5",
528    "h6",
529    "i",
530    "li",
531    "ol",
532    "p",
533    "strong",
534    "ul",
535]
536
537VERSION_PARTS = (
538    # list of common version strings
539    "version",
540    "live",
541    "edit",
542    "remix",
543    "mix",
544    "acoustic",
545    "instrumental",
546    "karaoke",
547    "remaster",
548    "remastered",
549    "versie",
550    "unplugged",
551    "disco",
552    "akoestisch",
553    "deluxe",
554    "video",
555    "radio",
556    "extended",
557    "single",
558    "edition",
559    "anniversary",
560    "stereo",
561    "album",
562    "bonus",
563    "release",
564)
565IGNORE_TITLE_PARTS = (
566    # strings that may be stripped off a title part
567    # (most important the featuring parts)
568    "feat.",
569    "featuring",
570    "ft.",
571    "with ",
572    "explicit",
573)
574WITH_TITLE_WORDS = (
575    # words that, when following "with", indicate this is part of the song title
576    # not a featuring credit.
577    "someone",
578    "the",
579    "u",
580    "you",
581    "no",
582)
583
584# Keywords for aggressive search cleaning (includes featuring).
585_VERSION_PATTERN = "|".join(re.escape(v) for v in VERSION_PARTS)
586_FEAT_PATTERN = r"feat(?:uring)?|ft"
587_SEARCH_PATTERN = rf"{_VERSION_PATTERN}|{_FEAT_PATTERN}"
588
589_SEARCH_PAREN_PATTERN = re.compile(
590    rf"[\(\[][^\)\]]*\b({_SEARCH_PATTERN})\b[^\)\]]*[\)\]]",
591    re.IGNORECASE,
592)
593_SEARCH_HYPHEN_PATTERN = re.compile(
594    rf"(\s*-\s*(\d{{4}}|{_SEARCH_PATTERN}).*)$",
595    re.IGNORECASE,
596)
597
598# Superfluous suffixes to strip for display (video/audio markers, etc.)
599_DISPLAY_STRIP_PATTERN = re.compile(
600    r"\s*[\(\[]"
601    r"(official\s+)?(lyric\s+|music\s+)?(video|audio|visualizer|clip)"
602    r"[\)\]]$",
603    re.IGNORECASE,
604)
605
606# Featuring patterns for stripping from titles (not in parentheses).
607_FEATURING_PATTERNS = (
608    " featuring ",
609    " feat. ",
610    " feat ",
611    " ft. ",
612    " ft ",
613)
614
615
616def filename_from_string(string: str) -> str:
617    """Create filename from unsafe string."""
618    keepcharacters = (" ", ".", "_")
619    return "".join(c for c in string if c.isalnum() or c in keepcharacters).rstrip()
620
621
622# aiohttp rejects the full C0 control character range plus DEL in response headers
623# to prevent header injection attacks (see aiohttp http_writer._FORBIDDEN_HEADER_CHARS_RE)
624_FORBIDDEN_HEADER_CHARS_RE = re.compile(r"[\x00-\x1f\x7f]")
625
626
627def sanitize_http_header_value(value: str) -> str:
628    """Replace control characters that are not allowed in HTTP header values."""
629    return _FORBIDDEN_HEADER_CHARS_RE.sub(" ", value).strip()
630
631
632def try_parse_int(possible_int: Any, default: int | None = 0) -> int | None:
633    """Try to parse an int."""
634    try:
635        return int(float(possible_int))
636    except TypeError, ValueError:
637        return default
638
639
640def try_parse_float(possible_float: Any, default: float | None = 0.0) -> float | None:
641    """Try to parse a float."""
642    try:
643        return float(possible_float)
644    except TypeError, ValueError:
645        return default
646
647
648def try_parse_bool(possible_bool: Any) -> bool:
649    """Try to parse a bool."""
650    if isinstance(possible_bool, bool):
651        return possible_bool
652    return possible_bool in ["true", "True", "1", "on", "ON", 1]
653
654
655def try_parse_duration(duration_str: str) -> float:
656    """Try to parse a duration in seconds from a duration (HH:MM:SS) string."""
657    milliseconds = (
658        float("0." + duration_str.rsplit(".", maxsplit=1)[-1]) if "." in duration_str else 0.0
659    )
660    duration_parts = duration_str.split(".", maxsplit=1)[0].split(",", maxsplit=1)[0].split(":")
661    if len(duration_parts) == 3:
662        seconds = sum(x * int(t) for x, t in zip([3600, 60, 1], duration_parts, strict=False))
663    elif len(duration_parts) == 2:
664        seconds = sum(x * int(t) for x, t in zip([60, 1], duration_parts, strict=False))
665    else:
666        seconds = int(duration_parts[0])
667    return seconds + milliseconds
668
669
670def normalize_unicode(value: str | None) -> str | None:
671    """
672    Normalize Unicode strings to NFC form for consistent handling.
673
674    This ensures that Unicode characters like "é" are stored as single
675    codepoints rather than "e" + combining accent mark, which prevents
676    issues with string comparisons and memory bloat.
677
678    :param value: String to normalize, or None.
679    """
680    if value is None:
681        return None
682    return unicodedata.normalize("NFC", value)
683
684
685@functools.lru_cache(maxsize=2048)
686def parse_title_and_version(
687    title: str,
688    track_version: str | None = None,
689    strip_for_search: bool = False,
690    strip_for_display: bool = False,
691) -> tuple[str, str]:
692    """
693    Parse version from the title and optionally clean for search or display.
694
695    :param title: The title to parse.
696    :param track_version: Optional existing version string.
697    :param strip_for_search: Aggressively strip for search matching.
698    :param strip_for_display: Strip superfluous suffixes for display.
699    """
700    version_parts = [track_version] if track_version else []
701    version_keys = {track_version.casefold()} if track_version else set()
702
703    # Strip featuring, bracketed version info, and hyphen suffixes (e.g. "- Remastered 2019")
704    if strip_for_search:
705        title = _SEARCH_PAREN_PATTERN.sub("", title)
706        title = _SEARCH_HYPHEN_PATTERN.sub("", title)
707        # Strip bare featuring credits (not in parentheses)
708        title_lower = title.lower()
709        for pattern in _FEATURING_PATTERNS:
710            if pattern in title_lower:
711                idx = title_lower.find(pattern)
712                title = title[:idx]
713                break
714        # Clean up dangling hyphens and extra spaces
715        title = re.sub(r"\s*-\s*$", "", title)
716        title = re.sub(r"\s+", " ", title).strip()
717        return title, track_version or ""
718
719    # Strip video/audio suffixes like "(Official Video)"
720    if strip_for_display:
721        title = _DISPLAY_STRIP_PATTERN.sub("", title).strip()
722        return title, track_version or ""
723
724    # Standard version parsing
725    for parts in (
726        _balanced_bracket_groups(title, "(", ")"),
727        _balanced_bracket_groups(title, "[", "]"),
728        re.findall(r" - .*", title),
729    ):
730        for title_part in parts:
731            # Extract the content without brackets/dashes for checking
732            clean_part = title_part.translate(str.maketrans("", "", "()[]-")).strip().lower()
733
734            # Check if this should be ignored (featuring/explicit parts)
735            should_ignore = False
736            for ignore_str in IGNORE_TITLE_PARTS:
737                if clean_part.startswith(ignore_str):
738                    # Special handling for "with " - check if followed by title words
739                    if ignore_str == "with ":
740                        # Extract the word after "with "
741                        after_with = (
742                            clean_part[len("with ") :].split()[0]
743                            if len(clean_part) > len("with ")
744                            else ""
745                        )
746                        if after_with in WITH_TITLE_WORDS:
747                            # This is part of the title (e.g., "with you"), don't ignore
748                            break
749                    # Remove this part from the title
750                    title = title.replace(title_part, "").strip()
751                    should_ignore = True
752                    break
753
754            if should_ignore:
755                continue
756
757            # Check if this part is a version
758            for version_str in VERSION_PARTS:
759                if version_str in clean_part:
760                    # Preserve original casing (and any nested brackets) for output
761                    version_part = _strip_outer_markers(title_part)
762                    if version_part.casefold() not in version_keys:
763                        version_parts.append(version_part)
764                        version_keys.add(version_part.casefold())
765                    title = title.replace(title_part, "").strip()
766                    break
767    title = re.sub(r"\s{2,}", " ", title).strip()
768    return title, " ".join(version_parts)
769
770
771def _balanced_bracket_groups(text: str, open_char: str, close_char: str) -> list[str]:
772    """
773    Return the top-level balanced bracketed substrings, including the outer brackets.
774
775    :param text: The text to scan.
776    :param open_char: The opening bracket character.
777    :param close_char: The closing bracket character.
778    """
779    groups: list[str] = []
780    depth = 0
781    start = -1
782    for idx, char in enumerate(text):
783        if char == open_char:
784            if depth == 0:
785                start = idx
786            depth += 1
787        elif char == close_char and depth > 0:
788            depth -= 1
789            if depth == 0:
790                groups.append(text[start : idx + 1])
791    return groups
792
793
794def _strip_outer_markers(part: str) -> str:
795    """
796    Strip the outer brackets or leading hyphen from a parsed title part.
797
798    :param part: The raw title part as matched from the title.
799    """
800    part = part.strip()
801    # only strip a single outer bracket pair so nested brackets stay intact
802    if part[:1] in "([" and part[-1:] in ")]":
803        return part[1:-1].strip()
804    return part.lstrip("- ").strip()
805
806
807def infer_album_type(title: str, version: str) -> AlbumType:
808    """Infer album type by looking for live or soundtrack indicators."""
809    combined = f"{title} {version}".lower()
810    for pat in LIVE_INDICATORS:
811        if re.search(pat, combined):
812            return AlbumType.LIVE
813    for pat in SOUNDTRACK_INDICATORS:
814        if re.search(pat, combined):
815            return AlbumType.SOUNDTRACK
816    return AlbumType.UNKNOWN
817
818
819def strip_ads(line: str) -> str:
820    """Strip Ads from line."""
821    if ad_pattern.search(line):
822        return "Advert"
823    return line
824
825
826def strip_url(line: str) -> str:
827    """Strip URL from line."""
828    return (
829        " ".join([p for p in line.split() if (not urlparse(p).scheme or not urlparse(p).netloc)])
830    ).rstrip()
831
832
833def strip_dotcom(line: str) -> str:
834    """Strip scheme-less netloc from line."""
835    return dot_com_pattern.sub("", line)
836
837
838def strip_end_junk(line: str) -> str:
839    """Strip non-word info from end of line."""
840    return end_junk_pattern.sub(r"\1", line)
841
842
843def swap_title_artist_order(line: str) -> str:
844    """Swap title/artist order in line."""
845    return title_artist_order_pattern.sub(r"\g<artist> - \g<title>", line)
846
847
848def strip_multi_space(line: str) -> str:
849    """Strip multi-whitespace from line."""
850    return multi_space_pattern.sub(" ", line)
851
852
853def html_to_markdown(line: str) -> str:
854    """Convert the safe subset of HTML in a string to markdown, stripping other tags."""
855    # unescape first so entity-encoded markup (e.g. "&lt;p&gt;") is handled too
856    return markdownify(
857        html.unescape(line),
858        convert=MARKDOWN_SAFE_TAGS,
859        escape_asterisks=False,
860        escape_underscores=False,
861        escape_misc=False,
862    ).strip()
863
864
865def multi_strip(line: str) -> str:
866    """Strip assorted junk from line."""
867    return strip_multi_space(
868        swap_title_artist_order(strip_end_junk(strip_dotcom(strip_url(strip_ads(line)))))
869    ).rstrip()
870
871
872def parse_quoted_stream_title(line: str) -> tuple[str, str, str | None] | None:
873    """
874    Parse stream titles that name the track in natural language with a quoted title.
875
876    Recognises '"Track" by Artist from "Album"' (album optional) and the German
877    '"Track" von Artist'.
878
879    :param line: Raw (uncleaned) stream title.
880    :returns: Tuple of (title, artist, album), or None when the line is not in one of
881        these formats. ``album`` is None when the station omits it.
882    """
883    stripped = line.strip()
884    if match := english_by_pattern.match(stripped):
885        title = multi_strip(match.group("title"))
886        artist = multi_strip(match.group("artist")).strip('"')
887        album_raw = match.group("album")
888        album = multi_strip(album_raw).strip('"') if album_raw else None
889        if title and artist:
890            return title, artist, album or None
891    if match := german_von_pattern.match(stripped):
892        title = multi_strip(match.group("title"))
893        artist = multi_strip(match.group("artist")).strip('"')
894        if title and artist:
895            return title, artist, None
896    return None
897
898
899def clean_stream_title(line: str) -> str:
900    """Strip junk text from radio streamtitle."""
901    title: str = ""
902    artist: str = ""
903
904    if not keyword_pattern.search(line):
905        if parsed := parse_quoted_stream_title(line):
906            track_name, artist_name, _ = parsed
907            return f"{artist_name} - {track_name}"
908        return multi_strip(line)
909
910    if match := title_pattern.search(line):
911        title = multi_strip(match.group("title"))
912
913    if match := artist_pattern.search(line):
914        possible_artist = multi_strip(match.group("artist"))
915        if possible_artist and possible_artist != title:
916            artist = possible_artist
917
918    if not title and not artist:
919        return ""
920
921    if title:
922        if re.search(" - ", title) or not artist:
923            return title
924        if artist:
925            return f"{artist} - {title}"
926
927    if artist:
928        return artist
929
930    return line
931
932
933# cache for get_ip_addresses: enumerating the network adapters involves a thread hop,
934# socket probes and a full adapter walk, while the result rarely (if ever) changes
935IP_ADDRESSES_CACHE_TTL = 30
936_ip_addresses_cache: dict[tuple[bool, bool], tuple[float, tuple[str, ...]]] = {}
937_ip_addresses_pending: dict[tuple[bool, bool], asyncio.Task[tuple[str, ...]]] = {}
938
939# Interfaces that only ever carry container, VM or VPN traffic, so a device on the local
940# network can never reach us on their addresses.
941_VIRTUAL_INTERFACE_PREFIXES = (
942    "cali",
943    "cni",
944    "docker",
945    "flannel",
946    "hassio",
947    "incusbr",
948    "lxcbr",
949    "lxdbr",
950    "nordlynx",
951    "podman",
952    "ppp",
953    "tailscale",
954    "tap",
955    "tun",
956    "utun",
957    "vboxnet",
958    "veth",
959    "virbr",
960    "vmnet",
961    "wg",
962    "zt",
963)
964# Docker names its user-defined bridges br-<12 hex> and the macOS host-only bridges of
965# Docker Desktop, Parallels and VMware start at bridge100. Both are matched in full, so a
966# hand-named LAN bridge (br-lan on OpenWrt, a second macOS bridge1) is left alone - as are
967# the regular LAN bridge names br0, vmbr0 and bond0.
968_VIRTUAL_INTERFACE_NAMES = re.compile(r"br-[0-9a-f]{12}|bridge\d{3}")
969
970
971async def get_ip_addresses(include_ipv6: bool = False) -> tuple[str, ...]:
972    """
973    Return all IP addresses of all network interfaces.
974
975    Always returns at least one address: when no routable address is found
976    (e.g. offline host), the loopback address is returned as fallback.
977    Results are cached for a short while, so an IP/interface change may take up to
978    IP_ADDRESSES_CACHE_TTL seconds to be reflected.
979
980    :param include_ipv6: Whether to include IPv6 addresses in the result.
981    """
982    return await _get_ip_addresses(include_ipv6, publish_candidates_only=False)
983
984
985async def get_publish_ip_candidates(include_ipv6: bool = False) -> tuple[str, ...]:
986    """
987    Return the IP addresses a device on the local network may reach this host on.
988
989    Same as get_ip_addresses, minus the addresses of container, VM and VPN interfaces -
990    unless the host holds no other address at all.
991
992    :param include_ipv6: Whether to include IPv6 addresses in the result.
993    """
994    return await _get_ip_addresses(include_ipv6, publish_candidates_only=True)
995
996
997async def _get_ip_addresses(include_ipv6: bool, publish_candidates_only: bool) -> tuple[str, ...]:
998    """Return the host's IP addresses, enumerating the adapters at most once per TTL."""
999    cache_key = (include_ipv6, publish_candidates_only)
1000    if cached := _ip_addresses_cache.get(cache_key):
1001        cached_at, addresses = cached
1002        if (time.monotonic() - cached_at) < IP_ADDRESSES_CACHE_TTL:
1003            return addresses
1004
1005    async def _probe() -> tuple[str, ...]:
1006        try:
1007            addresses = await asyncio.to_thread(
1008                _enumerate_ip_addresses, include_ipv6, publish_candidates_only
1009            )
1010            _ip_addresses_cache[cache_key] = (time.monotonic(), addresses)
1011            return addresses
1012        finally:
1013            _ip_addresses_pending.pop(cache_key, None)
1014
1015    # single-flight: no await between the pending-check and storing the task,
1016    # so concurrent callers always end up awaiting the same probe
1017    if not (pending := _ip_addresses_pending.get(cache_key)):
1018        pending = asyncio.create_task(_probe())
1019        pending.add_done_callback(_log_ip_probe_failure)
1020        _ip_addresses_pending[cache_key] = pending
1021    return await join_task(pending)
1022
1023
1024def _log_ip_probe_failure(probe: asyncio.Task[tuple[str, ...]]) -> None:
1025    """Log (and thereby retrieve) the exception of a finished address probe, if any."""
1026    if probe.cancelled():
1027        return
1028    # every waiter that is still around reports the failure itself, so a debug line is
1029    # enough here; retrieving the exception is what keeps asyncio from reporting it as
1030    # "Task exception was never retrieved" once the probe is garbage collected
1031    if (err := probe.exception()) is not None:
1032        LOGGER.debug("Enumerating IP addresses failed: %s", err)
1033
1034
1035def _enumerate_ip_addresses(include_ipv6: bool, publish_candidates_only: bool) -> tuple[str, ...]:
1036    """Enumerate all IP addresses of all network interfaces (blocking)."""
1037    result: list[tuple[int, str]] = []
1038    # the same addresses, without the ones no device on the local network can reach
1039    lan_result: list[tuple[int, str]] = []
1040    # try to get the primary IP address
1041    # this is the IP address of the default route
1042    primary_ip = ""
1043    # try IPv4 first
1044    _sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
1045    _sock.settimeout(0)
1046    try:
1047        # doesn't even have to be reachable
1048        _sock.connect(("10.254.254.254", 1))
1049        primary_ip = _sock.getsockname()[0]
1050    except Exception:
1051        primary_ip = ""
1052    finally:
1053        _sock.close()
1054    # fall back to IPv6 if no IPv4 primary found (e.g. IPv6-only networks)
1055    if not primary_ip:
1056        _sock6 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
1057        _sock6.settimeout(0)
1058        try:
1059            _sock6.connect(("2001:db8::1", 1))
1060            primary_ip = _sock6.getsockname()[0]
1061        except Exception:
1062            primary_ip = ""
1063        finally:
1064            _sock6.close()
1065    # get all IP addresses of all network interfaces
1066    adapters = ifaddr.get_adapters()
1067    for adapter in adapters:
1068        adapter_is_virtual = _is_virtual_interface(adapter.name) or _is_virtual_interface(
1069            adapter.nice_name
1070        )
1071        for ip in adapter.ips:
1072            if ip.is_IPv6 and not include_ipv6:
1073                continue
1074            # ifaddr returns IPv6 addresses as (address, flowinfo, scope_id) tuples
1075            ip_str = ip.ip[0] if isinstance(ip.ip, tuple) else ip.ip
1076            if ip_str.startswith(("127", "169.254")):
1077                # filter out IPv4 loopback/APIPA address
1078                continue
1079            if ip_str.startswith(("::1", "::ffff:", "fe80")):
1080                # filter out IPv6 loopback/link-local address
1081                continue
1082            if ip_str == primary_ip:
1083                score = 10
1084            elif ip_str.startswith(("192.168.",)):
1085                # we rank the 192.168 range a bit higher as its most
1086                # often used as the private network subnet
1087                score = 2
1088            elif ip_str.startswith(("172.", "10.", "192.")):
1089                # we rank the 172 range a bit lower as its most
1090                # often used as the private docker network
1091                score = 1
1092            else:
1093                score = 0
1094            result.append((score, ip_str))
1095            if not adapter_is_virtual:
1096                lan_result.append((score, ip_str))
1097    # a host that is only reachable over a tunnel or bridge still has to publish something
1098    selected = (lan_result or result) if publish_candidates_only else result
1099    selected.sort(key=lambda x: x[0], reverse=True)
1100    if not selected:
1101        # no routable addresses found (e.g. offline host with only loopback/link-local):
1102        # fall back to loopback so callers that rely on at least one address keep working
1103        return ("127.0.0.1",)
1104    return tuple(ip[1] for ip in selected)
1105
1106
1107def _is_virtual_interface(name: str) -> bool:
1108    """Return whether the named interface belongs to a container, VM or VPN network."""
1109    name = name.lower()
1110    return name.startswith(_VIRTUAL_INTERFACE_PREFIXES) or bool(
1111        _VIRTUAL_INTERFACE_NAMES.fullmatch(name)
1112    )
1113
1114
1115def interface_name_for_ip(ip: str) -> str | None:
1116    """
1117    Return the name of the network interface that holds the given IP, or None.
1118
1119    Used to map a bind/publish IP to its interface name for components that select
1120    their mDNS/zeroconf advertisement interface by name (e.g. shairport-sync and
1121    go-librespot), so the advertisement stays on the intended network.
1122
1123    :param ip: The IPv4/IPv6 address to look up.
1124    """
1125    for adapter in ifaddr.get_adapters():
1126        for ip_config in adapter.ips:
1127            addr = ip_config.ip if isinstance(ip_config.ip, str) else ip_config.ip[0]
1128            if addr == ip:
1129                return adapter.name
1130    return None
1131
1132
1133async def is_port_in_use(port: int, host: str | None = None) -> bool:
1134    """
1135    Check if a port is in use.
1136
1137    :param port: Port number to check.
1138    :param host: Optional bind address to probe. When omitted, both IPv4 and IPv6
1139        wildcard addresses are checked.
1140    """
1141
1142    def _is_port_in_use() -> bool:
1143        candidates: tuple[tuple[socket.AddressFamily, str], ...]
1144        if host is not None:
1145            candidates = ((socket.AF_INET6 if ":" in host else socket.AF_INET, host),)
1146        else:
1147            # Try both IPv4 and IPv6 to support single-stack and dual-stack systems.
1148            # A port is considered free if it can be bound on at least one address family.
1149            candidates = ((socket.AF_INET, "0.0.0.0"), (socket.AF_INET6, "::"))
1150        for family, addr in candidates:
1151            try:
1152                with socket.socket(family, socket.SOCK_STREAM) as _sock:
1153                    # Set SO_REUSEADDR to match asyncio.start_server behavior
1154                    # This allows binding to ports in TIME_WAIT state
1155                    _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1156                    _sock.bind((addr, port))
1157                    return False
1158            except OSError:
1159                continue
1160        return True
1161
1162    return await asyncio.to_thread(_is_port_in_use)
1163
1164
1165# In-process reservations for ports handed out by select_free_port. Provider
1166# instances (and reloads) frequently call select_free_port at nearly the same
1167# moment and only bind the returned port asynchronously afterwards, so a port
1168# that was just handed out is not yet detectable as "in use". Keeping a
1169# short-lived reservation per returned port stops concurrent/successive callers
1170# from picking the same one. Reservations expire automatically after the grace
1171# period so the range is never permanently exhausted across reloads.
1172_PORT_RESERVATION_TTL = 60.0
1173_reserved_ports: dict[int, float] = {}
1174_select_free_port_lock = asyncio.Lock()
1175
1176
1177async def select_free_port(range_start: int, range_end: int, host: str | None = None) -> int:
1178    """
1179    Find and reserve a free port within the given range.
1180
1181    The returned port is reserved so concurrent or successive callers are not
1182    handed the same port.
1183
1184    :param range_start: First port (inclusive) of the range to search.
1185    :param range_end: Port to stop before (exclusive) when searching the range.
1186    :param host: Optional bind address to probe for availability.
1187    """
1188    async with _select_free_port_lock:
1189        now = time.monotonic()
1190        # drop expired reservations so their ports become reusable again
1191        for reserved_port, deadline in list(_reserved_ports.items()):
1192            if deadline <= now:
1193                del _reserved_ports[reserved_port]
1194        for port in range(range_start, range_end):
1195            if port in _reserved_ports:
1196                continue
1197            if not await is_port_in_use(port, host=host):
1198                _reserved_ports[port] = now + _PORT_RESERVATION_TTL
1199                return port
1200    msg = f"No free port available in range {range_start}-{range_end - 1}"
1201    raise OSError(msg)
1202
1203
1204async def get_ip_from_host(dns_name: str) -> str | None:
1205    """Resolve (first) IP-address for given dns name."""
1206
1207    def _resolve() -> str | None:
1208        try:
1209            # use getaddrinfo to support both IPv4 and IPv6 resolution
1210            results = socket.getaddrinfo(dns_name, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
1211            if results:
1212                return str(results[0][4][0])
1213        except Exception:
1214            # fail gracefully!
1215            return None
1216        return None
1217
1218    return await asyncio.to_thread(_resolve)
1219
1220
1221async def get_source_ip_for_target(target_ip: str) -> str:
1222    """
1223    Return the local interface address the routing table would egress to ``target_ip`` from.
1224
1225    Empty when no route to the target can be determined.
1226
1227    :param target_ip: IP address of the device the traffic is meant for.
1228    """
1229
1230    def _routing_lookup() -> str:
1231        try:
1232            is_ipv6_target = ip_address(target_ip).version == 6
1233        except ValueError:
1234            is_ipv6_target = False
1235        route_family = socket.AF_INET6 if is_ipv6_target else socket.AF_INET
1236        route_target: tuple[str, int] | tuple[str, int, int, int] = (
1237            (target_ip, 80, 0, 0) if is_ipv6_target else (target_ip, 80)
1238        )
1239        with socket.socket(route_family, socket.SOCK_DGRAM) as _sock:
1240            try:
1241                _sock.settimeout(1.0)
1242                _sock.connect(route_target)
1243                routed_ip = str(_sock.getsockname()[0])
1244                if routed_ip and routed_ip not in WILDCARD_BIND_IPS:
1245                    return routed_ip
1246            except OSError:
1247                pass
1248        return ""
1249
1250    return await asyncio.to_thread(_routing_lookup)
1251
1252
1253async def get_ip_pton(ip_string: str) -> bytes:
1254    """Return socket pton for a local ip."""
1255    try:
1256        return await asyncio.to_thread(socket.inet_pton, socket.AF_INET, ip_string)
1257    except OSError:
1258        return await asyncio.to_thread(socket.inet_pton, socket.AF_INET6, ip_string)
1259
1260
1261def format_ip_for_url(ip_address: str) -> str:
1262    """Wrap IPv6 addresses in brackets for use in URLs (RFC 2732)."""
1263    if ":" in ip_address:
1264        return f"[{ip_address}]"
1265    return ip_address
1266
1267
1268async def get_folder_size(folderpath: str) -> float:
1269    """Return folder size in gb."""
1270
1271    def _get_folder_size(folderpath: str) -> float:
1272        total_size = 0
1273        for dirpath, _dirnames, filenames in os.walk(folderpath):
1274            for _file in filenames:
1275                _fp = os.path.join(dirpath, _file)
1276                total_size += Path(_fp).stat().st_size
1277        return total_size / float(1 << 30)
1278
1279    return await asyncio.to_thread(_get_folder_size, folderpath)
1280
1281
1282def get_changed_keys(
1283    dict1: dict[str, Any],
1284    dict2: dict[str, Any],
1285    recursive: bool = False,
1286) -> set[str]:
1287    """Compare 2 dicts and return set of changed keys."""
1288    # TODO: Check with Marcel whether we should calculate new dicts based on ignore_keys
1289    return set(get_changed_dict_values(dict1, dict2, recursive).keys())
1290    # return set(get_changed_dict_values(dict1, dict2, ignore_keys, recursive).keys())
1291
1292
1293def get_changed_dict_values(
1294    dict1: dict[str, Any],
1295    dict2: dict[str, Any],
1296    recursive: bool = False,
1297) -> dict[str, tuple[Any, Any]]:
1298    """
1299    Compare 2 dicts and return dict of changed values.
1300
1301    dict key is the changed key, value is tuple of old and new values.
1302    """
1303    if not dict1 and not dict2:
1304        return {}
1305    if not dict1:
1306        return {key: (None, value) for key, value in dict2.items()}
1307    if not dict2:
1308        return {key: (None, value) for key, value in dict1.items()}
1309    changed_values = {}
1310    for key, value in dict2.items():
1311        if isinstance(value, dict) and isinstance(dict1[key], dict) and recursive:
1312            changed_subvalues = get_changed_dict_values(dict1[key], value, recursive)
1313            for subkey, subvalue in changed_subvalues.items():
1314                changed_values[f"{key}.{subkey}"] = subvalue
1315            continue
1316        if key not in dict1:
1317            changed_values[key] = (None, value)
1318            continue
1319        if dict1[key] != value:
1320            changed_values[key] = (dict1[key], value)
1321    return changed_values
1322
1323
1324def empty_queue[T](q: asyncio.Queue[T]) -> None:
1325    """Empty an asyncio Queue."""
1326    for _ in range(q.qsize()):
1327        try:
1328            q.get_nowait()
1329            q.task_done()
1330        except asyncio.QueueEmpty, ValueError:
1331            pass
1332
1333
1334async def install_package(package: str) -> None:
1335    """Install package with pip, raise when install failed."""
1336    LOGGER.debug("Installing python package %s", package)
1337    args = ["uv", "pip", "install", "--no-cache", package]
1338    return_code, output = await check_output(*args)
1339    if return_code != 0:
1340        msg = f"Failed to install package {package}\n{output.decode()}"
1341        raise RuntimeError(msg)
1342
1343
1344async def get_package_version(pkg_name: str) -> str | None:
1345    """
1346    Return the version of an installed (python) package.
1347
1348    Will return None if the package is not found.
1349    """
1350    try:
1351        return await asyncio.to_thread(pkg_version, pkg_name)
1352    except PackageNotFoundError:
1353        return None
1354
1355
1356async def is_hass_supervisor() -> bool:
1357    """Return if we're running inside the HA Supervisor (e.g. HAOS)."""
1358    # Fast path: check for HA supervisor token environment variable
1359    # This is always set when running inside the HA supervisor
1360    if not os.environ.get("SUPERVISOR_TOKEN"):
1361        return False
1362
1363    # Token exists, verify the supervisor is actually reachable
1364    def _check() -> bool:
1365        try:
1366            urllib.request.urlopen("http://supervisor/core", timeout=1)
1367        except urllib.error.URLError as err:
1368            # this should return a 401 unauthorized if it exists
1369            return getattr(err, "code", 999) == 401
1370        except Exception:
1371            return False
1372        return False
1373
1374    return await asyncio.to_thread(_check)
1375
1376
1377# CPython holds a lock per module while importing it, so two threads importing modules with
1378# overlapping dependency graphs (e.g. two providers that both pull in `requests`) can end up
1379# waiting on each other's module locks. The import machinery then bails out at one of them with
1380# a _DeadlockError ("deadlock detected by _ModuleLock(...)") instead of hanging, which surfaces
1381# as a provider that failed to load and stays broken until it is reloaded by hand.
1382# A single-worker executor keeps imports serialized without parking a thread from the default
1383# pool while waiting; only the import itself is serialized, providers still load concurrently.
1384_IMPORT_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="module_import")
1385
1386# requirements verified this session, so repeated (config) loads skip the version check
1387_checked_requirements: set[str] = set()
1388
1389
1390async def import_module_in_thread(name: str, package: str | None = None) -> ModuleType:
1391    """
1392    Import a module in a thread, serialized against all other imports done this way.
1393
1394    :param name: Name of the module to import, may be relative to the given package.
1395    :param package: Package to resolve the name against, required for a relative name.
1396    """
1397    loop = asyncio.get_running_loop()
1398    try:
1399        return await loop.run_in_executor(_IMPORT_EXECUTOR, importlib.import_module, name, package)
1400    except RuntimeError as err:
1401        # threads we do not control (a library importing lazily in its own thread) can still
1402        # cross a module lock with ours; the import machinery reports that as a deadlock at
1403        # whoever detects it. The other import has finished by now, so a single retry sticks.
1404        if "deadlock detected" not in str(err):
1405            raise
1406        LOGGER.warning("Retrying import of %s after a module lock collision: %s", name, err)
1407        return await loop.run_in_executor(_IMPORT_EXECUTOR, importlib.import_module, name, package)
1408
1409
1410async def load_provider_module(domain: str, requirements: list[str]) -> ProviderModuleType:
1411    """Return module for given provider domain and make sure the requirements are met."""
1412
1413    async def _get_provider_module() -> ProviderModuleType:
1414        module = await import_module_in_thread(f".{domain}", "music_assistant.providers")
1415        return cast("ProviderModuleType", module)
1416
1417    # ensure module requirements are met
1418    for requirement in requirements:
1419        if requirement in _checked_requirements:
1420            continue
1421        if "==" not in requirement:
1422            # we should really get rid of unpinned requirements
1423            continue
1424        package_name, version = requirement.split("==", 1)
1425        # importlib.metadata can't resolve extras (e.g. aiosendspin[server]), so strip them
1426        package_name = package_name.split("[", 1)[0]
1427        installed_version = await get_package_version(package_name)
1428        if installed_version == "0.0.0":
1429            # ignore editable installs
1430            _checked_requirements.add(requirement)
1431            continue
1432        if installed_version != version:
1433            await install_package(requirement)
1434        _checked_requirements.add(requirement)
1435
1436    # try to load the module
1437    try:
1438        return await _get_provider_module()
1439    except ImportError:
1440        # (re)install ALL requirements
1441        for requirement in requirements:
1442            await install_package(requirement)
1443    # try loading the provider again to be safe
1444    # this will fail if something else is wrong (as it should)
1445    return await _get_provider_module()
1446
1447
1448async def has_tmpfs_mount() -> bool:
1449    """Check if we have a tmpfs mount."""
1450
1451    def _has_tmpfs_mount() -> bool:
1452        """Check if we have a tmpfs mount."""
1453        try:
1454            with open("/proc/mounts") as file:
1455                for line in file:
1456                    if "tmpfs /tmp tmpfs rw" in line:
1457                        return True
1458        except FileNotFoundError, OSError, PermissionError:
1459            pass
1460        return False
1461
1462    return await asyncio.to_thread(_has_tmpfs_mount)
1463
1464
1465async def get_free_space(folder: str) -> float:
1466    """Return free space on given folderpath in GB."""
1467
1468    def _get_free_space(folder: str) -> float:
1469        """Return free space on given folderpath in GB."""
1470        try:
1471            res = shutil.disk_usage(folder)
1472            return res.free / float(1 << 30)
1473        except FileNotFoundError, OSError, PermissionError:
1474            return 0.0
1475
1476    return await asyncio.to_thread(_get_free_space, folder)
1477
1478
1479async def get_free_space_percentage(folder: str) -> float:
1480    """Return free space on given folderpath in percentage."""
1481
1482    def _get_free_space(folder: str) -> float:
1483        """Return free space on given folderpath in GB."""
1484        try:
1485            res = shutil.disk_usage(folder)
1486            return res.free / res.total * 100
1487        except FileNotFoundError, OSError, PermissionError:
1488            return 0.0
1489
1490    return await asyncio.to_thread(_get_free_space, folder)
1491
1492
1493async def has_enough_space(folder: str, size: int) -> bool:
1494    """Check if folder has enough free space."""
1495    return await get_free_space(folder) > size
1496
1497
1498def divide_chunks(data: bytes, chunk_size: int) -> Iterator[bytes]:
1499    """Chunk bytes data into smaller chunks."""
1500    for i in range(0, len(data), chunk_size):
1501        yield data[i : i + chunk_size]
1502
1503
1504async def remove_file(file_path: str) -> None:
1505    """Remove file path (if it exists)."""
1506    if not await asyncio.to_thread(os.path.exists, file_path):
1507        return
1508    await asyncio.to_thread(os.remove, file_path)
1509    LOGGER.log(VERBOSE_LOG_LEVEL, "Removed file: %s", file_path)
1510
1511
1512def get_primary_ip_address_from_zeroconf(
1513    discovery_info: AsyncServiceInfo,
1514    prefer_ipv6: bool = False,
1515) -> str | None:
1516    """
1517    Get primary IP address from zeroconf discovery info.
1518
1519    :param discovery_info: The zeroconf service info to extract the address from.
1520    :param prefer_ipv6: If True, prefer IPv6 addresses over IPv4.
1521    """
1522    if prefer_ipv6:
1523        order = [IPVersion.V6Only, IPVersion.V4Only]
1524    else:
1525        order = [IPVersion.V4Only, IPVersion.V6Only]
1526    for version in order:
1527        for addr in discovery_info.ip_addresses_by_version(version):
1528            if addr.is_loopback or addr.is_link_local or addr.is_unspecified:
1529                continue
1530            return str(addr)
1531    return None
1532
1533
1534def get_port_from_zeroconf(discovery_info: AsyncServiceInfo) -> int | None:
1535    """Get port from zeroconf discovery info."""
1536    return discovery_info.port
1537
1538
1539def get_zeroconf_args(
1540    use_all_interfaces: bool = False,
1541) -> dict[str, Any]:
1542    """
1543    Determine optimal zeroconf IPVersion and interfaces from system adapters.
1544
1545    Inspects available network adapters to determine the correct IP version
1546    and interface configuration, similar to Home Assistant's approach.
1547
1548    :param use_all_interfaces: If True, use all interfaces (user override).
1549    """
1550    adapters = ifaddr.get_adapters()
1551    has_ipv4 = False
1552    has_ipv6 = False
1553    interface_ips: list[str] = []
1554    for adapter in adapters:
1555        for ip_config in adapter.ips:
1556            if ip_config.is_IPv6:
1557                ip_tuple = cast("tuple[str, int, int]", ip_config.ip)
1558                addr = ip_address(ip_tuple[0])
1559                if (
1560                    isinstance(addr, IPv6Address)
1561                    and not addr.is_loopback
1562                    and not addr.is_link_local
1563                ):
1564                    has_ipv6 = True
1565                    if not addr.is_global:
1566                        interface_ips.append(f"{ip_tuple[0]}%{ip_tuple[2]}")
1567            else:
1568                ip_str = cast("str", ip_config.ip)
1569                addr = ip_address(ip_str)
1570                if isinstance(addr, IPv4Address) and not addr.is_loopback:
1571                    has_ipv4 = True
1572                    interface_ips.append(ip_str)
1573
1574    # Determine IP version based on available addresses.
1575    # On macOS/FreeBSD, zeroconf's IPVersion.All creates an AF_INET6 listen socket
1576    # that cannot join IPv4 multicast groups, silently breaking discovery of
1577    # IPv4-only devices. Fall back to V4Only on those platforms.
1578    has_functional_dual_stack = not sys.platform.startswith(("freebsd", "darwin"))
1579    if has_ipv4 and has_ipv6 and has_functional_dual_stack:
1580        ip_version = IPVersion.All
1581    elif has_ipv4:
1582        ip_version = IPVersion.V4Only
1583    elif has_ipv6:
1584        ip_version = IPVersion.V6Only
1585    else:
1586        ip_version = IPVersion.V4Only
1587
1588    if use_all_interfaces:
1589        # User explicitly requested all interfaces — pass explicit IP list
1590        # to avoid issues with InterfaceChoice.Default on multi-interface hosts.
1591        if interface_ips:
1592            return {"ip_version": ip_version, "interfaces": interface_ips}
1593        return {"ip_version": ip_version, "interfaces": InterfaceChoice.All}
1594
1595    # Default mode: use InterfaceChoice.Default for IPv4-only single-interface,
1596    # otherwise pass explicit interface list for reliability.
1597    if ip_version == IPVersion.V4Only:
1598        return {"ip_version": ip_version, "interfaces": InterfaceChoice.Default}
1599    if interface_ips:
1600        return {"ip_version": ip_version, "interfaces": interface_ips}
1601    return {"ip_version": ip_version, "interfaces": InterfaceChoice.All}
1602
1603
1604async def close_async_generator(agen: AsyncGenerator[Any]) -> None:
1605    """Force close an async generator."""
1606    task = asyncio.create_task(agen.__anext__())
1607    task.cancel()
1608    with suppress(asyncio.CancelledError, StopAsyncIteration):
1609        await task
1610    await agen.aclose()
1611
1612
1613async def detect_charset(data: bytes, fallback: str = "utf-8", preferred: str | None = None) -> str:
1614    """
1615    Detect the charset to decode the given raw text with.
1616
1617    :param data: The raw text bytes to inspect.
1618    :param fallback: Charset to return when the charset can not be determined.
1619    :param preferred: Charset declared by the source, taken over detection when usable.
1620    """
1621    # a BOM outranks the declared charset: it names the very same UTF-8 but, unlike
1622    # the declared name, also gets the marker itself stripped off the decoded text
1623    if data.startswith(codecs.BOM_UTF8):
1624        return "utf-8-sig"
1625
1626    if preferred:
1627        # a declared charset is only worth anything if Python can actually decode text with
1628        # it: servers do send misspelled or plain made-up names in their Content-Type, and a
1629        # handful of names that do resolve to a codec still cannot decode text (base64, idna)
1630        try:
1631            data[:16].decode(preferred, errors="replace")
1632        except (LookupError, ValueError) as err:
1633            LOGGER.debug("Ignoring unusable charset %s: %s", preferred, err)
1634        else:
1635            return preferred
1636
1637    try:
1638        data.decode()
1639    except UnicodeDecodeError:
1640        pass
1641    else:
1642        # valid UTF-8 is never a legacy charset by accident, so skip detection
1643        return "utf-8"
1644
1645    # imported here to keep the detector out of the idle import footprint:
1646    # it is only needed for the rare text that is not UTF-8
1647    import chardet  # noqa: PLC0415
1648    from chardet.enums import EncodingEra  # noqa: PLC0415
1649
1650    # the reported confidence is deliberately not gated on: CUE sheets and playlists
1651    # are nearly all ASCII keywords, which holds the score far below any usable
1652    # threshold even though the charset itself is named correctly (support #6093).
1653    # With no score to weigh them against, DOS and mainframe codepages are dropped from
1654    # the candidates so a stray weak match cannot outrank the Windows codepage these
1655    # files are really written in. Only a superset is guaranteed to decode the bytes
1656    # past the window the detector samples, so it wins ties over its subsets.
1657    try:
1658        detected = await asyncio.to_thread(
1659            chardet.detect,
1660            data,
1661            encoding_era=EncodingEra.ALL & ~(EncodingEra.DOS | EncodingEra.MAINFRAME),
1662            prefer_superset=True,
1663            no_match_encoding=fallback,
1664        )
1665    except Exception as err:
1666        LOGGER.debug("Failed to detect charset: %s", err)
1667        return fallback
1668    if not (encoding := detected["encoding"]):
1669        return fallback
1670    LOGGER.debug("Detected charset %s (confidence %.2f)", encoding, detected["confidence"])
1671    return encoding
1672
1673
1674def parse_optional_bool(value: Any) -> bool | None:
1675    """Parse an optional boolean value from various input types."""
1676    if value is None:
1677        return None
1678    if isinstance(value, bool):
1679        return value
1680    if isinstance(value, str):
1681        value_lower = value.strip().lower()
1682        if value_lower in ("true", "1", "yes", "on"):
1683            return True
1684        if value_lower in ("false", "0", "no", "off"):
1685            return False
1686    if isinstance(value, (int, float)):
1687        return bool(value)
1688    return None
1689
1690
1691def merge_dict(
1692    base_dict: dict[Any, Any],
1693    new_dict: dict[Any, Any],
1694    allow_overwite: bool = False,
1695) -> dict[Any, Any]:
1696    """Merge dict without overwriting existing values."""
1697    final_dict = base_dict.copy()
1698    for key, value in new_dict.items():
1699        if final_dict.get(key) and isinstance(value, dict):
1700            final_dict[key] = merge_dict(final_dict[key], value)
1701        if final_dict.get(key) and isinstance(value, tuple):
1702            final_dict[key] = merge_tuples(final_dict[key], value)
1703        if final_dict.get(key) and isinstance(value, list):
1704            final_dict[key] = merge_lists(final_dict[key], value)
1705        elif not final_dict.get(key) or allow_overwite:
1706            final_dict[key] = value
1707    return final_dict
1708
1709
1710def merge_tuples(base: tuple[Any, ...], new: tuple[Any, ...]) -> tuple[Any, ...]:
1711    """Merge 2 tuples."""
1712    return tuple(x for x in base if x not in new) + tuple(new)
1713
1714
1715def merge_lists(base: list[Any], new: list[Any]) -> list[Any]:
1716    """Merge 2 lists."""
1717    return [x for x in base if x not in new] + list(new)
1718
1719
1720def percentage(part: float, whole: float) -> int:
1721    """Calculate percentage."""
1722    return int(100 * float(part) / float(whole))
1723
1724
1725def validate_announcement_chime_url(url: str) -> bool:
1726    """Validate announcement chime URL format."""
1727    if not url or not url.strip():
1728        return True  # Empty URL is valid
1729
1730    if url == ANNOUNCE_ALERT_FILE:
1731        return True  # Built-in chime file is valid
1732
1733    try:
1734        parsed = urlparse(url.strip())
1735
1736        if parsed.scheme not in ("http", "https"):
1737            return False
1738
1739        if not parsed.netloc:
1740            return False
1741
1742        path_lower = parsed.path.lower()
1743        audio_extensions = (".mp3", ".wav", ".flac", ".ogg", ".m4a", ".aac")
1744
1745        return any(path_lower.endswith(ext) for ext in audio_extensions)
1746
1747    except Exception:
1748        return False
1749
1750
1751async def get_mac_address(ip_address: str) -> str | None:
1752    """Get MAC address for given IP address via ARP lookup."""
1753    try:
1754        from getmac import get_mac_address as getmac_lookup  # noqa: PLC0415
1755
1756        return await asyncio.to_thread(getmac_lookup, ip=ip_address)
1757    except ImportError:
1758        LOGGER.debug("getmac module not available, cannot resolve MAC from IP")
1759        return None
1760    except Exception as err:
1761        LOGGER.debug("Failed to resolve MAC address for %s: %s", ip_address, err)
1762        return None
1763
1764
1765def is_locally_administered_mac(mac_address: str) -> bool:
1766    """
1767    Check if a MAC address is locally administered (virtual/randomized).
1768
1769    Locally administered addresses have bit 1 of the first octet set to 1.
1770    These are often used by devices for virtual interfaces or protocol-specific
1771    addresses (e.g., AirPlay, DLNA may use different virtual MACs than the real hardware MAC).
1772
1773    :param mac_address: MAC address in any common format (with :, -, or no separator).
1774    :return: True if locally administered, False if globally unique (real hardware MAC).
1775    """
1776    # Normalize MAC address
1777    mac_clean = mac_address.upper().replace(":", "").replace("-", "")
1778    if len(mac_clean) < 2:
1779        return False
1780
1781    # Get first octet and check bit 1 (second bit from right)
1782    try:
1783        first_octet = int(mac_clean[:2], 16)
1784        return bool(first_octet & 0x02)
1785    except ValueError:
1786        return False
1787
1788
1789def normalize_mac_for_matching(mac_address: str) -> str:
1790    """
1791    Normalize a MAC address for device matching by masking out the locally-administered bit.
1792
1793    Some protocols (like AirPlay) report a locally-administered MAC address variant where
1794    bit 1 of the first octet is set. For example:
1795    - Real hardware MAC: 54:78:C9:E6:0D:A0 (first byte 0x54 = 01010100)
1796    - AirPlay reports:   56:78:C9:E6:0D:A0 (first byte 0x56 = 01010110)
1797
1798    These represent the same device but differ only in the locally-administered bit.
1799    This function normalizes the MAC by clearing bit 1 of the first octet, allowing
1800    both variants to match the same device.
1801
1802    :param mac_address: MAC address in any common format (with :, -, or no separator).
1803    :return: Normalized MAC address in lowercase without separators, with the
1804             locally-administered bit cleared.
1805    """
1806    # Normalize MAC address (remove separators, lowercase)
1807    mac_clean = mac_address.lower().replace(":", "").replace("-", "")
1808    if len(mac_clean) != 12:
1809        # Invalid MAC length, return as-is
1810        return mac_clean
1811
1812    try:
1813        # Parse first octet and clear bit 1 (the locally-administered bit)
1814        first_octet = int(mac_clean[:2], 16)
1815        first_octet_normalized = first_octet & ~0x02  # Clear bit 1
1816        # Reconstruct the MAC with the normalized first octet
1817        return f"{first_octet_normalized:02x}{mac_clean[2:]}"
1818    except ValueError:
1819        # Invalid hex, return as-is
1820        return mac_clean
1821
1822
1823def is_valid_mac_address(mac_address: str | None) -> bool:
1824    """
1825    Check if a MAC address is valid and usable for device identification.
1826
1827    Invalid MAC addresses include:
1828    - None or empty strings
1829    - Null MAC: 00:00:00:00:00:00
1830    - Broadcast MAC: ff:ff:ff:ff:ff:ff
1831    - Any MAC that doesn't follow the expected pattern
1832
1833    :param mac_address: MAC address to validate.
1834    :return: True if valid and usable, False otherwise.
1835    """
1836    if not mac_address:
1837        return False
1838
1839    # Normalize MAC address (remove separators and convert to lowercase)
1840    normalized = mac_address.lower().replace(":", "").replace("-", "")
1841
1842    # Check for invalid/reserved MAC addresses
1843    if normalized in ("000000000000", "ffffffffffff"):
1844        return False
1845
1846    # Check length and hex validity
1847    if len(normalized) != 12:
1848        return False
1849
1850    try:
1851        int(normalized, 16)
1852        return True
1853    except ValueError:
1854        return False
1855
1856
1857def normalize_ip_address(ip_address: str | None) -> str | None:
1858    """
1859    Normalize IP address for comparison.
1860
1861    Handles IPv6-mapped IPv4 addresses (e.g., ::ffff:192.168.1.64 -> 192.168.1.64).
1862
1863    :param ip_address: IP address to normalize.
1864    :return: Normalized IP address or None if invalid.
1865    """
1866    if not ip_address:
1867        return None
1868
1869    # Handle IPv6-mapped IPv4 addresses
1870    if ip_address.startswith("::ffff:"):
1871        # Extract the IPv4 part
1872        return ip_address[7:]
1873
1874    return ip_address
1875
1876
1877async def resolve_real_mac_address(reported_mac: str | None, ip_address: str | None) -> str | None:
1878    """
1879    Resolve the real MAC address for a device.
1880
1881    Some devices report different virtual MAC addresses per protocol (AirPlay, DLNA,
1882    Chromecast). This function tries to resolve the actual hardware MAC via ARP
1883    when the reported MAC appears to be locally administered (virtual).
1884
1885    :param reported_mac: The MAC address reported by the protocol.
1886    :param ip_address: The IP address of the device (for ARP lookup).
1887    :return: The real MAC address if found, or None if it couldn't be resolved.
1888    """
1889    if not ip_address:
1890        return None
1891
1892    # If no MAC reported or it's a locally administered one, try ARP lookup
1893    if not reported_mac or is_locally_administered_mac(reported_mac):
1894        real_mac = await get_mac_address(ip_address)
1895        if real_mac and is_valid_mac_address(real_mac):
1896            return real_mac.upper()
1897
1898    return None
1899
1900
1901async def enrich_device_mac_address(
1902    device_info: DeviceInfo,
1903    logger: logging.Logger | None = None,
1904) -> None:
1905    """
1906    Enrich a player's device_info with a real MAC address via ARP.
1907
1908    Called automatically during player registration. It validates the existing MAC,
1909    normalizes IPv6-mapped IPv4 addresses, and always performs an ARP lookup when
1910    an IP is available. The ARP result replaces the reported MAC because it reflects
1911    the true hardware address and reliably unifies protocols on the same device -
1912    even when different protocols report different valid MACs (e.g., Yamaha devices
1913    where DLNA and AirPlay MACs differ by 1 in the last octet).
1914
1915    :param device_info: The player's DeviceInfo to enrich in-place.
1916    :param logger: Optional logger for debug messages.
1917    """
1918    identifiers = device_info.identifiers
1919    reported_mac = identifiers.get(IdentifierType.MAC_ADDRESS)
1920    ip_address = identifiers.get(IdentifierType.IP_ADDRESS)
1921
1922    # Blank out invalid MAC addresses (00:00:00:00:00:00, ff:ff:ff:ff:ff:ff, etc.)
1923    # so they can't cause false matches in protocol linking.
1924    if reported_mac and not is_valid_mac_address(reported_mac):
1925        if logger:
1926            logger.debug("Removing invalid MAC address: %s", reported_mac)
1927        device_info.add_identifier(IdentifierType.MAC_ADDRESS, None)
1928        reported_mac = None
1929
1930    # Normalize IP address (handle IPv6-mapped IPv4 like ::ffff:192.168.1.64)
1931    if ip_address:
1932        normalized_ip = normalize_ip_address(ip_address)
1933        if normalized_ip and normalized_ip != ip_address:
1934            device_info.add_identifier(IdentifierType.IP_ADDRESS, normalized_ip)
1935            if logger:
1936                logger.debug(
1937                    "Normalized IP address: %s -> %s",
1938                    ip_address,
1939                    normalized_ip,
1940                )
1941            ip_address = normalized_ip
1942
1943    # Skip ARP enrichment if no IP available (can't do ARP lookup)
1944    if not ip_address:
1945        return
1946
1947    # Always attempt ARP lookup when we have an IP address.
1948    # Some devices (e.g., Yamaha MusicCast) report different valid globally-unique
1949    # MACs per protocol (DLNA vs AirPlay differ by 1 in the last octet).
1950    # ARP resolves the true hardware MAC which reliably unifies all protocols.
1951    # The result is cached in player config so subsequent restarts are fast.
1952    real_mac = await resolve_real_mac_address(reported_mac, ip_address)
1953    if real_mac and real_mac.upper() != (reported_mac or "").upper():
1954        device_info.add_identifier(IdentifierType.MAC_ADDRESS, real_mac)
1955        if logger:
1956            logger.debug(
1957                "Resolved MAC via ARP: %s -> %s",
1958                reported_mac or "none",
1959                real_mac,
1960            )
1961    elif not reported_mac:
1962        # ARP failed and no reported MAC - nothing we can do
1963        if logger:
1964            logger.debug("ARP lookup failed for %s and no reported MAC", ip_address)
1965
1966
1967class TaskManager:
1968    """
1969    Helper class to run many tasks at once.
1970
1971    This is basically an alternative to asyncio.TaskGroup but this will not
1972    cancel all operations when one of the tasks fails.
1973    Logging of exceptions is done by the mass.create_task helper.
1974    """
1975
1976    def __init__(self, mass: MusicAssistant, limit: int = 0):
1977        """Initialize the TaskManager."""
1978        self.mass = mass
1979        self._tasks: list[asyncio.Task[None]] = []
1980        self._semaphore = asyncio.Semaphore(limit) if limit else None
1981
1982    def create_task(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[None]:
1983        """Create a new task and add it to the manager."""
1984        task = self.mass.create_task(coro)
1985        self._tasks.append(task)
1986        return task
1987
1988    async def create_task_with_limit(self, coro: Coroutine[Any, Any, Any]) -> None:
1989        """Create a new task with semaphore limit."""
1990        assert self._semaphore is not None
1991
1992        def task_done_callback(_task: asyncio.Task[None]) -> None:
1993            assert self._semaphore is not None  # for type checking
1994            self._tasks.remove(task)
1995            self._semaphore.release()
1996
1997        await self._semaphore.acquire()
1998        task: asyncio.Task[None] = self.create_task(coro)
1999        task.add_done_callback(task_done_callback)
2000
2001    async def __aenter__(self) -> Self:
2002        """Enter context manager."""
2003        return self
2004
2005    async def __aexit__(
2006        self,
2007        exc_type: type[BaseException] | None,
2008        exc_val: BaseException | None,
2009        exc_tb: TracebackType | None,
2010    ) -> bool | None:
2011        """Exit context manager."""
2012        if len(self._tasks) > 0:
2013            await asyncio.wait(self._tasks)
2014            self._tasks.clear()
2015        return None
2016
2017
2018_R = TypeVar("_R")
2019_P = ParamSpec("_P")
2020
2021
2022def lock[**P, R](  # type: ignore[valid-type]
2023    func: Callable[_P, Awaitable[_R]],
2024) -> Callable[_P, Coroutine[Any, Any, _R]]:
2025    """
2026    Call async function using a per-instance Lock.
2027
2028    Each instance gets its own lock so that e.g. SyncGroupPlayer A
2029    does not block SyncGroupPlayer B when both call set_members().
2030    """
2031    # Per-instance lock storage (weak refs so locks are GC'd with their instance)
2032    instance_locks: weakref.WeakKeyDictionary[Any, asyncio.Lock] = weakref.WeakKeyDictionary()
2033    # Fallback lock for non-method (no self) usage
2034    fallback_lock: asyncio.Lock | None = None
2035
2036    @functools.wraps(func)
2037    async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
2038        """Call async function using a per-instance Lock."""
2039        nonlocal fallback_lock
2040        instance = args[0] if args else None
2041        if instance is not None:
2042            try:
2043                func_lock = instance_locks.setdefault(instance, asyncio.Lock())
2044            except TypeError:
2045                # instance is not weakly referenceable, use fallback
2046                if fallback_lock is None:
2047                    fallback_lock = asyncio.Lock()
2048                func_lock = fallback_lock
2049        else:
2050            if fallback_lock is None:
2051                fallback_lock = asyncio.Lock()
2052            func_lock = fallback_lock
2053        async with func_lock:
2054            return await func(*args, **kwargs)
2055
2056    return wrapper
2057
2058
2059class TimedAsyncGenerator:
2060    """
2061    Async iterable that times out after a given time.
2062
2063    Source: https://medium.com/@dmitry8912/implementing-timeouts-in-pythons-asynchronous-generators-f7cbaa6dc1e9
2064    """
2065
2066    def __init__(self, iterable: AsyncIterator[Any], timeout: int = 0):
2067        """
2068        Initialize the AsyncTimedIterable.
2069
2070        Args:
2071            iterable: The async iterable to wrap.
2072            timeout: The timeout in seconds for each iteration.
2073        """
2074
2075        class AsyncTimedIterator:
2076            def __init__(self) -> None:
2077                self._iterator = iterable.__aiter__()
2078
2079            async def __anext__(self) -> Any:
2080                result = await asyncio.wait_for(self._iterator.__anext__(), int(timeout))
2081                if not result:
2082                    raise StopAsyncIteration
2083                return result
2084
2085        self._factory = AsyncTimedIterator
2086
2087    def __aiter__(self):  # type: ignore[no-untyped-def]
2088        """Return the async iterator."""
2089        return self._factory()
2090
2091
2092async def join_task[T](task: asyncio.Future[T], timeout: float | None = None) -> T:
2093    """
2094    Wait for a task started elsewhere and return its result.
2095
2096    Cancelling the waiter leaves the task running, so work that is shared between callers -
2097    or that must outlive a caller's deadline - keeps going and still reaches every other
2098    waiter. A task that can lose all its waiters needs a done callback that retrieves its
2099    exception (as mass.create_task installs) to keep asyncio quiet about it.
2100
2101    :param task: The task (or future) to wait for.
2102    :param timeout: Optional number of seconds to wait before giving up.
2103    :raises TimeoutError: If the task did not complete within the timeout.
2104    :raises asyncio.CancelledError: If the task itself was cancelled.
2105    :return: The task's result.
2106    """
2107    if not task.done():
2108        # awaiting the task directly would hold it as this coroutine's fut_waiter, so
2109        # cancelling the waiter would cancel the task itself. asyncio.shield achieves the
2110        # same isolation, but as of Python 3.14 a cancelled waiter makes it report the task's
2111        # exception through loop.call_exception_handler, even when another waiter already
2112        # handled it.
2113        await asyncio.wait((task,), timeout=timeout)
2114    if not task.done():
2115        raise TimeoutError
2116    return task.result()
2117
2118
2119# Bound for guard_single_request: it only needs ``.mass``, so a structural protocol
2120# lets it decorate providers, core controllers and media controllers alike without
2121# coupling to their concrete base classes.
2122class _SupportsMass(Protocol):
2123    """Structural type for objects exposing a MusicAssistant reference."""
2124
2125    mass: MusicAssistant
2126
2127
2128def guard_single_request[SelfT: _SupportsMass, **P, R](
2129    func: Callable[Concatenate[SelfT, P], Coroutine[Any, Any, R]],
2130) -> Callable[Concatenate[SelfT, P], Coroutine[Any, Any, R]]:
2131    """
2132    Ensure concurrent calls with identical arguments result in a single request.
2133
2134    Callers arriving while an identical call is already in flight await that same call and
2135    receive its result. Cancelling one caller leaves both the request and the other callers
2136    unaffected. Calls count as identical when they are made on the same object with equal
2137    arguments, no matter whether those were passed positionally or by keyword; the request
2138    runs with the arguments of the caller that started it.
2139
2140    Every argument must be a scalar or an object identified by its ``uri``, so that equal
2141    arguments are guaranteed to produce an equal key.
2142
2143    :param func: The coroutine method to guard.
2144    """
2145    signature = inspect.signature(func)
2146
2147    @functools.wraps(func)
2148    async def wrapper(self: SelfT, *args: P.args, **kwargs: P.kwargs) -> R:
2149        mass = self.mass
2150        # create a task_id dynamically based on the bound method and args/kwargs.
2151        # the instance is part of the key because a decorated method may be inherited by
2152        # multiple subclasses (all media controllers share
2153        # MediaControllerBase.get_provider_item) and a class may have multiple instances
2154        # (e.g. a provider set up twice), which must never join each other's flight.
2155        # id(self) is stable while a flight is live because the task references self;
2156        # the class name only serves to keep the task_id readable while debugging.
2157        # binding the arguments to their parameter names and filling in the defaults keys a
2158        # call the same however it was spelled; repr of the resulting tuple keeps the parts
2159        # apart, so an id that itself contains punctuation cannot run into the next one.
2160        bound = signature.bind(self, *args, **kwargs)
2161        bound.apply_defaults()
2162        task_id = repr(
2163            (
2164                type(self).__name__,
2165                id(self),
2166                func.__qualname__,
2167                # skip the instance: it is the first parameter and is keyed by id() above
2168                *(
2169                    (name, _canonical_key_part(value))
2170                    for name, value in islice(bound.arguments.items(), 1, None)
2171                ),
2172            )
2173        )
2174        task: asyncio.Task[R] = mass.create_task(
2175            func,
2176            self,
2177            *args,
2178            task_id=task_id,
2179            abort_existing=False,
2180            eager_start=True,
2181            # every caller awaits the flight below and so sees the failure itself; the
2182            # task's own exception log would report a handled error as an unhandled one
2183            log_exceptions=False,
2184            **kwargs,
2185        )
2186        return await join_task(task)
2187
2188    return wrapper
2189
2190
2191def _canonical_key_part(value: Any) -> Any:
2192    """Return a stable stand-in for a single argument of a guarded request."""
2193    if (uri := getattr(value, "uri", None)) is not None:
2194        # a media item renders as a multi-kilobyte dataclass repr in which the set-typed
2195        # fields (provider_mappings, external_ids) can iterate in different orders for two
2196        # equal items. the uri identifies the item, and the type travels with it because a
2197        # full item and an ItemMapping for that same item are not handled the same.
2198        return (type(value).__name__, uri)
2199    return value
2200