Coverage for src / beaverbunch / network / protocol.py: 100.0%
540 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 19:37 +0000
1from __future__ import annotations
3from dataclasses import asdict, dataclass
4from typing import Any, Callable, ClassVar, Literal
6from beaverbunch.core.game_settings import GameSettings
7from beaverbunch.network.session import Session
9RequestType = Literal[
10 "create_session",
11 "join_session",
12 "start_session",
13 "get_session",
14 "leave_session",
15 "close_session",
16 "get_game",
17 "peek_initial",
18 "draw_card",
19 "draw_discard",
20 "keep_card",
21 "discard_drawn",
22 "peek_own",
23 "swap_card",
24 "skip_bonus",
25 "snap",
26 "beaver",
27 "king_bonus",
28]
29ResponseType = Literal[
30 "create_session_response",
31 "join_session_response",
32 "start_session_response",
33 "get_session_response",
34 "leave_session_response",
35 "close_session_response",
36 "error",
37 "get_game_response",
38 "peek_initial_response",
39 "draw_card_response",
40 "draw_discard_response",
41 "keep_card_response",
42 "discard_drawn_response",
43 "peek_own_response",
44 "swap_card_response",
45 "skip_bonus_response",
46 "snap_response",
47 "beaver_response",
48 "king_bonus_response",
49]
52@dataclass(frozen=True)
53class SessionSettingsPayload:
54 """Serializable subset of GameSettings used by the lobby protocol."""
56 initial_hand_size: int
57 initial_cards_known: int
58 min_players: int
59 max_players: int
61 @classmethod
62 def from_settings(cls, settings: GameSettings) -> SessionSettingsPayload:
63 return cls(
64 initial_hand_size=settings.initial_hand_size,
65 initial_cards_known=settings.initial_cards_known,
66 min_players=settings.min_players,
67 max_players=settings.max_players,
68 )
70 @classmethod
71 def from_dict(cls, payload: dict[str, Any]) -> SessionSettingsPayload:
72 return cls(
73 initial_hand_size=payload.get("initial_hand_size", GameSettings.initial_hand_size),
74 initial_cards_known=payload.get("initial_cards_known", GameSettings.initial_cards_known),
75 min_players=payload.get("min_players", GameSettings.min_players),
76 max_players=payload.get("max_players", GameSettings.max_players),
77 )
79 def to_settings(self) -> GameSettings:
80 return GameSettings(
81 initial_hand_size=self.initial_hand_size,
82 initial_cards_known=self.initial_cards_known,
83 min_players=self.min_players,
84 max_players=self.max_players,
85 )
87 def to_dict(self) -> dict[str, Any]:
88 return asdict(self)
91@dataclass(frozen=True)
92class SessionSnapshot:
93 """Serializable view of the current lobby state."""
95 join_code: str
96 host_name: str | None
97 player_names: list[str]
98 player_count: int
99 lobby_state: str
100 can_start: bool
101 settings: SessionSettingsPayload
102 game_phase: str | None = None
104 @classmethod
105 def from_session(cls, session: Session) -> SessionSnapshot:
106 game_phase = None
107 if session.game is not None:
108 game_phase = session.game.state.phase.name
110 return cls(
111 join_code=session.join_code,
112 host_name=session.host_name,
113 player_names=[player.name for player in session.players],
114 player_count=session.player_count,
115 lobby_state=session.lobby_state.name,
116 can_start=session.can_start(),
117 settings=SessionSettingsPayload.from_settings(session.settings),
118 game_phase=game_phase,
119 )
121 @classmethod
122 def from_dict(cls, payload: dict[str, Any]) -> SessionSnapshot:
123 return cls(
124 join_code=payload["join_code"],
125 host_name=payload.get("host_name"),
126 player_names=list(payload.get("player_names", [])),
127 player_count=payload["player_count"],
128 lobby_state=payload["lobby_state"],
129 can_start=payload["can_start"],
130 settings=SessionSettingsPayload.from_dict(payload["settings"]),
131 game_phase=payload.get("game_phase"),
132 )
134 def to_dict(self) -> dict[str, Any]:
135 return {
136 "join_code": self.join_code,
137 "host_name": self.host_name,
138 "player_names": list(self.player_names),
139 "player_count": self.player_count,
140 "lobby_state": self.lobby_state,
141 "can_start": self.can_start,
142 "settings": self.settings.to_dict(),
143 "game_phase": self.game_phase,
144 }
147@dataclass(frozen=True)
148class CreateSessionRequest:
149 type: ClassVar[str] = "create_session"
150 host_name: str | None = None
151 settings: SessionSettingsPayload | None = None
153 @classmethod
154 def from_dict(cls, payload: dict[str, Any]) -> CreateSessionRequest:
155 _require_type(payload, cls.type)
156 raw_settings = payload.get("settings")
157 return cls(
158 host_name=payload.get("host_name"),
159 settings=SessionSettingsPayload.from_dict(raw_settings) if raw_settings is not None else None,
160 )
162 def to_dict(self) -> dict[str, Any]:
163 result: dict[str, Any] = {"type": self.type, "host_name": self.host_name}
164 if self.settings is not None:
165 result["settings"] = self.settings.to_dict()
166 return result
169@dataclass(frozen=True)
170class JoinSessionRequest:
171 type: ClassVar[str] = "join_session"
172 join_code: str = ""
173 player_name: str = ""
175 @classmethod
176 def from_dict(cls, payload: dict[str, Any]) -> JoinSessionRequest:
177 _require_type(payload, cls.type)
178 return cls(join_code=payload["join_code"], player_name=payload["player_name"])
180 def to_dict(self) -> dict[str, Any]:
181 return {"type": self.type, "join_code": self.join_code, "player_name": self.player_name}
184@dataclass(frozen=True)
185class StartSessionRequest:
186 type: ClassVar[str] = "start_session"
187 join_code: str = ""
188 player_token: str = ""
190 @classmethod
191 def from_dict(cls, payload: dict[str, Any]) -> StartSessionRequest:
192 _require_type(payload, cls.type)
193 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
195 def to_dict(self) -> dict[str, Any]:
196 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
199@dataclass(frozen=True)
200class GetSessionRequest:
201 type: ClassVar[str] = "get_session"
202 join_code: str = ""
204 @classmethod
205 def from_dict(cls, payload: dict[str, Any]) -> GetSessionRequest:
206 _require_type(payload, cls.type)
207 return cls(join_code=payload["join_code"])
209 def to_dict(self) -> dict[str, Any]:
210 return {"type": self.type, "join_code": self.join_code}
213@dataclass(frozen=True)
214class LeaveSessionRequest:
215 type: ClassVar[str] = "leave_session"
216 join_code: str = ""
217 player_token: str = ""
219 @classmethod
220 def from_dict(cls, payload: dict[str, Any]) -> LeaveSessionRequest:
221 _require_type(payload, cls.type)
222 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
224 def to_dict(self) -> dict[str, Any]:
225 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
228@dataclass(frozen=True)
229class CloseSessionRequest:
230 type: ClassVar[str] = "close_session"
231 join_code: str = ""
232 player_token: str = ""
234 @classmethod
235 def from_dict(cls, payload: dict[str, Any]) -> CloseSessionRequest:
236 _require_type(payload, cls.type)
237 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
239 def to_dict(self) -> dict[str, Any]:
240 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
243@dataclass(frozen=True)
244class CreateSessionResponse:
245 type: ClassVar[str] = "create_session_response"
246 session: SessionSnapshot
247 player_token: str | None = None
249 @classmethod
250 def from_dict(cls, payload: dict[str, Any]) -> CreateSessionResponse:
251 _require_type(payload, cls.type)
252 return cls(
253 session=SessionSnapshot.from_dict(payload["session"]),
254 player_token=payload.get("player_token"),
255 )
257 def to_dict(self) -> dict[str, Any]:
258 return {"type": self.type, "player_token": self.player_token, "session": self.session.to_dict()}
261@dataclass(frozen=True)
262class JoinSessionResponse:
263 type: ClassVar[str] = "join_session_response"
264 session: SessionSnapshot
265 player_token: str = ""
267 @classmethod
268 def from_dict(cls, payload: dict[str, Any]) -> JoinSessionResponse:
269 _require_type(payload, cls.type)
270 return cls(
271 session=SessionSnapshot.from_dict(payload["session"]),
272 player_token=payload.get("player_token", ""),
273 )
275 def to_dict(self) -> dict[str, Any]:
276 return {"type": self.type, "player_token": self.player_token, "session": self.session.to_dict()}
279@dataclass(frozen=True)
280class StartSessionResponse:
281 type: ClassVar[str] = "start_session_response"
282 session: SessionSnapshot
284 @classmethod
285 def from_dict(cls, payload: dict[str, Any]) -> StartSessionResponse:
286 _require_type(payload, cls.type)
287 return cls(session=SessionSnapshot.from_dict(payload["session"]))
289 def to_dict(self) -> dict[str, Any]:
290 return {"type": self.type, "session": self.session.to_dict()}
293@dataclass(frozen=True)
294class GetSessionResponse:
295 type: ClassVar[str] = "get_session_response"
296 session: SessionSnapshot
298 @classmethod
299 def from_dict(cls, payload: dict[str, Any]) -> GetSessionResponse:
300 _require_type(payload, cls.type)
301 return cls(session=SessionSnapshot.from_dict(payload["session"]))
303 def to_dict(self) -> dict[str, Any]:
304 return {"type": self.type, "session": self.session.to_dict()}
307@dataclass(frozen=True)
308class LeaveSessionResponse:
309 type: ClassVar[str] = "leave_session_response"
310 session: SessionSnapshot | None
312 @classmethod
313 def from_dict(cls, payload: dict[str, Any]) -> LeaveSessionResponse:
314 _require_type(payload, cls.type)
315 raw = payload.get("session")
316 return cls(session=SessionSnapshot.from_dict(raw) if raw is not None else None)
318 def to_dict(self) -> dict[str, Any]:
319 result: dict[str, Any] = {
320 "type": self.type, "session": self.session.to_dict() if self.session is not None else None,
321 }
322 return result
325@dataclass(frozen=True)
326class CloseSessionResponse:
327 type: ClassVar[str] = "close_session_response"
328 session: SessionSnapshot
330 @classmethod
331 def from_dict(cls, payload: dict[str, Any]) -> CloseSessionResponse:
332 _require_type(payload, cls.type)
333 return cls(session=SessionSnapshot.from_dict(payload["session"]))
335 def to_dict(self) -> dict[str, Any]:
336 return {"type": self.type, "session": self.session.to_dict()}
339@dataclass(frozen=True)
340class ErrorResponse:
341 type: ClassVar[str] = "error"
342 error_code: str
343 message: str
345 @classmethod
346 def from_dict(cls, payload: dict[str, Any]) -> ErrorResponse:
347 _require_type(payload, cls.type)
348 return cls(error_code=payload["error_code"], message=payload["message"])
350 def to_dict(self) -> dict[str, Any]:
351 return {"type": self.type, "error_code": self.error_code, "message": self.message}
354def request_from_dict(
355 payload: dict[str, Any],
356) -> (CreateSessionRequest | JoinSessionRequest | StartSessionRequest | GetSessionRequest | LeaveSessionRequest |
357 CloseSessionRequest | GetGameRequest | PeekInitialRequest | DrawCardRequest | DrawDiscardRequest |
358 KeepCardRequest | DiscardDrawnRequest | PeekOwnRequest | SwapCardRequest | SkipBonusRequest | SnapRequest |
359 BeaverRequest | KingBonusRequest):
360 """Parse a protocol request dict into its typed dataclass."""
361 message_type = payload.get("type")
362 if not isinstance(message_type, str):
363 raise ValueError(f"Unknown request type: {message_type}")
365 request_types: dict[str, Callable[[dict[str, Any]], Any]] = {
366 CreateSessionRequest.type: CreateSessionRequest.from_dict,
367 JoinSessionRequest.type: JoinSessionRequest.from_dict,
368 StartSessionRequest.type: StartSessionRequest.from_dict,
369 GetSessionRequest.type: GetSessionRequest.from_dict,
370 LeaveSessionRequest.type: LeaveSessionRequest.from_dict,
371 CloseSessionRequest.type: CloseSessionRequest.from_dict,
372 GetGameRequest.type: GetGameRequest.from_dict,
373 PeekInitialRequest.type: PeekInitialRequest.from_dict,
374 DrawCardRequest.type: DrawCardRequest.from_dict,
375 DrawDiscardRequest.type: DrawDiscardRequest.from_dict,
376 KeepCardRequest.type: KeepCardRequest.from_dict,
377 DiscardDrawnRequest.type: DiscardDrawnRequest.from_dict,
378 PeekOwnRequest.type: PeekOwnRequest.from_dict,
379 SwapCardRequest.type: SwapCardRequest.from_dict,
380 SkipBonusRequest.type: SkipBonusRequest.from_dict,
381 SnapRequest.type: SnapRequest.from_dict,
382 BeaverRequest.type: BeaverRequest.from_dict,
383 KingBonusRequest.type: KingBonusRequest.from_dict,
384 }
385 request_parser = request_types.get(message_type)
386 if request_parser is None:
387 raise ValueError(f"Unknown request type: {message_type}")
388 return request_parser(payload)
391def response_from_dict(
392 payload: dict[str, Any],
393) -> (CreateSessionResponse | JoinSessionResponse | StartSessionResponse | GetSessionResponse | LeaveSessionResponse |
394 CloseSessionResponse | ErrorResponse | GetGameResponse | PeekInitialResponse | DrawCardResponse |
395 DrawDiscardResponse | KeepCardResponse | DiscardDrawnResponse | PeekOwnResponse | SwapCardResponse |
396 SkipBonusResponse | SnapResponse | BeaverResponse | KingBonusResponse):
397 """Parse a protocol response dict into its typed dataclass."""
398 message_type = payload.get("type")
399 if not isinstance(message_type, str):
400 raise ValueError(f"Unknown response type: {message_type}")
402 response_types: dict[str, Callable[[dict[str, Any]], Any]] = {
403 CreateSessionResponse.type: CreateSessionResponse.from_dict,
404 JoinSessionResponse.type: JoinSessionResponse.from_dict,
405 StartSessionResponse.type: StartSessionResponse.from_dict,
406 GetSessionResponse.type: GetSessionResponse.from_dict,
407 LeaveSessionResponse.type: LeaveSessionResponse.from_dict,
408 CloseSessionResponse.type: CloseSessionResponse.from_dict,
409 ErrorResponse.type: ErrorResponse.from_dict,
410 GetGameResponse.type: GetGameResponse.from_dict,
411 PeekInitialResponse.type: PeekInitialResponse.from_dict,
412 DrawCardResponse.type: DrawCardResponse.from_dict,
413 DrawDiscardResponse.type: DrawDiscardResponse.from_dict,
414 KeepCardResponse.type: KeepCardResponse.from_dict,
415 DiscardDrawnResponse.type: DiscardDrawnResponse.from_dict,
416 PeekOwnResponse.type: PeekOwnResponse.from_dict,
417 SwapCardResponse.type: SwapCardResponse.from_dict,
418 SkipBonusResponse.type: SkipBonusResponse.from_dict,
419 SnapResponse.type: SnapResponse.from_dict,
420 BeaverResponse.type: BeaverResponse.from_dict,
421 KingBonusResponse.type: KingBonusResponse.from_dict,
422 }
423 response_parser = response_types.get(message_type)
424 if response_parser is None:
425 raise ValueError(f"Unknown response type: {message_type}")
426 return response_parser(payload)
429def _require_type(payload: dict[str, Any], expected_type: str) -> None:
430 actual_type = payload.get("type")
431 if actual_type != expected_type:
432 raise ValueError(f"Expected message type '{expected_type}', got '{actual_type}'")
435# ---------------------------------------------------------------------------
436# Game-phase protocol types
437# ---------------------------------------------------------------------------
439@dataclass(frozen=True)
440class GameCardSlot:
441 """Serializable view of a single card slot in a player's hand.
443 ``suit`` and ``value`` are ``None`` when the card is unknown to the
444 requesting player (i.e. face-down from their perspective).
445 """
447 known: bool
448 suit: str | None = None # e.g. "Hearts", None when unknown
449 value: int | None = None # 1–13 (or 0 for Joker), None when unknown
451 def to_dict(self) -> dict[str, Any]:
452 return {"known": self.known, "suit": self.suit, "value": self.value}
454 @classmethod
455 def from_dict(cls, payload: dict[str, Any]) -> GameCardSlot:
456 return cls(
457 known=payload["known"],
458 suit=payload.get("suit"),
459 value=payload.get("value"),
460 )
463@dataclass(frozen=True)
464class PlayerGameView:
465 """Serializable view of one player from another player's perspective."""
467 name: str
468 hand: list[GameCardSlot]
469 is_current_turn: bool
471 def to_dict(self) -> dict[str, Any]:
472 return {
473 "name": self.name,
474 "hand": [s.to_dict() for s in self.hand],
475 "is_current_turn": self.is_current_turn,
476 }
478 @classmethod
479 def from_dict(cls, payload: dict[str, Any]) -> PlayerGameView:
480 return cls(
481 name=payload["name"],
482 hand=[GameCardSlot.from_dict(s) for s in payload["hand"]],
483 is_current_turn=payload["is_current_turn"],
484 )
487@dataclass(frozen=True)
488class GameSnapshot:
489 """Perspective-aware view of the running game for a specific player."""
491 phase: str
492 current_player_name: str
493 draw_pile_size: int
494 top_discard: GameCardSlot | None
495 drawn_card: GameCardSlot | None # only set for the requesting player
496 pending_bonus: str | None # "JACK" | "QUEEN" | "KING" | None
497 players: list[PlayerGameView]
498 scores: list[dict[str, Any]] | None # only present when phase == "FINISHED"
500 def to_dict(self) -> dict[str, Any]:
501 return {
502 "phase": self.phase,
503 "current_player_name": self.current_player_name,
504 "draw_pile_size": self.draw_pile_size,
505 "top_discard": self.top_discard.to_dict() if self.top_discard else None,
506 "drawn_card": self.drawn_card.to_dict() if self.drawn_card else None,
507 "pending_bonus": self.pending_bonus,
508 "players": [p.to_dict() for p in self.players],
509 "scores": self.scores,
510 }
512 @classmethod
513 def from_dict(cls, payload: dict[str, Any]) -> GameSnapshot:
514 return cls(
515 phase=payload["phase"],
516 current_player_name=payload["current_player_name"],
517 draw_pile_size=payload["draw_pile_size"],
518 top_discard=GameCardSlot.from_dict(payload["top_discard"]) if payload.get("top_discard") else None,
519 drawn_card=GameCardSlot.from_dict(payload["drawn_card"]) if payload.get("drawn_card") else None,
520 pending_bonus=payload.get("pending_bonus"),
521 players=[PlayerGameView.from_dict(p) for p in payload["players"]],
522 scores=payload.get("scores"),
523 )
525 @classmethod
526 def from_game(
527 cls,
528 game: "Any", # beaverbunch.core.game.Game
529 requesting_player_name: str,
530 ) -> "GameSnapshot":
531 """Build a perspective-aware snapshot for ``requesting_player_name``."""
532 from beaverbunch.core.game import GamePhase
534 state = game.state
536 def _card_slot(card: "Any", known: bool) -> GameCardSlot:
537 if card is None:
538 return GameCardSlot(known=False)
539 if known:
540 suit_str = card.suit.value if card.suit is not None else None
541 return GameCardSlot(known=True, suit=suit_str, value=card.value)
542 return GameCardSlot(known=False)
544 player_views: list[PlayerGameView] = []
545 for player in state.players:
546 is_me = player.name == requesting_player_name
547 slots = [
548 _card_slot(slot.card, slot.known if is_me else False)
549 for slot in player.hand.slots
550 ]
551 player_views.append(
552 PlayerGameView(
553 name=player.name,
554 hand=slots,
555 is_current_turn=(state.current_player.name == player.name),
556 ),
557 )
559 # Top discard is always visible
560 top_discard_slot: GameCardSlot | None = None
561 if state.top_discard is not None:
562 top_discard_slot = _card_slot(state.top_discard, known=True)
564 # Drawn card is only shown to the player currently holding it
565 drawn_card_slot: GameCardSlot | None = None
566 if state.drawn_card is not None and state.current_player.name == requesting_player_name:
567 drawn_card_slot = _card_slot(state.drawn_card, known=True)
569 pending_bonus_str: str | None = None
570 if state.pending_bonus is not None:
571 pending_bonus_str = state.pending_bonus.name # "JACK" | "QUEEN" | "KING"
573 scores: list[dict[str, Any]] | None = None
574 if state.phase == GamePhase.FINISHED:
575 scores = [{"name": name, "score": score} for name, score in game.get_scores()]
577 return cls(
578 phase=state.phase.name,
579 current_player_name=state.current_player.name,
580 draw_pile_size=len(state.deck),
581 top_discard=top_discard_slot,
582 drawn_card=drawn_card_slot,
583 pending_bonus=pending_bonus_str,
584 players=player_views,
585 scores=scores,
586 )
589# --- Game request/response dataclasses ---
591@dataclass(frozen=True)
592class GetGameRequest:
593 type: ClassVar[str] = "get_game"
594 join_code: str = ""
595 player_token: str = ""
597 @classmethod
598 def from_dict(cls, payload: dict[str, Any]) -> GetGameRequest:
599 _require_type(payload, cls.type)
600 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
602 def to_dict(self) -> dict[str, Any]:
603 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
606@dataclass(frozen=True)
607class GetGameResponse:
608 type: ClassVar[str] = "get_game_response"
609 game: GameSnapshot
611 @classmethod
612 def from_dict(cls, payload: dict[str, Any]) -> GetGameResponse:
613 _require_type(payload, cls.type)
614 return cls(game=GameSnapshot.from_dict(payload["game"]))
616 def to_dict(self) -> dict[str, Any]:
617 return {"type": self.type, "game": self.game.to_dict()}
620@dataclass(frozen=True)
621class PeekInitialRequest:
622 type: ClassVar[str] = "peek_initial"
623 join_code: str = ""
624 player_token: str = ""
625 indices: list[int] = () # type: ignore[assignment]
627 @classmethod
628 def from_dict(cls, payload: dict[str, Any]) -> PeekInitialRequest:
629 _require_type(payload, cls.type)
630 return cls(
631 join_code=payload["join_code"],
632 player_token=payload["player_token"],
633 indices=list(payload.get("indices", [])),
634 )
636 def to_dict(self) -> dict[str, Any]:
637 return {
638 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
639 "indices": list(self.indices),
640 }
643@dataclass(frozen=True)
644class PeekInitialResponse:
645 type: ClassVar[str] = "peek_initial_response"
646 cards: list[GameCardSlot]
647 game: GameSnapshot
649 @classmethod
650 def from_dict(cls, payload: dict[str, Any]) -> PeekInitialResponse:
651 _require_type(payload, cls.type)
652 return cls(
653 cards=[GameCardSlot.from_dict(c) for c in payload["cards"]],
654 game=GameSnapshot.from_dict(payload["game"]),
655 )
657 def to_dict(self) -> dict[str, Any]:
658 return {"type": self.type, "cards": [c.to_dict() for c in self.cards], "game": self.game.to_dict()}
661@dataclass(frozen=True)
662class DrawCardRequest:
663 type: ClassVar[str] = "draw_card"
664 join_code: str = ""
665 player_token: str = ""
667 @classmethod
668 def from_dict(cls, payload: dict[str, Any]) -> DrawCardRequest:
669 _require_type(payload, cls.type)
670 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
672 def to_dict(self) -> dict[str, Any]:
673 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
676@dataclass(frozen=True)
677class DrawCardResponse:
678 type: ClassVar[str] = "draw_card_response"
679 card: GameCardSlot
680 game: GameSnapshot
682 @classmethod
683 def from_dict(cls, payload: dict[str, Any]) -> DrawCardResponse:
684 _require_type(payload, cls.type)
685 return cls(card=GameCardSlot.from_dict(payload["card"]), game=GameSnapshot.from_dict(payload["game"]))
687 def to_dict(self) -> dict[str, Any]:
688 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()}
691@dataclass(frozen=True)
692class DrawDiscardRequest:
693 type: ClassVar[str] = "draw_discard"
694 join_code: str = ""
695 player_token: str = ""
697 @classmethod
698 def from_dict(cls, payload: dict[str, Any]) -> DrawDiscardRequest:
699 _require_type(payload, cls.type)
700 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
702 def to_dict(self) -> dict[str, Any]:
703 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
706@dataclass(frozen=True)
707class DrawDiscardResponse:
708 type: ClassVar[str] = "draw_discard_response"
709 card: GameCardSlot
710 game: GameSnapshot
712 @classmethod
713 def from_dict(cls, payload: dict[str, Any]) -> DrawDiscardResponse:
714 _require_type(payload, cls.type)
715 return cls(card=GameCardSlot.from_dict(payload["card"]), game=GameSnapshot.from_dict(payload["game"]))
717 def to_dict(self) -> dict[str, Any]:
718 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()}
721@dataclass(frozen=True)
722class KeepCardRequest:
723 type: ClassVar[str] = "keep_card"
724 join_code: str = ""
725 player_token: str = ""
726 hand_index: int = 0
728 @classmethod
729 def from_dict(cls, payload: dict[str, Any]) -> KeepCardRequest:
730 _require_type(payload, cls.type)
731 return cls(
732 join_code=payload["join_code"],
733 player_token=payload["player_token"],
734 hand_index=int(payload["hand_index"]),
735 )
737 def to_dict(self) -> dict[str, Any]:
738 return {
739 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
740 "hand_index": self.hand_index,
741 }
744@dataclass(frozen=True)
745class KeepCardResponse:
746 type: ClassVar[str] = "keep_card_response"
747 discarded_card: GameCardSlot
748 game: GameSnapshot
750 @classmethod
751 def from_dict(cls, payload: dict[str, Any]) -> KeepCardResponse:
752 _require_type(payload, cls.type)
753 return cls(
754 discarded_card=GameCardSlot.from_dict(payload["discarded_card"]),
755 game=GameSnapshot.from_dict(payload["game"]),
756 )
758 def to_dict(self) -> dict[str, Any]:
759 return {"type": self.type, "discarded_card": self.discarded_card.to_dict(), "game": self.game.to_dict()}
762@dataclass(frozen=True)
763class DiscardDrawnRequest:
764 type: ClassVar[str] = "discard_drawn"
765 join_code: str = ""
766 player_token: str = ""
768 @classmethod
769 def from_dict(cls, payload: dict[str, Any]) -> DiscardDrawnRequest:
770 _require_type(payload, cls.type)
771 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
773 def to_dict(self) -> dict[str, Any]:
774 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
777@dataclass(frozen=True)
778class DiscardDrawnResponse:
779 type: ClassVar[str] = "discard_drawn_response"
780 bonus: str | None # "JACK" | "QUEEN" | "KING" | None
781 game: GameSnapshot
783 @classmethod
784 def from_dict(cls, payload: dict[str, Any]) -> DiscardDrawnResponse:
785 _require_type(payload, cls.type)
786 return cls(bonus=payload.get("bonus"), game=GameSnapshot.from_dict(payload["game"]))
788 def to_dict(self) -> dict[str, Any]:
789 return {"type": self.type, "bonus": self.bonus, "game": self.game.to_dict()}
792@dataclass(frozen=True)
793class PeekOwnRequest:
794 type: ClassVar[str] = "peek_own"
795 join_code: str = ""
796 player_token: str = ""
797 hand_index: int = 0
799 @classmethod
800 def from_dict(cls, payload: dict[str, Any]) -> PeekOwnRequest:
801 _require_type(payload, cls.type)
802 return cls(
803 join_code=payload["join_code"],
804 player_token=payload["player_token"],
805 hand_index=int(payload["hand_index"]),
806 )
808 def to_dict(self) -> dict[str, Any]:
809 return {
810 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
811 "hand_index": self.hand_index,
812 }
815@dataclass(frozen=True)
816class PeekOwnResponse:
817 type: ClassVar[str] = "peek_own_response"
818 card: GameCardSlot
819 game: GameSnapshot
821 @classmethod
822 def from_dict(cls, payload: dict[str, Any]) -> PeekOwnResponse:
823 _require_type(payload, cls.type)
824 return cls(card=GameCardSlot.from_dict(payload["card"]), game=GameSnapshot.from_dict(payload["game"]))
826 def to_dict(self) -> dict[str, Any]:
827 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()}
830@dataclass(frozen=True)
831class SwapCardRequest:
832 type: ClassVar[str] = "swap_card"
833 join_code: str = ""
834 player_token: str = ""
835 own_index: int = 0
836 target_player_name: str = ""
837 target_index: int = 0
839 @classmethod
840 def from_dict(cls, payload: dict[str, Any]) -> SwapCardRequest:
841 _require_type(payload, cls.type)
842 return cls(
843 join_code=payload["join_code"],
844 player_token=payload["player_token"],
845 own_index=int(payload["own_index"]),
846 target_player_name=payload["target_player_name"],
847 target_index=int(payload["target_index"]),
848 )
850 def to_dict(self) -> dict[str, Any]:
851 return {
852 "type": self.type,
853 "join_code": self.join_code,
854 "player_token": self.player_token,
855 "own_index": self.own_index,
856 "target_player_name": self.target_player_name,
857 "target_index": self.target_index,
858 }
861@dataclass(frozen=True)
862class SwapCardResponse:
863 type: ClassVar[str] = "swap_card_response"
864 game: GameSnapshot
866 @classmethod
867 def from_dict(cls, payload: dict[str, Any]) -> SwapCardResponse:
868 _require_type(payload, cls.type)
869 return cls(game=GameSnapshot.from_dict(payload["game"]))
871 def to_dict(self) -> dict[str, Any]:
872 return {"type": self.type, "game": self.game.to_dict()}
875@dataclass(frozen=True)
876class SkipBonusRequest:
877 type: ClassVar[str] = "skip_bonus"
878 join_code: str = ""
879 player_token: str = ""
881 @classmethod
882 def from_dict(cls, payload: dict[str, Any]) -> SkipBonusRequest:
883 _require_type(payload, cls.type)
884 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
886 def to_dict(self) -> dict[str, Any]:
887 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
890@dataclass(frozen=True)
891class SkipBonusResponse:
892 type: ClassVar[str] = "skip_bonus_response"
893 game: GameSnapshot
895 @classmethod
896 def from_dict(cls, payload: dict[str, Any]) -> SkipBonusResponse:
897 _require_type(payload, cls.type)
898 return cls(game=GameSnapshot.from_dict(payload["game"]))
900 def to_dict(self) -> dict[str, Any]:
901 return {"type": self.type, "game": self.game.to_dict()}
904@dataclass(frozen=True)
905class SnapRequest:
906 type: ClassVar[str] = "snap"
907 join_code: str = ""
908 player_token: str = ""
909 hand_index: int = 0
911 @classmethod
912 def from_dict(cls, payload: dict[str, Any]) -> SnapRequest:
913 _require_type(payload, cls.type)
914 return cls(
915 join_code=payload["join_code"],
916 player_token=payload["player_token"],
917 hand_index=int(payload["hand_index"]),
918 )
920 def to_dict(self) -> dict[str, Any]:
921 return {
922 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
923 "hand_index": self.hand_index,
924 }
927@dataclass(frozen=True)
928class SnapResponse:
929 type: ClassVar[str] = "snap_response"
930 correct: bool
931 game: GameSnapshot
933 @classmethod
934 def from_dict(cls, payload: dict[str, Any]) -> SnapResponse:
935 _require_type(payload, cls.type)
936 return cls(correct=payload["correct"], game=GameSnapshot.from_dict(payload["game"]))
938 def to_dict(self) -> dict[str, Any]:
939 return {"type": self.type, "correct": self.correct, "game": self.game.to_dict()}
942@dataclass(frozen=True)
943class BeaverRequest:
944 """Player declares 'Beaver!' to trigger the last round."""
945 type: ClassVar[str] = "beaver"
946 join_code: str = ""
947 player_token: str = ""
949 @classmethod
950 def from_dict(cls, payload: dict[str, Any]) -> BeaverRequest:
951 _require_type(payload, cls.type)
952 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
954 def to_dict(self) -> dict[str, Any]:
955 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
958@dataclass(frozen=True)
959class BeaverResponse:
960 type: ClassVar[str] = "beaver_response"
961 game: GameSnapshot
963 @classmethod
964 def from_dict(cls, payload: dict[str, Any]) -> BeaverResponse:
965 _require_type(payload, cls.type)
966 return cls(game=GameSnapshot.from_dict(payload["game"]))
968 def to_dict(self) -> dict[str, Any]:
969 return {"type": self.type, "game": self.game.to_dict()}
972@dataclass(frozen=True)
973class KingBonusRequest:
974 """Player executes the mandatory King bonus (draw again)."""
975 type: ClassVar[str] = "king_bonus"
976 join_code: str = ""
977 player_token: str = ""
979 @classmethod
980 def from_dict(cls, payload: dict[str, Any]) -> KingBonusRequest:
981 _require_type(payload, cls.type)
982 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
984 def to_dict(self) -> dict[str, Any]:
985 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
988@dataclass(frozen=True)
989class KingBonusResponse:
990 type: ClassVar[str] = "king_bonus_response"
991 game: GameSnapshot
993 @classmethod
994 def from_dict(cls, payload: dict[str, Any]) -> KingBonusResponse:
995 _require_type(payload, cls.type)
996 return cls(game=GameSnapshot.from_dict(payload["game"]))
998 def to_dict(self) -> dict[str, Any]:
999 return {"type": self.type, "game": self.game.to_dict()}