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    # wait for the shared probe instead of awaiting it directly: a caller awaiting a task
1014    # holds it as its fut_waiter, so cancelling that caller would otherwise cancel the probe
1015    # for all other callers. asyncio.shield achieves the same, but as of Python 3.14 a
1016    # cancelled caller makes it report the probe's exception through
1017    # loop.call_exception_handler, even when another caller already handled it.
1018    if not pending.done():
1019        await asyncio.wait((pending,))
1020    return pending.result()
1021
1022
1023def _log_ip_probe_failure(probe: asyncio.Task[tuple[str, ...]]) -> None:
1024    """Log (and thereby retrieve) the exception of a finished address probe, if any."""
1025    if probe.cancelled():
1026        return
1027    # every waiter that is still around reports the failure itself, so a debug line is
1028    # enough here; retrieving the exception is what keeps asyncio from reporting it as
1029    # "Task exception was never retrieved" once the probe is garbage collected
1030    if (err := probe.exception()) is not None:
1031        LOGGER.debug("Enumerating IP addresses failed: %s", err)
1032
1033
1034def _enumerate_ip_addresses(include_ipv6: bool, publish_candidates_only: bool) -> tuple[str, ...]:
1035    """Enumerate all IP addresses of all network interfaces (blocking)."""
1036    result: list[tuple[int, str]] = []
1037    # the same addresses, without the ones no device on the local network can reach
1038    lan_result: list[tuple[int, str]] = []
1039    # try to get the primary IP address
1040    # this is the IP address of the default route
1041    primary_ip = ""
1042    # try IPv4 first
1043    _sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
1044    _sock.settimeout(0)
1045    try:
1046        # doesn't even have to be reachable
1047        _sock.connect(("10.254.254.254", 1))
1048        primary_ip = _sock.getsockname()[0]
1049    except Exception:
1050        primary_ip = ""
1051    finally:
1052        _sock.close()
1053    # fall back to IPv6 if no IPv4 primary found (e.g. IPv6-only networks)
1054    if not primary_ip:
1055        _sock6 = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
1056        _sock6.settimeout(0)
1057        try:
1058            _sock6.connect(("2001:db8::1", 1))
1059            primary_ip = _sock6.getsockname()[0]
1060        except Exception:
1061            primary_ip = ""
1062        finally:
1063            _sock6.close()
1064    # get all IP addresses of all network interfaces
1065    adapters = ifaddr.get_adapters()
1066    for adapter in adapters:
1067        adapter_is_virtual = _is_virtual_interface(adapter.name) or _is_virtual_interface(
1068            adapter.nice_name
1069        )
1070        for ip in adapter.ips:
1071            if ip.is_IPv6 and not include_ipv6:
1072                continue
1073            # ifaddr returns IPv6 addresses as (address, flowinfo, scope_id) tuples
1074            ip_str = ip.ip[0] if isinstance(ip.ip, tuple) else ip.ip
1075            if ip_str.startswith(("127", "169.254")):
1076                # filter out IPv4 loopback/APIPA address
1077                continue
1078            if ip_str.startswith(("::1", "::ffff:", "fe80")):
1079                # filter out IPv6 loopback/link-local address
1080                continue
1081            if ip_str == primary_ip:
1082                score = 10
1083            elif ip_str.startswith(("192.168.",)):
1084                # we rank the 192.168 range a bit higher as its most
1085                # often used as the private network subnet
1086                score = 2
1087            elif ip_str.startswith(("172.", "10.", "192.")):
1088                # we rank the 172 range a bit lower as its most
1089                # often used as the private docker network
1090                score = 1
1091            else:
1092                score = 0
1093            result.append((score, ip_str))
1094            if not adapter_is_virtual:
1095                lan_result.append((score, ip_str))
1096    # a host that is only reachable over a tunnel or bridge still has to publish something
1097    selected = (lan_result or result) if publish_candidates_only else result
1098    selected.sort(key=lambda x: x[0], reverse=True)
1099    if not selected:
1100        # no routable addresses found (e.g. offline host with only loopback/link-local):
1101        # fall back to loopback so callers that rely on at least one address keep working
1102        return ("127.0.0.1",)
1103    return tuple(ip[1] for ip in selected)
1104
1105
1106def _is_virtual_interface(name: str) -> bool:
1107    """Return whether the named interface belongs to a container, VM or VPN network."""
1108    name = name.lower()
1109    return name.startswith(_VIRTUAL_INTERFACE_PREFIXES) or bool(
1110        _VIRTUAL_INTERFACE_NAMES.fullmatch(name)
1111    )
1112
1113
1114def interface_name_for_ip(ip: str) -> str | None:
1115    """
1116    Return the name of the network interface that holds the given IP, or None.
1117
1118    Used to map a bind/publish IP to its interface name for components that select
1119    their mDNS/zeroconf advertisement interface by name (e.g. shairport-sync and
1120    go-librespot), so the advertisement stays on the intended network.
1121
1122    :param ip: The IPv4/IPv6 address to look up.
1123    """
1124    for adapter in ifaddr.get_adapters():
1125        for ip_config in adapter.ips:
1126            addr = ip_config.ip if isinstance(ip_config.ip, str) else ip_config.ip[0]
1127            if addr == ip:
1128                return adapter.name
1129    return None
1130
1131
1132async def is_port_in_use(port: int, host: str | None = None) -> bool:
1133    """
1134    Check if a port is in use.
1135
1136    :param port: Port number to check.
1137    :param host: Optional bind address to probe. When omitted, both IPv4 and IPv6
1138        wildcard addresses are checked.
1139    """
1140
1141    def _is_port_in_use() -> bool:
1142        candidates: tuple[tuple[socket.AddressFamily, str], ...]
1143        if host is not None:
1144            candidates = ((socket.AF_INET6 if ":" in host else socket.AF_INET, host),)
1145        else:
1146            # Try both IPv4 and IPv6 to support single-stack and dual-stack systems.
1147            # A port is considered free if it can be bound on at least one address family.
1148            candidates = ((socket.AF_INET, "0.0.0.0"), (socket.AF_INET6, "::"))
1149        for family, addr in candidates:
1150            try:
1151                with socket.socket(family, socket.SOCK_STREAM) as _sock:
1152                    # Set SO_REUSEADDR to match asyncio.start_server behavior
1153                    # This allows binding to ports in TIME_WAIT state
1154                    _sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
1155                    _sock.bind((addr, port))
1156                    return False
1157            except OSError:
1158                continue
1159        return True
1160
1161    return await asyncio.to_thread(_is_port_in_use)
1162
1163
1164# In-process reservations for ports handed out by select_free_port. Provider
1165# instances (and reloads) frequently call select_free_port at nearly the same
1166# moment and only bind the returned port asynchronously afterwards, so a port
1167# that was just handed out is not yet detectable as "in use". Keeping a
1168# short-lived reservation per returned port stops concurrent/successive callers
1169# from picking the same one. Reservations expire automatically after the grace
1170# period so the range is never permanently exhausted across reloads.
1171_PORT_RESERVATION_TTL = 60.0
1172_reserved_ports: dict[int, float] = {}
1173_select_free_port_lock = asyncio.Lock()
1174
1175
1176async def select_free_port(range_start: int, range_end: int, host: str | None = None) -> int:
1177    """
1178    Find and reserve a free port within the given range.
1179
1180    The returned port is reserved so concurrent or successive callers are not
1181    handed the same port.
1182
1183    :param range_start: First port (inclusive) of the range to search.
1184    :param range_end: Port to stop before (exclusive) when searching the range.
1185    :param host: Optional bind address to probe for availability.
1186    """
1187    async with _select_free_port_lock:
1188        now = time.monotonic()
1189        # drop expired reservations so their ports become reusable again
1190        for reserved_port, deadline in list(_reserved_ports.items()):
1191            if deadline <= now:
1192                del _reserved_ports[reserved_port]
1193        for port in range(range_start, range_end):
1194            if port in _reserved_ports:
1195                continue
1196            if not await is_port_in_use(port, host=host):
1197                _reserved_ports[port] = now + _PORT_RESERVATION_TTL
1198                return port
1199    msg = f"No free port available in range {range_start}-{range_end - 1}"
1200    raise OSError(msg)
1201
1202
1203async def get_ip_from_host(dns_name: str) -> str | None:
1204    """Resolve (first) IP-address for given dns name."""
1205
1206    def _resolve() -> str | None:
1207        try:
1208            # use getaddrinfo to support both IPv4 and IPv6 resolution
1209            results = socket.getaddrinfo(dns_name, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
1210            if results:
1211                return str(results[0][4][0])
1212        except Exception:
1213            # fail gracefully!
1214            return None
1215        return None
1216
1217    return await asyncio.to_thread(_resolve)
1218
1219
1220async def get_source_ip_for_target(target_ip: str) -> str:
1221    """
1222    Return the local interface address the routing table would egress to ``target_ip`` from.
1223
1224    Empty when no route to the target can be determined.
1225
1226    :param target_ip: IP address of the device the traffic is meant for.
1227    """
1228
1229    def _routing_lookup() -> str:
1230        try:
1231            is_ipv6_target = ip_address(target_ip).version == 6
1232        except ValueError:
1233            is_ipv6_target = False
1234        route_family = socket.AF_INET6 if is_ipv6_target else socket.AF_INET
1235        route_target: tuple[str, int] | tuple[str, int, int, int] = (
1236            (target_ip, 80, 0, 0) if is_ipv6_target else (target_ip, 80)
1237        )
1238        with socket.socket(route_family, socket.SOCK_DGRAM) as _sock:
1239            try:
1240                _sock.settimeout(1.0)
1241                _sock.connect(route_target)
1242                routed_ip = str(_sock.getsockname()[0])
1243                if routed_ip and routed_ip not in WILDCARD_BIND_IPS:
1244                    return routed_ip
1245            except OSError:
1246                pass
1247        return ""
1248
1249    return await asyncio.to_thread(_routing_lookup)
1250
1251
1252async def get_ip_pton(ip_string: str) -> bytes:
1253    """Return socket pton for a local ip."""
1254    try:
1255        return await asyncio.to_thread(socket.inet_pton, socket.AF_INET, ip_string)
1256    except OSError:
1257        return await asyncio.to_thread(socket.inet_pton, socket.AF_INET6, ip_string)
1258
1259
1260def format_ip_for_url(ip_address: str) -> str:
1261    """Wrap IPv6 addresses in brackets for use in URLs (RFC 2732)."""
1262    if ":" in ip_address:
1263        return f"[{ip_address}]"
1264    return ip_address
1265
1266
1267async def get_folder_size(folderpath: str) -> float:
1268    """Return folder size in gb."""
1269
1270    def _get_folder_size(folderpath: str) -> float:
1271        total_size = 0
1272        for dirpath, _dirnames, filenames in os.walk(folderpath):
1273            for _file in filenames:
1274                _fp = os.path.join(dirpath, _file)
1275                total_size += Path(_fp).stat().st_size
1276        return total_size / float(1 << 30)
1277
1278    return await asyncio.to_thread(_get_folder_size, folderpath)
1279
1280
1281def get_changed_keys(
1282    dict1: dict[str, Any],
1283    dict2: dict[str, Any],
1284    recursive: bool = False,
1285) -> set[str]:
1286    """Compare 2 dicts and return set of changed keys."""
1287    # TODO: Check with Marcel whether we should calculate new dicts based on ignore_keys
1288    return set(get_changed_dict_values(dict1, dict2, recursive).keys())
1289    # return set(get_changed_dict_values(dict1, dict2, ignore_keys, recursive).keys())
1290
1291
1292def get_changed_dict_values(
1293    dict1: dict[str, Any],
1294    dict2: dict[str, Any],
1295    recursive: bool = False,
1296) -> dict[str, tuple[Any, Any]]:
1297    """
1298    Compare 2 dicts and return dict of changed values.
1299
1300    dict key is the changed key, value is tuple of old and new values.
1301    """
1302    if not dict1 and not dict2:
1303        return {}
1304    if not dict1:
1305        return {key: (None, value) for key, value in dict2.items()}
1306    if not dict2:
1307        return {key: (None, value) for key, value in dict1.items()}
1308    changed_values = {}
1309    for key, value in dict2.items():
1310        if isinstance(value, dict) and isinstance(dict1[key], dict) and recursive:
1311            changed_subvalues = get_changed_dict_values(dict1[key], value, recursive)
1312            for subkey, subvalue in changed_subvalues.items():
1313                changed_values[f"{key}.{subkey}"] = subvalue
1314            continue
1315        if key not in dict1:
1316            changed_values[key] = (None, value)
1317            continue
1318        if dict1[key] != value:
1319            changed_values[key] = (dict1[key], value)
1320    return changed_values
1321
1322
1323def empty_queue[T](q: asyncio.Queue[T]) -> None:
1324    """Empty an asyncio Queue."""
1325    for _ in range(q.qsize()):
1326        try:
1327            q.get_nowait()
1328            q.task_done()
1329        except asyncio.QueueEmpty, ValueError:
1330            pass
1331
1332
1333async def install_package(package: str) -> None:
1334    """Install package with pip, raise when install failed."""
1335    LOGGER.debug("Installing python package %s", package)
1336    args = ["uv", "pip", "install", "--no-cache", package]
1337    return_code, output = await check_output(*args)
1338    if return_code != 0:
1339        msg = f"Failed to install package {package}\n{output.decode()}"
1340        raise RuntimeError(msg)
1341
1342
1343async def get_package_version(pkg_name: str) -> str | None:
1344    """
1345    Return the version of an installed (python) package.
1346
1347    Will return None if the package is not found.
1348    """
1349    try:
1350        return await asyncio.to_thread(pkg_version, pkg_name)
1351    except PackageNotFoundError:
1352        return None
1353
1354
1355async def is_hass_supervisor() -> bool:
1356    """Return if we're running inside the HA Supervisor (e.g. HAOS)."""
1357    # Fast path: check for HA supervisor token environment variable
1358    # This is always set when running inside the HA supervisor
1359    if not os.environ.get("SUPERVISOR_TOKEN"):
1360        return False
1361
1362    # Token exists, verify the supervisor is actually reachable
1363    def _check() -> bool:
1364        try:
1365            urllib.request.urlopen("http://supervisor/core", timeout=1)
1366        except urllib.error.URLError as err:
1367            # this should return a 401 unauthorized if it exists
1368            return getattr(err, "code", 999) == 401
1369        except Exception:
1370            return False
1371        return False
1372
1373    return await asyncio.to_thread(_check)
1374
1375
1376# CPython holds a lock per module while importing it, so two threads importing modules with
1377# overlapping dependency graphs (e.g. two providers that both pull in `requests`) can end up
1378# waiting on each other's module locks. The import machinery then bails out at one of them with
1379# a _DeadlockError ("deadlock detected by _ModuleLock(...)") instead of hanging, which surfaces
1380# as a provider that failed to load and stays broken until it is reloaded by hand.
1381# A single-worker executor keeps imports serialized without parking a thread from the default
1382# pool while waiting; only the import itself is serialized, providers still load concurrently.
1383_IMPORT_EXECUTOR = ThreadPoolExecutor(max_workers=1, thread_name_prefix="module_import")
1384
1385# requirements verified this session, so repeated (config) loads skip the version check
1386_checked_requirements: set[str] = set()
1387
1388
1389async def import_module_in_thread(name: str, package: str | None = None) -> ModuleType:
1390    """
1391    Import a module in a thread, serialized against all other imports done this way.
1392
1393    :param name: Name of the module to import, may be relative to the given package.
1394    :param package: Package to resolve the name against, required for a relative name.
1395    """
1396    loop = asyncio.get_running_loop()
1397    try:
1398        return await loop.run_in_executor(_IMPORT_EXECUTOR, importlib.import_module, name, package)
1399    except RuntimeError as err:
1400        # threads we do not control (a library importing lazily in its own thread) can still
1401        # cross a module lock with ours; the import machinery reports that as a deadlock at
1402        # whoever detects it. The other import has finished by now, so a single retry sticks.
1403        if "deadlock detected" not in str(err):
1404            raise
1405        LOGGER.warning("Retrying import of %s after a module lock collision: %s", name, err)
1406        return await loop.run_in_executor(_IMPORT_EXECUTOR, importlib.import_module, name, package)
1407
1408
1409async def load_provider_module(domain: str, requirements: list[str]) -> ProviderModuleType:
1410    """Return module for given provider domain and make sure the requirements are met."""
1411
1412    async def _get_provider_module() -> ProviderModuleType:
1413        module = await import_module_in_thread(f".{domain}", "music_assistant.providers")
1414        return cast("ProviderModuleType", module)
1415
1416    # ensure module requirements are met
1417    for requirement in requirements:
1418        if requirement in _checked_requirements:
1419            continue
1420        if "==" not in requirement:
1421            # we should really get rid of unpinned requirements
1422            continue
1423        package_name, version = requirement.split("==", 1)
1424        # importlib.metadata can't resolve extras (e.g. aiosendspin[server]), so strip them
1425        package_name = package_name.split("[", 1)[0]
1426        installed_version = await get_package_version(package_name)
1427        if installed_version == "0.0.0":
1428            # ignore editable installs
1429            _checked_requirements.add(requirement)
1430            continue
1431        if installed_version != version:
1432            await install_package(requirement)
1433        _checked_requirements.add(requirement)
1434
1435    # try to load the module
1436    try:
1437        return await _get_provider_module()
1438    except ImportError:
1439        # (re)install ALL requirements
1440        for requirement in requirements:
1441            await install_package(requirement)
1442    # try loading the provider again to be safe
1443    # this will fail if something else is wrong (as it should)
1444    return await _get_provider_module()
1445
1446
1447async def has_tmpfs_mount() -> bool:
1448    """Check if we have a tmpfs mount."""
1449
1450    def _has_tmpfs_mount() -> bool:
1451        """Check if we have a tmpfs mount."""
1452        try:
1453            with open("/proc/mounts") as file:
1454                for line in file:
1455                    if "tmpfs /tmp tmpfs rw" in line:
1456                        return True
1457        except FileNotFoundError, OSError, PermissionError:
1458            pass
1459        return False
1460
1461    return await asyncio.to_thread(_has_tmpfs_mount)
1462
1463
1464async def get_free_space(folder: str) -> float:
1465    """Return free space on given folderpath in GB."""
1466
1467    def _get_free_space(folder: str) -> float:
1468        """Return free space on given folderpath in GB."""
1469        try:
1470            res = shutil.disk_usage(folder)
1471            return res.free / float(1 << 30)
1472        except FileNotFoundError, OSError, PermissionError:
1473            return 0.0
1474
1475    return await asyncio.to_thread(_get_free_space, folder)
1476
1477
1478async def get_free_space_percentage(folder: str) -> float:
1479    """Return free space on given folderpath in percentage."""
1480
1481    def _get_free_space(folder: str) -> float:
1482        """Return free space on given folderpath in GB."""
1483        try:
1484            res = shutil.disk_usage(folder)
1485            return res.free / res.total * 100
1486        except FileNotFoundError, OSError, PermissionError:
1487            return 0.0
1488
1489    return await asyncio.to_thread(_get_free_space, folder)
1490
1491
1492async def has_enough_space(folder: str, size: int) -> bool:
1493    """Check if folder has enough free space."""
1494    return await get_free_space(folder) > size
1495
1496
1497def divide_chunks(data: bytes, chunk_size: int) -> Iterator[bytes]:
1498    """Chunk bytes data into smaller chunks."""
1499    for i in range(0, len(data), chunk_size):
1500        yield data[i : i + chunk_size]
1501
1502
1503async def remove_file(file_path: str) -> None:
1504    """Remove file path (if it exists)."""
1505    if not await asyncio.to_thread(os.path.exists, file_path):
1506        return
1507    await asyncio.to_thread(os.remove, file_path)
1508    LOGGER.log(VERBOSE_LOG_LEVEL, "Removed file: %s", file_path)
1509
1510
1511def get_primary_ip_address_from_zeroconf(
1512    discovery_info: AsyncServiceInfo,
1513    prefer_ipv6: bool = False,
1514) -> str | None:
1515    """
1516    Get primary IP address from zeroconf discovery info.
1517
1518    :param discovery_info: The zeroconf service info to extract the address from.
1519    :param prefer_ipv6: If True, prefer IPv6 addresses over IPv4.
1520    """
1521    if prefer_ipv6:
1522        order = [IPVersion.V6Only, IPVersion.V4Only]
1523    else:
1524        order = [IPVersion.V4Only, IPVersion.V6Only]
1525    for version in order:
1526        for addr in discovery_info.ip_addresses_by_version(version):
1527            if addr.is_loopback or addr.is_link_local or addr.is_unspecified:
1528                continue
1529            return str(addr)
1530    return None
1531
1532
1533def get_port_from_zeroconf(discovery_info: AsyncServiceInfo) -> int | None:
1534    """Get port from zeroconf discovery info."""
1535    return discovery_info.port
1536
1537
1538def get_zeroconf_args(
1539    use_all_interfaces: bool = False,
1540) -> dict[str, Any]:
1541    """
1542    Determine optimal zeroconf IPVersion and interfaces from system adapters.
1543
1544    Inspects available network adapters to determine the correct IP version
1545    and interface configuration, similar to Home Assistant's approach.
1546
1547    :param use_all_interfaces: If True, use all interfaces (user override).
1548    """
1549    adapters = ifaddr.get_adapters()
1550    has_ipv4 = False
1551    has_ipv6 = False
1552    interface_ips: list[str] = []
1553    for adapter in adapters:
1554        for ip_config in adapter.ips:
1555            if ip_config.is_IPv6:
1556                ip_tuple = cast("tuple[str, int, int]", ip_config.ip)
1557                addr = ip_address(ip_tuple[0])
1558                if (
1559                    isinstance(addr, IPv6Address)
1560                    and not addr.is_loopback
1561                    and not addr.is_link_local
1562                ):
1563                    has_ipv6 = True
1564                    if not addr.is_global:
1565                        interface_ips.append(f"{ip_tuple[0]}%{ip_tuple[2]}")
1566            else:
1567                ip_str = cast("str", ip_config.ip)
1568                addr = ip_address(ip_str)
1569                if isinstance(addr, IPv4Address) and not addr.is_loopback:
1570                    has_ipv4 = True
1571                    interface_ips.append(ip_str)
1572
1573    # Determine IP version based on available addresses.
1574    # On macOS/FreeBSD, zeroconf's IPVersion.All creates an AF_INET6 listen socket
1575    # that cannot join IPv4 multicast groups, silently breaking discovery of
1576    # IPv4-only devices. Fall back to V4Only on those platforms.
1577    has_functional_dual_stack = not sys.platform.startswith(("freebsd", "darwin"))
1578    if has_ipv4 and has_ipv6 and has_functional_dual_stack:
1579        ip_version = IPVersion.All
1580    elif has_ipv4:
1581        ip_version = IPVersion.V4Only
1582    elif has_ipv6:
1583        ip_version = IPVersion.V6Only
1584    else:
1585        ip_version = IPVersion.V4Only
1586
1587    if use_all_interfaces:
1588        # User explicitly requested all interfaces — pass explicit IP list
1589        # to avoid issues with InterfaceChoice.Default on multi-interface hosts.
1590        if interface_ips:
1591            return {"ip_version": ip_version, "interfaces": interface_ips}
1592        return {"ip_version": ip_version, "interfaces": InterfaceChoice.All}
1593
1594    # Default mode: use InterfaceChoice.Default for IPv4-only single-interface,
1595    # otherwise pass explicit interface list for reliability.
1596    if ip_version == IPVersion.V4Only:
1597        return {"ip_version": ip_version, "interfaces": InterfaceChoice.Default}
1598    if interface_ips:
1599        return {"ip_version": ip_version, "interfaces": interface_ips}
1600    return {"ip_version": ip_version, "interfaces": InterfaceChoice.All}
1601
1602
1603async def close_async_generator(agen: AsyncGenerator[Any]) -> None:
1604    """Force close an async generator."""
1605    task = asyncio.create_task(agen.__anext__())
1606    task.cancel()
1607    with suppress(asyncio.CancelledError, StopAsyncIteration):
1608        await task
1609    await agen.aclose()
1610
1611
1612async def detect_charset(data: bytes, fallback: str = "utf-8") -> str:
1613    """Detect charset of raw data."""
1614    # imported here to keep chardet (~18MB) out of the idle import footprint:
1615    # it is only needed on the rarely-hit playlist/radio charset fallback path
1616    import chardet  # noqa: PLC0415
1617
1618    try:
1619        detected: ResultDict = await asyncio.to_thread(chardet.detect, data)
1620        if detected and detected["encoding"] and detected["confidence"] > 0.75:
1621            assert isinstance(detected["encoding"], str)  # for type checking
1622            return detected["encoding"]
1623    except Exception as err:
1624        LOGGER.debug("Failed to detect charset: %s", err)
1625    return fallback
1626
1627
1628def parse_optional_bool(value: Any) -> bool | None:
1629    """Parse an optional boolean value from various input types."""
1630    if value is None:
1631        return None
1632    if isinstance(value, bool):
1633        return value
1634    if isinstance(value, str):
1635        value_lower = value.strip().lower()
1636        if value_lower in ("true", "1", "yes", "on"):
1637            return True
1638        if value_lower in ("false", "0", "no", "off"):
1639            return False
1640    if isinstance(value, (int, float)):
1641        return bool(value)
1642    return None
1643
1644
1645def merge_dict(
1646    base_dict: dict[Any, Any],
1647    new_dict: dict[Any, Any],
1648    allow_overwite: bool = False,
1649) -> dict[Any, Any]:
1650    """Merge dict without overwriting existing values."""
1651    final_dict = base_dict.copy()
1652    for key, value in new_dict.items():
1653        if final_dict.get(key) and isinstance(value, dict):
1654            final_dict[key] = merge_dict(final_dict[key], value)
1655        if final_dict.get(key) and isinstance(value, tuple):
1656            final_dict[key] = merge_tuples(final_dict[key], value)
1657        if final_dict.get(key) and isinstance(value, list):
1658            final_dict[key] = merge_lists(final_dict[key], value)
1659        elif not final_dict.get(key) or allow_overwite:
1660            final_dict[key] = value
1661    return final_dict
1662
1663
1664def merge_tuples(base: tuple[Any, ...], new: tuple[Any, ...]) -> tuple[Any, ...]:
1665    """Merge 2 tuples."""
1666    return tuple(x for x in base if x not in new) + tuple(new)
1667
1668
1669def merge_lists(base: list[Any], new: list[Any]) -> list[Any]:
1670    """Merge 2 lists."""
1671    return [x for x in base if x not in new] + list(new)
1672
1673
1674def percentage(part: float, whole: float) -> int:
1675    """Calculate percentage."""
1676    return int(100 * float(part) / float(whole))
1677
1678
1679def validate_announcement_chime_url(url: str) -> bool:
1680    """Validate announcement chime URL format."""
1681    if not url or not url.strip():
1682        return True  # Empty URL is valid
1683
1684    if url == ANNOUNCE_ALERT_FILE:
1685        return True  # Built-in chime file is valid
1686
1687    try:
1688        parsed = urlparse(url.strip())
1689
1690        if parsed.scheme not in ("http", "https"):
1691            return False
1692
1693        if not parsed.netloc:
1694            return False
1695
1696        path_lower = parsed.path.lower()
1697        audio_extensions = (".mp3", ".wav", ".flac", ".ogg", ".m4a", ".aac")
1698
1699        return any(path_lower.endswith(ext) for ext in audio_extensions)
1700
1701    except Exception:
1702        return False
1703
1704
1705async def get_mac_address(ip_address: str) -> str | None:
1706    """Get MAC address for given IP address via ARP lookup."""
1707    try:
1708        from getmac import get_mac_address as getmac_lookup  # noqa: PLC0415
1709
1710        return await asyncio.to_thread(getmac_lookup, ip=ip_address)
1711    except ImportError:
1712        LOGGER.debug("getmac module not available, cannot resolve MAC from IP")
1713        return None
1714    except Exception as err:
1715        LOGGER.debug("Failed to resolve MAC address for %s: %s", ip_address, err)
1716        return None
1717
1718
1719def is_locally_administered_mac(mac_address: str) -> bool:
1720    """
1721    Check if a MAC address is locally administered (virtual/randomized).
1722
1723    Locally administered addresses have bit 1 of the first octet set to 1.
1724    These are often used by devices for virtual interfaces or protocol-specific
1725    addresses (e.g., AirPlay, DLNA may use different virtual MACs than the real hardware MAC).
1726
1727    :param mac_address: MAC address in any common format (with :, -, or no separator).
1728    :return: True if locally administered, False if globally unique (real hardware MAC).
1729    """
1730    # Normalize MAC address
1731    mac_clean = mac_address.upper().replace(":", "").replace("-", "")
1732    if len(mac_clean) < 2:
1733        return False
1734
1735    # Get first octet and check bit 1 (second bit from right)
1736    try:
1737        first_octet = int(mac_clean[:2], 16)
1738        return bool(first_octet & 0x02)
1739    except ValueError:
1740        return False
1741
1742
1743def normalize_mac_for_matching(mac_address: str) -> str:
1744    """
1745    Normalize a MAC address for device matching by masking out the locally-administered bit.
1746
1747    Some protocols (like AirPlay) report a locally-administered MAC address variant where
1748    bit 1 of the first octet is set. For example:
1749    - Real hardware MAC: 54:78:C9:E6:0D:A0 (first byte 0x54 = 01010100)
1750    - AirPlay reports:   56:78:C9:E6:0D:A0 (first byte 0x56 = 01010110)
1751
1752    These represent the same device but differ only in the locally-administered bit.
1753    This function normalizes the MAC by clearing bit 1 of the first octet, allowing
1754    both variants to match the same device.
1755
1756    :param mac_address: MAC address in any common format (with :, -, or no separator).
1757    :return: Normalized MAC address in lowercase without separators, with the
1758             locally-administered bit cleared.
1759    """
1760    # Normalize MAC address (remove separators, lowercase)
1761    mac_clean = mac_address.lower().replace(":", "").replace("-", "")
1762    if len(mac_clean) != 12:
1763        # Invalid MAC length, return as-is
1764        return mac_clean
1765
1766    try:
1767        # Parse first octet and clear bit 1 (the locally-administered bit)
1768        first_octet = int(mac_clean[:2], 16)
1769        first_octet_normalized = first_octet & ~0x02  # Clear bit 1
1770        # Reconstruct the MAC with the normalized first octet
1771        return f"{first_octet_normalized:02x}{mac_clean[2:]}"
1772    except ValueError:
1773        # Invalid hex, return as-is
1774        return mac_clean
1775
1776
1777def is_valid_mac_address(mac_address: str | None) -> bool:
1778    """
1779    Check if a MAC address is valid and usable for device identification.
1780
1781    Invalid MAC addresses include:
1782    - None or empty strings
1783    - Null MAC: 00:00:00:00:00:00
1784    - Broadcast MAC: ff:ff:ff:ff:ff:ff
1785    - Any MAC that doesn't follow the expected pattern
1786
1787    :param mac_address: MAC address to validate.
1788    :return: True if valid and usable, False otherwise.
1789    """
1790    if not mac_address:
1791        return False
1792
1793    # Normalize MAC address (remove separators and convert to lowercase)
1794    normalized = mac_address.lower().replace(":", "").replace("-", "")
1795
1796    # Check for invalid/reserved MAC addresses
1797    if normalized in ("000000000000", "ffffffffffff"):
1798        return False
1799
1800    # Check length and hex validity
1801    if len(normalized) != 12:
1802        return False
1803
1804    try:
1805        int(normalized, 16)
1806        return True
1807    except ValueError:
1808        return False
1809
1810
1811def normalize_ip_address(ip_address: str | None) -> str | None:
1812    """
1813    Normalize IP address for comparison.
1814
1815    Handles IPv6-mapped IPv4 addresses (e.g., ::ffff:192.168.1.64 -> 192.168.1.64).
1816
1817    :param ip_address: IP address to normalize.
1818    :return: Normalized IP address or None if invalid.
1819    """
1820    if not ip_address:
1821        return None
1822
1823    # Handle IPv6-mapped IPv4 addresses
1824    if ip_address.startswith("::ffff:"):
1825        # Extract the IPv4 part
1826        return ip_address[7:]
1827
1828    return ip_address
1829
1830
1831async def resolve_real_mac_address(reported_mac: str | None, ip_address: str | None) -> str | None:
1832    """
1833    Resolve the real MAC address for a device.
1834
1835    Some devices report different virtual MAC addresses per protocol (AirPlay, DLNA,
1836    Chromecast). This function tries to resolve the actual hardware MAC via ARP
1837    when the reported MAC appears to be locally administered (virtual).
1838
1839    :param reported_mac: The MAC address reported by the protocol.
1840    :param ip_address: The IP address of the device (for ARP lookup).
1841    :return: The real MAC address if found, or None if it couldn't be resolved.
1842    """
1843    if not ip_address:
1844        return None
1845
1846    # If no MAC reported or it's a locally administered one, try ARP lookup
1847    if not reported_mac or is_locally_administered_mac(reported_mac):
1848        real_mac = await get_mac_address(ip_address)
1849        if real_mac and is_valid_mac_address(real_mac):
1850            return real_mac.upper()
1851
1852    return None
1853
1854
1855async def enrich_device_mac_address(
1856    device_info: DeviceInfo,
1857    logger: logging.Logger | None = None,
1858) -> None:
1859    """
1860    Enrich a player's device_info with a real MAC address via ARP.
1861
1862    Called automatically during player registration. It validates the existing MAC,
1863    normalizes IPv6-mapped IPv4 addresses, and always performs an ARP lookup when
1864    an IP is available. The ARP result replaces the reported MAC because it reflects
1865    the true hardware address and reliably unifies protocols on the same device -
1866    even when different protocols report different valid MACs (e.g., Yamaha devices
1867    where DLNA and AirPlay MACs differ by 1 in the last octet).
1868
1869    :param device_info: The player's DeviceInfo to enrich in-place.
1870    :param logger: Optional logger for debug messages.
1871    """
1872    identifiers = device_info.identifiers
1873    reported_mac = identifiers.get(IdentifierType.MAC_ADDRESS)
1874    ip_address = identifiers.get(IdentifierType.IP_ADDRESS)
1875
1876    # Blank out invalid MAC addresses (00:00:00:00:00:00, ff:ff:ff:ff:ff:ff, etc.)
1877    # so they can't cause false matches in protocol linking.
1878    if reported_mac and not is_valid_mac_address(reported_mac):
1879        if logger:
1880            logger.debug("Removing invalid MAC address: %s", reported_mac)
1881        device_info.add_identifier(IdentifierType.MAC_ADDRESS, None)
1882        reported_mac = None
1883
1884    # Normalize IP address (handle IPv6-mapped IPv4 like ::ffff:192.168.1.64)
1885    if ip_address:
1886        normalized_ip = normalize_ip_address(ip_address)
1887        if normalized_ip and normalized_ip != ip_address:
1888            device_info.add_identifier(IdentifierType.IP_ADDRESS, normalized_ip)
1889            if logger:
1890                logger.debug(
1891                    "Normalized IP address: %s -> %s",
1892                    ip_address,
1893                    normalized_ip,
1894                )
1895            ip_address = normalized_ip
1896
1897    # Skip ARP enrichment if no IP available (can't do ARP lookup)
1898    if not ip_address:
1899        return
1900
1901    # Always attempt ARP lookup when we have an IP address.
1902    # Some devices (e.g., Yamaha MusicCast) report different valid globally-unique
1903    # MACs per protocol (DLNA vs AirPlay differ by 1 in the last octet).
1904    # ARP resolves the true hardware MAC which reliably unifies all protocols.
1905    # The result is cached in player config so subsequent restarts are fast.
1906    real_mac = await resolve_real_mac_address(reported_mac, ip_address)
1907    if real_mac and real_mac.upper() != (reported_mac or "").upper():
1908        device_info.add_identifier(IdentifierType.MAC_ADDRESS, real_mac)
1909        if logger:
1910            logger.debug(
1911                "Resolved MAC via ARP: %s -> %s",
1912                reported_mac or "none",
1913                real_mac,
1914            )
1915    elif not reported_mac:
1916        # ARP failed and no reported MAC - nothing we can do
1917        if logger:
1918            logger.debug("ARP lookup failed for %s and no reported MAC", ip_address)
1919
1920
1921class TaskManager:
1922    """
1923    Helper class to run many tasks at once.
1924
1925    This is basically an alternative to asyncio.TaskGroup but this will not
1926    cancel all operations when one of the tasks fails.
1927    Logging of exceptions is done by the mass.create_task helper.
1928    """
1929
1930    def __init__(self, mass: MusicAssistant, limit: int = 0):
1931        """Initialize the TaskManager."""
1932        self.mass = mass
1933        self._tasks: list[asyncio.Task[None]] = []
1934        self._semaphore = asyncio.Semaphore(limit) if limit else None
1935
1936    def create_task(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[None]:
1937        """Create a new task and add it to the manager."""
1938        task = self.mass.create_task(coro)
1939        self._tasks.append(task)
1940        return task
1941
1942    async def create_task_with_limit(self, coro: Coroutine[Any, Any, Any]) -> None:
1943        """Create a new task with semaphore limit."""
1944        assert self._semaphore is not None
1945
1946        def task_done_callback(_task: asyncio.Task[None]) -> None:
1947            assert self._semaphore is not None  # for type checking
1948            self._tasks.remove(task)
1949            self._semaphore.release()
1950
1951        await self._semaphore.acquire()
1952        task: asyncio.Task[None] = self.create_task(coro)
1953        task.add_done_callback(task_done_callback)
1954
1955    async def __aenter__(self) -> Self:
1956        """Enter context manager."""
1957        return self
1958
1959    async def __aexit__(
1960        self,
1961        exc_type: type[BaseException] | None,
1962        exc_val: BaseException | None,
1963        exc_tb: TracebackType | None,
1964    ) -> bool | None:
1965        """Exit context manager."""
1966        if len(self._tasks) > 0:
1967            await asyncio.wait(self._tasks)
1968            self._tasks.clear()
1969        return None
1970
1971
1972_R = TypeVar("_R")
1973_P = ParamSpec("_P")
1974
1975
1976def lock[**P, R](  # type: ignore[valid-type]
1977    func: Callable[_P, Awaitable[_R]],
1978) -> Callable[_P, Coroutine[Any, Any, _R]]:
1979    """
1980    Call async function using a per-instance Lock.
1981
1982    Each instance gets its own lock so that e.g. SyncGroupPlayer A
1983    does not block SyncGroupPlayer B when both call set_members().
1984    """
1985    # Per-instance lock storage (weak refs so locks are GC'd with their instance)
1986    instance_locks: weakref.WeakKeyDictionary[Any, asyncio.Lock] = weakref.WeakKeyDictionary()
1987    # Fallback lock for non-method (no self) usage
1988    fallback_lock: asyncio.Lock | None = None
1989
1990    @functools.wraps(func)
1991    async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
1992        """Call async function using a per-instance Lock."""
1993        nonlocal fallback_lock
1994        instance = args[0] if args else None
1995        if instance is not None:
1996            try:
1997                func_lock = instance_locks.setdefault(instance, asyncio.Lock())
1998            except TypeError:
1999                # instance is not weakly referenceable, use fallback
2000                if fallback_lock is None:
2001                    fallback_lock = asyncio.Lock()
2002                func_lock = fallback_lock
2003        else:
2004            if fallback_lock is None:
2005                fallback_lock = asyncio.Lock()
2006            func_lock = fallback_lock
2007        async with func_lock:
2008            return await func(*args, **kwargs)
2009
2010    return wrapper
2011
2012
2013class TimedAsyncGenerator:
2014    """
2015    Async iterable that times out after a given time.
2016
2017    Source: https://medium.com/@dmitry8912/implementing-timeouts-in-pythons-asynchronous-generators-f7cbaa6dc1e9
2018    """
2019
2020    def __init__(self, iterable: AsyncIterator[Any], timeout: int = 0):
2021        """
2022        Initialize the AsyncTimedIterable.
2023
2024        Args:
2025            iterable: The async iterable to wrap.
2026            timeout: The timeout in seconds for each iteration.
2027        """
2028
2029        class AsyncTimedIterator:
2030            def __init__(self) -> None:
2031                self._iterator = iterable.__aiter__()
2032
2033            async def __anext__(self) -> Any:
2034                result = await asyncio.wait_for(self._iterator.__anext__(), int(timeout))
2035                if not result:
2036                    raise StopAsyncIteration
2037                return result
2038
2039        self._factory = AsyncTimedIterator
2040
2041    def __aiter__(self):  # type: ignore[no-untyped-def]
2042        """Return the async iterator."""
2043        return self._factory()
2044
2045
2046async def join_task[T](task: asyncio.Future[T], timeout: float | None = None) -> T:
2047    """
2048    Wait for a task started elsewhere and return its result.
2049
2050    Cancelling the waiter leaves the task running, so work that is shared between callers -
2051    or that must outlive a caller's deadline - keeps going and still reaches every other
2052    waiter. A task that can lose all its waiters needs a done callback that retrieves its
2053    exception (as mass.create_task installs) to keep asyncio quiet about it.
2054
2055    :param task: The task (or future) to wait for.
2056    :param timeout: Optional number of seconds to wait before giving up.
2057    :raises TimeoutError: If the task did not complete within the timeout.
2058    :raises asyncio.CancelledError: If the task itself was cancelled.
2059    :return: The task's result.
2060    """
2061    if not task.done():
2062        # awaiting the task directly would hold it as this coroutine's fut_waiter, so
2063        # cancelling the waiter would cancel the task itself. asyncio.shield achieves the
2064        # same isolation, but as of Python 3.14 a cancelled waiter makes it report the task's
2065        # exception through loop.call_exception_handler, even when another waiter already
2066        # handled it.
2067        await asyncio.wait((task,), timeout=timeout)
2068    if not task.done():
2069        raise TimeoutError
2070    return task.result()
2071
2072
2073# Bound for guard_single_request: it only needs ``.mass``, so a structural protocol
2074# lets it decorate providers, core controllers and media controllers alike without
2075# coupling to their concrete base classes.
2076class _SupportsMass(Protocol):
2077    """Structural type for objects exposing a MusicAssistant reference."""
2078
2079    mass: MusicAssistant
2080
2081
2082def guard_single_request[SelfT: _SupportsMass, **P, R](
2083    func: Callable[Concatenate[SelfT, P], Coroutine[Any, Any, R]],
2084) -> Callable[Concatenate[SelfT, P], Coroutine[Any, Any, R]]:
2085    """
2086    Ensure concurrent calls with identical arguments result in a single request.
2087
2088    Callers arriving while an identical call is already in flight await that same call and
2089    receive its result. Cancelling one caller leaves both the request and the other callers
2090    unaffected. Calls count as identical when they are made on the same object with equally
2091    represented arguments.
2092
2093    :param func: The coroutine method to guard.
2094    """
2095
2096    @functools.wraps(func)
2097    async def wrapper(self: SelfT, *args: P.args, **kwargs: P.kwargs) -> R:
2098        mass = self.mass
2099        # create a task_id dynamically based on the bound method and args/kwargs.
2100        # the instance is part of the key because a decorated method may be inherited by
2101        # multiple subclasses (all media controllers share
2102        # MediaControllerBase.get_provider_item) and a class may have multiple instances
2103        # (e.g. a provider set up twice), which must never join each other's flight.
2104        # id(self) is stable while a flight is live because the task references self;
2105        # the class name only serves to keep the task_id readable while debugging.
2106        cache_key_parts = [type(self).__name__, id(self), func.__qualname__, *args]
2107        for key in sorted(kwargs.keys()):
2108            cache_key_parts.append(f"{key}{kwargs[key]}")
2109        task_id = ".".join(map(str, cache_key_parts))
2110        task: asyncio.Task[R] = mass.create_task(
2111            func,
2112            self,
2113            *args,
2114            task_id=task_id,
2115            abort_existing=False,
2116            eager_start=True,
2117            **kwargs,
2118        )
2119        # wait for the shared task instead of awaiting it directly: a caller awaiting a
2120        # task holds it as its fut_waiter, so cancelling that caller would cancel the
2121        # request for every other caller too. asyncio.shield achieves the same, but as of
2122        # Python 3.14 a cancelled caller makes it report the request's exception through
2123        # loop.call_exception_handler, even when another caller already handled it.
2124        if not task.done():
2125            await asyncio.wait((task,))
2126        return task.result()
2127
2128    return wrapper
2129