[Small] Feature/game repo

src/app/game_manager.py

@@ -0,0 +1,53 @@
+from typing import Protocol, runtime_checkable
+
+from src.app.game_repo.base import BaseGameStateRepo
+from src.models.game_state import GameState
+from src.models.ids import GameId
+from src.models.message import GameToPlayerMessage, PlayerToGameMessage, ToGameMessage, InternalMessage, FromGameMessage
+
+
+@runtime_checkable
+class CanReceiveGameToPlayerMessages(Protocol):
+    # A generic stub for the front end implementation
+    def handle_player_messages(self, msgs: list[GameToPlayerMessage]) -> None: ...
+
+
+@runtime_checkable
+class CanReceiveToGameMessage(Protocol):
+    # A generic stub for the game engine implementation
+    @classmethod
+    def handle_message(cls, game_state: GameState, msg: ToGameMessage) -> tuple[GameState, list[FromGameMessage]]: ...
+
+
+class GameManager:
+    def __init__(
+        self,
+        game_repo: BaseGameStateRepo,
+        game_engine: CanReceiveToGameMessage,
+        front_end: CanReceiveGameToPlayerMessages,
+    ) -> None:
+        assert isinstance(game_repo, BaseGameStateRepo)
+        assert isinstance(game_engine, CanReceiveToGameMessage)
+        assert isinstance(front_end, CanReceiveGameToPlayerMessages)
+        self.game_repo = game_repo
+        self.game_engine = game_engine
+        self.front_end = front_end
+
+    def handle_player_message(self, game_id: GameId, msg: PlayerToGameMessage) -> None:
+        # TODO Make this atomic
+        game_state = self.game_repo.get_game_state(game_id)
+        updated_game_state = self.handle_message(game_state=game_state, msg=msg)
+        self.game_repo.update_game_state(updated_game_state)
+
+    def handle_message(self, game_state: GameState, msg: ToGameMessage) -> GameState:
+        game_state, messages = self.game_engine.handle_message(game_state=game_state, msg=msg)
+
+        msgs_to_players = [e for e in messages if isinstance(e, GameToPlayerMessage)]
+        self.front_end.handle_player_messages(msgs=msgs_to_players)
+
+        msgs_to_self = [e for e in messages if isinstance(e, InternalMessage)]
+        if not len(msgs_to_self):
+            return game_state
+
+        assert len(msgs_to_self) == 1, "There should be at most one internal message generated by the engine."
+        return self.handle_message(game_state=game_state, msg=msgs_to_self[0])

src/app/game_repo/base.py

@@ -0,0 +1,30 @@
+from abc import ABC, abstractmethod
+
+from src.models.game_state import GameState
+from src.models.ids import GameId
+
+
+class BaseGameStateRepo(ABC):
+    @abstractmethod
+    def generate_game_id(self) -> GameId:
+        pass
+
+    @abstractmethod
+    def add_game_state(self, game: GameState) -> None:
+        pass
+
+    @abstractmethod
+    def update_game_state(self, game: GameState) -> None:
+        pass
+
+    @abstractmethod
+    def get_game_state(self, game_id: GameId) -> GameState:
+        pass
+
+    @abstractmethod
+    def list_game_ids(self) -> list[GameId]:
+        pass
+
+    @abstractmethod
+    def delete_game_state(self, game_id: GameId, missing_ok: bool = True) -> None:
+        pass

src/app/game_repo/file_game_repo.py

@@ -0,0 +1,58 @@
+from pathlib import Path
+
+from src.app.game_repo.base import BaseGameStateRepo
+from src.directories import game_cache_dir
+from src.models.game_state import GameState
+from src.models.ids import GameId
+from src.tools.serialization import serialize, deserialize
+
+
+class FileGameStateRepo(BaseGameStateRepo):
+    def __init__(self, cache_dir: Path = game_cache_dir) -> None:
+        self.cache_dir = cache_dir
+
+    def generate_game_id(self) -> GameId:
+        game_ids = self.list_game_ids()
+        return GameId(max(game_ids).as_int() + 1 if game_ids else 0)
+
+    def add_game_state(self, game: GameState) -> None:
+        path = self.game_id_to_file_path(game.game_id)
+        if path.exists():  # Todo make these checks threadsafe
+            raise FileExistsError(f"Game with ID {game.game_id} already exists.")
+        with open(path, "w") as file:
+            file.write(serialize(game))
+
+    def update_game_state(self, game: GameState) -> None:
+        path = self.game_id_to_file_path(game.game_id)
+        if not path.exists():
+            raise FileNotFoundError(f"Game with ID {game.game_id} does not exist.")
+        with open(path, "w") as file:
+            file.write(serialize(game))
+
+    def get_game_state(self, game_id: GameId) -> GameState:
+        path = self.game_id_to_file_path(game_id)
+        if not path.exists():
+            raise FileNotFoundError(f"Game with ID {game_id} does not exist.")
+        with open(path, "r") as file:
+            data = file.read()
+        return deserialize(x=data, cls=GameState)
+
+    def list_game_ids(self) -> list[GameId]:
+        game_files = self.cache_dir.glob("game_*.json")
+        return [self.file_path_to_game_id(file_path) for file_path in game_files if file_path.is_file()]
+
+    def delete_game_state(self, game_id: GameId, missing_ok: bool = True) -> None:
+        path = self.game_id_to_file_path(game_id)
+        path.unlink(missing_ok=missing_ok)
+
+    def game_id_to_file_path(self, game_id: GameId) -> Path:
+        """Get the file path for a specific game ID."""
+        return self.cache_dir / f"game_{game_id}.json"
+
+    @staticmethod
+    def file_path_to_game_id(file_path: Path) -> GameId:
+        """Extract the game ID from a file path."""
+        if not file_path.name.startswith("game_") or not file_path.name.endswith(".json"):
+            raise ValueError(f"Invalid game file name: {file_path.name}")
+        game_id_str = file_path.name[len("game_") : -len(".json")]
+        return GameId(int(game_id_str))

