# ==============================================================================
# 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 notifications."""
import inspect
from diplomacy.engine.game import Game
from diplomacy.engine.message import Message
from diplomacy.utils import common, exceptions, parsing, strings
from diplomacy.utils.network_data import NetworkData
from diplomacy.utils.constants import OrderSettings
from diplomacy.utils.game_phase_data import GamePhaseData
class _AbstractNotification(NetworkData):
""" Base notification object """
__slots__ = ['notification_id', 'token']
header = {
strings.NOTIFICATION_ID: str,
strings.NAME: str,
strings.TOKEN: str,
}
params = {}
id_field = strings.NOTIFICATION_ID
level = None
def __init__(self, **kwargs):
self.notification_id = None # type: str
self.token = None # type: str
super(_AbstractNotification, self).__init__(**kwargs)
@classmethod
def validate_params(cls):
""" Hack: we just use it to validate level. """
assert cls.level in strings.ALL_COMM_LEVELS
class _ChannelNotification(_AbstractNotification):
""" Channel notification (intended to be sent to a channel). """
__slots__ = []
level = strings.CHANNEL
class _GameNotification(_AbstractNotification):
""" Game notification (intended to be sent to a game). """
__slots__ = ['game_id', 'game_role', 'power_name']
header = parsing.update_model(_AbstractNotification.header, {
strings.GAME_ID: str,
strings.GAME_ROLE: str,
strings.POWER_NAME: parsing.OptionalValueType(str),
})
level = strings.GAME
def __init__(self, **kwargs):
self.game_id = None # type: str
self.game_role = None # type: str
self.power_name = None # type: str
super(_GameNotification, self).__init__(**kwargs)
[docs]class AccountDeleted(_ChannelNotification):
""" Notification about an account deleted. """
__slots__ = []
[docs]class OmniscientUpdated(_GameNotification):
""" Notification about a grade updated. Sent at channel level.
Properties:
- **grade_update**: :class:`str` One of 'promote' or 'demote'.
- **game**: :class:`parsing.JsonableClassType(Game)` a :class:`diplomacy.engine.game.Game` object.
"""
__slots__ = ['grade_update', 'game']
params = {
strings.GRADE_UPDATE: parsing.EnumerationType(strings.ALL_GRADE_UPDATES),
strings.GAME: parsing.JsonableClassType(Game)
}
def __init__(self, **kwargs):
self.grade_update = ''
self.game = None # type: Game
super(OmniscientUpdated, self).__init__(**kwargs)
[docs]class ClearedCenters(_GameNotification):
""" Notification about centers cleared. """
__slots__ = []
[docs]class ClearedOrders(_GameNotification):
""" Notification about orders cleared. """
__slots__ = []
[docs]class ClearedUnits(_GameNotification):
""" Notification about units cleared. """
__slots__ = []
[docs]class VoteCountUpdated(_GameNotification):
""" Notification about new count of draw votes for a game (for observers).
Properties:
- **count_voted**: :class:`int` number of powers that have voted.
- **count_expected**: :class:`int` number of powers to be expected to vote.
"""
__slots__ = ['count_voted', 'count_expected']
params = {
strings.COUNT_VOTED: int,
strings.COUNT_EXPECTED: int,
}
def __init__(self, **kwargs):
self.count_voted = None # type: int
self.count_expected = None # type: int
super(VoteCountUpdated, self).__init__(**kwargs)
[docs]class VoteUpdated(_GameNotification):
""" Notification about votes updated for a game (for omniscient observers).
Properties:
- **vote**: :class:`Dict` mapping a power name to a Vote (:class:`str`) object representing power vote.
Possible votes are: yes, no, neutral.
"""
__slots__ = ['vote']
params = {
strings.VOTE: parsing.DictType(str, parsing.EnumerationType(strings.ALL_VOTE_DECISIONS))
}
def __init__(self, **kwargs):
self.vote = None # type: dict{str, str}
super(VoteUpdated, self).__init__(**kwargs)
[docs]class PowerVoteUpdated(VoteCountUpdated):
""" Notification about a new vote for a specific game power (for player games).
Properties:
- **vote**: :class:`str` vote object representing associated power vote. Can be yes, no, neutral.
"""
__slots__ = ['vote']
params = parsing.extend_model(VoteCountUpdated.params, {
strings.VOTE: parsing.EnumerationType(strings.ALL_VOTE_DECISIONS)
})
def __init__(self, **kwargs):
self.vote = None # type: str
super(PowerVoteUpdated, self).__init__(**kwargs)
[docs]class PowersControllers(_GameNotification):
""" Notification about current controller for each power in a game.
Properties:
- **powers**: A :class:`Dict` that maps a power_name to a controller_name :class:`str`.
- **timestamps**: A :class:`Dict` that maps a power_name to timestamp where the controller took over.
"""
__slots__ = ['powers', 'timestamps']
params = {
# {power_name => controller_name}
strings.POWERS: parsing.DictType(str, parsing.OptionalValueType(str)),
# {power_name => controller timestamp}
strings.TIMESTAMPS: parsing.DictType(str, int)
}
def __init__(self, **kwargs):
self.powers = {}
self.timestamps = {}
super(PowersControllers, self).__init__(**kwargs)
[docs]class GameDeleted(_GameNotification):
""" Notification about a game deleted. """
__slots__ = []
[docs]class GameProcessed(_GameNotification):
""" Notification about a game phase update. Sent after game has processed a phase.
Properties:
- **previous_phase_data**: :class:`diplomacy.utils.game_phase_data.GamePhaseData` of the previous phase
- **current_phase_data**: :class:`diplomacy.utils.game_phase_data.GamePhaseData` of the current phase
"""
__slots__ = ['previous_phase_data', 'current_phase_data']
params = {
strings.PREVIOUS_PHASE_DATA: parsing.JsonableClassType(GamePhaseData),
strings.CURRENT_PHASE_DATA: parsing.JsonableClassType(GamePhaseData),
}
def __init__(self, **kwargs):
self.previous_phase_data = None # type: GamePhaseData
self.current_phase_data = None # type: GamePhaseData
super(GameProcessed, self).__init__(**kwargs)
[docs]class GamePhaseUpdate(_GameNotification):
""" Notification about a game phase update.
Properties:
- **phase_data**: :class:`diplomacy.utils.game_phase_data.GamePhaseData` of the updated phase
- **phase_data_type**: :class:`str`. One of 'state_history', 'state', 'phase'
"""
__slots__ = ['phase_data', 'phase_data_type']
params = {
strings.PHASE_DATA: parsing.JsonableClassType(GamePhaseData),
strings.PHASE_DATA_TYPE: strings.ALL_STATE_TYPES
}
def __init__(self, **kwargs):
self.phase_data = None # type: GamePhaseData
self.phase_data_type = None # type: str
super(GamePhaseUpdate, self).__init__(**kwargs)
[docs]class GameStatusUpdate(_GameNotification):
""" Notification about a game status update.
Properties:
-**status**: :class:`str`. One of 'forming', 'active', 'paused', 'completed', 'canceled'
"""
__slots__ = ['status']
params = {
strings.STATUS: parsing.EnumerationType(strings.ALL_GAME_STATUSES),
}
def __init__(self, **kwargs):
self.status = None
super(GameStatusUpdate, self).__init__(**kwargs)
[docs]class GameMessageReceived(_GameNotification):
""" Notification about a game message received.
Properties:
- **message**: :class:`diplomacy.engine.message.Message` received.
"""
__slots__ = ['message']
params = {
strings.MESSAGE: parsing.JsonableClassType(Message),
}
def __init__(self, **kwargs):
self.message = None # type: Message
super(GameMessageReceived, self).__init__(**kwargs)
[docs]class PowerOrdersUpdate(_GameNotification):
""" Notification about a power order update.
Properties:
- **orders**: List of updated orders (i.e. :class:`str`)
"""
__slots__ = ['orders']
params = {
strings.ORDERS: parsing.OptionalValueType(parsing.SequenceType(str)),
}
def __init__(self, **kwargs):
self.orders = None # type: set
super(PowerOrdersUpdate, self).__init__(**kwargs)
[docs]class PowerOrdersFlag(_GameNotification):
""" Notification about a power order flag update.
Properties:
- **order_is_set**: :class:`int`. O = ORDER_NOT_SET, 1 = ORDER_SET_EMPTY, 2 = ORDER_SET.
"""
__slots__ = ['order_is_set']
params = {
strings.ORDER_IS_SET: parsing.EnumerationType(OrderSettings.ALL_SETTINGS),
}
def __init__(self, **kwargs):
self.order_is_set = 0
super(PowerOrdersFlag, self).__init__(**kwargs)
[docs]class PowerWaitFlag(_GameNotification):
""" Notification about a power wait flag update.
Properties:
- **wait**: :class:`bool` that indicates to wait until the deadline is reached before proceeding. Otherwise
if all powers are not waiting, the game is processed as soon as all non-eliminated powers have submitted
their orders.
"""
__slots__ = ['wait']
params = {
strings.WAIT: bool,
}
def __init__(self, **kwargs):
self.wait = None # type: bool
super(PowerWaitFlag, self).__init__(**kwargs)
[docs]def parse_dict(json_notification):
""" Parse a JSON expected to represent a notification. Raise an exception if parsing failed.
:param json_notification: JSON dictionary.
:return: a notification class instance.
"""
assert isinstance(json_notification, dict), 'Notification parser expects a dict.'
name = json_notification.get(strings.NAME, None)
if name is None:
raise exceptions.NotificationException()
expected_class_name = common.snake_case_to_upper_camel_case(name)
notification_class = globals()[expected_class_name]
assert inspect.isclass(notification_class) and issubclass(notification_class, _AbstractNotification)
return notification_class.from_dict(json_notification)