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

1from __future__ import annotations 

2 

3from dataclasses import asdict, dataclass 

4from typing import Any, Callable, ClassVar, Literal 

5 

6from beaverbunch.core.game_settings import GameSettings 

7from beaverbunch.network.session import Session 

8 

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] 

48 

49 

50@dataclass(frozen=True) 

51class SessionSettingsPayload: 

52 """Serializable subset of GameSettings used by the lobby protocol.""" 

53 

54 initial_hand_size: int 

55 initial_cards_known: int 

56 min_players: int 

57 max_players: int 

58 

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 ) 

67 

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 ) 

76 

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 ) 

84 

85 def to_dict(self) -> dict[str, Any]: 

86 return asdict(self) 

87 

88 

89@dataclass(frozen=True) 

90class SessionSnapshot: 

91 """Serializable view of the current lobby state.""" 

92 

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 

101 

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 

107 

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 ) 

118 

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 ) 

131 

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 } 

143 

144 

145@dataclass(frozen=True) 

146class CreateSessionRequest: 

147 type: ClassVar[str] = "create_session" 

148 host_name: str | None = None 

149 settings: SessionSettingsPayload | None = None 

150 

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 ) 

159 

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 

165 

166 

167@dataclass(frozen=True) 

168class JoinSessionRequest: 

169 type: ClassVar[str] = "join_session" 

170 join_code: str = "" 

171 player_name: str = "" 

172 

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"]) 

177 

178 def to_dict(self) -> dict[str, Any]: 

179 return {"type": self.type, "join_code": self.join_code, "player_name": self.player_name} 

180 

181 

182@dataclass(frozen=True) 

183class StartSessionRequest: 

184 type: ClassVar[str] = "start_session" 

185 join_code: str = "" 

186 player_token: str = "" 

187 

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"]) 

192 

193 def to_dict(self) -> dict[str, Any]: 

194 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

195 

196 

197@dataclass(frozen=True) 

198class GetSessionRequest: 

199 type: ClassVar[str] = "get_session" 

200 join_code: str = "" 

201 

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"]) 

206 

207 def to_dict(self) -> dict[str, Any]: 

208 return {"type": self.type, "join_code": self.join_code} 

209 

210 

211@dataclass(frozen=True) 

212class LeaveSessionRequest: 

213 type: ClassVar[str] = "leave_session" 

214 join_code: str = "" 

215 player_token: str = "" 

216 

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"]) 

221 

222 def to_dict(self) -> dict[str, Any]: 

223 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

224 

225 

226@dataclass(frozen=True) 

227class CloseSessionRequest: 

228 type: ClassVar[str] = "close_session" 

229 join_code: str = "" 

230 player_token: str = "" 

231 

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"]) 

236 

237 def to_dict(self) -> dict[str, Any]: 

238 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

239 

240 

241@dataclass(frozen=True) 

242class CreateSessionResponse: 

243 type: ClassVar[str] = "create_session_response" 

244 session: SessionSnapshot 

245 player_token: str | None = None 

246 

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 ) 

254 

255 def to_dict(self) -> dict[str, Any]: 

256 return {"type": self.type, "player_token": self.player_token, "session": self.session.to_dict()} 

257 

258 

259@dataclass(frozen=True) 

260class JoinSessionResponse: 

261 type: ClassVar[str] = "join_session_response" 

262 session: SessionSnapshot 

263 player_token: str = "" 

264 

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 ) 

272 

273 def to_dict(self) -> dict[str, Any]: 

274 return {"type": self.type, "player_token": self.player_token, "session": self.session.to_dict()} 

275 

276 

277@dataclass(frozen=True) 

278class StartSessionResponse: 

279 type: ClassVar[str] = "start_session_response" 

280 session: SessionSnapshot 

281 

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"])) 

286 

287 def to_dict(self) -> dict[str, Any]: 

288 return {"type": self.type, "session": self.session.to_dict()} 

289 

290 

291@dataclass(frozen=True) 

292class GetSessionResponse: 

293 type: ClassVar[str] = "get_session_response" 

294 session: SessionSnapshot 

295 

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"])) 

300 

301 def to_dict(self) -> dict[str, Any]: 