src/directories.py

@@ -1,3 +1,4 @@
 from pathlib import Path
 
 root_dir = Path(__file__).parent
+game_cache_dir = root_dir / "game_cache"

src/engine/engine.py

@@ -1,102 +1,104 @@
-from src.models.event import (
-    Event,
+from src.models.game_state import GameState
+from src.models.message import (
     UpdateBidRequest,
     BuyAssetRequest,
     EndTurn,
     UpdateBidResponse,
-    BuyAssetResponse, NewPhase
+    BuyAssetResponse,
+    NewPhase,
+    ToGameMessage,
+    FromGameMessage,
 )
-from src.models.game_state import GameState
 
 
 class Engine:
     @classmethod
-    def handle_event(cls, game_state: GameState, event: Event) -> tuple[GameState, list[Event]]:
+    def handle_message(cls, game_state: GameState, msg: ToGameMessage) -> tuple[GameState, list[FromGameMessage]]:
         """
-        Events happen every time a player takes an action or a timer runs out.
-        Every time an event occurs, the engine is informed and it can then:
+        Messages can come from players or from the game itself
+        Every time a message occurs, the engine is informed and it can then:
         -Update the game state
-        -Send messages back to the players OR to itself to trigger a new phase
+        -Send messages back to the players OR to itself
         :param game_state: The current state of the game
-        :param event: The triggering event
-        :return: The new game state and a list of events to be sent
+        :param msg: The triggering message
+        :return: The new game state and a list of messages to be sent
         """
-        # Handle the event based on its type
-        if isinstance(event, NewPhase):
-            return cls.handle_new_phase_event(game_state, event)
-        elif isinstance(event, UpdateBidRequest):
-            return cls.handle_update_bid_event(game_state, event)
-        elif isinstance(event, BuyAssetRequest):
-            return cls.handle_buy_asset_event(game_state, event)
-        elif isinstance(event, EndTurn):
-            return cls.handle_end_turn_event(game_state, event)
+        # Handle the message based on its type
+        if isinstance(msg, NewPhase):
+            return cls.handle_new_phase_message(game_state, msg)
+        elif isinstance(msg, UpdateBidRequest):
+            return cls.handle_update_bid_message(game_state, msg)
+        elif isinstance(msg, BuyAssetRequest):
+            return cls.handle_buy_asset_message(game_state, msg)
+        elif isinstance(msg, EndTurn):
+            return cls.handle_end_turn_message(game_state, msg)
         else:
-            raise NotImplementedError(f"Event type {type(event)} not implemented.")
+            raise NotImplementedError(f"message type {type(msg)} not implemented.")
 
     @classmethod
-    def handle_new_phase_event(
-            cls,
-            game_state: GameState,
-            event: NewPhase,
-    ) -> tuple[GameState, list[Event]]:
+    def handle_new_phase_message(
+        cls,
+        game_state: GameState,
+        msg: NewPhase,
+    ) -> tuple[GameState, list[FromGameMessage]]:
         """
-        Handle a new phase event.
+        Handle a new phase message.
         :param game_state: The current state of the game
-        :param event: The triggering event
-        :return: The new game state and a list of events to be sent to the player interface
+        :param msg: The triggering message
+        :return: The new game state and a list of messages to be sent to the player interface
         """
         # TODO Do something depending on what phase we are in
         # TODO if we are in the da_auction phase, we need to run the market coupling algorithm
         raise NotImplementedError()
 
     @classmethod
-    def handle_update_bid_event(
-            cls,
-            game_state: GameState,
-            event: UpdateBidRequest,
+    def handle_update_bid_message(
+        cls,
+        game_state: GameState,
+        msg: UpdateBidRequest,
     ) -> tuple[GameState, list[UpdateBidResponse]]:
         """
-        Handle an update bid event.
+        Handle an update bid message.
         :param game_state: The current state of the game
-        :param event: The triggering event
-        :return: The new game state and a list of events to be sent to the player interface
+        :param msg: The triggering message
+        :return: The new game state and a list of messages to be sent to the player interface
         """
         # TODO Check if the bid is valid (including if the player can afford it).
         # TODO Update bids in the game state
-        # TODO Return one UpdateBidResponseEvent for the player who made the bid
+        # TODO Return one UpdateBidResponse for the player who made the bid
         raise NotImplementedError()
 
     @classmethod
-    def handle_buy_asset_event(
-            cls,
-            game_state: GameState,
-            event: BuyAssetRequest,
+    def handle_buy_asset_message(
+        cls,
+        game_state: GameState,
+        msg: BuyAssetRequest,
     ) -> tuple[GameState, list[BuyAssetResponse]]:
         """
-        Handle a buy asset event.
+        Handle a buy asset message.
         :param game_state: The current state of the game
-        :param event: The triggering event
-        :return: The new game state and a list of events to be sent to the player interface
+        :param msg: The triggering message
+        :return: The new game state and a list of messages to be sent to the player interface
         """
         # TODO Check if the request is valid (including if the asset is for sale and the player can afford it).
         # TODO Update asset ownership
         # TODO Return a BuyAssetResponse
         raise NotImplementedError()
 
     @classmethod
-    def handle_end_turn_event(
-            cls,
-            game_state: GameState,
-            event: EndTurn,
+    def handle_end_turn_message(
+        cls,
+        game_state: GameState,
+        msg: EndTurn,
     ) -> tuple[GameState, list[NewPhase]]:
         """
-        Handle an end turn event.
+        Handle an end turn message.
         :param game_state: The current state of the game
-        :param event: The triggering event
-        :return: The new game state and a list of events to be sent to the player interface
+        :param msg: The triggering message
+        :return: The new game state and a list of messages to be sent to the player interface
         """
         # TODO Update the player to indicate that their turn has ended
         # TODO If this phase requires players to play one by one (Do we need such a phase?) Then cycle to the next player
         # TODO Check if all players have ended their turns and we need to move on to the next phase
-        # TODO If necessary, return an event to signal yourself to go to a different phase
+        # TODO If necessary, return an message to signal yourself to go to a different phase
         raise NotImplementedError()

src/engine/new_game.py

@@ -1,45 +1,18 @@
 from src.models.game_settings import GameSettings
-from src.models.ids import (
-    GameId,
-    PlayerId,
-    BusId,
-    AssetId,
-    TransmissionId
-)
-from src.models.game_state import (
-    GameState,
-    Phase
-)
-from src.models.player import (
-    Player,
-    PlayerRepo
-)
-from src.models.buses import (
-    BusRepo,
-    Bus
-)
-from src.models.assets import (
-    AssetRepo,
-    AssetInfo
-)
-from src.models.transmission import (
-    TransmissionRepo,
-    TransmissionInfo
-)
-from typing import (
-    List,
-    Dict
-)
+from src.models.ids import GameId, PlayerId, BusId, AssetId, TransmissionId
+from src.models.game_state import GameState, Phase
+from src.models.player import Player, PlayerRepo
+from src.models.buses import BusRepo, Bus
+from src.models.assets import AssetRepo, AssetInfo
+from src.models.transmission import TransmissionRepo, TransmissionInfo
+from typing import List, Dict
 
 
 __all__ = ["create_new_game"]
 
 
 def create_new_game(
-        game_id: GameId,
-        settings: GameSettings,
-        player_names: List[str],
-        player_colors: List[str]
+    game_id: GameId, settings: GameSettings, player_names: List[str], player_colors: List[str]
 ) -> GameState:
     """
     Create a new game state with the given game ID and settings.
@@ -102,8 +75,11 @@ def _assign_bus_location(i: int) -> Dict[str, float]:
         return {'x': float(i), 'y': float(i)}  # TODO define logic for bus location (e.g., from map library)
 
     buses = [
-        Bus(id=BusId(i), player_id=PlayerId(i), **_assign_bus_location(i)) if i < n_players
-        else Bus(id=BusId(i), **_assign_bus_location(i))  # Neutral bus not owned by any player
+        (
+            Bus(id=BusId(i), player_id=PlayerId(i), **_assign_bus_location(i))
+            if i < n_players
+            else Bus(id=BusId(i), **_assign_bus_location(i))
+        )  # Neutral bus not owned by any player
         for i in range(1, settings.n_buses + 1)
     ]
     return BusRepo(buses)

src/models/game_settings.py

@@ -5,6 +5,7 @@
 @dataclass(frozen=True)
 class GameSettings:
     """A class to hold game settings."""
+
     n_buses: int = 5
     max_rounds: int = 20
     n_ice_cream: int = 5
@@ -31,4 +32,3 @@ def from_simple_dict(cls, simple_dict: dict) -> Self:
             initial_funds=simple_dict["initial_funds"],
             max_connections_per_bus=simple_dict["max_connections_per_bus"],
         )
-