music-assistant-server

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