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