/
/
/
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