/
/
/
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.exceptions import SoCoException, SoCoUPnPException
13
14from .constants import UID_POSTFIX, UID_PREFIX
15
16if TYPE_CHECKING:
17 from soco import SoCo
18
19 from .player import SonosPlayer
20
21
22_LOGGER = logging.getLogger(__name__)
23
24_T = TypeVar("_T", bound="SonosPlayer")
25_R = TypeVar("_R")
26_P = ParamSpec("_P")
27
28_FuncType = Callable[Concatenate[_T, _P], _R]
29_ReturnFuncType = Callable[Concatenate[_T, _P], _R | None]
30
31
32class SonosUpdateError(PlayerCommandFailed):
33 """Update failed."""
34
35
36@overload
37def soco_error(
38 errorcodes: None = ...,
39) -> Callable[[_FuncType[_T, _P, _R]], _FuncType[_T, _P, _R]]: ...
40
41
42@overload
43def soco_error(
44 errorcodes: list[str],
45) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]: ...
46
47
48def soco_error(
49 errorcodes: list[str] | None = None,
50) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]:
51 """Filter out specified UPnP errors and raise exceptions for service calls."""
52
53 def decorator(funct: _FuncType[_T, _P, _R]) -> _ReturnFuncType[_T, _P, _R]:
54 """Decorate functions."""
55 if inspect.iscoroutinefunction(funct):
56
57 @functools.wraps(funct)
58 async def async_wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> Any:
59 """Await the call so soco UPnP exceptions surface inside the try block."""
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)
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 try:
72 result = funct(self, *args, **kwargs)
73 except (OSError, SoCoException, SoCoUPnPException, TimeoutError) as err:
74 _handle_soco_error(self, err, funct.__qualname__, errorcodes)
75 return None
76
77 return result
78
79 return wrapper
80
81 return decorator
82
83
84def hostname_to_uid(hostname: str) -> str:
85 """Convert a Sonos hostname to a uid."""
86 if hostname.startswith("Sonos-"):
87 baseuid = hostname.removeprefix("Sonos-").replace(".local.", "")
88 elif hostname.startswith("sonos"):
89 baseuid = hostname.removeprefix("sonos").replace(".local.", "")
90 else:
91 msg = f"{hostname} is not a sonos device."
92 raise ValueError(msg)
93 return f"{UID_PREFIX}{baseuid}{UID_POSTFIX}"
94
95
96def sync_get_visible_zones(soco: SoCo) -> set[SoCo]:
97 """Ensure I/O attributes are cached and return visible zones."""
98 _ = soco.household_id
99 _ = soco.uid
100 return soco.visible_zones or set()
101
102
103def _handle_soco_error(
104 instance: SonosPlayer,
105 err: Exception,
106 function: str,
107 errorcodes: list[str] | None,
108) -> None:
109 """Ignore a filtered UPnP error code or raise the error as a SonosUpdateError."""
110 error_code = getattr(err, "error_code", None)
111 if errorcodes and error_code in errorcodes:
112 _LOGGER.debug("Error code %s ignored in call to %s", error_code, function)
113 return
114
115 soco = instance.soco
116 # Only use attributes with no I/O
117 target = soco._player_name or soco.ip_address
118 message = f"Error calling {function} on {target}: {err}"
119 raise SonosUpdateError(message) from err
120