diff options
Diffstat (limited to 'diplomacy/daide/notifications.py')
| -rw-r--r-- | diplomacy/daide/notifications.py | 449 |
1 files changed, 449 insertions, 0 deletions
diff --git a/diplomacy/daide/notifications.py b/diplomacy/daide/notifications.py new file mode 100644 index 0000000..e9f6366 --- /dev/null +++ b/diplomacy/daide/notifications.py @@ -0,0 +1,449 @@ +# ============================================================================== +# Copyright (C) 2019 - Philip Paquette +# +# 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/>. +# ============================================================================== +""" DAIDE Notifications - Contains a list of responses sent by the server to the client """ +from diplomacy import Map +from diplomacy.daide.clauses import String, Power, Province, Turn, Unit, add_parentheses, strip_parentheses, \ + parse_string +from diplomacy.daide import tokens +from diplomacy.daide.tokens import Token +from diplomacy.daide.utils import bytes_to_str, str_to_bytes + +class DaideNotification(): + """ Represents a DAIDE response. """ + def __init__(self, **kwargs): + """ Constructor """ + del kwargs # Unused kwargs + self._bytes = b'' + self._str = '' + + def __bytes__(self): + """ Returning the bytes representation of the response """ + return self._bytes + + def __str__(self): + """ Returning the string representation of the response """ + return bytes_to_str(self._bytes) + + def to_bytes(self): + """ Returning the bytes representation of the response """ + return bytes(self) + + def to_string(self): + """ Returning the string representation of the response """ + return str(self) + +class MapNameNotification(DaideNotification): + """ Represents a MAP DAIDE response. Sends the name of the current map to the client. + Syntax: + MAP ('name') + """ + def __init__(self, map_name, **kwargs): + """ Builds the response + :param map_name: String. The name of the current map. + """ + super(MapNameNotification, self).__init__(**kwargs) + self._bytes = bytes(tokens.MAP) \ + + bytes(parse_string(String, map_name)) + +class HelloNotification(DaideNotification): + """ Represents a HLO DAIDE response. Sends the power to be played by the client with the passcode to rejoin the + game and the details of the game. + Syntax: + HLO (power) (passcode) (variant) (variant) ... + Variant syntax: + LVL n # Level of the syntax accepted + MTL seconds # Movement time limit + RTL seconds # Retreat time limit + BTL seconds # Build time limit + DSD # Disables the time limit when a client disconects + AOA # Any orders accepted + LVL 10: + Variant syntax: + PDA # Accept partial draws + NPR # No press during retreat phases + NPB # No press during build phases + PTL seconds # Press time limit + """ + def __init__(self, power_name, passcode, level, deadline, rules, **kwargs): + """ Builds the response + :param power_name: The name of the power being played. + :param passcode: Integer. A passcode to rejoin the game. + :param level: Integer. The daide syntax level of the game + :param deadline: Integer. The number of seconds per turn (0 to disable) + :param rules: The list of game rules. + """ + super(HelloNotification, self).__init__(**kwargs) + power = parse_string(Power, power_name) + passcode = Token(from_int=passcode) + + if 'NO_PRESS' in rules: + level = 0 + variants = add_parentheses(bytes(tokens.LVL) + bytes(Token(from_int=level))) + + if deadline > 0: + variants += add_parentheses(bytes(tokens.MTL) + bytes(Token(from_int=deadline))) + variants += add_parentheses(bytes(tokens.RTL) + bytes(Token(from_int=deadline))) + variants += add_parentheses(bytes(tokens.BTL) + bytes(Token(from_int=deadline))) + + if 'NO_CHECK' in rules: + variants += add_parentheses(bytes(tokens.AOA)) + + self._bytes = bytes(tokens.HLO) \ + + add_parentheses(bytes(power)) \ + + add_parentheses(bytes(passcode)) \ + + add_parentheses(bytes(variants)) + +class SupplyCenterNotification(DaideNotification): + """ Represents a SCO DAIDE notification. Sends the current supply centre ownership. + Syntax: + SCO (power centre centre ...) (power centre centre ...) ... + """ + def __init__(self, powers_centers, map_name, **kwargs): + """ Builds the notification + :param powers_centers: A dict of {power_name: centers} objects + :param map_name: The name of the map + """ + super(SupplyCenterNotification, self).__init__(**kwargs) + remaining_scs = Map(map_name).scs[:] + all_powers_bytes = [] + + # Parsing each power + for power_name in sorted(powers_centers): + centers = sorted(powers_centers[power_name]) + power_clause = parse_string(Power, power_name) + power_bytes = bytes(power_clause) + + for center in centers: + sc_clause = parse_string(Province, center) + power_bytes += bytes(sc_clause) + remaining_scs.remove(center) + + all_powers_bytes += [power_bytes] + + # Parsing unowned center + uno_token = tokens.UNO + power_bytes = bytes(uno_token) + + for center in remaining_scs: + sc_clause = parse_string(Province, center) + power_bytes += bytes(sc_clause) + + all_powers_bytes += [power_bytes] + + # Storing full response + self._bytes = bytes(tokens.SCO) \ + + b''.join([add_parentheses(power_bytes) for power_bytes in all_powers_bytes]) + +class CurrentPositionNotification(DaideNotification): + """ Represents a NOW DAIDE notification. Sends the current turn, and the current unit positions. + Syntax: + NOW (turn) (unit) (unit) ... + Unit syntax: + power unit_type province + power unit_type province MRT (province province ...) + """ + def __init__(self, phase_name, powers_units, powers_retreats, **kwargs): + """ Builds the notification + :param phase_name: The name of the current phase (e.g. 'S1901M') + :param powers: A list of `diplomacy.engine.power.Power` objects + """ + super(CurrentPositionNotification, self).__init__(**kwargs) + units_bytes_buffer = [] + + # Turn + turn_clause = parse_string(Turn, phase_name) + + # Units + for power_name, units in sorted(powers_units.items()): + # Regular units + for unit in units: + unit_clause = parse_string(Unit, '%s %s' % (power_name, unit)) + units_bytes_buffer += [bytes(unit_clause)] + + # Dislodged units + for unit, retreat_provinces in sorted(powers_retreats[power_name].items()): + unit_clause = parse_string(Unit, '%s %s' % (power_name, unit)) + retreat_clauses = [parse_string(Province, province) for province in retreat_provinces] + units_bytes_buffer += [add_parentheses(strip_parentheses(bytes(unit_clause)) + + bytes(tokens.MRT) + + add_parentheses(b''.join([bytes(province) + for province in retreat_clauses])))] + + # Storing full response + self._bytes = bytes(tokens.NOW) + bytes(turn_clause) + b''.join(units_bytes_buffer) + +class MissingOrdersNotification(DaideNotification): + """ Represents a MIS DAIDE response. Sends the list of unit for which an order is missing or indication about + required disbands or builds. + Syntax: + MIS (unit) (unit) ... + MIS (unit MRT (province province ...)) (unit MRT (province province ...)) ... + MIS (number) + """ + def __init__(self, phase_name, power, **kwargs): + """ Builds the response + :param phase_name: The name of the current phase (e.g. 'S1901M') + :param power: The power to check for missing orders + :type power: diplomacy.engine.power.Power + """ + super(MissingOrdersNotification, self).__init__(**kwargs) + assert phase_name[-1] in 'MRA', 'Invalid phase "%s"' & phase_name + {'M': self._build_movement_phase, + 'R': self._build_retreat_phase, + 'A': self._build_adjustment_phase}[phase_name[-1]](power) + + def _build_movement_phase(self, power): + """ Builds the missing orders response for a movement phase """ + units_with_no_order = [unit for unit in power.units] + + # Removing units for which we have orders + for key, value in power.orders.items(): + unit = key # Regular game {e.g. 'A PAR': '- BUR') + if key[0] in 'RIO': # No-check game (key is INVALID, ORDER x, REORDER x) + unit = ' '.join(value.split()[:2]) + if unit in units_with_no_order: + units_with_no_order.remove(unit) + + # Storing full response + self._bytes = bytes(tokens.MIS) + \ + b''.join([bytes(parse_string(Unit, '%s %s' % (power.name, unit))) + for unit in units_with_no_order]) + + def _build_retreat_phase(self, power): + """ Builds the missing orders response for a retreat phase """ + units_bytes_buffer = [] + + units_with_no_order = {unit: retreat_provinces for unit, retreat_provinces in power.retreats.items()} + + # Removing units for which we have orders + for key, value in power.orders.items(): + unit = key # Regular game {e.g. 'A PAR': '- BUR') + if key[0] in 'RIO': # No-check game (key is INVALID, ORDER x, REORDER x) + unit = ' '.join(value.split()[:2]) + if unit in units_with_no_order: + del units_with_no_order[unit] + + for unit, retreat_provinces in sorted(units_with_no_order.items(), + key=lambda key_val: key_val[0].split()[-1]): + unit_clause = parse_string(Unit, '%s %s' % (power.name, unit)) + retreat_clauses = [parse_string(Province, province) + for province in retreat_provinces] + units_bytes_buffer += [add_parentheses(strip_parentheses(bytes(unit_clause)) + + bytes(tokens.MRT) + + add_parentheses(b''.join([bytes(province) + for province in retreat_clauses])))] + + self._bytes = bytes(tokens.MIS) + b''.join(units_bytes_buffer) + + def _build_adjustment_phase(self, power): + """ Builds the missing orders response for a build phase """ + disbands_status = len(power.units) - len(power.centers) + + if disbands_status < 0: + available_homes = power.homes[:] + + # Removing centers for which it's impossible to build + for unit in [unit.split() for unit in power.units]: + province = unit[1] + if province in available_homes: + available_homes.remove(province) + + disbands_status = max(-len(available_homes), disbands_status) + + self._bytes += bytes(tokens.MIS) + add_parentheses(bytes(Token(from_int=disbands_status))) + +class OrderResultNotification(DaideNotification): + """ Represents a ORD DAIDE response. Sends the result of an order after the turn has been processed. + Syntax: + ORD (turn) (order) (result) + ORD (turn) (order) (result RET) + Result syntax: + SUC # Order succeeded (can apply to any order). + BNC # Move bounced (only for MTO, CTO or RTO orders). + CUT # Support cut (only for SUP orders). + DSR # Move via convoy failed due to dislodged convoying fleet (only for CTO orders). + NSO # No such order (only for SUP, CVY or CTO orders). + RET # Unit was dislodged and must retreat. + """ + def __init__(self, phase_name, order_bytes, results, **kwargs): + """ Builds the response + :param phase_name: The name of the current phase (e.g. 'S1901M') + :param order_bytes: The bytes received for the order + :param results: An array containing the error codes. + """ + super(OrderResultNotification, self).__init__(**kwargs) + turn_clause = parse_string(Turn, phase_name) + if not results or 0 in results: # Order success response + result_clause = tokens.SUC + else: # Generic order failure response + result_clause = tokens.NSO + + self._bytes = bytes(tokens.ORD) \ + + bytes(turn_clause) \ + + add_parentheses(order_bytes) \ + + add_parentheses(bytes(result_clause)) + +class TimeToDeadlineNotification(DaideNotification): + """ Represents a TME DAIDE response. Sends the time to the next deadline. + Syntax: + TME (seconds) + """ + def __init__(self, seconds, **kwargs): + """ Builds the response + :param seconds: Integer. The number of seconds before deadline + """ + super(TimeToDeadlineNotification, self).__init__(**kwargs) + self._bytes = bytes(tokens.TME) + add_parentheses(bytes(tokens.Token(from_int=seconds))) + +class PowerInCivilDisorderNotification(DaideNotification): + """ Represents a CCD DAIDE response. Sends the name of the power in civil disorder. + Syntax: + CCD (power) + """ + def __init__(self, power_name, **kwargs): + """ Builds the response + :param power_name: The name of the power being played. + """ + super(PowerInCivilDisorderNotification, self).__init__(**kwargs) + power = parse_string(Power, power_name) + self._bytes = bytes(tokens.CCD) + add_parentheses(bytes(power)) + +class PowerIsEliminatedNotification(DaideNotification): + """ Represents a OUT DAIDE response. Sends the name of the power eliminated. + Syntax: + OUT (power) + """ + def __init__(self, power_name, **kwargs): + """ Builds the response + :param power_name: The name of the power being played. + """ + super(PowerIsEliminatedNotification, self).__init__(**kwargs) + power = parse_string(Power, power_name) + self._bytes = bytes(tokens.OUT) + add_parentheses(bytes(power)) + +class DrawNotification(DaideNotification): + """ Represents a DRW DAIDE response. Indicates that the game has ended due to a draw + Syntax: + DRW + """ + def __init__(self, **kwargs): + """ Builds the response + """ + super(DrawNotification, self).__init__(**kwargs) + self._bytes = bytes(tokens.DRW) + +class MessageFromNotification(DaideNotification): + """ Represents a FRM DAIDE response. Indicates that the game has ended due to a draw + Syntax: + FRM (power) (power power ...) (press_message) + FRM (power) (power power ...) (reply) + """ + def __init__(self, from_power_name, to_power_names, message, **kwargs): + """ Builds the response + """ + super(MessageFromNotification, self).__init__(**kwargs) + + from_power_clause = bytes(parse_string(Power, from_power_name)) + to_powers_clause = b''.join([bytes(parse_string(Power, power_name)) for power_name in to_power_names]) + message_clause = str_to_bytes(message) + + self._bytes = bytes(tokens.FRM) \ + + b''.join([add_parentheses(clause) + for clause in [from_power_clause, to_powers_clause, message_clause]]) + +class SoloNotification(DaideNotification): + """ Represents a SLO DAIDE response. Indicates that the game has ended due to a solo by the specified power + Syntax: + SLO (power) + """ + def __init__(self, power_name, **kwargs): + """ Builds the response + :param power_name: The name of the power being solo. + """ + super(SoloNotification, self).__init__(**kwargs) + power = parse_string(Power, power_name) + self._bytes = bytes(tokens.SLO) + add_parentheses(bytes(power)) + +class SummaryNotification(DaideNotification): + """ Represents a SMR DAIDE response. Sends the summary for each power at the end of the game + Syntax: + SMR (turn) (power_summary) ... + power_summary syntax: + power ('name') ('version') number_of_centres + power ('name') ('version') number_of_centres year_of_elimination + """ + def __init__(self, phase_name, powers, daide_users, years_of_elimination, **kwargs): + """ Builds the Notification """ + super(SummaryNotification, self).__init__(**kwargs) + powers_smrs_clause = [] + + # Turn + turn_clause = parse_string(Turn, phase_name) + + for power, daide_user, year_of_elimination in zip(powers, daide_users, years_of_elimination): + power_smr_clause = [] + + name = daide_user.client_name if daide_user else power.get_controller() + version = daide_user.client_version if daide_user else 'v0.0.0' + + power_name_clause = bytes(parse_string(Power, power.name)) + power_smr_clause.append(power_name_clause) + + # (name) + name_clause = bytes(parse_string(String, name)) + power_smr_clause.append(name_clause) + + # (version) + version_clause = bytes(parse_string(String, version)) + power_smr_clause.append(version_clause) + + number_of_centres_clause = bytes(Token(from_int=len(power.centers))) + power_smr_clause.append(number_of_centres_clause) + + if not power.centers: + year_of_elimination_clause = bytes(Token(from_int=year_of_elimination)) + power_smr_clause.append(year_of_elimination_clause) + + power_smr_clause = add_parentheses(b''.join(power_smr_clause)) + powers_smrs_clause.append(power_smr_clause) + + self._bytes = bytes(tokens.SMR) + bytes(turn_clause) + b''.join(powers_smrs_clause) + +class TurnOffNotification(DaideNotification): + """ Represents an OFF DAIDE response. Requests a client to exit + Syntax: + OFF + """ + def __init__(self, **kwargs): + """ Builds the response """ + super(TurnOffNotification, self).__init__(**kwargs) + self._bytes = bytes(tokens.OFF) + +MAP = MapNameNotification +HLO = HelloNotification +SCO = SupplyCenterNotification +NOW = CurrentPositionNotification +MIS = MissingOrdersNotification +ORD = OrderResultNotification +TME = TimeToDeadlineNotification +CCD = PowerInCivilDisorderNotification +OUT = PowerIsEliminatedNotification +DRW = DrawNotification +FRM = MessageFromNotification +SLO = SoloNotification +SMR = SummaryNotification +OFF = TurnOffNotification |