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
685def parse_title_and_version(
686    title: str,
687    track_version: str | None = None,
688    strip_for_search: bool = False,
689    strip_for_display: bool = False,
690) -> tuple[str, str]:
691    """
692    Parse version from the title and optionally clean for search or display.
693
694    :param title: The title to parse.
695    :param track_version: Optional existing version string.
696    :param strip_for_search: Aggressively strip for search matching.
697    :param strip_for_display: Strip superfluous suffixes for display.
698    """
699    version = track_version or ""
700
701    # Strip featuring, bracketed version info, and hyphen suffixes (e.g. "- Remastered 2019")
702    if strip_for_search:
703        title = _SEARCH_PAREN_PATTERN.sub("", title)
704        title = _SEARCH_HYPHEN_PATTERN.sub("", title)
705        # Strip bare featuring credits (not in parentheses)
706        title_lower = title.lower()
707        for pattern in _FEATURING_PATTERNS:
708            if pattern in title_lower:
709                idx = title_lower.find(pattern)
710                title = title[:idx]
711                break
712        # Clean up dangling hyphens and extra spaces
713        title = re.sub(r"\s*-\s*$", "", title)
714        title = re.sub(r"\s+", " ", title).strip()
715        return title, version
716
717    # Strip video/audio suffixes like "(Official Video)"
718    if strip_for_display:
719        title = _DISPLAY_STRIP_PATTERN.sub("", title).strip()
720        return title, version
721
722    # Standard version parsing
723    for parts in (
724        _balanced_bracket_groups(title, "(", ")"),
725        _balanced_bracket_groups(title, "[", "]"),
726        re.findall(r" - .*", title),
727    ):
728        for title_part in parts:
729            # Extract the content without brackets/dashes for checking
730            clean_part = title_part.translate(str.maketrans("", "", "()[]-")).strip().lower()
731
732            # Check if this should be ignored (featuring/explicit parts)
733            should_ignore = False
734            for ignore_str in IGNORE_TITLE_PARTS:
735                if clean_part.startswith(ignore_str):
736                    # Special handling for "with " - check if followed by title words
737                    if ignore_str == "with ":
738                        # Extract the word after "with "
739                        after_with = (
740                            clean_part[len("with ") :].split()[0]
741                            if len(clean_part) > len("with ")
742                            else ""
743                        )
744                        if after_with in WITH_TITLE_WORDS:
745                            # This is part of the title (e.g., "with you"), don't ignore
746                            break
747                    # Remove this part from the title
748                    title = title.replace(title_part, "").strip()
749                    should_ignore = True
750                    break
751
752            if should_ignore:
753                continue
754
755            # Check if this part is a version
756            for version_str in VERSION_PARTS:
757                if version_str in clean_part:
758                    # Preserve original casing (and any nested brackets) for output
759                    version = _strip_outer_markers(title_part)
760                    title = title.replace(title_part, "").strip()
761                    return title, version
762    return title, version
763
764
765def _balanced_bracket_groups(text: str, open_char: str, close_char: str) -> list[str]:
766    """
767    Return the top-level balanced bracketed substrings, including the outer brackets.
768
769    :param text: The text to scan.
770    :param open_char: The opening bracket character.
771    :param close_char: The closing bracket character.
772    """
773    groups: list[str] = []
774    depth = 0
775    start = -1
776    for idx, char in enumerate(text):
777        if char == open_char:
778            if depth == 0:
779                start = idx
780            depth += 1
781        elif char == close_char and depth > 0:
782            depth -= 1
783            if depth == 0:
784                groups.append(text[start : idx + 1])
785    return groups
786
787
788def _strip_outer_markers(part: str) -> str:
789    """
790    Strip the outer brackets or leading hyphen from a parsed title part.
791
792    :param part: The raw title part as matched from the title.
793    """
794    part = part.strip()
795    # only strip a single outer bracket pair so nested brackets stay intact
796    if part[:1] in "([" and part[-1:] in ")]":
797        return part[1:-1].strip()
798    return part.lstrip("- ").strip()
799
800
801def infer_album_type(title: str, version: str) -> AlbumType:
802    """Infer album type by looking for live or soundtrack indicators."""
803    combined = f"{title} {version}".lower()
804    for pat in LIVE_INDICATORS:
805        if re.search(pat, combined):
806            return AlbumType.LIVE
807    for pat in SOUNDTRACK_INDICATORS:
808        if re.search(pat, combined):
809            return AlbumType.SOUNDTRACK
810    return AlbumType.UNKNOWN
811
812
813def strip_ads(line: str) -> str:
814    """Strip Ads from line."""
815    if ad_pattern.search(line):
816        return "Advert"
817    return line
818
819
820def strip_url(line: str) -> str:
821    """Strip URL from line."""
822    return (
823        " ".join([p for p in line.split() if (not urlparse(p).scheme or not urlparse(p).netloc)])
824    ).rstrip()
825
826
827def strip_dotcom(line: str) -> str:
828    """Strip scheme-less netloc from line."""
829    return dot_com_pattern.sub("", line)
830
831
832def strip_end_junk(line: str) -> str:
833    """Strip non-word info from end of line."""
834    return end_junk_pattern.sub(r"\1", line)
835
836
837def swap_title_artist_order(line: str) -> str:
838    """Swap title/artist order in line."""
839    return title_artist_order_pattern.sub(r"\g<artist> - \g<title>", line)
840
841
842def strip_multi_space(line: str) -> str:
843    """Strip multi-whitespace from line."""
844    return multi_space_pattern.sub(" ", line)
845
846
847def html_to_markdown(line: str) -> str:
848    """Convert the safe subset of HTML in a string to markdown, stripping other tags."""
849    # unescape first so entity-encoded markup (e.g. "&lt;p&gt;") is handled too
850    return markdownify(
851        html.unescape(line),
852        convert=MARKDOWN_SAFE_TAGS,
853        escape_asterisks=False,
854        escape_underscores=False,
855        escape_misc=False,
856    ).strip()
857
858
859def multi_strip(line: str) -> str:
860    """Strip assorted junk from line."""
861    return strip_multi_space(
862        swap_title_artist_order(strip_end_junk(strip_dotcom(strip_url(strip_ads(line)))))
863    ).rstrip()
864
865
866def parse_quoted_stream_title(line: str) -> tuple[str, str, str | None] | None:
867    """
868    Parse stream titles that name the track in natural language with a quoted title.
869
870    Recognises '"Track" by Artist from "Album"' (album optional) and the German
871    '"Track" von Artist'.
872
873    :param line: Raw (uncleaned) stream title.
874    :returns: Tuple of (title, artist, album), or None when the line is not in one of
875        these formats. ``album`` is None when the station omits it.
876    """
877    stripped = line.strip()
878    if match := english_by_pattern.match(stripped):
879        title = multi_strip(match.group("title"))
880        artist = multi_strip(match.group("artist")).strip('"')
881        album_raw = match.group("album")
882        album = multi_strip(album_raw).strip('"') if album_raw else None
883        if title and artist:
884            return title, artist, album or None
885    if match := german_von_pattern.match(stripped):
886        title = multi_strip(match.group("title"))
887        artist = multi_strip(match.group("artist")).strip('"')
888        if title and artist:
889            return title, artist, None
890    return None
891
892
893def clean_stream_title(line: str) -> str:
894    """Strip junk text from radio streamtitle."""
895    title: str = ""
896    artist: str = ""
897
898    if not keyword_pattern.search(line):
899        if parsed := parse_quoted_stream_title(line):
900            track_name, artist_name, _ = parsed
901            return f"{artist_name} - {track_name}"
902        return multi_strip(line)
903
904    if match := title_pattern.search(line):
905        title = multi_strip(match.group("title"))
906
907    if match := artist_pattern.search(line):
908        possible_artist = multi_strip(match.group("artist"))
909        if possible_artist and possible_artist != title:
910            artist = possible_artist
911
912    if not title and not artist:
913        return ""
914
915    if title:
916        if re.search(" - ", title) or not artist:
917            return title
918        if artist:
919            return f"{artist} - {title}"
920
921    if artist:
922        return artist
923
924    return line
925
926
927# cache for get_ip_addresses: enumerating the network adapters involves a thread hop,
928# socket probes and a full adapter walk, while the result rarely (if ever) changes
929IP_ADDRESSES_CACHE_TTL = 30
930_ip_addresses_cache: dict[tuple[bool, bool], tuple[float, tuple[str, ...]]] = {}
931_ip_addresses_pending: dict[tuple[bool, bool], asyncio.Task[tuple[str, ...]]] = {}
932
933# Interfaces that only ever carry container, VM or VPN traffic, so a device on the local
934# network can never reach us on their addresses.
935_VIRTUAL_INTERFACE_PREFIXES = (
936    "cali",
937    "cni",
938    "docker",
939    "flannel",
940    "hassio",
941    "incusbr",
942    "lxcbr",
943    "lxdbr",
944    "nordlynx",
945    "podman",
946    "ppp",
947    "tailscale",
948    "tap",
949    "tun",
950    "utun",
951    "vboxnet",
952    "veth",
953    "virbr",
954    "vmnet",
955    "wg",
956    "zt",
957)
958# Docker names its user-defined bridges br-<12 hex> and the macOS host-only bridges of
959# Docker Desktop, Parallels and VMware start at bridge100. Both are matched in full, so a
960# hand-named LAN bridge (br-lan on OpenWrt, a second macOS bridge1) is left alone - as are
961# the regular LAN bridge names br0, vmbr0 and bond0.
962_VIRTUAL_INTERFACE_NAMES = re.compile(r"br-[0-9a-f]{12}|bridge\d{3}")
963
964
965async def get_ip_addresses(include_ipv6: bool = False) -> tuple[str, ...]:
966    """
967    Return all IP addresses of all network interfaces.
968
969    Always returns at least one address: when no routable address is found
970    (e.g. offline host), the loopback address is returned as fallback.
971    Results are cached for a short while, so an IP/interface change may take up to
972    IP_ADDRESSES_CACHE_TTL seconds to be reflected.
973
974    :param include_ipv6: Whether to include IPv6 addresses in the result.
975    """
976    return await _get_ip_addresses(include_ipv6, publish_candidates_only=False)
977
978
979async def get_publish_ip_candidates(include_ipv6: bool = False) -> tuple[str, ...]:
980    """
981    Return the IP addresses a device on the local network may reach this host on.
982
983    Same as get_ip_addresses, minus the addresses of container, VM and VPN interfaces -
984    unless the host holds no other address at all.
985
986    :param include_ipv6: Whether to include IPv6 addresses in the result.
987    """
988    return await _get_ip_addresses(include_ipv6, publish_candidates_only=True)
989
990
991async def _get_ip_addresses(include_ipv6: bool, publish_candidates_only: bool) -> tuple[str, ...]:
992    """Return the host's IP addresses, enumerating the adapters at most once per TTL."""
993    cache_key = (include_ipv6, publish_candidates_only)
994    if cached := _ip_addresses_cache.get(cache_key):
995        cached_at, addresses = cached
996        if (time.monotonic() - cached_at) < IP_ADDRESSES_CACHE_TTL:
997            return addresses
998
999    async def _probe() -> tuple[str, ...]:
1000        try:
1001            addresses = await asyncio.to_thread(
1002                _enumerate_ip_addresses, include_ipv6, publish_candidates_only
1003            )
1004            _ip_addresses_cache[cache_key] = (time.monotonic(), addresses)
1005            return addresses
1006        finally:
1007            _ip_addresses_pending.pop(cache_key, None)
1008
1009    # single-flight: no await between the pending-check and storing the task,
1010    # so concurrent callers always end up awaiting the same probe
1011    if not (pending := _ip_addresses_pending.get(cache_key)):
1012        pending = asyncio.create_task(_probe())
1013        pending.add_done_callback(_log_ip_probe_failure)
1014        _ip_addresses_pending[cache_key] = pending
1015    return await join_task(pending)
1016
1017
1018def _log_ip_probe_failure(probe: asyncio.Task[tuple[str, ...]]) -> None:
1019    """Log (and thereby retrieve) the exception of a finished address probe, if any."""
1020    if probe.cancelled():
1021        return
1022    # every waiter that is still around reports the failure itself, so a debug line is
1023    # enough here; retrieving the exception is what keeps asyncio from reporting it as
1024    # "Task exception was never retrieved" once the probe is garbage collected
1025    if (err := probe.exception()) is not None:
1026        LOGGER.debug("Enumerating IP addresses failed: %s", err)
1027
1028
1029def _enumerate_ip_addresses(include_ipv6: bool, publish_candidates_only: bool) -> tuple[str, ...]:
1030    """Enumerate all IP addresses of all network interfaces (blocking)."""
1031    result: list[tuple[int, str]] = []
1032    # the same addresses, without the ones no device on the local network can reach
1033    lan_result: list[tuple[int, str]] = []
1034    # try to get the primary IP address
1035    # this is the IP address of the default route
1036    primary_ip = ""
1037    # try IPv4 first
1038    _sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
1039    _sock.settimeout(0)
1040    try:
1041        # doesn't even have to be reachable
1042        _sock.connect(("10.254.254.254", 1))
1043        primary_ip = _sock.getsockname()[0]
1044    except Exception:
1045        primary_ip = ""
1046    finally:
1047        _sock.close()
1048    # fall back to IPv6 if no IPv4 primary found (e.g. IPv6-only networks)
1049    if not primary_ip:
1050        _sock6 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
1051        _sock6.settimeout(0)
1052        try:
1053            _sock6.connect(("2001:db8::1", 1))
1054            primary_ip = _sock6.getsockname()[0]
1055        except Exception:
1056            primary_ip = ""
1057        finally:
1058            _sock6.close()
1059    # get all IP addresses of all network interfaces
1060    adapters = ifaddr.get_adapters()
1061    for adapter in adapters:
1062        adapter_is_virtual = _is_virtual_interface(adapter.name) or _is_virtual_interface(
1063            adapter.nice_name
1064        )
1065        for ip in adapter.ips:
1066            if ip.is_IPv6 and not include_ipv6:
1067                continue
1068            # ifaddr returns IPv6 addresses as (address, flowinfo, scope_id) tuples
1069            ip_str = ip.ip[0] if isinstance(ip.ip, tuple) else ip.ip
1070            if ip_str.startswith(("127", "169.254")):
1071                # filter out IPv4 loopback/APIPA address
1072                continue
1073            if ip_str.startswith(("::1", "::ffff:", "fe80")):
1074                # filter out IPv6 loopback/link-local address
1075                continue
1076            if ip_str == primary_ip:
1077                score = 10
1078            elif ip_str.startswith(("192.168.",)):
1079                # we rank the 192.168 range a bit higher as its most
1080                # often used as the private network subnet
1081                score = 2
1082            elif ip_str.startswith(("172.", "10.", "192.")):
1083                # we rank the 172 range a bit lower as its most
1084                # often used as the private docker network
1085                score = 1
1086            else:
1087                score = 0
1088            result.append((score, ip_str))
1089            if not adapter_is_virtual:
1090                lan_result.append((score, ip_str))
1091    # a host that is only reachable over a tunnel or bridge still has to publish something
1092    selected = (lan_result or result) if publish_candidates_only else result
1093    selected.sort(key=lambda x: x[0], reverse=True)
1094    if not selected:
1095        # no routable addresses found (e.g. offline host with only loopback/link-local):
1096        # fall back to loopback so callers that rely on at least one address keep working
1097        return ("127.0.0.1",)
1098    return tuple(ip[1] for ip in selected)
1099
1100
1101def _is_virtual_interface(name: str) -> bool:
1102    """Return whether the named interface belongs to a container, VM or VPN network."""
1103    name = name.lower()
1104    return name.startswith(_VIRTUAL_INTERFACE_PREFIXES) or bool(
1105        _VIRTUAL_INTERFACE_NAMES.fullmatch(name)
1106    )
1107
1108
1109def interface_name_for_ip(ip: str) -> str | None:
1110    """
1111    Return the name of the network interface that holds the given IP, or None.
1112
1113    Used to map a bind/publish IP to its interface name for components that select
1114    their mDNS/zeroconf advertisement interface by name (e.g. shairport-sync and
1115    go-librespot), so the advertisement stays on the intended network.
1116
1117    :param ip: The IPv4/IPv6 address to look up.
1118    """
1119    for adapter in ifaddr.get_adapters():
1120        for ip_config in adapter.ips:
1121            addr = ip_config.ip if isinstance(ip_config.ip, str) else ip_config.ip[0]
1122            if addr == ip:
1123                return adapter.name
1124    return None
1125
1126
1127async def is_port_in_use(port: int, host: str | None = None) -> bool:
1128    """
1129    Check if a port is in use.
1130
1131    :param port: Port number to check.
1132    :param host: Optional bind address to probe. When omitted, both IPv4 and IPv6
1133        wildcard addresses are checked.
1134    """
1135
1136    def _is_port_in_use() -> bool:
1137        candidates: tuple[tuple[socket.AddressFamily, str], ...]
1138        if host is not None:
1139            candidates = ((socket.AF_INET6 if ":" in host else socket.AF_INET, host),)
1140        else:
1141            # Try both IPv4 and IPv6 to support single-stack and dual-stack systems.
1142            # A port is considered free if it can be bound on at least one address family.
1143            candidates = ((socket.AF_INET, "0.0.0.0"), (socket.AF_INET6, "::"))
1144        for family, addr in candidates:
1145            try:
1146                with socket.socket(family, socket.SOCK_STREAM) as _sock:
1147                    # Set SO_REUSEADDR to match asyncio.start_server behavior
1148                    # This allows binding to ports in TIME_WAIT state
1149                    _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1150                    _sock.bind((addr, port))
1151                    return False
1152            except OSError:
1153                continue
1154        return True
1155
1156    return await asyncio.to_thread(_is_port_in_use)
1157
1158
1159# In-process reservations for ports handed out by select_free_port. Provider
1160# instances (and reloads) frequently call select_free_port at nearly the same
1161# moment and only bind the returned port asynchronously afterwards, so a port
1162# that was just handed out is not yet detectable as "in use". Keeping a
1163# short-lived reservation per returned port stops concurrent/successive callers
1164# from picking the same one. Reservations expire automatically after the grace
1165# period so the range is never permanently exhausted across reloads.
1166_PORT_RESERVATION_TTL = 60.0
1167_reserved_ports: dict[int, float] = {}
1168_select_free_port_lock = asyncio.Lock()
1169
1170
1171async def select_free_port(range_start: int, range_end: int, host: str | None = None) -> int:
1172    """
1173    Find and reserve a free port within the given range.
1174
1175    The returned port is reserved so concurrent or successive callers are not
1176    handed the same port.
1177
1178    :param range_start: First port (inclusive) of the range to search.
1179    :param range_end: Port to stop before (exclusive) when searching the range.
1180    :param host: Optional bind address to probe for availability.
1181    """
1182    async with _select_free_port_lock:
1183        now = time.monotonic()
1184        # drop expired reservations so their ports become reusable again
1185        for reserved_port, deadline in list(_reserved_ports.items()):
1186            if deadline <= now:
1187                del _reserved_ports[reserved_port]
1188        for port in range(range_start, range_end):
1189            if port in _reserved_ports:
1190                continue
1191            if not await is_port_in_use(port, host=host):
1192                _reserved_ports[port] = now + _PORT_RESERVATION_TTL
1193                return port
1194    msg = f"No free port available in range {range_start}-{range_end - 1}"
1195    raise OSError(msg)
1196
1197
1198async def get_ip_from_host(dns_name: str) -> str | None:
1199    """Resolve (first) IP-address for given dns name."""
1200
1201    def _resolve() -> str | None:
1202        try:
1203            # use getaddrinfo to support both IPv4 and IPv6 resolution
1204            results = socket.getaddrinfo(dns_name, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
1205            if results:
1206                return str(results[0][4][0])
1207        except Exception:
1208            # fail gracefully!
1209            return None
1210        return None
1211
1212    return await asyncio.to_thread(_resolve)
1213
1214
1215async def get_source_ip_for_target(target_ip: str) -> str:
1216    """
1217    Return the local interface address the routing table would egress to ``target_ip`` from.
1218
1219    Empty when no route to the target can be determined.
1220
1221    :param target_ip: IP address of the device the traffic is meant for.
1222    """
1223
1224    def _routing_lookup() -> str:
1225        try:
1226            is_ipv6_target = ip_address(target_ip).version == 6
1227        except ValueError:
1228            is_ipv6_target = False
1229        route_family = socket.AF_INET6 if is_ipv6_target else socket.AF_INET
1230        route_target: tuple[str, int] | tuple[str, int, int, int] = (
1231            (target_ip, 80, 0, 0) if is_ipv6_target else (target_ip, 80)
1232        )
1233        with socket.socket(route_family, socket.SOCK_DGRAM) as _sock:
1234            try:
1235                _sock.settimeout(1.0)
1236                _sock.connect(route_target)
1237                routed_ip = str(_sock.getsockname()[0])
1238                if routed_ip and routed_ip not in WILDCARD_BIND_IPS:
1239                    return routed_ip
1240            except OSError:
1241                pass
1242        return ""
1243
1244    return await asyncio.to_thread(_routing_lookup)
1245
1246
1247async def get_ip_pton(ip_string: str) -> bytes:
1248    """Return socket pton for a local ip."""
1249    try:
1250        return await asyncio.to_thread(socket.inet_pton, socket.AF_INET, ip_string)
1251    except OSError:
1252        return await asyncio.to_thread(socket.inet_pton, socket.AF_INET6, ip_string)
1253
1254
1255def format_ip_for_url(ip_address: str) -> str:
1256    """Wrap IPv6 addresses in brackets for use in URLs (RFC 2732)."""
1257    if ":" in ip_address:
1258        return f"[{ip_address}]"
1259    return ip_address
1260
1261
1262async def get_folder_size(folderpath: str) -> float:
1263    """Return folder size in gb."""
1264
1265    def _get_folder_size(folderpath: str) -> float:
1266        total_size = 0
1267        for dirpath, _dirnames, filenames in os.walk(folderpath):
1268            for _file in filenames:
1269                _fp = os.path.join(dirpath, _file)
1270                total_size += Path(_fp).stat().st_size
1271        return total_size / float(1 << 30)
1272
1273    return await asyncio.to_thread(_get_folder_size, folderpath)
1274
1275
1276def get_changed_keys(
1277    dict1: dict[str, Any],
1278    dict2: dict[str, Any],
1279    recursive: bool = False,
1280) -> set[str]:
1281    """Compare 2 dicts and return set of changed keys."""
1282    # TODO: Check with Marcel whether we should calculate new dicts based on ignore_keys
1283    return set(get_changed_dict_values(dict1, dict2, recursive).keys())
1284    # return set(get_changed_dict_values(dict1, dict2, ignore_keys, recursive).keys())
1285
1286
1287def get_changed_dict_values(
1288    dict1: dict[str, Any],
1289    dict2: dict[str, Any],
1290    recursive: bool = False,
1291) -> dict[str, tuple[Any, Any]]:
1292    """
1293    Compare 2 dicts and return dict of changed values.
1294
1295    dict key is the changed key, value is tuple of old and new values.
1296    """
1297    if not dict1 and not dict2:
1298        return {}
1299    if not dict1:
1300        return {key: (None, value) for key, value in dict2.items()}
1301    if not dict2:
1302        return {key: (None, value) for key, value in dict1.items()}
1303    changed_values = {}
1304    for key, value in dict2.items():
1305        if isinstance(value, dict) and isinstance(dict1[key], dict) and recursive:
1306            changed_subvalues = get_changed_dict_values(dict1[key], value, recursive)
1307            for subkey, subvalue in changed_subvalues.items():
1308                changed_values[f"{key}.{subkey}"] = subvalue
1309            continue
1310        if key not in dict1:
1311            changed_values[key] = (None, value)
1312            continue
1313        if dict1[key] != value:
1314            changed_values[key] = (dict1[key], value)
1315    return changed_values
1316
1317
1318def empty_queue[T](q: asyncio.Queue[T]) -> None:
1319    """Empty an asyncio Queue."""
1320    for _ in range(q.qsize()):
1321        try:
1322            q.get_nowait()
1323            q.task_done()
1324        except asyncio.QueueEmpty, ValueError:
1325            pass
1326
1327
1328async def install_package(package: str) -> None:
1329    """Install package with pip, raise when install failed."""
1330    LOGGER.debug("Installing python package %s", package)
1331    args = ["uv", "pip", "install", "--no-cache", package]
1332    return_code, output = await check_output(*args)
1333    if return_code != 0:
1334        msg = f"Failed to install package {package}\n{output.decode()}"
1335        raise RuntimeError(msg)
1336
1337
1338async def get_package_version(pkg_name: str) -> str | None:
1339    """
1340    Return the version of an installed (python) package.
1341
1342    Will return None if the package is not found.
1343    """
1344    try:
1345        return await asyncio.to_thread(pkg_version, pkg_name)
1346    except PackageNotFoundError:
1347        return None
1348
1349
1350async def is_hass_supervisor() -> bool:
1351    """Return if we're running inside the HA Supervisor (e.g. HAOS)."""
1352    # Fast path: check for HA supervisor token environment variable
1353    # This is always set when running inside the HA supervisor
1354    if not os.environ.get("SUPERVISOR_TOKEN"):
1355        return False
1356
1357    # Token exists, verify the supervisor is actually reachable
1358    def _check() -> bool:
1359        try:
1360            urllib.request.urlopen("http://supervisor/core", timeout=1)
1361        except urllib.error.URLError as err:
1362            # this should return a 401 unauthorized if it exists
1363            return getattr(err, "code", 999) == 401
1364        except Exception:
1365            return False
1366        return False
1367
1368    return await asyncio.to_thread(_check)
1369
1370
1371# CPython holds a lock per module while importing it, so two threads importing modules with
1372# overlapping dependency graphs (e.g. two providers that both pull in `requests`) can end up
1373# waiting on each other's module locks. The import machinery then bails out at one of them with
1374# a _DeadlockError ("deadlock detected by _ModuleLock(...)") instead of hanging, which surfaces
1375# as a provider that failed to load and stays broken until it is reloaded by hand.
1376# A single-worker executor keeps imports serialized without parking a thread from the default
1377# pool while waiting; only the import itself is serialized, providers still load concurrently.
1378_IMPORT_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="module_import")
1379
1380# requirements verified this session, so repeated (config) loads skip the version check
1381_checked_requirements: set[str] = set()
1382
1383
1384async def import_module_in_thread(name: str, package: str | None = None) -> ModuleType:
1385    """
1386    Import a module in a thread, serialized against all other imports done this way.
1387
1388    :param name: Name of the module to import, may be relative to the given package.
1389    :param package: Package to resolve the name against, required for a relative name.
1390    """
1391    loop = asyncio.get_running_loop()
1392    try:
1393        return await loop.run_in_executor(_IMPORT_EXECUTOR, importlib.import_module, name, package)
1394    except RuntimeError as err:
1395        # threads we do not control (a library importing lazily in its own thread) can still
1396        # cross a module lock with ours; the import machinery reports that as a deadlock at
1397        # whoever detects it. The other import has finished by now, so a single retry sticks.
1398        if "deadlock detected" not in str(err):
1399            raise
1400        LOGGER.warning("Retrying import of %s after a module lock collision: %s", name, err)
1401        return await loop.run_in_executor(_IMPORT_EXECUTOR, importlib.import_module, name, package)
1402
1403
1404async def load_provider_module(domain: str, requirements: list[str]) -> ProviderModuleType:
1405    """Return module for given provider domain and make sure the requirements are met."""
1406
1407    async def _get_provider_module() -> ProviderModuleType:
1408        module = await import_module_in_thread(f".{domain}", "music_assistant.providers")
1409        return cast("ProviderModuleType", module)
1410
1411    # ensure module requirements are met
1412    for requirement in requirements:
1413        if requirement in _checked_requirements:
1414            continue
1415        if "==" not in requirement:
1416            # we should really get rid of unpinned requirements
1417            continue
1418        package_name, version = requirement.split("==", 1)
1419        # importlib.metadata can't resolve extras (e.g. aiosendspin[server]), so strip them
1420        package_name = package_name.split("[", 1)[0]
1421        installed_version = await get_package_version(package_name)
1422        if installed_version == "0.0.0":
1423            # ignore editable installs
1424            _checked_requirements.add(requirement)
1425            continue
1426        if installed_version != version:
1427            await install_package(requirement)
1428        _checked_requirements.add(requirement)
1429
1430    # try to load the module
1431    try:
1432        return await _get_provider_module()
1433    except ImportError:
1434        # (re)install ALL requirements
1435        for requirement in requirements:
1436            await install_package(requirement)
1437    # try loading the provider again to be safe
1438    # this will fail if something else is wrong (as it should)
1439    return await _get_provider_module()
1440
1441
1442async def has_tmpfs_mount() -> bool:
1443    """Check if we have a tmpfs mount."""
1444
1445    def _has_tmpfs_mount() -> bool:
1446        """Check if we have a tmpfs mount."""
1447        try:
1448            with open("/proc/mounts") as file:
1449                for line in file:
1450                    if "tmpfs /tmp tmpfs rw" in line:
1451                        return True
1452        except FileNotFoundError, OSError, PermissionError:
1453            pass
1454        return False
1455
1456    return await asyncio.to_thread(_has_tmpfs_mount)
1457
1458
1459async def get_free_space(folder: str) -> float:
1460    """Return free space on given folderpath in GB."""
1461
1462    def _get_free_space(folder: str) -> float:
1463        """Return free space on given folderpath in GB."""
1464        try:
1465            res = shutil.disk_usage(folder)
1466            return res.free / float(1 << 30)
1467        except FileNotFoundError, OSError, PermissionError:
1468            return 0.0
1469
1470    return await asyncio.to_thread(_get_free_space, folder)
1471
1472
1473async def get_free_space_percentage(folder: str) -> float:
1474    """Return free space on given folderpath in percentage."""
1475
1476    def _get_free_space(folder: str) -> float:
1477        """Return free space on given folderpath in GB."""
1478        try:
1479            res = shutil.disk_usage(folder)
1480            return res.free / res.total * 100
1481        except FileNotFoundError, OSError, PermissionError:
1482            return 0.0
1483
1484    return await asyncio.to_thread(_get_free_space, folder)
1485
1486
1487async def has_enough_space(folder: str, size: int) -> bool:
1488    """Check if folder has enough free space."""
1489    return await get_free_space(folder) > size
1490
1491
1492def divide_chunks(data: bytes, chunk_size: int) -> Iterator[bytes]:
1493    """Chunk bytes data into smaller chunks."""
1494    for i in range(0, len(data), chunk_size):
1495        yield data[i : i + chunk_size]
1496
1497
1498async def remove_file(file_path: str) -> None:
1499    """Remove file path (if it exists)."""
1500    if not await asyncio.to_thread(os.path.exists, file_path):
1501        return
1502    await asyncio.to_thread(os.remove, file_path)
1503    LOGGER.log(VERBOSE_LOG_LEVEL, "Removed file: %s", file_path)
1504
1505
1506def get_primary_ip_address_from_zeroconf(
1507    discovery_info: AsyncServiceInfo,
1508    prefer_ipv6: bool = False,
1509) -> str | None:
1510    """
1511    Get primary IP address from zeroconf discovery info.
1512
1513    :param discovery_info: The zeroconf service info to extract the address from.
1514    :param prefer_ipv6: If True, prefer IPv6 addresses over IPv4.
1515    """
1516    if prefer_ipv6:
1517        order = [IPVersion.V6Only, IPVersion.V4Only]
1518    else:
1519        order = [IPVersion.V4Only, IPVersion.V6Only]
1520    for version in order:
1521        for addr in discovery_info.ip_addresses_by_version(version):
1522            if addr.is_loopback or addr.is_link_local or addr.is_unspecified:
1523                continue
1524            return str(addr)
1525    return None
1526
1527
1528def get_port_from_zeroconf(discovery_info: AsyncServiceInfo) -> int | None:
1529    """Get port from zeroconf discovery info."""
1530    return discovery_info.port
1531
1532
1533def get_zeroconf_args(
1534    use_all_interfaces: bool = False,
1535) -> dict[str, Any]:
1536    """
1537    Determine optimal zeroconf IPVersion and interfaces from system adapters.
1538
1539    Inspects available network adapters to determine the correct IP version
1540    and interface configuration, similar to Home Assistant's approach.
1541
1542    :param use_all_interfaces: If True, use all interfaces (user override).
1543    """
1544    adapters = ifaddr.get_adapters()
1545    has_ipv4 = False
1546    has_ipv6 = False
1547    interface_ips: list[str] = []
1548    for adapter in adapters:
1549        for ip_config in adapter.ips:
1550            if ip_config.is_IPv6:
1551                ip_tuple = cast("tuple[str, int, int]", ip_config.ip)
1552                addr = ip_address(ip_tuple[0])
1553                if (
1554                    isinstance(addr, IPv6Address)
1555                    and not addr.is_loopback
1556                    and not addr.is_link_local
1557                ):
1558                    has_ipv6 = True
1559                    if not addr.is_global:
1560                        interface_ips.append(f"{ip_tuple[0]}%{ip_tuple[2]}")
1561            else:
1562                ip_str = cast("str", ip_config.ip)
1563                addr = ip_address(ip_str)
1564                if isinstance(addr, IPv4Address) and not addr.is_loopback:
1565                    has_ipv4 = True
1566                    interface_ips.append(ip_str)
1567
1568    # Determine IP version based on available addresses.
1569    # On macOS/FreeBSD, zeroconf's IPVersion.All creates an AF_INET6 listen socket
1570    # that cannot join IPv4 multicast groups, silently breaking discovery of
1571    # IPv4-only devices. Fall back to V4Only on those platforms.
1572    has_functional_dual_stack = not sys.platform.startswith(("freebsd", "darwin"))
1573    if has_ipv4 and has_ipv6 and has_functional_dual_stack:
1574        ip_version = IPVersion.All
1575    elif has_ipv4:
1576        ip_version = IPVersion.V4Only
1577    elif has_ipv6:
1578        ip_version = IPVersion.V6Only
1579    else:
1580        ip_version = IPVersion.V4Only
1581
1582    if use_all_interfaces:
1583        # User explicitly requested all interfaces — pass explicit IP list
1584        # to avoid issues with InterfaceChoice.Default on multi-interface hosts.
1585        if interface_ips:
1586            return {"ip_version": ip_version, "interfaces": interface_ips}
1587        return {"ip_version": ip_version, "interfaces": InterfaceChoice.All}
1588
1589    # Default mode: use InterfaceChoice.Default for IPv4-only single-interface,
1590    # otherwise pass explicit interface list for reliability.
1591    if ip_version == IPVersion.V4Only:
1592        return {"ip_version": ip_version, "interfaces": InterfaceChoice.Default}
1593    if interface_ips:
1594        return {"ip_version": ip_version, "interfaces": interface_ips}
1595    return {"ip_version": ip_version, "interfaces": InterfaceChoice.All}
1596
1597
1598async def close_async_generator(agen: AsyncGenerator[Any]) -> None:
1599    """Force close an async generator."""
1600    task = asyncio.create_task(agen.__anext__())
1601    task.cancel()
1602    with suppress(asyncio.CancelledError, StopAsyncIteration):
1603        await task
1604    await agen.aclose()
1605
1606
1607async def detect_charset(data: bytes, fallback: str = "utf-8", preferred: str | None = None) -> str:
1608    """
1609    Detect the charset to decode the given raw text with.
1610
1611    :param data: The raw text bytes to inspect.
1612    :param fallback: Charset to return when the charset can not be determined.
1613    :param preferred: Charset declared by the source, taken over detection when usable.
1614    """
1615    # a BOM outranks the declared charset: it names the very same UTF-8 but, unlike
1616    # the declared name, also gets the marker itself stripped off the decoded text
1617    if data.startswith(codecs.BOM_UTF8):
1618        return "utf-8-sig"
1619
1620    if preferred:
1621        # a declared charset is only worth anything if Python can actually decode text with
1622        # it: servers do send misspelled or plain made-up names in their Content-Type, and a
1623        # handful of names that do resolve to a codec still cannot decode text (base64, idna)
1624        try:
1625            data[:16].decode(preferred, errors="replace")
1626        except (LookupError, ValueError) as err:
1627            LOGGER.debug("Ignoring unusable charset %s: %s", preferred, err)
1628        else:
1629            return preferred
1630
1631    try:
1632        data.decode()
1633    except UnicodeDecodeError:
1634        pass
1635    else:
1636        # valid UTF-8 is never a legacy charset by accident, so skip detection
1637        return "utf-8"
1638
1639    # imported here to keep the detector out of the idle import footprint:
1640    # it is only needed for the rare text that is not UTF-8
1641    import chardet  # noqa: PLC0415
1642    from chardet.enums import EncodingEra  # noqa: PLC0415
1643
1644    # the reported confidence is deliberately not gated on: CUE sheets and playlists
1645    # are nearly all ASCII keywords, which holds the score far below any usable
1646    # threshold even though the charset itself is named correctly (support #6093).
1647    # With no score to weigh them against, DOS and mainframe codepages are dropped from
1648    # the candidates so a stray weak match cannot outrank the Windows codepage these
1649    # files are really written in. Only a superset is guaranteed to decode the bytes
1650    # past the window the detector samples, so it wins ties over its subsets.
1651    try:
1652        detected = await asyncio.to_thread(
1653            chardet.detect,
1654            data,
1655            encoding_era=EncodingEra.ALL & ~(EncodingEra.DOS | EncodingEra.MAINFRAME),
1656            prefer_superset=True,
1657            no_match_encoding=fallback,
1658        )
1659    except Exception as err:
1660        LOGGER.debug("Failed to detect charset: %s", err)
1661        return fallback
1662    if not (encoding := detected["encoding"]):
1663        return fallback
1664    LOGGER.debug("Detected charset %s (confidence %.2f)", encoding, detected["confidence"])
1665    return encoding
1666
1667
1668def parse_optional_bool(value: Any) -> bool | None:
1669    """Parse an optional boolean value from various input types."""
1670    if value is None:
1671        return None
1672    if isinstance(value, bool):
1673        return value
1674    if isinstance(value, str):
1675        value_lower = value.strip().lower()
1676        if value_lower in ("true", "1", "yes", "on"):
1677            return True
1678        if value_lower in ("false", "0", "no", "off"):
1679            return False
1680    if isinstance(value, (int, float)):
1681        return bool(value)
1682    return None
1683
1684
1685def merge_dict(
1686    base_dict: dict[Any, Any],
1687    new_dict: dict[Any, Any],
1688    allow_overwite: bool = False,
1689) -> dict[Any, Any]:
1690    """Merge dict without overwriting existing values."""
1691    final_dict = base_dict.copy()
1692    for key, value in new_dict.items():
1693        if final_dict.get(key) and isinstance(value, dict):
1694            final_dict[key] = merge_dict(final_dict[key], value)
1695        if final_dict.get(key) and isinstance(value, tuple):
1696            final_dict[key] = merge_tuples(final_dict[key], value)
1697        if final_dict.get(key) and isinstance(value, list):
1698            final_dict[key] = merge_lists(final_dict[key], value)
1699        elif not final_dict.get(key) or allow_overwite:
1700            final_dict[key] = value
1701    return final_dict
1702
1703
1704def merge_tuples(base: tuple[Any, ...], new: tuple[Any, ...]) -> tuple[Any, ...]:
1705    """Merge 2 tuples."""
1706    return tuple(x for x in base if x not in new) + tuple(new)
1707
1708
1709def merge_lists(base: list[Any], new: list[Any]) -> list[Any]:
1710    """Merge 2 lists."""
1711    return [x for x in base if x not in new] + list(new)
1712
1713
1714def percentage(part: float, whole: float) -> int:
1715    """Calculate percentage."""
1716    return int(100 * float(part) / float(whole))
1717
1718
1719def validate_announcement_chime_url(url: str) -> bool:
1720    """Validate announcement chime URL format."""
1721    if not url or not url.strip():
1722        return True  # Empty URL is valid
1723
1724    if url == ANNOUNCE_ALERT_FILE:
1725        return True  # Built-in chime file is valid
1726
1727    try:
1728        parsed = urlparse(url.strip())
1729
1730        if parsed.scheme not in ("http", "https"):
1731            return False
1732
1733        if not parsed.netloc:
1734            return False
1735
1736        path_lower = parsed.path.lower()
1737        audio_extensions = (".mp3", ".wav", ".flac", ".ogg", ".m4a", ".aac")
1738
1739        return any(path_lower.endswith(ext) for ext in audio_extensions)
1740
1741    except Exception:
1742        return False
1743
1744
1745async def get_mac_address(ip_address: str) -> str | None:
1746    """Get MAC address for given IP address via ARP lookup."""
1747    try:
1748        from getmac import get_mac_address as getmac_lookup  # noqa: PLC0415
1749
1750        return await asyncio.to_thread(getmac_lookup, ip=ip_address)
1751    except ImportError:
1752        LOGGER.debug("getmac module not available, cannot resolve MAC from IP")
1753        return None
1754    except Exception as err:
1755        LOGGER.debug("Failed to resolve MAC address for %s: %s", ip_address, err)
1756        return None
1757
1758
1759def is_locally_administered_mac(mac_address: str) -> bool:
1760    """
1761    Check if a MAC address is locally administered (virtual/randomized).
1762
1763    Locally administered addresses have bit 1 of the first octet set to 1.
1764    These are often used by devices for virtual interfaces or protocol-specific
1765    addresses (e.g., AirPlay, DLNA may use different virtual MACs than the real hardware MAC).
1766
1767    :param mac_address: MAC address in any common format (with :, -, or no separator).
1768    :return: True if locally administered, False if globally unique (real hardware MAC).
1769    """
1770    # Normalize MAC address
1771    mac_clean = mac_address.upper().replace(":", "").replace("-", "")
1772    if len(mac_clean) < 2:
1773        return False
1774
1775    # Get first octet and check bit 1 (second bit from right)
1776    try:
1777        first_octet = int(mac_clean[:2], 16)
1778        return bool(first_octet & 0x02)
1779    except ValueError:
1780        return False
1781
1782
1783def normalize_mac_for_matching(mac_address: str) -> str:
1784    """
1785    Normalize a MAC address for device matching by masking out the locally-administered bit.
1786
1787    Some protocols (like AirPlay) report a locally-administered MAC address variant where
1788    bit 1 of the first octet is set. For example:
1789    - Real hardware MAC: 54:78:C9:E6:0D:A0 (first byte 0x54 = 01010100)
1790    - AirPlay reports:   56:78:C9:E6:0D:A0 (first byte 0x56 = 01010110)
1791
1792    These represent the same device but differ only in the locally-administered bit.
1793    This function normalizes the MAC by clearing bit 1 of the first octet, allowing
1794    both variants to match the same device.
1795
1796    :param mac_address: MAC address in any common format (with :, -, or no separator).
1797    :return: Normalized MAC address in lowercase without separators, with the
1798             locally-administered bit cleared.
1799    """
1800    # Normalize MAC address (remove separators, lowercase)
1801    mac_clean = mac_address.lower().replace(":", "").replace("-", "")
1802    if len(mac_clean) != 12:
1803        # Invalid MAC length, return as-is
1804        return mac_clean
1805
1806    try:
1807        # Parse first octet and clear bit 1 (the locally-administered bit)
1808        first_octet = int(mac_clean[:2], 16)
1809        first_octet_normalized = first_octet & ~0x02  # Clear bit 1
1810        # Reconstruct the MAC with the normalized first octet
1811        return f"{first_octet_normalized:02x}{mac_clean[2:]}"
1812    except ValueError:
1813        # Invalid hex, return as-is
1814        return mac_clean
1815
1816
1817def is_valid_mac_address(mac_address: str | None) -> bool:
1818    """
1819    Check if a MAC address is valid and usable for device identification.
1820
1821    Invalid MAC addresses include:
1822    - None or empty strings
1823    - Null MAC: 00:00:00:00:00:00
1824    - Broadcast MAC: ff:ff:ff:ff:ff:ff
1825    - Any MAC that doesn't follow the expected pattern
1826
1827    :param mac_address: MAC address to validate.
1828    :return: True if valid and usable, False otherwise.
1829    """
1830    if not mac_address:
1831        return False
1832
1833    # Normalize MAC address (remove separators and convert to lowercase)
1834    normalized = mac_address.lower().replace(":", "").replace("-", "")
1835
1836    # Check for invalid/reserved MAC addresses
1837    if normalized in ("000000000000", "ffffffffffff"):
1838        return False
1839
1840    # Check length and hex validity
1841    if len(normalized) != 12:
1842        return False
1843
1844    try:
1845        int(normalized, 16)
1846        return True
1847    except ValueError:
1848        return False
1849
1850
1851def normalize_ip_address(ip_address: str | None) -> str | None:
1852    """
1853    Normalize IP address for comparison.
1854
1855    Handles IPv6-mapped IPv4 addresses (e.g., ::ffff:192.168.1.64 -> 192.168.1.64).
1856
1857    :param ip_address: IP address to normalize.
1858    :return: Normalized IP address or None if invalid.
1859    """
1860    if not ip_address:
1861        return None
1862
1863    # Handle IPv6-mapped IPv4 addresses
1864    if ip_address.startswith("::ffff:"):
1865        # Extract the IPv4 part
1866        return ip_address[7:]
1867
1868    return ip_address
1869
1870
1871async def resolve_real_mac_address(reported_mac: str | None, ip_address: str | None) -> str | None:
1872    """
1873    Resolve the real MAC address for a device.
1874
1875    Some devices report different virtual MAC addresses per protocol (AirPlay, DLNA,
1876    Chromecast). This function tries to resolve the actual hardware MAC via ARP
1877    when the reported MAC appears to be locally administered (virtual).
1878
1879    :param reported_mac: The MAC address reported by the protocol.
1880    :param ip_address: The IP address of the device (for ARP lookup).
1881    :return: The real MAC address if found, or None if it couldn't be resolved.
1882    """
1883    if not ip_address:
1884        return None
1885
1886    # If no MAC reported or it's a locally administered one, try ARP lookup
1887    if not reported_mac or is_locally_administered_mac(reported_mac):
1888        real_mac = await get_mac_address(ip_address)
1889        if real_mac and is_valid_mac_address(real_mac):
1890            return real_mac.upper()
1891
1892    return None
1893
1894
1895async def enrich_device_mac_address(
1896    device_info: DeviceInfo,
1897    logger: logging.Logger | None = None,
1898) -> None:
1899    """
1900    Enrich a player's device_info with a real MAC address via ARP.
1901
1902    Called automatically during player registration. It validates the existing MAC,
1903    normalizes IPv6-mapped IPv4 addresses, and always performs an ARP lookup when
1904    an IP is available. The ARP result replaces the reported MAC because it reflects
1905    the true hardware address and reliably unifies protocols on the same device -
1906    even when different protocols report different valid MACs (e.g., Yamaha devices
1907    where DLNA and AirPlay MACs differ by 1 in the last octet).
1908
1909    :param device_info: The player's DeviceInfo to enrich in-place.
1910    :param logger: Optional logger for debug messages.
1911    """
1912    identifiers = device_info.identifiers
1913    reported_mac = identifiers.get(IdentifierType.MAC_ADDRESS)
1914    ip_address = identifiers.get(IdentifierType.IP_ADDRESS)
1915
1916    # Blank out invalid MAC addresses (00:00:00:00:00:00, ff:ff:ff:ff:ff:ff, etc.)
1917    # so they can't cause false matches in protocol linking.
1918    if reported_mac and not is_valid_mac_address(reported_mac):
1919        if logger:
1920            logger.debug("Removing invalid MAC address: %s", reported_mac)
1921        device_info.add_identifier(IdentifierType.MAC_ADDRESS, None)
1922        reported_mac = None
1923
1924    # Normalize IP address (handle IPv6-mapped IPv4 like ::ffff:192.168.1.64)
1925    if ip_address:
1926        normalized_ip = normalize_ip_address(ip_address)
1927        if normalized_ip and normalized_ip != ip_address:
1928            device_info.add_identifier(IdentifierType.IP_ADDRESS, normalized_ip)
1929            if logger:
1930                logger.debug(
1931                    "Normalized IP address: %s -> %s",
1932                    ip_address,
1933                    normalized_ip,
1934                )
1935            ip_address = normalized_ip
1936
1937    # Skip ARP enrichment if no IP available (can't do ARP lookup)
1938    if not ip_address:
1939        return
1940
1941    # Always attempt ARP lookup when we have an IP address.
1942    # Some devices (e.g., Yamaha MusicCast) report different valid globally-unique
1943    # MACs per protocol (DLNA vs AirPlay differ by 1 in the last octet).
1944    # ARP resolves the true hardware MAC which reliably unifies all protocols.
1945    # The result is cached in player config so subsequent restarts are fast.
1946    real_mac = await resolve_real_mac_address(reported_mac, ip_address)
1947    if real_mac and real_mac.upper() != (reported_mac or "").upper():
1948        device_info.add_identifier(IdentifierType.MAC_ADDRESS, real_mac)
1949        if logger:
1950            logger.debug(
1951                "Resolved MAC via ARP: %s -> %s",
1952                reported_mac or "none",
1953                real_mac,
1954            )
1955    elif not reported_mac:
1956        # ARP failed and no reported MAC - nothing we can do
1957        if logger:
1958            logger.debug("ARP lookup failed for %s and no reported MAC", ip_address)
1959
1960
1961class TaskManager:
1962    """
1963    Helper class to run many tasks at once.
1964
1965    This is basically an alternative to asyncio.TaskGroup but this will not
1966    cancel all operations when one of the tasks fails.
1967    Logging of exceptions is done by the mass.create_task helper.
1968    """
1969
1970    def __init__(self, mass: MusicAssistant, limit: int = 0):
1971        """Initialize the TaskManager."""
1972        self.mass = mass
1973        self._tasks: list[asyncio.Task[None]] = []
1974        self._semaphore = asyncio.Semaphore(limit) if limit else None
1975
1976    def create_task(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[None]:
1977        """Create a new task and add it to the manager."""
1978        task = self.mass.create_task(coro)
1979        self._tasks.append(task)
1980        return task
1981
1982    async def create_task_with_limit(self, coro: Coroutine[Any, Any, Any]) -> None:
1983        """Create a new task with semaphore limit."""
1984        assert self._semaphore is not None
1985
1986        def task_done_callback(_task: asyncio.Task[None]) -> None:
1987            assert self._semaphore is not None  # for type checking
1988            self._tasks.remove(task)
1989            self._semaphore.release()
1990
1991        await self._semaphore.acquire()
1992        task: asyncio.Task[None] = self.create_task(coro)
1993        task.add_done_callback(task_done_callback)
1994
1995    async def __aenter__(self) -> Self:
1996        """Enter context manager."""
1997        return self
1998
1999    async def __aexit__(
2000        self,
2001        exc_type: type[BaseException] | None,
2002        exc_val: BaseException | None,
2003        exc_tb: TracebackType | None,
2004    ) -> bool | None:
2005        """Exit context manager."""
2006        if len(self._tasks) > 0:
2007            await asyncio.wait(self._tasks)
2008            self._tasks.clear()
2009        return None
2010
2011
2012_R = TypeVar("_R")
2013_P = ParamSpec("_P")
2014
2015
2016def lock[**P, R](  # type: ignore[valid-type]
2017    func: Callable[_P, Awaitable[_R]],
2018) -> Callable[_P, Coroutine[Any, Any, _R]]:
2019    """
2020    Call async function using a per-instance Lock.
2021
2022    Each instance gets its own lock so that e.g. SyncGroupPlayer A
2023    does not block SyncGroupPlayer B when both call set_members().
2024    """
2025    # Per-instance lock storage (weak refs so locks are GC'd with their instance)
2026    instance_locks: weakref.WeakKeyDictionary[Any, asyncio.Lock] = weakref.WeakKeyDictionary()
2027    # Fallback lock for non-method (no self) usage
2028    fallback_lock: asyncio.Lock | None = None
2029
2030    @functools.wraps(func)
2031    async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
2032        """Call async function using a per-instance Lock."""
2033        nonlocal fallback_lock
2034        instance = args[0] if args else None
2035        if instance is not None:
2036            try:
2037                func_lock = instance_locks.setdefault(instance, asyncio.Lock())
2038            except TypeError:
2039                # instance is not weakly referenceable, use fallback
2040                if fallback_lock is None:
2041                    fallback_lock = asyncio.Lock()
2042                func_lock = fallback_lock
2043        else:
2044            if fallback_lock is None:
2045                fallback_lock = asyncio.Lock()
2046            func_lock = fallback_lock
2047        async with func_lock:
2048            return await func(*args, **kwargs)
2049
2050    return wrapper
2051
2052
2053class TimedAsyncGenerator:
2054    """
2055    Async iterable that times out after a given time.
2056
2057    Source: https://medium.com/@dmitry8912/implementing-timeouts-in-pythons-asynchronous-generators-f7cbaa6dc1e9
2058    """
2059
2060    def __init__(self, iterable: AsyncIterator[Any], timeout: int = 0):
2061        """
2062        Initialize the AsyncTimedIterable.
2063
2064        Args:
2065            iterable: The async iterable to wrap.
2066            timeout: The timeout in seconds for each iteration.
2067        """
2068
2069        class AsyncTimedIterator:
2070            def __init__(self) -> None:
2071                self._iterator = iterable.__aiter__()
2072
2073            async def __anext__(self) -> Any:
2074                result = await asyncio.wait_for(self._iterator.__anext__(), int(timeout))
2075                if not result:
2076                    raise StopAsyncIteration
2077                return result
2078
2079        self._factory = AsyncTimedIterator
2080
2081    def __aiter__(self):  # type: ignore[no-untyped-def]
2082        """Return the async iterator."""
2083        return self._factory()
2084
2085
2086async def join_task[T](task: asyncio.Future[T], timeout: float | None = None) -> T:
2087    """
2088    Wait for a task started elsewhere and return its result.
2089
2090    Cancelling the waiter leaves the task running, so work that is shared between callers -
2091    or that must outlive a caller's deadline - keeps going and still reaches every other
2092    waiter. A task that can lose all its waiters needs a done callback that retrieves its
2093    exception (as mass.create_task installs) to keep asyncio quiet about it.
2094
2095    :param task: The task (or future) to wait for.
2096    :param timeout: Optional number of seconds to wait before giving up.
2097    :raises TimeoutError: If the task did not complete within the timeout.
2098    :raises asyncio.CancelledError: If the task itself was cancelled.
2099    :return: The task's result.
2100    """
2101    if not task.done():
2102        # awaiting the task directly would hold it as this coroutine's fut_waiter, so
2103        # cancelling the waiter would cancel the task itself. asyncio.shield achieves the
2104        # same isolation, but as of Python 3.14 a cancelled waiter makes it report the task's
2105        # exception through loop.call_exception_handler, even when another waiter already
2106        # handled it.
2107        await asyncio.wait((task,), timeout=timeout)
2108    if not task.done():
2109        raise TimeoutError
2110    return task.result()
2111
2112
2113# Bound for guard_single_request: it only needs ``.mass``, so a structural protocol
2114# lets it decorate providers, core controllers and media controllers alike without
2115# coupling to their concrete base classes.
2116class _SupportsMass(Protocol):
2117    """Structural type for objects exposing a MusicAssistant reference."""
2118
2119    mass: MusicAssistant
2120
2121
2122def guard_single_request[SelfT: _SupportsMass, **P, R](
2123    func: Callable[Concatenate[SelfT, P], Coroutine[Any, Any, R]],
2124) -> Callable[Concatenate[SelfT, P], Coroutine[Any, Any, R]]:
2125    """
2126    Ensure concurrent calls with identical arguments result in a single request.
2127
2128    Callers arriving while an identical call is already in flight await that same call and
2129    receive its result. Cancelling one caller leaves both the request and the other callers
2130    unaffected. Calls count as identical when they are made on the same object with equal
2131    arguments, no matter whether those were passed positionally or by keyword; the request
2132    runs with the arguments of the caller that started it.
2133
2134    Every argument must be a scalar or an object identified by its ``uri``, so that equal
2135    arguments are guaranteed to produce an equal key.
2136
2137    :param func: The coroutine method to guard.
2138    """
2139    signature = inspect.signature(func)
2140
2141    @functools.wraps(func)
2142    async def wrapper(self: SelfT, *args: P.args, **kwargs: P.kwargs) -> R:
2143        mass = self.mass
2144        # create a task_id dynamically based on the bound method and args/kwargs.
2145        # the instance is part of the key because a decorated method may be inherited by
2146        # multiple subclasses (all media controllers share
2147        # MediaControllerBase.get_provider_item) and a class may have multiple instances
2148        # (e.g. a provider set up twice), which must never join each other's flight.
2149        # id(self) is stable while a flight is live because the task references self;
2150        # the class name only serves to keep the task_id readable while debugging.
2151        # binding the arguments to their parameter names and filling in the defaults keys a
2152        # call the same however it was spelled; repr of the resulting tuple keeps the parts
2153        # apart, so an id that itself contains punctuation cannot run into the next one.
2154        bound = signature.bind(self, *args, **kwargs)
2155        bound.apply_defaults()
2156        task_id = repr(
2157            (
2158                type(self).__name__,
2159                id(self),
2160                func.__qualname__,
2161                # skip the instance: it is the first parameter and is keyed by id() above
2162                *(
2163                    (name, _canonical_key_part(value))
2164                    for name, value in islice(bound.arguments.items(), 1, None)
2165                ),
2166            )
2167        )
2168        task: asyncio.Task[R] = mass.create_task(
2169            func,
2170            self,
2171            *args,
2172            task_id=task_id,
2173            abort_existing=False,
2174            eager_start=True,
2175            # every caller awaits the flight below and so sees the failure itself; the
2176            # task's own exception log would report a handled error as an unhandled one
2177            log_exceptions=False,
2178            **kwargs,
2179        )
2180        return await join_task(task)
2181
2182    return wrapper
2183
2184
2185def _canonical_key_part(value: Any) -> Any:
2186    """Return a stable stand-in for a single argument of a guarded request."""
2187    if (uri := getattr(value, "uri", None)) is not None:
2188        # a media item renders as a multi-kilobyte dataclass repr in which the set-typed
2189        # fields (provider_mappings, external_ids) can iterate in different orders for two
2190        # equal items. the uri identifies the item, and the type travels with it because a
2191        # full item and an ItemMapping for that same item are not handled the same.
2192        return (type(value).__name__, uri)
2193    return value
2194