# ============================================================================== # 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 . # ============================================================================== """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)