Release v1.0.0 - Diplomacy Game Engine - AGPL v3+ License

1 files changed, 218 insertions, 0 deletions

diff --git a/diplomacy/communication/responses.py b/diplomacy/communication/responses.py
new file mode 100644
index 0000000..a928720
--- /dev/null
+++ b/diplomacy/communication/responses.py
@@ -0,0 +1,218 @@
+# ==============================================================================
+# Copyright (C) 2019 - Philip Paquette, Steven Bocco
+#
+#  This program is free software: you can redistribute it and/or modify it under
+#  the terms of the GNU Affero General Public License as published by the Free
+#  Software Foundation, either version 3 of the License, or (at your option) any
+#  later version.
+#
+#  This program is distributed in the hope that it will be useful, but WITHOUT
+#  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+#  FOR A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more
+#  details.
+#
+#  You should have received a copy of the GNU Affero General Public License along
+#  with this program.  If not, see <https://www.gnu.org/licenses/>.
+# ==============================================================================
+""" Server -> Client responses sent by server when it received a request. """
+import inspect
+
+from diplomacy.engine.game import Game
+from diplomacy.utils import common, parsing, strings
+from diplomacy.utils.scheduler_event import SchedulerEvent
+from diplomacy.utils.exceptions import ResponseException
+from diplomacy.utils.network_data import NetworkData
+from diplomacy.utils.game_phase_data import GamePhaseData
+
+class _AbstractResponse(NetworkData):
+    """ Base response object """
+    __slots__ = ['request_id']
+    header = {
+        strings.REQUEST_ID: str,
+        strings.NAME: str,
+    }
+    id_field = strings.REQUEST_ID
+
+    def __init__(self, **kwargs):
+        self.request_id = None
+        super(_AbstractResponse, self).__init__(**kwargs)
+
+class Error(_AbstractResponse):
+    """ Error response sent when an error occurred on server-side while handling a request. """
+    __slots__ = ['message']
+    params = {
+        strings.MESSAGE: str
+    }
+
+    def __init__(self, **kwargs):
+        self.message = None
+        super(Error, self).__init__(**kwargs)
+
+class Ok(_AbstractResponse):
+    """ Ok response sent by default after handling a request. """
+    __slots__ = []
+
+class DataGameSchedule(_AbstractResponse):
+    """ Response with info about current scheduling for a game. """
+    __slots__ = ['game_id', 'phase', 'schedule']
+    params = {
+        'game_id': str,
+        'phase': str,
+        'schedule': parsing.JsonableClassType(SchedulerEvent)
+    }
+
+    def __init__(self, **kwargs):
+        self.game_id = ''
+        self.phase = ''
+        self.schedule = None  # type: SchedulerEvent
+        super(DataGameSchedule, self).__init__(**kwargs)
+
+class DataGameInfo(_AbstractResponse):
+    """ Response containing information about a game, to be used when no entire game object is required. """
+    __slots__ = ['game_id', 'phase', 'timestamp', 'map_name', 'rules', 'status', 'n_players', 'n_controls',
+                 'deadline', 'registration_password', 'observer_level', 'controlled_powers']
+    params = {
+        strings.GAME_ID: str,
+        strings.PHASE: str,
+        strings.TIMESTAMP: int,
+        strings.MAP_NAME: parsing.OptionalValueType(str),
+        strings.OBSERVER_LEVEL: parsing.OptionalValueType(
+            parsing.EnumerationType((strings.MASTER_TYPE, strings.OMNISCIENT_TYPE, strings.OBSERVER_TYPE))),
+        strings.CONTROLLED_POWERS: parsing.OptionalValueType(parsing.SequenceType(str)),
+        strings.RULES: parsing.OptionalValueType(parsing.SequenceType(str)),
+        strings.STATUS: parsing.OptionalValueType(parsing.EnumerationType(strings.ALL_GAME_STATUSES)),
+        strings.N_PLAYERS: parsing.OptionalValueType(int),
+        strings.N_CONTROLS: parsing.OptionalValueType(int),
+        strings.DEADLINE: parsing.OptionalValueType(int),
+        strings.REGISTRATION_PASSWORD: parsing.OptionalValueType(bool)
+    }
+
+    def __init__(self, **kwargs):
+        self.game_id = None  # type: str
+        self.phase = None  # type: str
+        self.timestamp = None  # type: int
+        self.map_name = None  # type: str
+        self.observer_level = None  # type: str
+        self.controlled_powers = None  # type: list
+        self.rules = None  # type: list
+        self.status = None  # type: str
+        self.n_players = None  # type: int
+        self.n_controls = None  # type: int
+        self.deadline = None  # type: int
+        self.registration_password = None  # type: bool
+        super(DataGameInfo, self).__init__(**kwargs)
+
+class DataPossibleOrders(_AbstractResponse):
+    """ Response containing a dict of all possibles orders per location and a dict of all orderable locations per power
+        for a game phase.
+    """
+    __slots__ = ['possible_orders', 'orderable_locations']
+    params = {
+        # {location => [orders]}
+        strings.POSSIBLE_ORDERS: parsing.DictType(str, parsing.SequenceType(str)),
+        # {power name => [locations]}
+        strings.ORDERABLE_LOCATIONS: parsing.DictType(str, parsing.SequenceType(str)),
+    }
+    def __init__(self, **kwargs):
+        self.possible_orders = {}
+        self.orderable_locations = {}
+        super(DataPossibleOrders, self).__init__(**kwargs)
+
+class UniqueData(_AbstractResponse):
+    """ Response containing only 1 field named `data`.
+        `params` must have exactly one field named DATA.
+    """
+    __slots__ = ['data']
+
+    @classmethod
+    def validate_params(cls):
+        assert len(cls.params) == 1 and strings.DATA in cls.params
+
+    def __init__(self, **kwargs):
+        self.data = None
+        super(UniqueData, self).__init__(**kwargs)
+
+class DataToken(UniqueData):
+    """ Unique data containing a token. """
+    __slots__ = []
+    params = {
+        strings.DATA: str
+    }
+
+class DataMaps(UniqueData):
+    """ Unique data containing maps info. """
+    __slots__ = []
+    params = {
+        # {map_id => {'powers': [power names], 'supply centers' => [supply centers], 'loc_type' => {loc => type}}}
+        strings.DATA: dict
+    }
+
+class DataPowerNames(UniqueData):
+    """ Unique data containing a list of power names. """
+    __slots__ = []
+    params = {
+        strings.DATA: parsing.SequenceType(str)
+    }
+
+class DataGames(UniqueData):
+    """ Unique data containing a list of game info objects. """
+    __slots__ = []
+    params = {
+        strings.DATA: parsing.SequenceType(parsing.JsonableClassType(DataGameInfo))  # list of game info.
+    }
+
+class DataTimeStamp(UniqueData):
+    """ Unique data containing a timestamp. """
+    __slots__ = []
+    params = {
+        strings.DATA: int  # microseconds
+    }
+
+class DataGamePhases(UniqueData):
+    """ Unique data containing a list of GamePhaseData objects. """
+    __slots__ = []
+    params = {
+        strings.DATA: parsing.SequenceType(parsing.JsonableClassType(GamePhaseData))
+    }
+
+class DataGame(UniqueData):
+    """ Unique data containing a Game object. """
+    __slots__ = []
+    params = {
+        strings.DATA: parsing.JsonableClassType(Game)
+    }
+
+class DataSavedGame(UniqueData):
+    """ Unique data containing a game saved in JSON format. """
+    __slots__ = []
+    params = {
+        strings.DATA: dict
+    }
+
+class DataGamesToPowerNames(UniqueData):
+    """ Unique data containing a dict of game IDs associated to sequences of power names. """
+    __slots__ = []
+    params = {
+        strings.DATA: parsing.DictType(str, parsing.SequenceType(str))
+    }
+
+def parse_dict(json_response):
+    """ Parse a JSON dictionary expected to represent a response.
+        Raise an exception if either:
+        - parsing failed
+        - response received is an Error response. In such case, a ResponseException is raised
+          with the error message.
+        :param json_response: a JSON dict.
+        :return: a Response class instance.
+    """
+    assert isinstance(json_response, dict), 'Response parser expects a dict.'
+    name = json_response.get(strings.NAME, None)
+    if name is None:
+        raise ResponseException()
+    expected_class_name = common.snake_case_to_upper_camel_case(name)
+    response_class = globals()[expected_class_name]
+    assert inspect.isclass(response_class) and issubclass(response_class, _AbstractResponse)
+    response_object = response_class.from_dict(json_response)
+    if isinstance(response_object, Error):
+        raise ResponseException('%s: %s' % (response_object.name, response_object.message))
+    return response_object