/
/
/
1"""Game state helpers for the Music Quiz provider."""
2
3from __future__ import annotations
4
5from music_assistant_models.errors import InvalidDataError
6
7from music_assistant.providers.music_quiz.answer_types.base import (
8 QuizAnswerSubmission,
9 QuizAnswerType,
10)
11from music_assistant.providers.music_quiz.errors import (
12 MusicQuizNameTakenError,
13 MusicQuizNotActiveThisRoundError,
14 MusicQuizUnknownPlayerError,
15 MusicQuizWrongPhaseError,
16)
17from music_assistant.providers.music_quiz.models import (
18 MusicQuizGame,
19 MusicQuizPhase,
20 MusicQuizPlayer,
21 MusicQuizRound,
22)
23
24
25def add_player(
26 game: MusicQuizGame,
27 player: MusicQuizPlayer,
28) -> None:
29 """
30 Add a player to a game.
31
32 :param game: Game to mutate.
33 :param player: Player to add.
34 """
35 if player.player_id in game.players:
36 raise InvalidDataError("Player already exists")
37 normalized_name = player.name.casefold()
38 if any(existing.name.casefold() == normalized_name for existing in game.players.values()):
39 raise MusicQuizNameTakenError("Player name must be unique")
40 game.players[player.player_id] = player
41
42
43def remove_player(
44 game: MusicQuizGame,
45 player_id: str,
46 answer_type: QuizAnswerType,
47) -> None:
48 """
49 Remove a player and their answer-specific round state.
50
51 :param game: Game to mutate.
52 :param player_id: Player to remove.
53 :param answer_type: Answer strategy for the game.
54 """
55 if player_id not in game.players:
56 raise MusicQuizUnknownPlayerError("Unknown player")
57 for game_round in game.rounds:
58 answer_type.remove_player(game_round.answer_state, player_id)
59 del game.players[player_id]
60
61
62def submit_answer(
63 game: MusicQuizGame,
64 player_id: str,
65 submission: QuizAnswerSubmission,
66 submitted_at: float,
67 answer_type: QuizAnswerType,
68) -> None:
69 """
70 Submit a validated player answer for the current round.
71
72 :param game: Game to mutate.
73 :param player_id: Player submitting the answer.
74 :param submission: Validated answer submission.
75 :param submitted_at: Server timestamp of the submission.
76 :param answer_type: Answer strategy for the game.
77 """
78 if game.phase != MusicQuizPhase.ANSWERING:
79 raise MusicQuizWrongPhaseError("Answers can only be submitted during the answering phase")
80 current_round = get_current_round(game)
81 if player_id not in game.players:
82 raise MusicQuizUnknownPlayerError("Unknown player")
83 player = game.players[player_id]
84 if player.active_from_round > current_round.round_index:
85 raise MusicQuizNotActiveThisRoundError("Player is not active for this round")
86 answer_type.submit(game, current_round.answer_state, player, submission, submitted_at)
87
88
89def reveal_round(game: MusicQuizGame, answer_type: QuizAnswerType) -> None:
90 """
91 Reveal the current round and apply scores.
92
93 :param game: Game to mutate.
94 :param answer_type: Answer strategy for the game.
95 """
96 current_round = get_current_round(game)
97 if game.phase != MusicQuizPhase.ANSWERING:
98 raise MusicQuizWrongPhaseError("Round can only be revealed during the answering phase")
99 answer_timestamp = answer_type.reveal(game, current_round.answer_state)
100 current_round.ended_at = (
101 answer_timestamp if answer_timestamp is not None else current_round.started_at or 0
102 )
103 game.phase = MusicQuizPhase.REVEAL
104 for player in game.players.values():
105 player.ready = False
106
107
108def start_round(
109 game: MusicQuizGame,
110 music_quiz_round: MusicQuizRound,
111 started_at: float,
112 answer_type: QuizAnswerType,
113) -> None:
114 """
115 Start a new answering round.
116
117 :param game: Game to mutate.
118 :param music_quiz_round: Round to append and make current.
119 :param started_at: Round start timestamp.
120 :param answer_type: Answer strategy for the game.
121 """
122 if game.phase not in (MusicQuizPhase.LOBBY, MusicQuizPhase.REVEAL):
123 raise InvalidDataError("A round cannot be started from the current phase")
124 if len(game.rounds) >= game.config.round_count:
125 raise InvalidDataError("All configured rounds have already been played")
126 expected_index = len(game.rounds)
127 if music_quiz_round.round_index != expected_index:
128 raise InvalidDataError("Round index does not match the game state")
129 answer_type.validate_round(game, music_quiz_round.answer_state)
130 music_quiz_round.started_at = started_at
131 music_quiz_round.ended_at = None
132 game.rounds.append(music_quiz_round)
133 game.current_round_index = music_quiz_round.round_index
134 game.phase = MusicQuizPhase.ANSWERING
135 for player in game.players.values():
136 player.ready = False
137
138
139def mark_player_ready(game: MusicQuizGame, player_id: str) -> None:
140 """
141 Mark a player ready during the reveal/listening phase.
142
143 :param game: Game to mutate.
144 :param player_id: Player ID.
145 """
146 if game.phase != MusicQuizPhase.REVEAL:
147 raise MusicQuizWrongPhaseError("Players can only become ready during reveal")
148 if player_id not in game.players:
149 raise MusicQuizUnknownPlayerError("Unknown player")
150 game.players[player_id].ready = True
151
152
153def are_active_players_ready(game: MusicQuizGame) -> bool:
154 """
155 Return whether every player of the upcoming round is ready.
156
157 :param game: Game to inspect.
158 """
159 current_round = get_current_round(game)
160 # +1: during reveal the next round belongs to late joiners too, so the
161 # game must not advance from under them â capped at the last real round,
162 # because a final-reveal joiner will never play and must not block finish
163 gate_round_index = min(current_round.round_index + 1, game.config.round_count - 1)
164 players = active_players_for_round(game, gate_round_index)
165 return bool(players) and all(player.ready for player in players)
166
167
168def all_active_players_complete(game: MusicQuizGame, answer_type: QuizAnswerType) -> bool:
169 """
170 Return whether every active player completed the current round.
171
172 :param game: Game to inspect.
173 :param answer_type: Answer strategy for the game.
174 """
175 if game.phase != MusicQuizPhase.ANSWERING:
176 return False
177 current_round = get_current_round(game)
178 players = active_players_for_round(game, current_round.round_index)
179 return bool(game.players) and (
180 not players or answer_type.is_round_complete(current_round.answer_state, players)
181 )
182
183
184def finish_game(game: MusicQuizGame) -> None:
185 """
186 Mark a game finished.
187
188 :param game: Game to mutate.
189 """
190 if game.phase != MusicQuizPhase.REVEAL:
191 raise InvalidDataError("A game can only be finished from the reveal phase")
192 game.phase = MusicQuizPhase.FINISHED
193
194
195def reset_game(game: MusicQuizGame) -> None:
196 """
197 Reset a game for a new run with the same config and players.
198
199 :param game: Game to mutate.
200 """
201 game.phase = MusicQuizPhase.LOBBY
202 game.auto_start_at = None
203 game.rounds.clear()
204 game.current_round_index = None
205 for player in game.players.values():
206 player.score = 0
207 player.ready = False
208 player.active_from_round = 0
209
210
211def active_players_for_round(game: MusicQuizGame, round_index: int) -> list[MusicQuizPlayer]:
212 """
213 Return players active for a round.
214
215 :param game: Game to inspect.
216 :param round_index: Round index.
217 """
218 return [player for player in game.players.values() if player.active_from_round <= round_index]
219
220
221def get_current_round(game: MusicQuizGame) -> MusicQuizRound:
222 """
223 Return the current round.
224
225 :param game: Game to inspect.
226 """
227 if game.current_round_index is None:
228 raise InvalidDataError("No current round")
229 try:
230 return game.rounds[game.current_round_index]
231 except IndexError as err:
232 raise InvalidDataError("Current round does not exist") from err
233