1 files changed, 255 insertions, 0 deletions
diff --git a/diplomacy/communication/notifications.py b/diplomacy/communication/notifications.py
new file mode 100644
index 0000000..c88d526
--- /dev/null
+++ b/diplomacy/communication/notifications.py
@@ -0,0 +1,255 @@
+# ==============================================================================
+# 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)
+
+class AccountDeleted(_ChannelNotification):
+    """ Notification about an account deleted. """
+    __slots__ = []
+
+class OmniscientUpdated(_GameNotification):
+    """ Notification about a grade updated. Sent at channel level. """
+    __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)
+
+class ClearedCenters(_GameNotification):
+    """ Notification about centers cleared. """
+    __slots__ = []
+
+class ClearedOrders(_GameNotification):
+    """ Notification about orders cleared. """
+    __slots__ = []
+
+class ClearedUnits(_GameNotification):
+    """ Notification about units cleared. """
+    __slots__ = []
+
+class VoteCountUpdated(_GameNotification):
+    """ Notification about new count of draw votes for a game (for observers)
+        Properties:
+        - count_voted: number of powers that have voted.
+        - count_expected: 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)
+
+class VoteUpdated(_GameNotification):
+    """ Notification about votes updated for a game (for omniscient observers). Properties:
+        - vote: dictionary mapping a power name to a Vote object representing power vote.
+    """
+    __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)
+
+class PowerVoteUpdated(VoteCountUpdated):
+    """ Notification about a new vote for a specific game power (for player games). Properties:
+        - vote: vote object representing associated power vote.
+    """
+    __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)
+
+class PowersControllers(_GameNotification):
+    """ Notification about current controller for each power in a game. """
+    __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)
+
+class GameDeleted(_GameNotification):
+    """ Notification about a game deleted. """
+    __slots__ = []
+
+class GameProcessed(_GameNotification):
+    """ Notification about a game phase update. Sent after game had processed a 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)
+
+class GamePhaseUpdate(_GameNotification):
+    """ Notification about a game phase update. """
+    __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)
+
+class GameStatusUpdate(_GameNotification):
+    """ Notification about a game status update. """
+    __slots__ = ['status']
+    params = {
+        strings.STATUS: parsing.EnumerationType(strings.ALL_GAME_STATUSES),
+    }
+
+    def __init__(self, **kwargs):
+        self.status = None
+        super(GameStatusUpdate, self).__init__(**kwargs)
+
+class GameMessageReceived(_GameNotification):
+    """ Notification about a game message received. """
+    __slots__ = ['message']
+    params = {
+        strings.MESSAGE: parsing.JsonableClassType(Message),
+    }
+
+    def __init__(self, **kwargs):
+        self.message = None  # type: Message
+        super(GameMessageReceived, self).__init__(**kwargs)
+
+class PowerOrdersUpdate(_GameNotification):
+    """ Notification about a power order update. """
+    __slots__ = ['orders']
+    params = {
+        strings.ORDERS: parsing.OptionalValueType(parsing.SequenceType(str)),
+    }
+
+    def __init__(self, **kwargs):
+        self.orders = None  # type: set
+        super(PowerOrdersUpdate, self).__init__(**kwargs)
+
+class PowerOrdersFlag(_GameNotification):
+    """ Notification about a power order flag update. """
+    __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)
+
+class PowerWaitFlag(_GameNotification):
+    """ Notification about a power wait flag update. """
+    __slots__ = ['wait']
+    params = {
+        strings.WAIT: bool,
+    }
+
+    def __init__(self, **kwargs):
+        self.wait = None  # type: bool
+        super(PowerWaitFlag, self).__init__(**kwargs)
+
+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)