diff --git a/socha/api/networking/game_client.py b/socha/api/networking/game_client.py index 27f33ab..4d92e6d 100644 --- a/socha/api/networking/game_client.py +++ b/socha/api/networking/game_client.py @@ -8,42 +8,14 @@ from typing import List, Union from socha.api.networking.xml_protocol_interface import XMLProtocolInterface -from socha.api.plugin.penguins import game_state -from socha.api.plugin.penguins.board import Field, Move, CartesianCoordinate, TeamEnum, Penguin, HexCoordinate +from socha.api.plugin.penguins.board import Move from socha.api.plugin.penguins.game_state import GameState -from socha.api.protocol.protocol import State, Board, Data, \ - Error, From, Join, Joined, JoinPrepared, JoinRoom, To, Team, Room, Result, MoveRequest, Left, Errorpacket +from socha.api.plugin.penguins.utils import handle_move, if_last_game_state, if_not_last_game_state +from socha.api.protocol.protocol import State, Error, Join, Joined, JoinPrepared, JoinRoom, Room, Result, MoveRequest, \ + Left, Errorpacket from socha.api.protocol.protocol_packet import ProtocolPacket -def _convert_board(protocol_board: Board) -> game_state.Board: - """ - Converts a protocol Board to a usable game board for using in the logic. - :param protocol_board: A Board object in protocol format - :type protocol_board: Board - :return: A Board object in the format used by the game logic - :rtype: penguins.Board - """ - if not isinstance(protocol_board, Board): - raise TypeError("The input must be a Board object in protocol format") - - board_list = [] - for y, row in enumerate(protocol_board.list_value): - board_list.append([]) - for x, fields_value in enumerate(row.field_value): - coordinate = CartesianCoordinate(x, y).to_hex() - if type(fields_value) is int: - board_list[y].append(Field(coordinate, penguin=None, fish=fields_value)) - elif fields_value == "ONE": - board_list[y].append(Field(coordinate, penguin=Penguin(coordinate, TeamEnum.ONE), fish=0)) - elif fields_value == "TWO": - board_list[y].append(Field(coordinate, penguin=Penguin(coordinate, TeamEnum.TWO), fish=0)) - else: - raise ValueError(f"Invalid field value {fields_value} at coordinates {coordinate}") - - return game_state.Board(board_list) - - class IClientHandler: history: List[List[Union[GameState, Error, Result]]] = [] @@ -183,13 +155,11 @@ def _on_move_request(self, room_id): start_time = time.time() move_response = self._game_handler.calculate_move() if move_response: - from_pos = None - to_pos = To(x=move_response.to_value.x, y=move_response.to_value.y) - if move_response.from_value: - from_pos = From(x=move_response.from_value.x, y=move_response.from_value.y) - response = Data(class_value="move", from_value=from_pos, to=to_pos) + response = handle_move(move_response) logging.info(f"Sent {move_response} after {round(time.time() - start_time, ndigits=3)} seconds.") self.send_message_to_room(room_id, response) + else: + logging.error(f"{move_response} is not a valid move.") def _on_state(self, message): last_game_state = None @@ -198,34 +168,9 @@ def _on_state(self, message): last_game_state = item break if last_game_state: - from_value = None if not message.data.class_binding.last_move.from_value else HexCoordinate( - x=message.data.class_binding.last_move.from_value.x, - y=message.data.class_binding.last_move.from_value.y) - to_value = HexCoordinate(x=message.data.class_binding.last_move.to.x, - y=message.data.class_binding.last_move.to.y) - last_move = Move(team_enum=last_game_state.current_team.name, - from_value=from_value, - to_value=to_value) - _game_state = last_game_state.perform_move(last_move) + _game_state = if_last_game_state(message, last_game_state) else: - first_team = Team(TeamEnum.ONE, - fish=0 if not last_game_state else last_game_state.first_team.fish, - penguins=[] if not last_game_state else last_game_state.first_team.penguins, - moves=[] if not last_game_state else last_game_state.first_team.moves) - second_team = Team(TeamEnum.TWO, - 0 if not last_game_state else last_game_state.second_team.fish, - penguins=[] if not last_game_state else last_game_state.second_team.penguins, - moves=[] if not last_game_state else last_game_state.second_team.moves) - first_team.opponent = second_team - second_team.opponent = first_team - - _game_state = GameState( - board=_convert_board(message.data.class_binding.board), - turn=message.data.class_binding.turn, - first_team=first_team, - second_team=second_team, - last_move=None, - ) + _game_state = if_not_last_game_state(message) self._game_handler.history[-1].append(_game_state) self._game_handler.on_update(_game_state) @@ -290,6 +235,7 @@ def _client_loop(self): elif isinstance(response, ProtocolPacket): logging.debug(f"Received new object: {response}") if isinstance(response, Left): + self._game_handler.on_game_left() self._handle_left() else: self._on_object(response) diff --git a/socha/api/networking/xml_protocol_interface.py b/socha/api/networking/xml_protocol_interface.py index 61324c9..ed549fd 100644 --- a/socha/api/networking/xml_protocol_interface.py +++ b/socha/api/networking/xml_protocol_interface.py @@ -147,10 +147,4 @@ def _serialize_object(self, object_class: object) -> bytes: :param object_class: The ProtocolPacket child to serialize. :return: The serialized byte stream. """ - if isinstance(object_class, Move): - from_value = From(x=object_class.from_value.x, y=object_class.from_value.y) - to_value = To(x=object_class.to_value.x, y=object_class.to_value.y) - data = Data(class_value="move", from_value=from_value, to=to_value) - return self.serializer.render(data).encode("utf-8") - return self.serializer.render(object_class).encode("utf-8") diff --git a/socha/api/plugin/penguins/utils.py b/socha/api/plugin/penguins/utils.py new file mode 100644 index 0000000..f4e2b47 --- /dev/null +++ b/socha/api/plugin/penguins/utils.py @@ -0,0 +1,75 @@ +from typing import Union + +from socha.api.plugin.penguins import game_state +from socha.api.plugin.penguins.board import Field, Move, CartesianCoordinate, TeamEnum, Penguin, HexCoordinate +from socha.api.plugin.penguins.game_state import GameState +from socha.api.protocol.protocol import Board, Data, \ + From, To, Team + + +def _convert_board(protocol_board: Board) -> game_state.Board: + """ + Converts a protocol Board to a usable game board for using in the logic. + :param protocol_board: A Board object in protocol format + :type protocol_board: Board + :return: A Board object in the format used by the game logic + :rtype: penguins.Board + """ + if not isinstance(protocol_board, Board): + raise TypeError("The input must be a Board object in protocol format") + + def create_field(y: int, x: int, fields_value: Union[int, str]) -> Field: + coordinate = CartesianCoordinate(x, y).to_hex() + if isinstance(fields_value, int): + return Field(coordinate, penguin=None, fish=fields_value) + elif fields_value == "ONE": + return Field(coordinate, penguin=Penguin(coordinate, TeamEnum.ONE), fish=0) + elif fields_value == "TWO": + return Field(coordinate, penguin=Penguin(coordinate, TeamEnum.TWO), fish=0) + else: + raise ValueError(f"Invalid field value {fields_value} at coordinates {coordinate}") + + board_list = [[create_field(y, x, fields_value) for x, fields_value in enumerate(row.field_value)] for y, row in + enumerate(protocol_board.list_value)] + return game_state.Board(board_list) + + +def handle_move(move_response) -> Data: + from_pos = None + to_pos = To(x=move_response.to_value.x, y=move_response.to_value.y) + if move_response.from_value: + from_pos = From(x=move_response.from_value.x, y=move_response.from_value.y) + return Data(class_value="move", from_value=from_pos, to=to_pos) + + +def if_last_game_state(message, last_game_state: GameState) -> GameState: + from_value = None if not message.data.class_binding.last_move.from_value else HexCoordinate( + x=message.data.class_binding.last_move.from_value.x, + y=message.data.class_binding.last_move.from_value.y) + to_value = HexCoordinate(x=message.data.class_binding.last_move.to.x, + y=message.data.class_binding.last_move.to.y) + last_move = Move(team_enum=last_game_state.current_team.name, + from_value=from_value, + to_value=to_value) + return last_game_state.perform_move(last_move) + + +def if_not_last_game_state(message) -> GameState: + first_team = Team(TeamEnum.ONE, + fish=0, + penguins=[], + moves=[]) + second_team = Team(TeamEnum.TWO, + 0, + penguins=[], + moves=[]) + first_team.opponent = second_team + second_team.opponent = first_team + + return GameState( + board=_convert_board(message.data.class_binding.board), + turn=message.data.class_binding.turn, + first_team=first_team, + second_team=second_team, + last_move=None, + )