Coverage for backend/game.py: 94%
141 statements
« prev ^ index » next coverage.py v7.6.12, created at 2025-05-02 01:42 +0000
1import difflib
2import itertools
3import math
4import random
5import threading
6import typing
7from collections import deque
9import structlog
11from .constants import Settings
12from .game_state import Lobby, Player, TaggedMessage
13from .models import (
14 Burger,
15 Chat,
16 DayEnd,
17 Drink,
18 GameEnd,
19 GameStart,
20 Message,
21 NewOrder,
22 Order,
23 OrderComponent,
24 OrderScore,
25 OrderSubmission,
26 PlayerJoin,
27 PlayerLeave,
28 Role,
29 RoleAssignment,
30 Side,
31)
33logger = structlog.stdlib.get_logger(__file__)
35BURGER_INGREDIENTS = ["Patty", "Lettuce", "Onion", "Tomato", "Ketchup", "Mustard", "Cheese"]
36DRINK_COLORS = [
37 ["Blue", "#34C6F4"],
38 ["Green", "#99CA3C"],
39 ["Yellow", "#e2d700"],
40 ["Red", "#FF0000"],
41 ["Orange", "#F5841F"],
42 ["Purple", "#7E69AF"],
43]
44DRINK_SIZES = ["small", "medium", "large"]
45SIDE_TYPES = ["fries", "onionRings", "mozzarellaSticks"]
47MESSAGES_PER_LOOP = 5
48DAYS_PER_GAME = 5
50s = Settings()
53class GameLoop:
54 """Implements game logic."""
56 def __init__(self, lobby: Lobby) -> None:
57 self.lobby = lobby
59 # Backend stores game score and day
60 # Acts as source of truth in case any messages fail to send
61 self.day = 1
62 self.score = 0
64 self.customers = dict()
65 self.day_score = {}
67 self.orders = deque()
68 self.order: Order
70 self.started = False
72 def run(self) -> None:
73 """
74 Main game loop.
76 Processes messages in a loop.
77 """
78 while True:
79 for message in itertools.islice(self.lobby.messages(), MESSAGES_PER_LOOP):
80 match message.data:
81 case GameStart():
82 self.start_game(message.id)
83 case GameEnd():
84 return
85 case PlayerJoin():
86 self.lobby.broadcast(Message(data=message.data), exclude=[message.id])
87 case PlayerLeave(id=id):
88 self.lobby.players.pop(id)
89 self.lobby.broadcast(Message(data=message.data))
90 case Chat():
91 self.typing_indicator(message)
92 case OrderComponent() as component:
93 self.manager.send(Message(data=component))
94 case OrderSubmission(order=order):
95 logger.debug("Received order.", order=order)
96 self.handle_scoring(order=order)
97 self.handle_next_order()
98 case _:
99 logger.warning("Unimplemented message.", message=message.data)
101 def start_game(self, id: str) -> None:
102 """
103 Start game and generate the first order.
105 Args:
106 id: The id of the player that started the game.
107 """
108 if self.started:
109 return
111 logger.debug("Starting game.")
113 self.started = True
114 self.lobby.open = False
115 self.orders = get_orders(day=self.day, num_players=len(self.lobby.players))
117 self.lobby.broadcast(Message(data=GameStart()), exclude=[id])
118 self.assign_roles()
119 self.handle_next_order()
121 def assign_roles(self) -> None:
122 """Assign roles to players."""
123 players = self.lobby.players.values()
124 roles = list(Role)[: len(players)]
126 # Shuffle initial roles in cycle mode
127 if s.mode == "cycle":
128 random.shuffle(roles)
130 for player, role in zip(players, roles, strict=True):
131 player.role = role
132 player.send(Message(data=RoleAssignment(role=role)))
134 def rotate_roles(self) -> None:
135 """Rotate player roles such that no player has the same role as the last day."""
136 players = self.lobby.players.values()
137 roles = typing.cast(list[Role], [player.role for player in players])
139 # 4 players; efficiency isn't an issue
140 roles.append(roles.pop(0))
142 for player, role in zip(players, roles, strict=False):
143 player.role = role
144 player.send(Message(data=RoleAssignment(role=role)))
146 def handle_next_order(self) -> None:
147 """Give manager next order."""
148 if len(self.orders) == 0:
149 self.handle_new_day()
150 self.orders = get_orders(day=self.day, num_players=len(self.lobby.players))
152 self.order = self.orders.pop()
153 logger.debug("Generated order.", order=self.order)
154 self.manager.send(Message(data=NewOrder(order=self.order)))
156 def handle_new_day(self) -> None:
157 """Update current day."""
158 customers = self.customers[self.day]
159 day_score = self.day_score[self.day]
160 self.day += 1
161 if self.day == DAYS_PER_GAME + 1:
162 self.lobby.broadcast(Message(data=GameEnd()))
163 logger.debug("Game complete.")
164 else:
165 logger.debug("New day.", day=self.day)
166 # Rotate roles on cycle mode
167 if s.mode == "cycle":
168 self.rotate_roles()
169 self.lobby.broadcast(Message(data=DayEnd(day=self.day, customers_served=customers, score=day_score)))
171 def handle_scoring(self, order: Order) -> None:
172 """Updates all scores."""
173 score = self.grade_order(order)
174 self.customers[self.day] = self.customers.get(self.day, 0) + 1
175 self.day_score[self.day] = self.day_score.get(self.day, 0) + score
176 self.score += score
177 self.lobby.broadcast(Message(data=OrderScore(score=self.score)))
179 def grade_order(self, order: Order) -> int:
180 """
181 Grade order based on correctness.
183 See capstone-projects-2025-spring.github.io/aac-go-fish/docs/requirements/features-and-requirements#scoring
184 """
185 # Burgers are graded based on edit distance to the correct burger for up to 2 extra dollars + 3 base dollars
186 burger_score = 300
187 if order.burger is not None:
188 # this is never None, but the type checker doesn't know that
189 assert self.order.burger is not None
191 similarity = difflib.SequenceMatcher(None, order.burger.ingredients, self.order.burger.ingredients).ratio()
192 burger_score += int(200 * round(similarity, 2))
194 # Sides are graded based on completeness.
195 # Up to 2 extra dollars + 1 base dollar
196 side_score = 0
197 if self.order.side is not None:
198 side_score += 100
200 if self.order.side == order.side:
201 side_score += 200
203 # Drink attributes are equally weighted, with the fill percentage being
204 # graded on the square root of the error from the correct fill
205 # percentage. up to 2 bonus dollars + 2 base dollars
206 drink_score = 0
207 if not (self.order.drink is None or order.drink is None):
208 drink_score += 200
210 correct = (self.order.drink.size == order.drink.size) + (self.order.drink.color == order.drink.color)
211 drink_score += 50 * correct
213 drink_score += int(math.sqrt(1 - abs(1 - order.drink.fill / 100)) * 100)
215 return burger_score + side_score + drink_score
217 def typing_indicator(self, msg: TaggedMessage) -> None:
218 """Send an indicator that the manager is typing."""
219 self.lobby.broadcast(Message(data=msg.data), exclude=[msg.id])
221 @property
222 def manager(self) -> Player:
223 """The player with the manager role."""
224 return next(player for player in self.lobby.players.values() if player.role == Role.manager)
227def start_main_loop(lobby: Lobby) -> None:
228 """Start the main game loop."""
229 loop = GameLoop(lobby)
230 threading.Thread(target=loop.run).start()
233def get_orders(day: int, num_players: int) -> deque[Order]:
234 """Return a queue of orders for the next day."""
235 orders = deque()
236 for _ in range(_orders_on_day(day)):
237 orders.append(_generate_order(num_players))
238 return orders
241def _orders_on_day(day: int) -> int:
242 """Compute the number of orders on a given day."""
243 return day
246def _generate_order(num_players: int) -> Order:
247 """Generate an order based on the number of players."""
248 order = Order(
249 burger=Burger(
250 ingredients=["Bottom Bun"] + random.choices(BURGER_INGREDIENTS, k=random.randint(3, 5)) + ["Top Bun"]
251 ),
252 drink=None,
253 side=None,
254 )
256 if num_players >= 3:
257 order.drink = Drink(color=random.choice(DRINK_COLORS)[1], fill=100, size=random.choice(DRINK_SIZES))
259 if num_players >= 4:
260 order.side = Side(table_state=random.choice(SIDE_TYPES))
262 return order