/
/
/
1"""
2Helper utilities for the Player Controller.
3
4Contains decorators, type definitions, and utility functions used by the
5PlayerController that don't need direct access to the controller class.
6"""
7
8from __future__ import annotations
9
10import asyncio
11import functools
12from collections.abc import Awaitable, Callable, Coroutine
13from typing import TYPE_CHECKING, Any, Concatenate, TypedDict, overload
14
15from music_assistant_models.errors import (
16 InsufficientPermissions,
17 MusicAssistantError,
18 PlayerCommandFailed,
19)
20
21from music_assistant.controllers.players.constants import PlayerLockPurpose
22from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
23
24if TYPE_CHECKING:
25 import logging
26
27 from music_assistant_models.player_control import PlayerControl
28
29 from music_assistant.models.player import Player
30
31 from .controller import PlayerController
32
33
34class AnnounceData(TypedDict):
35 """Announcement data for play_announcement command."""
36
37 announcement_url: str
38 pre_announce: bool
39 pre_announce_url: str
40 # player that fetches the announcement stream when it is not the
41 # visible player itself (e.g. a linked protocol player)
42 announce_player_id: str | None
43
44
45@overload
46def handle_player_command[PlayerControllerT: "PlayerController", **P, R](
47 func: Callable[Concatenate[PlayerControllerT, P], Awaitable[R]],
48) -> Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]]: ...
49
50
51@overload
52def handle_player_command[PlayerControllerT: "PlayerController", **P, R](
53 func: None = None,
54 *,
55 lock: PlayerLockPurpose | None = None,
56) -> Callable[
57 [Callable[Concatenate[PlayerControllerT, P], Awaitable[R]]],
58 Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]],
59]: ...
60
61
62def handle_player_command[PlayerControllerT: "PlayerController", **P, R](
63 func: Callable[Concatenate[PlayerControllerT, P], Awaitable[R]] | None = None,
64 *,
65 lock: PlayerLockPurpose | None = None,
66) -> (
67 Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]]
68 | Callable[
69 [Callable[Concatenate[PlayerControllerT, P], Awaitable[R]]],
70 Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]],
71 ]
72):
73 """
74 Decorator to check and log commands to players.
75
76 Validates that the player exists and is available before executing the command.
77 Also checks user permissions and optionally acquires a per-player lock.
78
79 :param func: The function to wrap (when used without parentheses).
80 :param lock: PlayerLockPurpose to serialize commands in the same category per
81 player. Commands with the same lock purpose on the same player will not run
82 concurrently. None (default) means no locking.
83 """ # noqa: D401
84
85 def decorator(
86 fn: Callable[Concatenate[PlayerControllerT, P], Awaitable[R]],
87 ) -> Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]]:
88 @functools.wraps(fn)
89 async def wrapper(self: PlayerControllerT, *args: P.args, **kwargs: P.kwargs) -> None:
90 """Log and handle_player_command commands to players."""
91 player_id = kwargs.get("player_id") or args[0]
92 assert isinstance(player_id, str) # for type checking
93 if (player := self._players.get(player_id)) is None or not player.available:
94 self.logger.warning(
95 "Ignoring command %s for unavailable player %s",
96 fn.__name__,
97 player_id,
98 )
99 return
100
101 # this should not happen, but in case a player_id of a protocol player is used,
102 # auto-resolve it to the parent player
103 if player.protocol_parent_id and (
104 protocol_parent := self._players.get(player.protocol_parent_id)
105 ):
106 player = protocol_parent
107 if "player_id" in kwargs:
108 kwargs["player_id"] = protocol_parent.player_id
109 else:
110 args = (protocol_parent.player_id, *args[1:]) # type: ignore[assignment]
111 self.logger.debug(
112 "Auto-resolved protocol player %s to linked parent %s for command %s",
113 player_id,
114 protocol_parent.player_id,
115 fn.__name__,
116 )
117
118 current_user = get_current_user()
119 if (
120 current_user
121 and current_user.player_filter
122 and player.player_id not in current_user.player_filter
123 ):
124 msg = (
125 f"{current_user.username} does not have access to player {player.display_name}"
126 )
127 raise InsufficientPermissions(msg)
128
129 self.logger.debug(
130 "Handling command %s for player %s (%s)",
131 fn.__name__,
132 player.display_name,
133 f"by user {current_user.username}" if current_user else "unauthenticated",
134 )
135
136 try:
137 if lock:
138 async with self.get_player_lock(player.player_id, lock):
139 await fn(self, *args, **kwargs)
140 else:
141 await fn(self, *args, **kwargs)
142 except MusicAssistantError:
143 # A typed error already carries its own error code and translation
144 # (e.g. "this device needs a password"); re-wrapping it here would
145 # flatten every specific failure into the generic message.
146 raise
147 except Exception as err:
148 raise PlayerCommandFailed(str(err)) from err
149
150 return wrapper
151
152 # Support both @handle_player_command and @handle_player_command(lock=...)
153 if func is not None:
154 return decorator(func)
155 return decorator
156
157
158async def wait_for_power_on(
159 logger: logging.Logger,
160 player: Player,
161 player_control: PlayerControl | None = None,
162 timeout: float = 5.0,
163) -> None:
164 """
165 Wait for a player (or player control) to report powered on after a power on command.
166
167 :param logger: Logger instance for debug logging.
168 :param player: The player to wait for (checked when player_control is None).
169 :param player_control: Optional PlayerControl to check instead of the player.
170 :param timeout: Maximum time to wait in seconds.
171 """
172 try:
173 async with asyncio.timeout(timeout):
174 if player_control is not None:
175 while not player_control.power_state:
176 await asyncio.sleep(0.1)
177 else:
178 while not player.powered:
179 await asyncio.sleep(0.1)
180 except TimeoutError:
181 logger.debug(
182 "Player %s did not report powered on within %s seconds",
183 player.state.name,
184 timeout,
185 )
186