path: root/diplomacy/communication/notifications.py

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