/
/
/
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 current_user = get_current_user()
86 if (
87 current_user
88 and current_user.player_filter
89 and player.player_id not in current_user.player_filter
90 ):
91 msg = (
92 f"{current_user.username} does not have access to player {player.display_name}"
93 )
94 raise InsufficientPermissions(msg)
95
96 self.logger.debug(
97 "Handling command %s for player %s (%s)",
98 fn.__name__,
99 player.display_name,
100 f"by user {current_user.username}" if current_user else "unauthenticated",
101 )
102
103 async def execute() -> None:
104 async with self._player_throttlers[player_id]:
105 try:
106 await fn(self, *args, **kwargs)
107 except Exception as err:
108 raise PlayerCommandFailed(str(err)) from err
109
110 if lock:
111 # Acquire a lock specific to player_id and function name
112 lock_key = f"{fn.__name__}_{player_id}"
113 if lock_key not in self._player_command_locks:
114 self._player_command_locks[lock_key] = asyncio.Lock()
115 async with self._player_command_locks[lock_key]:
116 await execute()
117 else:
118 await execute()
119
120 return wrapper
121
122 # Support both @handle_player_command and @handle_player_command(lock=True)
123 if func is not None:
124 return decorator(func)
125 return decorator
126