302 return {"type": self.type, "session": self.session.to_dict()} 

303 

304 

305@dataclass(frozen=True) 

306class LeaveSessionResponse: 

307 type: ClassVar[str] = "leave_session_response" 

308 session: SessionSnapshot | None 

309 

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) 

315 

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 

321 

322 

323@dataclass(frozen=True) 

324class CloseSessionResponse: 

325 type: ClassVar[str] = "close_session_response" 

326 session: SessionSnapshot 

327 

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"])) 

332 

333 def to_dict(self) -> dict[str, Any]: 

334 return {"type": self.type, "session": self.session.to_dict()} 

335 

336 

337@dataclass(frozen=True) 

338class ErrorResponse: 

339 type: ClassVar[str] = "error" 

340 error_code: str 

341 message: str 

342 

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"]) 

347 

348 def to_dict(self) -> dict[str, Any]: 

349 return {"type": self.type, "error_code": self.error_code, "message": self.message} 

350 

351 

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}") 

362 

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) 

386 

387 

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}") 

398 

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) 

423 

424 

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}'") 

429 

430 

431# --------------------------------------------------------------------------- 

432# Game-phase protocol types 

433# --------------------------------------------------------------------------- 

434 

435@dataclass(frozen=True) 

436class GameCardSlot: 

437 """Serializable view of a single card slot in a player's hand. 

438 

439 ``suit`` and ``value`` are ``None`` when the card is unknown to the 

440 requesting player (i.e. face-down from their perspective). 

441 """ 

442 

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 

446 

447 def to_dict(self) -> dict[str, Any]: 

448 return {"known": self.known, "suit": self.suit, "value": self.value} 

449 

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 ) 

457 

458 

459@dataclass(frozen=True) 

460class PlayerGameView: 

461 """Serializable view of one player from another player's perspective.""" 

462 

463 name: str 

464 hand: list[GameCardSlot] 

465 is_current_turn: bool 

466 

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 } 

473 

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 ) 

481 

482 

483@dataclass(frozen=True) 

484class GameSnapshot: 

485 """Perspective-aware view of the running game for a specific player.""" 

486 

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" 

495 

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 } 

507 

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 ) 

520 

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 

529 

530 state = game.state 

531 

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) 

539 

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 ) 

554 

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) 

559 

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) 

564 

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" 

568 

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()] 

572 

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 ) 

583 

584 

585# --- Game request/response dataclasses --- 

586 

587@dataclass(frozen=True) 

588class GetGameRequest: 

589 type: ClassVar[str] = "get_game" 

590 join_code: str = "" 

591 player_token: str = "" 

592 

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"]) 

597 

598 def to_dict(self) -> dict[str, Any]: 

599 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

600 

601 

602@dataclass(frozen=True) 

603class GetGameResponse: 

604 type: ClassVar[str] = "get_game_response" 

605 game: GameSnapshot 

606 

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"])) 

611 

612 def to_dict(self) -> dict[str, Any]: 

613 return {"type": self.type, "game": self.game.to_dict()} 

614 

615 

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] 

622 

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 ) 

631 

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 } 

637 

638 

639@dataclass(frozen=True) 

640class PeekInitialResponse: 

641 type: ClassVar[str] = "peek_initial_response" 

642 cards: list[GameCardSlot] 

643 game: GameSnapshot 

644 

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 ) 

652 

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()} 

655 

656 

657@dataclass(frozen=True) 

658class DrawCardRequest: 

659 type: ClassVar[str] = "draw_card" 

660 join_code: str = "" 

661 player_token: str = "" 

662 

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"]) 

667 

668 def to_dict(self) -> dict[str, Any]: 

669 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

670 

671 

672@dataclass(frozen=True) 

673class DrawCardResponse: 

674 type: ClassVar[str] = "draw_card_response" 

675 card: GameCardSlot 

676 game: GameSnapshot 

677 

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"])) 

682 

683 def to_dict(self) -> dict[str, Any]: 

684 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()} 

685 

686 

687@dataclass(frozen=True) 

688class DrawDiscardRequest: 

689 type: ClassVar[str] = "draw_discard" 

690 join_code: str = "" 

691 player_token: str = "" 

