music-assistant-server

1"""Helper methods for common tasks."""
2
3from __future__ import annotations
4
5import logging
6from collections.abc import Callable
7from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar, overload
8
9from music_assistant_models.errors import PlayerCommandFailed
10from soco import SoCo
11from soco.exceptions import SoCoException, SoCoUPnPException
12
13from .constants import UID_POSTFIX, UID_PREFIX
14
15if TYPE_CHECKING:
16    from .player import SonosPlayer
17
18
19_LOGGER = logging.getLogger(__name__)
20
21_T = TypeVar("_T", bound="SonosPlayer")
22_R = TypeVar("_R")
23_P = ParamSpec("_P")
24
25_FuncType = Callable[Concatenate[_T, _P], _R]
26_ReturnFuncType = Callable[Concatenate[_T, _P], _R | None]
27
28
29class SonosUpdateError(PlayerCommandFailed):
30    """Update failed."""
31
32
33@overload
34def soco_error(
35    errorcodes: None = ...,
36) -> Callable[[_FuncType[_T, _P, _R]], _FuncType[_T, _P, _R]]: ...
37
38
39@overload
40def soco_error(
41    errorcodes: list[str],
42) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]: ...
43
44
45def soco_error(
46    errorcodes: list[str] | None = None,
47) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]:
48    """Filter out specified UPnP errors and raise exceptions for service calls."""
49
50    def decorator(funct: _FuncType[_T, _P, _R]) -> _ReturnFuncType[_T, _P, _R]:
51        """Decorate functions."""
52
53        def wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R | None:
54            """Wrap for all soco UPnP exception."""
55            args_soco = next((arg for arg in args if isinstance(arg, SoCo)), None)
56            try:
57                result = funct(self, *args, **kwargs)
58            except (OSError, SoCoException, SoCoUPnPException, TimeoutError) as err:
59                error_code = getattr(err, "error_code", None)
60                function = funct.__qualname__
61                if errorcodes and error_code in errorcodes:
62                    _LOGGER.debug("Error code %s ignored in call to %s", error_code, function)
63                    return None
64
65                if (target := _find_target_identifier(self, args_soco)) is None:
66                    msg = "Unexpected use of soco_error"
67                    raise RuntimeError(msg) from err
68
69                message = f"Error calling {function} on {target}: {err}"
70                raise SonosUpdateError(message) from err
71
72            return result
73
74        return wrapper
75
76    return decorator
77
78
79def _find_target_identifier(instance: Any, fallback_soco: SoCo | None) -> str | None:
80    """Extract the best available target identifier from the provided instance object."""
81    if zone_name := getattr(instance, "zone_name", None):
82        # SonosPlayer instance
83        return str(zone_name)
84    if soco := getattr(instance, "soco", fallback_soco):
85        # Holds a SoCo instance attribute
86        # Only use attributes with no I/O
87        return str(soco._player_name or soco.ip_address)
88    return None
89
90
91def hostname_to_uid(hostname: str) -> str:
92    """Convert a Sonos hostname to a uid."""
93    if hostname.startswith("Sonos-"):
94        baseuid = hostname.removeprefix("Sonos-").replace(".local.", "")
95    elif hostname.startswith("sonos"):
96        baseuid = hostname.removeprefix("sonos").replace(".local.", "")
97    else:
98        msg = f"{hostname} is not a sonos device."
99        raise ValueError(msg)
100    return f"{UID_PREFIX}{baseuid}{UID_POSTFIX}"
101
102
103def sync_get_visible_zones(soco: SoCo) -> set[SoCo]:
104    """Ensure I/O attributes are cached and return visible zones."""
105    _ = soco.household_id
106    _ = soco.uid
107    return soco.visible_zones or set()
108