/
/
/
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 InsufficientPermissions, PlayerCommandFailed
16
17from music_assistant.controllers.webserver.helpers.auth_middleware import get_current_user
18
19if TYPE_CHECKING:
20 from .controller import PlayerController
21
22
23class AnnounceData(TypedDict):
24 """Announcement data for play_announcement command."""
25
26 announcement_url: str
27 pre_announce: bool
28 pre_announce_url: str
29
30
31@overload
32def handle_player_command[PlayerControllerT: "PlayerController", **P, R](
33 func: Callable[Concatenate[PlayerControllerT, P], Awaitable[R]],
34) -> Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]]: ...
35
36
37@overload
38def handle_player_command[PlayerControllerT: "PlayerController", **P, R](
39 func: None = None,
40 *,
41 lock: bool = False,
42) -> Callable[
43 [Callable[Concatenate[PlayerControllerT, P], Awaitable[R]]],
44 Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]],
45]: ...
46
47
48def handle_player_command[PlayerControllerT: "PlayerController", **P, R](
49 func: Callable[Concatenate[PlayerControllerT, P], Awaitable[R]] | None = None,
50 *,
51 lock: bool = False,
52) -> (
53 Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]]
54 | Callable[
55 [Callable[Concatenate[PlayerControllerT, P], Awaitable[R]]],
56 Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]],
57 ]
58):
59 """
60 Decorator to check and log commands to players.
61
62 Validates that the player exists and is available before executing the command.
63 Also checks user permissions and optionally acquires a per-player lock.
64
65 :param func: The function to wrap (when used without parentheses).
66 :param lock: If True, acquire a lock per player_id and function name before executing.
67 """ # noqa: D401
68
69 def decorator(
70 fn: Callable[Concatenate[PlayerControllerT, P], Awaitable[R]],
71 ) -> Callable[Concatenate[PlayerControllerT, P], Coroutine[Any, Any, R | None]]:
72 @functools.wraps(fn)
73 async def wrapper(self: PlayerControllerT, *args: P.args, **kwargs: P.kwargs) -> None:
74 """Log and handle_player_command commands to players."""
75 player_id = kwargs.get("player_id") or args[0]
76 assert isinstance(player_id, str) # for type checking
77 if (player := self._players.get(player_id)) is None or not player.available:
78 self.logger.warning(
79 "Ignoring command %s for unavailable player %s",
80 fn.__name__,
81 player_id,
82 )
83 return
84
85 # this should not happen, but in case a player_id of a protocol player is used,
86 # auto-resolve it to the parent player
87 if player.protocol_parent_id and (
88 protocol_parent := self._players.get(player.protocol_parent_id)
89 ):
90 player = protocol_parent
91 if "player_id" in kwargs:
92 kwargs["player_id"] = protocol_parent.player_id
93 else:
94 args = (protocol_parent.player_id, *args[1:]) # type: ignore[assignment]
95 self.logger.debug(
96 "Auto-resolved protocol player %s to linked parent %s for command %s",
97 player_id,
98 protocol_parent.player_id,
99 fn.__name__,
100 )
101
102 current_user = get_current_user()
103 if (
104 current_user
105 and current_user.player_filter
106 and player.player_id not in current_user.player_filter
107 ):
108 msg = (
109 f"{current_user.username} does not have access to player {player.display_name}"
110 )
111 raise InsufficientPermissions(msg)
112
113 self.logger.debug(
114 "Handling command %s for player %s (%s)",
115 fn.__name__,
116 player.display_name,
117 f"by user {current_user.username}" if current_user else "unauthenticated",
118 )
119
120 async def execute() -> None:
121 async with self._player_throttlers[player.player_id]:
122 try:
123 await fn(self, *args, **kwargs)
124 except Exception as err:
125 raise PlayerCommandFailed(str(err)) from err
126
127 if lock:
128 # Acquire a lock specific to player_id and function name
129 lock_key = f"{fn.__name__}_{player_id}"
130 if lock_key not in self._player_command_locks:
131 self._player_command_locks[lock_key] = asyncio.Lock()
132 async with self._player_command_locks[lock_key]:
133 await execute()
134 else:
135 await execute()
136
137 return wrapper
138
139 # Support both @handle_player_command and @handle_player_command(lock=True)
140 if func is not None:
141 return decorator(func)
142 return decorator
143