692 

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"]) 

697 

698 def to_dict(self) -> dict[str, Any]: 

699 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

700 

701 

702@dataclass(frozen=True) 

703class DrawDiscardResponse: 

704 type: ClassVar[str] = "draw_discard_response" 

705 card: GameCardSlot 

706 game: GameSnapshot 

707 

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"])) 

712 

713 def to_dict(self) -> dict[str, Any]: 

714 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()} 

715 

716 

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 

723 

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 ) 

732 

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 } 

738 

739 

740@dataclass(frozen=True) 

741class KeepCardResponse: 

742 type: ClassVar[str] = "keep_card_response" 

743 discarded_card: GameCardSlot 

744 game: GameSnapshot 

745 

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 ) 

753 

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()} 

756 

757 

758@dataclass(frozen=True) 

759class DiscardDrawnRequest: 

760 type: ClassVar[str] = "discard_drawn" 

761 join_code: str = "" 

762 player_token: str = "" 

763 

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"]) 

768 

769 def to_dict(self) -> dict[str, Any]: 

770 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

771 

772 

773@dataclass(frozen=True) 

774class DiscardDrawnResponse: 

775 type: ClassVar[str] = "discard_drawn_response" 

776 bonus: str | None # "JACK" | "QUEEN" | "KING" | None 

777 game: GameSnapshot 

778 

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"])) 

783 

784 def to_dict(self) -> dict[str, Any]: 

785 return {"type": self.type, "bonus": self.bonus, "game": self.game.to_dict()} 

786 

787 

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 

794 

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 ) 

803 

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 } 

809 

810 

811@dataclass(frozen=True) 

812class PeekOwnResponse: 

813 type: ClassVar[str] = "peek_own_response" 

814 card: GameCardSlot 

815 game: GameSnapshot 

816 

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"])) 

821 

822 def to_dict(self) -> dict[str, Any]: 

823 return {"type": self.type, "card": self.card.to_dict(), "game": self.game.to_dict()} 

824 

825 

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 

834 

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 ) 

845 

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 } 

855 

856 

857@dataclass(frozen=True) 

858class SwapCardResponse: 

859 type: ClassVar[str] = "swap_card_response" 

860 game: GameSnapshot 

861 

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"])) 

866 

867 def to_dict(self) -> dict[str, Any]: 

868 return {"type": self.type, "game": self.game.to_dict()} 

869 

870 

871@dataclass(frozen=True) 

872class SkipBonusRequest: 

873 type: ClassVar[str] = "skip_bonus" 

874 join_code: str = "" 

875 player_token: str = "" 

876 

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"]) 

881 

882 def to_dict(self) -> dict[str, Any]: 

883 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

884 

885 

886@dataclass(frozen=True) 

887class SkipBonusResponse: 

888 type: ClassVar[str] = "skip_bonus_response" 

889 game: GameSnapshot 

890 

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"])) 

895 

896 def to_dict(self) -> dict[str, Any]: 

897 return {"type": self.type, "game": self.game.to_dict()} 

898 

899 

900@dataclass(frozen=True) 

901class SnapRequest: 

902 type: ClassVar[str] = "snap" 

903 join_code: str = "" 

904 player_token: str = "" 

905 hand_index: int = 0 

906 

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 ) 

915 

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 } 

921 

922 

923@dataclass(frozen=True) 

924class SnapResponse: 

925 type: ClassVar[str] = "snap_response" 

926 correct: bool 

927 game: GameSnapshot 

928 

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"])) 

933 

934 def to_dict(self) -> dict[str, Any]: 

935 return {"type": self.type, "correct": self.correct, "game": self.game.to_dict()} 

936 

937 

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 = "" 

944 

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"]) 

949 

950 def to_dict(self) -> dict[str, Any]: 

951 return {"type": self.type, "join_code": self.join_code, "player_token": self.player_token} 

952 

953 

954@dataclass(frozen=True) 

955class BeaverResponse: 

956 type: ClassVar[str] = "beaver_response" 

957 game: GameSnapshot 

958 

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"])) 

963 

964 def to_dict(self) -> dict[str, Any]: 

965 return {"type": self.type, "game": self.game.to_dict()}