Coverage for src / beaverbunch / network / protocol.py: 100.0%
519 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 00:24 +0000
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 00:24 +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]
28ResponseType = Literal[
29 "create_session_response",
30 "join_session_response",
31 "start_session_response",
32 "get_session_response",
33 "leave_session_response",
34 "close_session_response",
35 "error",
36 "get_game_response",
37 "peek_initial_response",
38 "draw_card_response",
39 "draw_discard_response",
40 "keep_card_response",
41 "discard_drawn_response",
42 "peek_own_response",
43 "swap_card_response",
44 "skip_bonus_response",
45 "snap_response",
46 "beaver_response",
47]
50@dataclass(frozen=True)
51class SessionSettingsPayload:
52 """Serializable subset of GameSettings used by the lobby protocol."""
54 initial_hand_size: int
55 initial_cards_known: int
56 min_players: int
57 max_players: int
59 @classmethod
60 def from_settings(cls, settings: GameSettings) -> SessionSettingsPayload:
61 return cls(
62 initial_hand_size=settings.initial_hand_size,
63 initial_cards_known=settings.initial_cards_known,
64 min_players=settings.min_players,
65 max_players=settings.max_players,
66 )
68 @classmethod
69 def from_dict(cls, payload: dict[str, Any]) -> SessionSettingsPayload:
70 return cls(
71 initial_hand_size=payload.get("initial_hand_size", GameSettings.initial_hand_size),
72 initial_cards_known=payload.get("initial_cards_known", GameSettings.initial_cards_known),
73 min_players=payload.get("min_players", GameSettings.min_players),
74 max_players=payload.get("max_players", GameSettings.max_players),
75 )
77 def to_settings(self) -> GameSettings:
78 return GameSettings(
79 initial_hand_size=self.initial_hand_size,
80 initial_cards_known=self.initial_cards_known,
81 min_players=self.min_players,
82 max_players=self.max_players,
83 )
85 def to_dict(self) -> dict[str, Any]:
86 return asdict(self)
89@dataclass(frozen=True)
90class SessionSnapshot:
91 """Serializable view of the current lobby state."""
93 join_code: str
94 host_name: str | None
95 player_names: list[str]
96 player_count: int
97 lobby_state: str
98 can_start: bool
99 settings: SessionSettingsPayload
100 game_phase: str | None = None
102 @classmethod
103 def from_session(cls, session: Session) -> SessionSnapshot:
104 game_phase = None
105 if session.game is not None:
106 game_phase = session.game.state.phase.name
108 return cls(
109 join_code=session.join_code,
110 host_name=session.host_name,
111 player_names=[player.name for player in session.players],
112 player_count=session.player_count,
113 lobby_state=session.lobby_state.name,
114 can_start=session.can_start(),
115 settings=SessionSettingsPayload.from_settings(session.settings),
116 game_phase=game_phase,
117 )
119 @classmethod
120 def from_dict(cls, payload: dict[str, Any]) -> SessionSnapshot:
121 return cls(
122 join_code=payload["join_code"],
123 host_name=payload.get("host_name"),
124 player_names=list(payload.get("player_names", [])),
125 player_count=payload["player_count"],
126 lobby_state=payload["lobby_state"],
127 can_start=payload["can_start"],
128 settings=SessionSettingsPayload.from_dict(payload["settings"]),
129 game_phase=payload.get("game_phase"),
130 )
132 def to_dict(self) -> dict[str, Any]:
133 return {
134 "join_code": self.join_code,
135 "host_name": self.host_name,
136 "player_names": list(self.player_names),
137 "player_count": self.player_count,
138 "lobby_state": self.lobby_state,
139 "can_start": self.can_start,
140 "settings": self.settings.to_dict(),
141 "game_phase": self.game_phase,
142 }
145@dataclass(frozen=True)
146class CreateSessionRequest:
147 type: ClassVar[str] = "create_session"
148 host_name: str | None = None
149 settings: SessionSettingsPayload | None = None
151 @classmethod
152 def from_dict(cls, payload: dict[str, Any]) -> CreateSessionRequest:
153 _require_type(payload, cls.type)
154 raw_settings = payload.get("settings")
155 return cls(
156 host_name=payload.get("host_name"),
157 settings=SessionSettingsPayload.from_dict(raw_settings) if raw_settings is not None else None,
158 )
160 def to_dict(self) -> dict[str, Any]:
161 result: dict[str, Any] = {"type": self.type, "host_name": self.host_name}
162 if self.settings is not None:
163 result["settings"] = self.settings.to_dict()
164 return result
167@dataclass(frozen=True)
168class JoinSessionRequest:
169 type: ClassVar[str] = "join_session"
170 join_code: str = ""
171 player_name: str = ""
173 @classmethod
174 def from_dict(cls, payload: dict[str, Any]) -> JoinSessionRequest:
175 _require_type(payload, cls.type)
176 return cls(join_code=payload["join_code"], player_name=payload["player_name"])
178 def to_dict(self) -> dict[str, Any]:
179 return {"type": self.type, "join_code": self.join_code, "player_name": self.player_name}
182@dataclass(frozen=True)
183class StartSessionRequest:
184 type: ClassVar[str] = "start_session"
185 join_code: str = ""
186 player_token: str = ""
188 @classmethod
189 def from_dict(cls, payload: dict[str, Any]) -> StartSessionRequest:
190 _require_type(payload, cls.type)
191 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
193 def to_dict(self) -> dict[str, Any]:
194 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
197@dataclass(frozen=True)
198class GetSessionRequest:
199 type: ClassVar[str] = "get_session"
200 join_code: str = ""
202 @classmethod
203 def from_dict(cls, payload: dict[str, Any]) -> GetSessionRequest:
204 _require_type(payload, cls.type)
205 return cls(join_code=payload["join_code"])
207 def to_dict(self) -> dict[str, Any]:
208 return {"type": self.type, "join_code": self.join_code}
211@dataclass(frozen=True)
212class LeaveSessionRequest:
213 type: ClassVar[str] = "leave_session"
214 join_code: str = ""
215 player_token: str = ""
217 @classmethod
218 def from_dict(cls, payload: dict[str, Any]) -> LeaveSessionRequest:
219 _require_type(payload, cls.type)
220 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
222 def to_dict(self) -> dict[str, Any]:
223 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
226@dataclass(frozen=True)
227class CloseSessionRequest:
228 type: ClassVar[str] = "close_session"
229 join_code: str = ""
230 player_token: str = ""
232 @classmethod
233 def from_dict(cls, payload: dict[str, Any]) -> CloseSessionRequest:
234 _require_type(payload, cls.type)
235 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
237 def to_dict(self) -> dict[str, Any]:
238 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
241@dataclass(frozen=True)
242class CreateSessionResponse:
243 type: ClassVar[str] = "create_session_response"
244 session: SessionSnapshot
245 player_token: str | None = None
247 @classmethod
248 def from_dict(cls, payload: dict[str, Any]) -> CreateSessionResponse:
249 _require_type(payload, cls.type)
250 return cls(
251 session=SessionSnapshot.from_dict(payload["session"]),
252 player_token=payload.get("player_token"),
253 )
255 def to_dict(self) -> dict[str, Any]:
256 return {"type": self.type, "player_token": self.player_token, "session": self.session.to_dict()}
259@dataclass(frozen=True)
260class JoinSessionResponse:
261 type: ClassVar[str] = "join_session_response"
262 session: SessionSnapshot
263 player_token: str = ""
265 @classmethod
266 def from_dict(cls, payload: dict[str, Any]) -> JoinSessionResponse:
267 _require_type(payload, cls.type)
268 return cls(
269 session=SessionSnapshot.from_dict(payload["session"]),
270 player_token=payload.get("player_token", ""),
271 )
273 def to_dict(self) -> dict[str, Any]:
274 return {"type": self.type, "player_token": self.player_token, "session": self.session.to_dict()}
277@dataclass(frozen=True)
278class StartSessionResponse:
279 type: ClassVar[str] = "start_session_response"
280 session: SessionSnapshot
282 @classmethod
283 def from_dict(cls, payload: dict[str, Any]) -> StartSessionResponse:
284 _require_type(payload, cls.type)
285 return cls(session=SessionSnapshot.from_dict(payload["session"]))
287 def to_dict(self) -> dict[str, Any]:
288 return {"type": self.type, "session": self.session.to_dict()}
291@dataclass(frozen=True)
292class GetSessionResponse:
293 type: ClassVar[str] = "get_session_response"
294 session: SessionSnapshot
296 @classmethod
297 def from_dict(cls, payload: dict[str, Any]) -> GetSessionResponse:
298 _require_type(payload, cls.type)
299 return cls(session=SessionSnapshot.from_dict(payload["session"]))
301 def to_dict(self) -> dict[str, Any]:
302 return {"type": self.type, "session": self.session.to_dict()}
305@dataclass(frozen=True)
306class LeaveSessionResponse:
307 type: ClassVar[str] = "leave_session_response"
308 session: SessionSnapshot | None
310 @classmethod
311 def from_dict(cls, payload: dict[str, Any]) -> LeaveSessionResponse:
312 _require_type(payload, cls.type)
313 raw = payload.get("session")
314 return cls(session=SessionSnapshot.from_dict(raw) if raw is not None else None)
316 def to_dict(self) -> dict[str, Any]:
317 result: dict[str, Any] = {
318 "type": self.type, "session": self.session.to_dict() if self.session is not None else None,
319 }
320 return result
323@dataclass(frozen=True)
324class CloseSessionResponse:
325 type: ClassVar[str] = "close_session_response"
326 session: SessionSnapshot
328 @classmethod
329 def from_dict(cls, payload: dict[str, Any]) -> CloseSessionResponse:
330 _require_type(payload, cls.type)
331 return cls(session=SessionSnapshot.from_dict(payload["session"]))
333 def to_dict(self) -> dict[str, Any]:
334 return {"type": self.type, "session": self.session.to_dict()}
337@dataclass(frozen=True)
338class ErrorResponse:
339 type: ClassVar[str] = "error"
340 error_code: str
341 message: str
343 @classmethod
344 def from_dict(cls, payload: dict[str, Any]) -> ErrorResponse:
345 _require_type(payload, cls.type)
346 return cls(error_code=payload["error_code"], message=payload["message"])
348 def to_dict(self) -> dict[str, Any]:
349 return {"type": self.type, "error_code": self.error_code, "message": self.message}
352def request_from_dict(
353 payload: dict[str, Any],
354) -> (CreateSessionRequest | JoinSessionRequest | StartSessionRequest | GetSessionRequest | LeaveSessionRequest |
355 CloseSessionRequest | GetGameRequest | PeekInitialRequest | DrawCardRequest | DrawDiscardRequest |
356 KeepCardRequest | DiscardDrawnRequest | PeekOwnRequest | SwapCardRequest | SkipBonusRequest | SnapRequest |
357 BeaverRequest):
358 """Parse a protocol request dict into its typed dataclass."""
359 message_type = payload.get("type")
360 if not isinstance(message_type, str):
361 raise ValueError(f"Unknown request type: {message_type}")
363 request_types: dict[str, Callable[[dict[str, Any]], Any]] = {
364 CreateSessionRequest.type: CreateSessionRequest.from_dict,
365 JoinSessionRequest.type: JoinSessionRequest.from_dict,
366 StartSessionRequest.type: StartSessionRequest.from_dict,
367 GetSessionRequest.type: GetSessionRequest.from_dict,
368 LeaveSessionRequest.type: LeaveSessionRequest.from_dict,
369 CloseSessionRequest.type: CloseSessionRequest.from_dict,
370 GetGameRequest.type: GetGameRequest.from_dict,
371 PeekInitialRequest.type: PeekInitialRequest.from_dict,
372 DrawCardRequest.type: DrawCardRequest.from_dict,
373 DrawDiscardRequest.type: DrawDiscardRequest.from_dict,
374 KeepCardRequest.type: KeepCardRequest.from_dict,
375 DiscardDrawnRequest.type: DiscardDrawnRequest.from_dict,
376 PeekOwnRequest.type: PeekOwnRequest.from_dict,
377 SwapCardRequest.type: SwapCardRequest.from_dict,
378 SkipBonusRequest.type: SkipBonusRequest.from_dict,
379 SnapRequest.type: SnapRequest.from_dict,
380 BeaverRequest.type: BeaverRequest.from_dict,
381 }
382 request_parser = request_types.get(message_type)
383 if request_parser is None:
384 raise ValueError(f"Unknown request type: {message_type}")
385 return request_parser(payload)
388def response_from_dict(
389 payload: dict[str, Any],
390) -> (CreateSessionResponse | JoinSessionResponse | StartSessionResponse | GetSessionResponse | LeaveSessionResponse |
391 CloseSessionResponse | ErrorResponse | GetGameResponse | PeekInitialResponse | DrawCardResponse |
392 DrawDiscardResponse | KeepCardResponse | DiscardDrawnResponse | PeekOwnResponse | SwapCardResponse |
393 SkipBonusResponse | SnapResponse | BeaverResponse):
394 """Parse a protocol response dict into its typed dataclass."""
395 message_type = payload.get("type")
396 if not isinstance(message_type, str):
397 raise ValueError(f"Unknown response type: {message_type}")
399 response_types: dict[str, Callable[[dict[str, Any]], Any]] = {
400 CreateSessionResponse.type: CreateSessionResponse.from_dict,
401 JoinSessionResponse.type: JoinSessionResponse.from_dict,
402 StartSessionResponse.type: StartSessionResponse.from_dict,
403 GetSessionResponse.type: GetSessionResponse.from_dict,
404 LeaveSessionResponse.type: LeaveSessionResponse.from_dict,
405 CloseSessionResponse.type: CloseSessionResponse.from_dict,
406 ErrorResponse.type: ErrorResponse.from_dict,
407 GetGameResponse.type: GetGameResponse.from_dict,
408 PeekInitialResponse.type: PeekInitialResponse.from_dict,
409 DrawCardResponse.type: DrawCardResponse.from_dict,
410 DrawDiscardResponse.type: DrawDiscardResponse.from_dict,
411 KeepCardResponse.type: KeepCardResponse.from_dict,
412 DiscardDrawnResponse.type: DiscardDrawnResponse.from_dict,
413 PeekOwnResponse.type: PeekOwnResponse.from_dict,
414 SwapCardResponse.type: SwapCardResponse.from_dict,
415 SkipBonusResponse.type: SkipBonusResponse.from_dict,
416 SnapResponse.type: SnapResponse.from_dict,
417 BeaverResponse.type: BeaverResponse.from_dict,
418 }
419 response_parser = response_types.get(message_type)
420 if response_parser is None:
421 raise ValueError(f"Unknown response type: {message_type}")
422 return response_parser(payload)
425def _require_type(payload: dict[str, Any], expected_type: str) -> None:
426 actual_type = payload.get("type")
427 if actual_type != expected_type:
428 raise ValueError(f"Expected message type '{expected_type}', got '{actual_type}'")
431# ---------------------------------------------------------------------------
432# Game-phase protocol types
433# ---------------------------------------------------------------------------
435@dataclass(frozen=True)
436class GameCardSlot:
437 """Serializable view of a single card slot in a player's hand.
439 ``suit`` and ``value`` are ``None`` when the card is unknown to the
440 requesting player (i.e. face-down from their perspective).
441 """
443 known: bool
444 suit: str | None = None # e.g. "Hearts", None when unknown
445 value: int | None = None # 1–13 (or 0 for Joker), None when unknown
447 def to_dict(self) -> dict[str, Any]:
448 return {"known": self.known, "suit": self.suit, "value": self.value}
450 @classmethod
451 def from_dict(cls, payload: dict[str, Any]) -> GameCardSlot:
452 return cls(
453 known=payload["known"],
454 suit=payload.get("suit"),
455 value=payload.get("value"),
456 )
459@dataclass(frozen=True)
460class PlayerGameView:
461 """Serializable view of one player from another player's perspective."""
463 name: str
464 hand: list[GameCardSlot]
465 is_current_turn: bool
467 def to_dict(self) -> dict[str, Any]:
468 return {
469 "name": self.name,
470 "hand": [s.to_dict() for s in self.hand],
471 "is_current_turn": self.is_current_turn,
472 }
474 @classmethod
475 def from_dict(cls, payload: dict[str, Any]) -> PlayerGameView:
476 return cls(
477 name=payload["name"],
478 hand=[GameCardSlot.from_dict(s) for s in payload["hand"]],
479 is_current_turn=payload["is_current_turn"],
480 )
483@dataclass(frozen=True)
484class GameSnapshot:
485 """Perspective-aware view of the running game for a specific player."""
487 phase: str
488 current_player_name: str
489 draw_pile_size: int
490 top_discard: GameCardSlot | None
491 drawn_card: GameCardSlot | None # only set for the requesting player
492 pending_bonus: str | None # "JACK" | "QUEEN" | "KING" | None
493 players: list[PlayerGameView]
494 scores: list[dict[str, Any]] | None # only present when phase == "FINISHED"
496 def to_dict(self) -> dict[str, Any]:
497 return {
498 "phase": self.phase,
499 "current_player_name": self.current_player_name,
500 "draw_pile_size": self.draw_pile_size,
501 "top_discard": self.top_discard.to_dict() if self.top_discard else None,
502 "drawn_card": self.drawn_card.to_dict() if self.drawn_card else None,
503 "pending_bonus": self.pending_bonus,
504 "players": [p.to_dict() for p in self.players],
505 "scores": self.scores,
506 }
508 @classmethod
509 def from_dict(cls, payload: dict[str, Any]) -> GameSnapshot:
510 return cls(
511 phase=payload["phase"],
512 current_player_name=payload["current_player_name"],
513 draw_pile_size=payload["draw_pile_size"],
514 top_discard=GameCardSlot.from_dict(payload["top_discard"]) if payload.get("top_discard") else None,
515 drawn_card=GameCardSlot.from_dict(payload["drawn_card"]) if payload.get("drawn_card") else None,
516 pending_bonus=payload.get("pending_bonus"),
517 players=[PlayerGameView.from_dict(p) for p in payload["players"]],
518 scores=payload.get("scores"),
519 )
521 @classmethod
522 def from_game(
523 cls,
524 game: "Any", # beaverbunch.core.game.Game
525 requesting_player_name: str,
526 ) -> "GameSnapshot":
527 """Build a perspective-aware snapshot for ``requesting_player_name``."""
528 from beaverbunch.core.game import GamePhase
530 state = game.state
532 def _card_slot(card: "Any", known: bool) -> GameCardSlot:
533 if card is None:
534 return GameCardSlot(known=False)
535 if known:
536 suit_str = card.suit.value if card.suit is not None else None
537 return GameCardSlot(known=True, suit=suit_str, value=card.value)
538 return GameCardSlot(known=False)
540 player_views: list[PlayerGameView] = []
541 for player in state.players:
542 is_me = player.name == requesting_player_name
543 slots = [
544 _card_slot(slot.card, slot.known if is_me else False)
545 for slot in player.hand.slots
546 ]
547 player_views.append(
548 PlayerGameView(
549 name=player.name,
550 hand=slots,
551 is_current_turn=(state.current_player.name == player.name),
552 ),
553 )
555 # Top discard is always visible
556 top_discard_slot: GameCardSlot | None = None
557 if state.top_discard is not None:
558 top_discard_slot = _card_slot(state.top_discard, known=True)
560 # Drawn card is only shown to the player currently holding it
561 drawn_card_slot: GameCardSlot | None = None
562 if state.drawn_card is not None and state.current_player.name == requesting_player_name:
563 drawn_card_slot = _card_slot(state.drawn_card, known=True)
565 pending_bonus_str: str | None = None
566 if state.pending_bonus is not None:
567 pending_bonus_str = state.pending_bonus.name # "JACK" | "QUEEN" | "KING"
569 scores: list[dict[str, Any]] | None = None
570 if state.phase == GamePhase.FINISHED:
571 scores = [{"name": name, "score": score} for name, score in game.get_scores()]
573 return cls(
574 phase=state.phase.name,
575 current_player_name=state.current_player.name,
576 draw_pile_size=len(state.deck),
577 top_discard=top_discard_slot,
578 drawn_card=drawn_card_slot,
579 pending_bonus=pending_bonus_str,
580 players=player_views,
581 scores=scores,
582 )
585# --- Game request/response dataclasses ---
587@dataclass(frozen=True)
588class GetGameRequest:
589 type: ClassVar[str] = "get_game"
590 join_code: str = ""
591 player_token: str = ""
593 @classmethod
594 def from_dict(cls, payload: dict[str, Any]) -> GetGameRequest:
595 _require_type(payload, cls.type)
596 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
598 def to_dict(self) -> dict[str, Any]:
599 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
602@dataclass(frozen=True)
603class GetGameResponse:
604 type: ClassVar[str] = "get_game_response"
605 game: GameSnapshot
607 @classmethod
608 def from_dict(cls, payload: dict[str, Any]) -> GetGameResponse:
609 _require_type(payload, cls.type)
610 return cls(game=GameSnapshot.from_dict(payload["game"]))
612 def to_dict(self) -> dict[str, Any]:
613 return {"type": self.type, "game": self.game.to_dict()}
616@dataclass(frozen=True)
617class PeekInitialRequest:
618 type: ClassVar[str] = "peek_initial"
619 join_code: str = ""
620 player_token: str = ""
621 indices: list[int] = () # type: ignore[assignment]
623 @classmethod
624 def from_dict(cls, payload: dict[str, Any]) -> PeekInitialRequest:
625 _require_type(payload, cls.type)
626 return cls(
627 join_code=payload["join_code"],
628 player_token=payload["player_token"],
629 indices=list(payload.get("indices", [])),
630 )
632 def to_dict(self) -> dict[str, Any]:
633 return {
634 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
635 "indices": list(self.indices),
636 }
639@dataclass(frozen=True)
640class PeekInitialResponse:
641 type: ClassVar[str] = "peek_initial_response"
642 cards: list[GameCardSlot]
643 game: GameSnapshot
645 @classmethod
646 def from_dict(cls, payload: dict[str, Any]) -> PeekInitialResponse:
647 _require_type(payload, cls.type)
648 return cls(
649 cards=[GameCardSlot.from_dict(c) for c in payload["cards"]],
650 game=GameSnapshot.from_dict(payload["game"]),
651 )
653 def to_dict(self) -> dict[str, Any]:
654 return {"type": self.type, "cards": [c.to_dict() for c in self.cards], "game": self.game.to_dict()}
657@dataclass(frozen=True)
658class DrawCardRequest:
659 type: ClassVar[str] = "draw_card"
660 join_code: str = ""
661 player_token: str = ""
663 @classmethod
664 def from_dict(cls, payload: dict[str, Any]) -> DrawCardRequest:
665 _require_type(payload, cls.type)
666 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
668 def to_dict(self) -> dict[str, Any]:
669 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
672@dataclass(frozen=True)
673class DrawCardResponse:
674 type: ClassVar[str] = "draw_card_response"
675 card: GameCardSlot
676 game: GameSnapshot
678 @classmethod
679 def from_dict(cls, payload: dict[str, Any]) -> DrawCardResponse:
680 _require_type(payload, cls.type)
681 return cls(card=GameCardSlot.from_dict(payload["card"]), game=GameSnapshot.from_dict(payload["game"]))
683 def to_dict(self) -> dict[str, Any]:
684 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()}
687@dataclass(frozen=True)
688class DrawDiscardRequest:
689 type: ClassVar[str] = "draw_discard"
690 join_code: str = ""
691 player_token: str = ""
693 @classmethod
694 def from_dict(cls, payload: dict[str, Any]) -> DrawDiscardRequest:
695 _require_type(payload, cls.type)
696 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
698 def to_dict(self) -> dict[str, Any]:
699 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
702@dataclass(frozen=True)
703class DrawDiscardResponse:
704 type: ClassVar[str] = "draw_discard_response"
705 card: GameCardSlot
706 game: GameSnapshot
708 @classmethod
709 def from_dict(cls, payload: dict[str, Any]) -> DrawDiscardResponse:
710 _require_type(payload, cls.type)
711 return cls(card=GameCardSlot.from_dict(payload["card"]), game=GameSnapshot.from_dict(payload["game"]))
713 def to_dict(self) -> dict[str, Any]:
714 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()}
717@dataclass(frozen=True)
718class KeepCardRequest:
719 type: ClassVar[str] = "keep_card"
720 join_code: str = ""
721 player_token: str = ""
722 hand_index: int = 0
724 @classmethod
725 def from_dict(cls, payload: dict[str, Any]) -> KeepCardRequest:
726 _require_type(payload, cls.type)
727 return cls(
728 join_code=payload["join_code"],
729 player_token=payload["player_token"],
730 hand_index=int(payload["hand_index"]),
731 )
733 def to_dict(self) -> dict[str, Any]:
734 return {
735 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
736 "hand_index": self.hand_index,
737 }
740@dataclass(frozen=True)
741class KeepCardResponse:
742 type: ClassVar[str] = "keep_card_response"
743 discarded_card: GameCardSlot
744 game: GameSnapshot
746 @classmethod
747 def from_dict(cls, payload: dict[str, Any]) -> KeepCardResponse:
748 _require_type(payload, cls.type)
749 return cls(
750 discarded_card=GameCardSlot.from_dict(payload["discarded_card"]),
751 game=GameSnapshot.from_dict(payload["game"]),
752 )
754 def to_dict(self) -> dict[str, Any]:
755 return {"type": self.type, "discarded_card": self.discarded_card.to_dict(), "game": self.game.to_dict()}
758@dataclass(frozen=True)
759class DiscardDrawnRequest:
760 type: ClassVar[str] = "discard_drawn"
761 join_code: str = ""
762 player_token: str = ""
764 @classmethod
765 def from_dict(cls, payload: dict[str, Any]) -> DiscardDrawnRequest:
766 _require_type(payload, cls.type)
767 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
769 def to_dict(self) -> dict[str, Any]:
770 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
773@dataclass(frozen=True)
774class DiscardDrawnResponse:
775 type: ClassVar[str] = "discard_drawn_response"
776 bonus: str | None # "JACK" | "QUEEN" | "KING" | None
777 game: GameSnapshot
779 @classmethod
780 def from_dict(cls, payload: dict[str, Any]) -> DiscardDrawnResponse:
781 _require_type(payload, cls.type)
782 return cls(bonus=payload.get("bonus"), game=GameSnapshot.from_dict(payload["game"]))
784 def to_dict(self) -> dict[str, Any]:
785 return {"type": self.type, "bonus": self.bonus, "game": self.game.to_dict()}
788@dataclass(frozen=True)
789class PeekOwnRequest:
790 type: ClassVar[str] = "peek_own"
791 join_code: str = ""
792 player_token: str = ""
793 hand_index: int = 0
795 @classmethod
796 def from_dict(cls, payload: dict[str, Any]) -> PeekOwnRequest:
797 _require_type(payload, cls.type)
798 return cls(
799 join_code=payload["join_code"],
800 player_token=payload["player_token"],
801 hand_index=int(payload["hand_index"]),
802 )
804 def to_dict(self) -> dict[str, Any]:
805 return {
806 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
807 "hand_index": self.hand_index,
808 }
811@dataclass(frozen=True)
812class PeekOwnResponse:
813 type: ClassVar[str] = "peek_own_response"
814 card: GameCardSlot
815 game: GameSnapshot
817 @classmethod
818 def from_dict(cls, payload: dict[str, Any]) -> PeekOwnResponse:
819 _require_type(payload, cls.type)
820 return cls(card=GameCardSlot.from_dict(payload["card"]), game=GameSnapshot.from_dict(payload["game"]))
822 def to_dict(self) -> dict[str, Any]:
823 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()}
826@dataclass(frozen=True)
827class SwapCardRequest:
828 type: ClassVar[str] = "swap_card"
829 join_code: str = ""
830 player_token: str = ""
831 own_index: int = 0
832 target_player_name: str = ""
833 target_index: int = 0
835 @classmethod
836 def from_dict(cls, payload: dict[str, Any]) -> SwapCardRequest:
837 _require_type(payload, cls.type)
838 return cls(
839 join_code=payload["join_code"],
840 player_token=payload["player_token"],
841 own_index=int(payload["own_index"]),
842 target_player_name=payload["target_player_name"],
843 target_index=int(payload["target_index"]),
844 )
846 def to_dict(self) -> dict[str, Any]:
847 return {
848 "type": self.type,
849 "join_code": self.join_code,
850 "player_token": self.player_token,
851 "own_index": self.own_index,
852 "target_player_name": self.target_player_name,
853 "target_index": self.target_index,
854 }
857@dataclass(frozen=True)
858class SwapCardResponse:
859 type: ClassVar[str] = "swap_card_response"
860 game: GameSnapshot
862 @classmethod
863 def from_dict(cls, payload: dict[str, Any]) -> SwapCardResponse:
864 _require_type(payload, cls.type)
865 return cls(game=GameSnapshot.from_dict(payload["game"]))
867 def to_dict(self) -> dict[str, Any]:
868 return {"type": self.type, "game": self.game.to_dict()}
871@dataclass(frozen=True)
872class SkipBonusRequest:
873 type: ClassVar[str] = "skip_bonus"
874 join_code: str = ""
875 player_token: str = ""
877 @classmethod
878 def from_dict(cls, payload: dict[str, Any]) -> SkipBonusRequest:
879 _require_type(payload, cls.type)
880 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
882 def to_dict(self) -> dict[str, Any]:
883 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
886@dataclass(frozen=True)
887class SkipBonusResponse:
888 type: ClassVar[str] = "skip_bonus_response"
889 game: GameSnapshot
891 @classmethod
892 def from_dict(cls, payload: dict[str, Any]) -> SkipBonusResponse:
893 _require_type(payload, cls.type)
894 return cls(game=GameSnapshot.from_dict(payload["game"]))
896 def to_dict(self) -> dict[str, Any]:
897 return {"type": self.type, "game": self.game.to_dict()}
900@dataclass(frozen=True)
901class SnapRequest:
902 type: ClassVar[str] = "snap"
903 join_code: str = ""
904 player_token: str = ""
905 hand_index: int = 0
907 @classmethod
908 def from_dict(cls, payload: dict[str, Any]) -> SnapRequest:
909 _require_type(payload, cls.type)
910 return cls(
911 join_code=payload["join_code"],
912 player_token=payload["player_token"],
913 hand_index=int(payload["hand_index"]),
914 )
916 def to_dict(self) -> dict[str, Any]:
917 return {
918 "type": self.type, "join_code": self.join_code, "player_token": self.player_token,
919 "hand_index": self.hand_index,
920 }
923@dataclass(frozen=True)
924class SnapResponse:
925 type: ClassVar[str] = "snap_response"
926 correct: bool
927 game: GameSnapshot
929 @classmethod
930 def from_dict(cls, payload: dict[str, Any]) -> SnapResponse:
931 _require_type(payload, cls.type)
932 return cls(correct=payload["correct"], game=GameSnapshot.from_dict(payload["game"]))
934 def to_dict(self) -> dict[str, Any]:
935 return {"type": self.type, "correct": self.correct, "game": self.game.to_dict()}
938@dataclass(frozen=True)
939class BeaverRequest:
940 """Player declares 'Beaver!' to trigger the last round."""
941 type: ClassVar[str] = "beaver"
942 join_code: str = ""
943 player_token: str = ""
945 @classmethod
946 def from_dict(cls, payload: dict[str, Any]) -> BeaverRequest:
947 _require_type(payload, cls.type)
948 return cls(join_code=payload["join_code"], player_token=payload["player_token"])
950 def to_dict(self) -> dict[str, Any]:
951 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token}
954@dataclass(frozen=True)
955class BeaverResponse:
956 type: ClassVar[str] = "beaver_response"
957 game: GameSnapshot
959 @classmethod
960 def from_dict(cls, payload: dict[str, Any]) -> BeaverResponse:
961 _require_type(payload, cls.type)
962 return cls(game=GameSnapshot.from_dict(payload["game"]))
964 def to_dict(self) -> dict[str, Any]:
965 return {"type": self.type, "game": self.game.to_dict()}