music-assistant-server

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