/
/
/
1"""Base classes for WAM features."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import Callable, Coroutine
7from functools import wraps
8from logging import Logger
9from typing import TYPE_CHECKING, Any
10
11from music_assistant_models.errors import PlayerCommandFailed
12from pywam.lib.exceptions import ApiCallTimeoutError, PywamError
13
14from music_assistant.providers.samsung_wam.consts import (
15 COMMAND_RETRY_ATTEMPTS,
16 COMMAND_RETRY_BACKOFF,
17)
18
19if TYPE_CHECKING:
20 from pywam.speaker import Speaker
21
22 from music_assistant.mass import MusicAssistant
23 from music_assistant.providers.samsung_wam.player import WamPlayer
24 from music_assistant.providers.samsung_wam.provider import SamsungWamProvider
25
26
27class WamProviderFeatureBase:
28 """Base class for provider feature handlers."""
29
30 def __init__(self, provider: SamsungWamProvider) -> None:
31 """
32 Initialize the feature base with the parent provider instance.
33
34 :param provider: The SamsungWamProvider instance.
35 """
36 self.provider = provider
37
38 @property
39 def mass(self) -> MusicAssistant:
40 """Return the MusicAssistant instance."""
41 return self.provider.mass
42
43 @property
44 def logger(self) -> Logger:
45 """Return the provider's dedicated logger."""
46 return self.provider.logger
47
48 @property
49 def players(self) -> list[WamPlayer]:
50 """Return all registered WAM players."""
51 return self.provider.get_players()
52
53
54class WamPlayerFeatureBase:
55 """Base class for player feature handlers."""
56
57 def __init__(self, player: WamPlayer) -> None:
58 """
59 Initialize the feature base with the parent player instance.
60
61 :param player: The WamPlayer instance.
62 """
63 self.player = player
64
65 @property
66 def mass(self) -> MusicAssistant:
67 """Return the MusicAssistant instance."""
68 return self.player.mass
69
70 @property
71 def speaker(self) -> Speaker:
72 """Return the pywam Speaker instance."""
73 return self.player.speaker
74
75 @property
76 def logger(self) -> Logger:
77 """Return the player's dedicated logger."""
78 return self.player.logger
79
80
81def retry_command(
82 attempts: int = COMMAND_RETRY_ATTEMPTS,
83 initial_backoff: float = COMMAND_RETRY_BACKOFF,
84) -> Callable[[Callable[..., Coroutine[Any, Any, Any]]], Callable[..., Coroutine[Any, Any, Any]]]:
85 """
86 Decorate a player command to retry on temporary failures.
87
88 :param attempts: Total number of attempts to make.
89 :param initial_backoff: Initial wait time in seconds before retrying.
90 """
91
92 def decorator(
93 func: Callable[..., Coroutine[Any, Any, Any]],
94 ) -> Callable[..., Coroutine[Any, Any, Any]]:
95 @wraps(func)
96 async def wrapper(self: WamPlayerFeatureBase, *args: Any, **kwargs: Any) -> Any:
97 backoff_delay = initial_backoff
98 last_error: Exception | None = None
99
100 for attempt in range(attempts):
101 try:
102 await self.player.state_sync.ensure_speaker_connected()
103 result = await func(self, *args, **kwargs)
104
105 if attempt > 0:
106 self.logger.debug(
107 "Command '%s' recovered on attempt %s/%s",
108 func.__name__,
109 attempt + 1,
110 attempts,
111 )
112 return result
113
114 except (TimeoutError, PlayerCommandFailed, PywamError) as err:
115 last_error = err
116 if attempt < attempts - 1:
117 self.logger.debug(
118 "Command '%s' failed on attempt %s/%s, retrying in %.1fs",
119 func.__name__,
120 attempt + 1,
121 attempts,
122 backoff_delay,
123 )
124 await asyncio.sleep(backoff_delay)
125 backoff_delay *= 2
126 else:
127 self.logger.debug(
128 "Command '%s' failed on final attempt %s/%s",
129 func.__name__,
130 attempt + 1,
131 attempts,
132 )
133
134 self.logger.warning("Command '%s' failed after %s attempts", func.__name__, attempts)
135 raise PlayerCommandFailed(
136 f"Command '{func.__name__}' failed after {attempts} attempts"
137 ) from last_error
138
139 return wrapper
140
141 return decorator
142
143
144def handle_pywam_errors(
145 func: Callable[..., Coroutine[Any, Any, Any]],
146) -> Callable[..., Coroutine[Any, Any, Any]]:
147 """Decorate a command to translate pywam exceptions into player command errors."""
148
149 @wraps(func)
150 async def wrapper(self: WamPlayerFeatureBase, *args: Any, **kwargs: Any) -> Any:
151 try:
152 return await func(self, *args, **kwargs)
153 except ConnectionError as err:
154 self.logger.warning("Command failed with a connection error")
155 raise PlayerCommandFailed("Connection lost") from err
156 except (ApiCallTimeoutError, TimeoutError, PywamError) as err:
157 raise PlayerCommandFailed(str(err)) from err
158
159 return wrapper
160