# ==============================================================================
# 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 as replies to requests. """
import inspect
from diplomacy.engine.game import Game
from diplomacy.utils import common, parsing, strings
from diplomacy.utils import exceptions
from diplomacy.utils.game_phase_data import GamePhaseData
from diplomacy.utils.network_data import NetworkData
from diplomacy.utils.scheduler_event import SchedulerEvent
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)
[docs]class Error(_AbstractResponse):
""" Error response sent when an error occurred on server-side while handling a request.
Properties:
- **error_type**: str - error type, containing the exception class name.
- **message**: str - error message
"""
__slots__ = ['message', 'error_type']
params = {
strings.MESSAGE: str,
strings.ERROR_TYPE: str
}
def __init__(self, **kwargs):
self.message = None
self.error_type = None
super(Error, self).__init__(**kwargs)
[docs] def throw(self):
""" Convert this error to an instance of a Diplomacy ResponseException class and raises it. """
# If error type is the name of a ResponseException class,
# convert it to related class and raise it.
if hasattr(exceptions, self.error_type):
symbol = getattr(exceptions, self.error_type)
if inspect.isclass(symbol) and issubclass(symbol, exceptions.ResponseException):
raise symbol(self.message)
# Otherwise, raise a generic ResponseException object.
raise exceptions.ResponseException('%s/%s' % (self.error_type, self.message))
[docs]class Ok(_AbstractResponse):
""" Ok response sent by default after handling a request. Contains nothing. """
__slots__ = []
[docs]class NoResponse(_AbstractResponse):
""" Placeholder response to indicate that no responses are required """
__slots__ = []
def __bool__(self):
""" This response always evaluate to false """
return False
[docs]class DataGameSchedule(_AbstractResponse):
""" Response with info about current scheduling for a game.
Properties:
- **game_id**: str - game ID
- **phase**: str - game phase
- **schedule**: :class:`.SchedulerEvent` - scheduling information about the 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)
[docs]class DataGameInfo(_AbstractResponse):
""" Response containing information about a game, to be used when no entire game object is required.
Properties:
- **game_id**: game ID
- **phase**: game phase
- **timestamp**: latest timestamp when data was saved into game on server
(ie. game state or message)
- **timestamp_created**: timestamp when game was created on server
- **map_name**: (optional) game map name
- **observer_level**: (optional) highest observer level allowed for the user who sends
the request. Either ``'observer_type'``, ``'omniscient_type'`` or ``'master_type'``.
- **controlled_powers**: (optional) list of power names controlled by the user who sends
the request.
- **rules**: (optional) game rules
- **status**: (optional) game status
- **n_players**: (optional) number of powers currently controlled in the game
- **n_controls**: (optional) number of controlled powers required by the game to be active
- **deadline**: (optional) game deadline - time to wait before processing a game phase
- **registration_password**: (optional) boolean - if True, a password is required to join the game
"""
__slots__ = ['game_id', 'phase', 'timestamp', 'map_name', 'rules', 'status', 'n_players',
'n_controls', 'deadline', 'registration_password', 'observer_level',
'controlled_powers', 'timestamp_created']
params = {
strings.GAME_ID: str,
strings.PHASE: str,
strings.TIMESTAMP: int,
strings.TIMESTAMP_CREATED: 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.timestamp_created = 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)
[docs]class DataPossibleOrders(_AbstractResponse):
""" Response containing information about possible orders for a game at its current phase.
Properties:
- **possible_orders**: dictionary mapping a location short name to all possible orders here
- **orderable_locations**: dictionary mapping a power name to its orderable locations
"""
__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)
[docs]class UniqueData(_AbstractResponse):
""" Response containing only 1 field named ``data``. A derived class will contain
a specific typed value in this field.
"""
# `params` must have exactly one field named DATA.
__slots__ = ['data']
[docs] @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)
[docs]class DataToken(UniqueData):
""" Unique data containing a token. """
__slots__ = []
params = {
strings.DATA: str
}
[docs]class DataMaps(UniqueData):
""" Unique data containing maps info
(dictionary mapping a map name to a dictionary with map information). """
__slots__ = []
params = {
# {map_id => {'powers': [power names], 'supply centers' => [supply centers], 'loc_type' => {loc => type}}}
strings.DATA: dict
}
[docs]class DataPowerNames(UniqueData):
""" Unique data containing a list of power names. """
__slots__ = []
params = {
strings.DATA: parsing.SequenceType(str)
}
[docs]class DataGames(UniqueData):
""" Unique data containing a list of :class:`.DataGameInfo` objects. """
__slots__ = []
params = {
# list of game info.
strings.DATA: parsing.SequenceType(parsing.JsonableClassType(DataGameInfo))
}
[docs]class DataPort(UniqueData):
""" Unique data containing a DAIDE port (integer). """
__slots__ = []
params = {
strings.DATA: int # DAIDE port
}
[docs]class DataTimeStamp(UniqueData):
""" Unique data containing a timestamp (integer). """
__slots__ = []
params = {
strings.DATA: int # microseconds
}
[docs]class DataGamePhases(UniqueData):
""" Unique data containing a list of :class:`.GamePhaseData` objects. """
__slots__ = []
params = {
strings.DATA: parsing.SequenceType(parsing.JsonableClassType(GamePhaseData))
}
[docs]class DataGame(UniqueData):
""" Unique data containing a :class:`.Game` object. """
__slots__ = []
params = {
strings.DATA: parsing.JsonableClassType(Game)
}
[docs]class DataSavedGame(UniqueData):
""" Unique data containing a game saved in JSON dictionary. """
__slots__ = []
params = {
strings.DATA: dict
}
[docs]class DataGamesToPowerNames(UniqueData):
""" Unique data containing a dictionary mapping a game ID to a list of power names. """
__slots__ = []
params = {
strings.DATA: parsing.DictType(str, parsing.SequenceType(str))
}
[docs]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 exceptions.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):
response_object.throw()
return response_object