Coverage for backend/game_state.py: 98%

1from __future__ import annotations 

2 

3import dataclasses 

4import queue 

5from collections.abc import Iterable 

6from dataclasses import dataclass 

7from uuid import uuid4 

8 

9from .models import Chat, GameStateUpdate, Initializer, LifecycleEvent, Message, Role 

10 

11 

12@dataclass 

13class Lobby: 

14 """ 

15 Maintains lobby state and allows communication to and from players. 

16 

17 Attributes: 

18 id: The lobby's internal ID 

19 players: Map of player id to Players 

20 channel: Message queue for incoming messages 

21 code: The code used to join the lobby 

22 loop_started: Whether the game loop has started 

23 open: Whether new players can still join 

24 """ 

25 

26 code: tuple[str, ...] 

27 players: dict[str, Player] = dataclasses.field(default_factory=dict) 

28 channel: queue.Queue[TaggedMessage] = dataclasses.field(default_factory=queue.Queue) 

29 id: str = dataclasses.field(init=False, default_factory=lambda: uuid4().hex) 

30 loop_started: bool = False 

31 open: bool = True 

32 

33 def broadcast(self, msg: Message, *, exclude: Iterable[str] = ()) -> None: 

34 """Send a message to all players except those in exclude.""" 

35 exclude = set(exclude) 

36 for player in self.players.values(): 

37 if player.id not in exclude: 

38 player.send(msg) 

39 

40 def messages(self) -> Iterable[TaggedMessage]: 

41 """Iterate over messages that are currently available.""" 

42 while True: 

43 try: 

44 yield self.channel.get_nowait() 

45 except queue.Empty: 

46 return 

47 

48 

49@dataclass 

50class Player: 

51 """ 

52 A player in a lobby. 

53 

54 Attributes: 

55 id: The player's internal ID 

56 role: The player's role. None if the game has not started yet. 

57 channel: Message queue for outgoing messages. 

58 """ 

59 

60 channel: queue.Queue 

61 role: Role | None 

62 id: str = dataclasses.field(init=False, default_factory=lambda: uuid4().hex) 

63 

64 def send(self, msg: Message) -> None: 

65 """Send a message to this player.""" 

66 self.channel.put(msg) 

67 

68 

69@dataclass(frozen=True) 

70class TaggedMessage: 

71 """A message with extra metadata attached.""" 

72 

73 data: Initializer | GameStateUpdate | LifecycleEvent | Chat 

74 id: str 

75 

76 

77class LobbyNotFoundError(ValueError): 

78 """The lobby does not exist.""" 

79 

80 

81class LobbyFullError(ValueError): 

82 """The lobby is full.""" 

83 

84 

85class LobbyClosedError(ValueError): 

86 """The lobby is not open."""