diff options
Diffstat (limited to 'diplomacy/integration/webdiplomacy_net')
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/__init__.py | 17 | ||||
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/api.py | 228 | ||||
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/game.py | 333 | ||||
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/orders.py | 593 | ||||
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/tests/__init__.py | 16 | ||||
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/tests/test_game.py | 185 | ||||
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/tests/test_orders.py | 385 | ||||
| -rw-r--r-- | diplomacy/integration/webdiplomacy_net/utils.py | 75 |
8 files changed, 1832 insertions, 0 deletions
diff --git a/diplomacy/integration/webdiplomacy_net/__init__.py b/diplomacy/integration/webdiplomacy_net/__init__.py new file mode 100644 index 0000000..6ee73ed --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/__init__.py @@ -0,0 +1,17 @@ +# ============================================================================== +# 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/>. +# ============================================================================== +from diplomacy.integration.webdiplomacy_net.api import API diff --git a/diplomacy/integration/webdiplomacy_net/api.py b/diplomacy/integration/webdiplomacy_net/api.py new file mode 100644 index 0000000..a5fa94a --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/api.py @@ -0,0 +1,228 @@ +# ============================================================================== +# 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/>. +# ============================================================================== +""" Contains an API class to send requests to webdiplomacy.net """ +import logging +import os +from urllib.parse import urlencode +import ujson as json +from tornado import gen +from tornado.httpclient import AsyncHTTPClient, HTTPRequest +from diplomacy.integration.webdiplomacy_net.game import state_dict_to_game_and_power +from diplomacy.integration.webdiplomacy_net.orders import Order +from diplomacy.integration.webdiplomacy_net.utils import CACHE, GameIdCountryId + +# Constants +LOGGER = logging.getLogger(__name__) +API_USER_AGENT = 'KestasBot / Philip Paquette v1.0' +API_WEBDIPLOMACY_NET = os.environ.get('API_WEBDIPLOMACY', 'https://webdiplomacy.net/api.php') + +class API(): + """ API to interact with webdiplomacy.net """ + + def __init__(self, api_key, connect_timeout=30, request_timeout=60): + """ Constructor + :param api_key: The API key to use for sending API requests + :param connect_timeout: The maximum amount of time to wait for the connection to be established + :param request_timeout: The maximum amount of time to wait for the request to be processed + """ + self.api_key = api_key + self.http_client = AsyncHTTPClient() + self.connect_timeout = connect_timeout + self.request_timeout = request_timeout + + @gen.coroutine + def list_games_with_players_in_cd(self): + """ Lists the game on the standard map where a player is in CD and the bots needs to submit orders + :return: List of GameIdCountryId tuples [(game_id, country_id), (game_id, country_id)] + """ + route = 'players/cd' + url = '%s?%s' % (API_WEBDIPLOMACY_NET, urlencode({'route': route})) + response = yield self._send_get_request(url) + return_val = [] + + # 200 - Response OK + if response.code == 200: + list_games_players = json.loads(response.body.decode('utf-8')) + for game_player in list_games_players: + return_val += [GameIdCountryId(game_id=game_player['gameID'], country_id=game_player['countryID'])] + + # Error Occurred + else: + LOGGER.warning('ERROR during "%s". Error code: %d. Body: %s.', route, response.code, response.body) + + # Returning + return return_val + + @gen.coroutine + def list_games_with_missing_orders(self): + """ Lists of the game on the standard where the user has not submitted orders yet. + :return: List of GameIdCountryId tuples [(game_id, country_id), (game_id, country_id)] + """ + route = 'players/missing_orders' + url = '%s?%s' % (API_WEBDIPLOMACY_NET, urlencode({'route': route})) + response = yield self._send_get_request(url) + return_val = [] + + # 200 - Response OK + if response.code == 200: + list_games_players = json.loads(response.body.decode('utf-8')) + for game_player in list_games_players: + return_val += [GameIdCountryId(game_id=game_player['gameID'], country_id=game_player['countryID'])] + + # Error Occurred + else: + LOGGER.warning('ERROR during "%s". Error code: %d. Body: %s.', route, response.code, response.body) + + # Returning + return return_val + + @gen.coroutine + def get_game_and_power(self, game_id, country_id, max_phases=None): + """ Returns the game and the power we are playing + :param game_id: The id of the game object (integer) + :param country_id: The id of the country for which we want the game state (integer) + :param max_phases: Optional. If set, improve speed by generating game only using the last 'x' phases. + :return: A tuple consisting of + 1) The diplomacy.Game object from the game state or None if an error occurred + 2) The power name (e.g. 'FRANCE') referred to by country_id + """ + route = 'game/status' + url = '%s?%s' % (API_WEBDIPLOMACY_NET, urlencode({'route': route, 'gameID': game_id, 'countryID': country_id})) + response = yield self._send_get_request(url) + return_val = None, None + + # 200 - Response OK + if response.code == 200: + state_dict = json.loads(response.body.decode('utf-8')) + game, power_name = state_dict_to_game_and_power(state_dict, country_id, max_phases=max_phases) + return_val = game, power_name + + # Error Occurred + else: + LOGGER.warning('ERROR during "%s". Error code: %d. Body: %s.', route, response.code, response.body) + + # Returning + return return_val + + @gen.coroutine + def set_orders(self, game, power_name, orders, wait=None): + """ Submits orders back to the server + :param game: A diplomacy.Game object representing the current state of the game + :param power_name: The name of the power submitting the orders (e.g. 'FRANCE') + :param orders: A list of strings representing the orders (e.g. ['A PAR H', 'F BRE - MAO']) + :param wait: Optional. If True, sets ready=False, if False sets ready=True. + :return: True for success, False for failure + :type game: diplomacy.Game + """ + # Logging orders + LOGGER.info('[%s/%s] - Submitting orders: %s', game.game_id, power_name, orders) + + # Converting orders to dict + orders_dict = [Order(order, map_name=game.map_name, phase_type=game.phase_type) for order in orders] + + # Recording submitted orders + submitted_orders = {} + for order in orders_dict: + unit = ' '.join(order.to_string().split()[:2]) + if order.to_string()[-2:] == ' D': + unit = '? ' + unit[2:] + submitted_orders[unit] = order.to_norm_string() + + # Getting other info + game_id = int(game.game_id) + country_id = CACHE[game.map_name]['power_to_ix'].get(power_name, -1) + current_phase = game.get_current_phase() + + if current_phase != 'COMPLETED': + season, current_year, phase_type = current_phase[0], int(current_phase[1:5]), current_phase[5] + nb_years = current_year - game.map.first_year + turn = 2 * nb_years + (0 if season == 'S' else 1) + phase = {'M': 'Diplomacy', 'R': 'Retreats', 'A': 'Builds'}[phase_type] + else: + turn = -1 + phase = 'Diplomacy' + + # Sending request + route = 'game/orders' + url = '%s?%s' % (API_WEBDIPLOMACY_NET, urlencode({'route': route})) + body = {'gameID': game_id, + 'turn': turn, + 'phase': phase, + 'countryID': country_id, + 'orders': [order.to_dict() for order in orders_dict if order]} + if wait is not None: + body['ready'] = 'Yes' if not wait else 'No' + body = json.dumps(body).encode('utf-8') + response = yield self._send_post_request(url, body) + + # Error Occurred + if response.code != 200: + LOGGER.warning('ERROR during "%s". Error code: %d. Body: %s.', route, response.code, response.body) + return False + + # No orders set - Was only setting the ready flag + if not orders: + return True + + # Otherwise, validating that received orders are the same as submitted orders + response_body = json.loads(response.body) + orders_dict = [Order(order, map_name=game.map_name, phase_type=game.phase_type) for order in response_body] + all_orders_set = True + + # Recording received orders + received_orders = {} + for order in orders_dict: + unit = ' '.join(order.to_string().split()[:2]) + if order.to_string()[-2:] == ' D': + unit = '? ' + unit[2:] + received_orders[unit] = order.to_norm_string() + + # Logging different orders + for unit in submitted_orders: + if submitted_orders[unit] != received_orders.get(unit, ''): + all_orders_set = False + LOGGER.warning('[%s/%s]. Submitted: "%s" - Server has: "%s".', + game.game_id, power_name, submitted_orders[unit], received_orders.get(unit, '')) + + # Returning status + return all_orders_set + + # ---- Helper methods ---- + @gen.coroutine + def _send_get_request(self, url): + """ Helper method to send a get request to the API endpoint """ + http_request = HTTPRequest(url=url, + method='GET', + headers={'Authorization': 'Bearer %s' % self.api_key}, + connect_timeout=self.connect_timeout, + request_timeout=self.request_timeout, + user_agent=API_USER_AGENT) + http_response = yield self.http_client.fetch(http_request, raise_error=False) + return http_response + + @gen.coroutine + def _send_post_request(self, url, body): + """ Helper method to send a post request to the API endpoint """ + http_request = HTTPRequest(url=url, + method='POST', + body=body, + headers={'Authorization': 'Bearer %s' % self.api_key}, + connect_timeout=self.connect_timeout, + request_timeout=self.request_timeout, + user_agent=API_USER_AGENT) + http_response = yield self.http_client.fetch(http_request, raise_error=False) + return http_response diff --git a/diplomacy/integration/webdiplomacy_net/game.py b/diplomacy/integration/webdiplomacy_net/game.py new file mode 100644 index 0000000..32c425f --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/game.py @@ -0,0 +1,333 @@ +# ============================================================================== +# Copyright 2019 - Philip Paquette. All Rights Reserved. +# +# NOTICE: All information contained herein is, and remains the property of the authors +# listed above. The intellectual and technical concepts contained herein are proprietary +# and may be covered by U.S. and Foreign Patents, patents in process, and are protected +# by trade secret or copyright law. Dissemination of this information or reproduction of +# this material is strictly forbidden unless prior written permission is obtained. +# ============================================================================== +""" Utility to convert a webdiplomacy.net game state to a game object """ +import logging +from diplomacy import Game +from diplomacy.integration.webdiplomacy_net.orders import Order +from diplomacy.integration.webdiplomacy_net.utils import CACHE + +# Constants +LOGGER = logging.getLogger(__name__) + + +def turn_to_phase(turn, phase): + """ Converts a turn and phase to a short phase name + e.g. turn 1 - phase 'Retreats' to 'F1901R' + """ + year = 1901 + turn // 2 + season = 'S' if turn % 2 == 0 else 'F' + if phase == 'Builds': + season = 'W' + phase_type = {'Diplomacy': 'M', 'Retreats': 'R', 'Builds': 'A'}[phase] + return season + str(year) + phase_type + + +# Format: +# {'unitType': 'Army'/'Fleet', +# 'terrID': integer, +# 'countryId': integer, +# 'retreating': 'Yes'/'No'} +def unit_dict_to_str(unit_dict, map_id=1): + """ Converts a unit from the dictionary format to the string format + e.g. {'unitType': 'Army', 'terrID': 6, 'countryId': 1, 'retreating': 'No'} + to 'ENGLAND', 'A LON' + + :param unit_dict: The unit in dictionary format from webdiplomacy.net + :return: A tuple consisting of: + 1) The power owning the unit (e.g. 'FRANCE') + 2) The unit in string format (with a leading * when dislodged) (e.g. '*A PAR') + """ + req_fields = ('unitType', 'terrID', 'countryID', 'retreating') + if [1 for field in req_fields if field not in unit_dict]: + LOGGER.error('The required fields for unit dict are %s. Cannot translate %s', req_fields, unit_dict) + return '', '' + + # Extracting information + unit_type = str(unit_dict['unitType']) + terr_id = int(unit_dict['terrID']) + country_id = int(unit_dict['countryID']) + retreating = str(unit_dict['retreating']) + + # Validating + if unit_type not in ('Army', 'Fleet'): + LOGGER.error('Unknown unitType "%s". Expected "Army" or "Fleet".', unit_type) + return '', '' + if terr_id not in CACHE[map_id]['ix_to_loc']: + LOGGER.error('Unknown terrID "%s" for mapID "%s".', terr_id, map_id) + return '', '' + if country_id not in CACHE[map_id]['ix_to_power']: + LOGGER.error('Unknown countryID "%s" for mapID "%s".', country_id, map_id) + return '', '' + if retreating not in ('Yes', 'No'): + LOGGER.error('Unknown retreating "%s". Expected "Yes" or "No".', retreating) + return '', '' + + # Translating names + loc_name = CACHE[map_id]['ix_to_loc'][terr_id] + power_name = CACHE[map_id]['ix_to_power'][country_id] + is_dislodged = bool(retreating == 'Yes') + + # Building unit and returning + unit = '%s%s %s' % ('*' if is_dislodged else '', unit_type[0], loc_name) + return power_name, unit + + +# Format: +# {'terrID': integer, +# 'countryId': integer} +def center_dict_to_str(center_dict, map_id=1): + """ Converts a supply center from the dictionary format to the string format + e.g. {'terrID': 6, 'countryId': 1} to 'ENGLAND', 'LON' + + :param center_dict: The center in dictionary format from webdiplomacy.net + :return: A tuple consisting of: + 1) The power owning the center (e.g. 'FRANCE') + 2) The location where the supply center is (e.g. 'PAR') + """ + req_fields = ('terrID', 'countryID') + if [1 for field in req_fields if field not in center_dict]: + LOGGER.error('The required fields for center dict are %s. Cannot translate %s', req_fields, center_dict) + return '', '' + + # Extracting information + terr_id = int(center_dict['terrID']) + country_id = int(center_dict['countryID']) + + # Validating + if terr_id not in CACHE[map_id]['ix_to_loc']: + LOGGER.error('Unknown terrID "%s" for mapID "%s".', terr_id, map_id) + return '', '' + if country_id not in CACHE[map_id]['ix_to_power']: + LOGGER.error('Unknown countryID "%s" for mapID "%s".', country_id, map_id) + return '', '' + + # Translating names + loc_name = CACHE[map_id]['ix_to_loc'][terr_id] + power_name = CACHE[map_id]['ix_to_power'][country_id] + + # Returning + return power_name, loc_name + + +# Format: +# {'turn': integer, +# 'phase': 'Diplomacy', 'Retreats', 'Builds', +# 'countryID': integer, +# 'terrID': integer, +# 'unitType': 'Army', 'Fleet', +# 'type': 'Hold', 'Move', 'Support hold', 'Support move', 'Convoy', +# 'Retreat', 'Disband', +# 'Build Army', 'Build Fleet', 'Wait', 'Destroy' +# 'toTerrID': integer, +# 'fromTerrID': integer, +# 'viaConvoy': 'Yes', 'No', +# 'success': 'Yes', 'No', +# 'dislodged': 'Yes', 'No'} +def order_dict_to_str(order_dict, phase, map_id=1): + """ Converts an order from the dictionary format to the string format + :param order_dict: The order in dictionary format from webdiplomacy.net + :param phase: The current phase ('Diplomacy', 'Retreats', 'Builds') + :return: A tuple consisting of: + 1) The power who submitted the order (e.g. 'FRANCE') + 2) The order in string format (e.g. 'A PAR H') + """ + req_fields = ('countryID',) + if [1 for field in req_fields if field not in order_dict]: + LOGGER.error('The required fields for order dict are %s. Cannot translate %s', req_fields, order_dict) + return '', '', '' + + # Extracting information + country_id = int(order_dict['countryID']) + + # Validating + if country_id not in CACHE[map_id]['ix_to_power']: + LOGGER.error('Unknown countryID "%s" for mapID "%s".', country_id, map_id) + return '', '' + + # Getting power name and phase_type + power_name = CACHE[map_id]['ix_to_power'][country_id] + phase_type = {'Diplomacy': 'M', 'Retreats': 'R', 'Builds': 'A'}[phase] + + # Getting order in string format + order = Order(order_dict, map_id=map_id, phase_type=phase_type) + if not order: + return '', '' + + # Returning + return power_name, order.to_string() + + +# Format: +# {'turn': integer, +# 'phase': 'Diplomacy', 'Retreats', 'Builds', +# 'units': [], +# 'centers': [], +# 'orders': []} +def process_phase_dict(phase_dict, map_id=1): + """ Converts a phase dict to its string representation """ + phase = turn_to_phase(phase_dict.get('turn', 0), phase_dict.get('phase', 'Diplomacy')) + + # Processing units + units_per_power = {} + for unit_dict in phase_dict.get('units', []): + power_name, unit = unit_dict_to_str(unit_dict, map_id=map_id) + if not power_name: + continue + if power_name not in units_per_power: + units_per_power[power_name] = [] + units_per_power[power_name].append(unit) + + # Processing centers + centers_per_power = {} + for center_dict in phase_dict.get('centers', []): + power_name, loc = center_dict_to_str(center_dict, map_id=map_id) + if not power_name: + continue + if power_name not in centers_per_power: + centers_per_power[power_name] = [] + centers_per_power[power_name].append(loc) + + # Processing orders + orders_per_power = {} + for order_dict in phase_dict.get('orders', []): + power_name, order = order_dict_to_str(order_dict, + phase=phase_dict.get('phase', 'Diplomacy'), + map_id=map_id) + if not power_name: + continue + if power_name not in orders_per_power: + orders_per_power[power_name] = [] + orders_per_power[power_name].append(order) + + # Returning + return {'name': phase, + 'units': units_per_power, + 'centers': centers_per_power, + 'orders': orders_per_power} + + +# Format: +# {'gameID': integer, +# 'variantID': integer, +# 'turn': integer, +# 'phase': 'Pre-Game', 'Diplomacy', 'Retreats', 'Builds', 'Finished', +# 'gameOver': 'No, 'Won', 'Drawn', +# 'phases': [], +# 'standoffs': []} +def state_dict_to_game_and_power(state_dict, country_id, max_phases=None): + """ Converts a game state from the dictionary format to an actual diplomacy.Game object with the related power. + :param state_dict: The game state in dictionary format from webdiplomacy.net + :param country_id: The country id we want to convert. + :param max_phases: Optional. If set, improve speed by only keeping the last 'x' phases to regenerate the game. + :return: A tuple of + 1) None, None - on error or if the conversion is not possible, or game is invalid / not-started / done + 2) game, power_name - on successful conversion + """ + if state_dict is None: + return None, None + + req_fields = ('gameID', 'variantID', 'turn', 'phase', 'gameOver', 'phases', 'standoffs', 'occupiedFrom') + if [1 for field in req_fields if field not in state_dict]: + LOGGER.error('The required fields for state dict are %s. Cannot translate %s', req_fields, state_dict) + return None, None + + # Extracting information + game_id = str(state_dict['gameID']) + map_id = int(state_dict['variantID']) + standoffs = state_dict['standoffs'] + occupied_from = state_dict['occupiedFrom'] + + # Parsing all phases + state_dict_phases = state_dict.get('phases', []) + if max_phases is not None and isinstance(max_phases, int): + state_dict_phases = state_dict_phases[-1 * max_phases:] + all_phases = [process_phase_dict(phase_dict, map_id=map_id) for phase_dict in state_dict_phases] + + # Building game - Replaying the last 6 phases + game = Game() + game.game_id = game_id + + for phase_to_replay in all_phases[:-1]: + game.set_current_phase(phase_to_replay['name']) + + # Units + game.clear_units() + for power_name, power_units in phase_to_replay['units'].items(): + if power_name == 'GLOBAL': + continue + game.set_units(power_name, power_units) + + # Centers + game.clear_centers() + for power_name, power_centers in phase_to_replay['centers'].items(): + if power_name == 'GLOBAL': + continue + game.set_centers(power_name, power_centers) + + # Orders + for power_name, power_orders in phase_to_replay['orders'].items(): + if power_name == 'GLOBAL': + continue + game.set_orders(power_name, power_orders) + + # Processing + game.process() + + # Setting the current phase + current_phase = all_phases[-1] + game.set_current_phase(current_phase['name']) + + # Units + game.clear_units() + for power_name, power_units in current_phase['units'].items(): + if power_name == 'GLOBAL': + continue + game.set_units(power_name, power_units) + + # Centers + game.clear_centers() + for power_name, power_centers in current_phase['centers'].items(): + if power_name == 'GLOBAL': + continue + game.set_centers(power_name, power_centers) + + # Setting retreat locs + if current_phase['name'][-1] == 'R': + invalid_retreat_locs = set() + attack_source = {} + + # Loc is occupied + for power in game.powers.values(): + for unit in power.units: + invalid_retreat_locs.add(unit[2:5]) + + # Loc was in standoff + if standoffs: + for loc_dict in standoffs: + _, loc = center_dict_to_str(loc_dict, map_id=map_id) + invalid_retreat_locs.add(loc[:3]) + + # Loc was attacked from + if occupied_from: + for loc_id, occupied_from_id in occupied_from.items(): + loc_name = CACHE[map_id]['ix_to_loc'][int(loc_id)][:3] + from_loc_name = CACHE[map_id]['ix_to_loc'][int(occupied_from_id)][:3] + attack_source[loc_name] = from_loc_name + + # Removing invalid retreat locs + for power in game.powers.values(): + for retreat_unit in power.retreats: + power.retreats[retreat_unit] = [loc for loc in power.retreats[retreat_unit] + if loc[:3] not in invalid_retreat_locs + and loc[:3] != attack_source.get(retreat_unit[2:5], '')] + + # Returning + power_name = CACHE[map_id]['ix_to_power'][country_id] + return game, power_name diff --git a/diplomacy/integration/webdiplomacy_net/orders.py b/diplomacy/integration/webdiplomacy_net/orders.py new file mode 100644 index 0000000..42c455f --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/orders.py @@ -0,0 +1,593 @@ +# ============================================================================== +# Copyright 2019 - Philip Paquette. All Rights Reserved. +# +# NOTICE: All information contained herein is, and remains the property of the authors +# listed above. The intellectual and technical concepts contained herein are proprietary +# and may be covered by U.S. and Foreign Patents, patents in process, and are protected +# by trade secret or copyright law. Dissemination of this information or reproduction of +# this material is strictly forbidden unless prior written permission is obtained. +# ============================================================================== +""" Orders - Contains utilities to convert orders between string format and webdiplomacy.net format """ +import logging +from diplomacy import Map +from diplomacy.integration.webdiplomacy_net.utils import CACHE + +# Constants +LOGGER = logging.getLogger(__name__) + + +class Order(): + """ Class to convert order from string representation to dictionary (webdiplomacy.net) representation """ + + def __init__(self, order, map_id=None, map_name=None, phase_type=None): + """ Constructor + :param order: An order (either as a string 'A PAR H' or as a dictionary) + :param map_id: Optional. The map id of the map where orders are submitted (webdiplomacy format) + :param map_name: Optional. The map name of the map where orders are submitted. + :param phase_type: Optional. The phase type ('M', 'R', 'A') to disambiguate orders to send. + """ + self.map_name = 'standard' + self.phase_type = 'M' + self.order_str = '' + self.order_dict = {} + + # Detecting the map name + if map_id is not None: + if map_id not in CACHE['ix_to_map']: + raise ValueError('Map with id %s is not supported.' % map_id) + self.map_name = CACHE['ix_to_map'][map_id] + elif map_name is not None: + if map_name not in CACHE['map_to_ix']: + raise ValueError('Map with name %s is not supported.' % map_name) + self.map_name = map_name + + # Detecting the phase type + if isinstance(phase_type, str) and phase_type in 'MRA': + self.phase_type = phase_type + + # Building the order + if isinstance(order, str): + self._build_from_string(order) + elif isinstance(order, dict): + self._build_from_dict(order) + else: + raise ValueError('Expected order to be a string or a dictionary.') + + def _build_from_string(self, order): + """ Builds this object from a string + :type order: str + """ + # pylint: disable=too-many-return-statements,too-many-branches,too-many-statements + words = order.split() + + # --- Wait / Waive --- + # [{"id": "56", "unitID": null, "type": "Wait", "toTerrID": "", "fromTerrID": "", "viaConvoy": ""}] + if len(words) == 1 and words[0] == 'WAIVE': + self.order_str = 'WAIVE' + self.order_dict = {'terrID': None, + 'unitType': '', + 'type': 'Wait', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + return + + # Validating + if len(words) < 3: + LOGGER.error('Unable to parse the order "%s". Require at least 3 words', order) + return + + short_unit_type, loc_name, order_type = words[:3] + if short_unit_type not in 'AF': + LOGGER.error('Unable to parse the order "%s". Valid unit types are "A" and "F".', order) + return + if order_type not in 'H-SCRBD': + LOGGER.error('Unable to parse the order "%s". Valid order types are H-SCRBD', order) + return + if loc_name not in CACHE[self.map_name]['loc_to_ix']: + LOGGER.error('Received invalid loc "%s" for map "%s".', loc_name, self.map_name) + return + + # Extracting territories + unit_type = {'A': 'Army', 'F': 'Fleet'}[short_unit_type] + terr_id = CACHE[self.map_name]['loc_to_ix'][loc_name] + + # --- Hold --- + # {"id": "76", "unitID": "19", "type": "Hold", "toTerrID": "", "fromTerrID": "", "viaConvoy": ""} + if order_type == 'H': + self.order_str = '%s %s H' % (short_unit_type, loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Hold', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Move --- + # {"id": "73", "unitID": "16", "type": "Move", "toTerrID": "25", "fromTerrID": "", "viaConvoy": "Yes", + # "convoyPath": ["22", "69"]}, + # {"id": "74", "unitID": "17", "type": "Move", "toTerrID": "69", "fromTerrID": "", "viaConvoy": "No"} + elif order_type == '-': + if len(words) < 4: + LOGGER.error('[Move] Unable to parse the move order "%s". Require at least 4 words', order) + LOGGER.error(order) + return + + # Building map + map_object = Map(self.map_name) + fleets_in_convoy = set() + convoy_path = [] + + # Getting destination + to_loc_name = words[3] + to_terr_id = CACHE[self.map_name]['loc_to_ix'].get(to_loc_name, None) + + # Deciding if this move is doable by convoy or not + if unit_type != 'Army': + via_flag = '' + else: + reachable_by_land = map_object.abuts('A', loc_name, '-', to_loc_name) + via_convoy = bool(words[-1] == 'VIA') or not reachable_by_land + via_flag = ' VIA' if via_convoy else '' + + # Finding at least one possible convoy path from src to dest + for nb_fleets in map_object.convoy_paths: + for start_loc, fleet_locs, dest_locs in map_object.convoy_paths[nb_fleets]: + if start_loc != loc_name or to_loc_name not in dest_locs: + continue + fleets_in_convoy |= fleet_locs + break + if fleets_in_convoy: + break + + # Converting to list of ints + if fleets_in_convoy: + convoy_path = [terr_id] + [CACHE[self.map_name]['loc_to_ix'][loc] for loc in fleets_in_convoy] + + if to_loc_name is None: + LOGGER.error('[Move] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + + self.order_str = '%s %s - %s%s' % (short_unit_type, loc_name, to_loc_name, via_flag) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Move', + 'toTerrID': to_terr_id, + 'fromTerrID': '', + 'viaConvoy': 'Yes' if via_flag else 'No'} + if convoy_path: + self.order_dict['convoyPath'] = [-1] + + # --- Support hold --- + # {"id": "73", "unitID": "16", "type": "Support hold", "toTerrID": "24", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'S' and '-' not in words: + if len(words) < 5: + LOGGER.error('[Support H] Unable to parse the support hold order "%s". Require at least 5 words', order) + LOGGER.error(order) + return + + # Getting supported unit + to_loc_name = words[4][:3] + to_terr_id = CACHE[self.map_name]['loc_to_ix'].get(to_loc_name, None) + + if to_loc_name is None: + LOGGER.error('[Support H] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + + self.order_str = '%s %s S %s' % (short_unit_type, loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Support hold', + 'toTerrID': to_terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Support move --- + # {"id": "73", "unitID": "16", "type": "Support move", "toTerrID": "24", "fromTerrID": "69", "viaConvoy": ""} + elif order_type == 'S': + if len(words) < 6: + LOGGER.error('Unable to parse the support move order "%s". Require at least 6 words', order) + return + + # Building map + map_object = Map(self.map_name) + fleets_in_convoy = set() + convoy_path = [] + + # Getting supported unit + move_index = words.index('-') + to_loc_name = words[move_index + 1][:3] # Removing coast from dest + from_loc_name = words[move_index - 1] + to_terr_id = CACHE[self.map_name]['loc_to_ix'].get(to_loc_name, None) + from_terr_id = CACHE[self.map_name]['loc_to_ix'].get(from_loc_name, None) + + if to_loc_name is None: + LOGGER.error('[Support M] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + if from_loc_name is None: + LOGGER.error('[Support M] Received invalid from loc "%s" for map "%s".', from_terr_id, self.map_name) + LOGGER.error(order) + return + + # Deciding if we are support a move by convoy or not + if words[move_index - 2] != 'F' and map_object.area_type(from_loc_name) == 'COAST': + + # Finding at least one possible convoy path from src to dest + for nb_fleets in map_object.convoy_paths: + for start_loc, fleet_locs, dest_locs in map_object.convoy_paths[nb_fleets]: + if start_loc != from_loc_name or to_loc_name not in dest_locs: + continue + fleets_in_convoy |= fleet_locs + break + if fleets_in_convoy: + break + + # Converting to list of ints + if fleets_in_convoy: + convoy_path = [from_terr_id] + [CACHE[self.map_name]['loc_to_ix'][loc] for loc in fleets_in_convoy] + + self.order_str = '%s %s S %s - %s' % (short_unit_type, loc_name, from_loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Support move', + 'toTerrID': to_terr_id, + 'fromTerrID': from_terr_id, + 'viaConvoy': ''} + if convoy_path: + self.order_dict['convoyPath'] = [-1] + + # --- Convoy --- + # {"id": "79", "unitID": "22", "type": "Convoy", "toTerrID": "24", "fromTerrID": "20", "viaConvoy": "", + # "convoyPath": ["20", "69"]} + elif order_type == 'C': + if len(words) < 6: + LOGGER.error('[Convoy] Unable to parse the convoy order "%s". Require at least 6 words', order) + LOGGER.error(order) + return + + # Building map + map_object = Map(self.map_name) + fleets_in_convoy = set() + convoy_path = [] + + # Getting supported unit + move_index = words.index('-') + to_loc_name = words[move_index + 1] + from_loc_name = words[move_index - 1] + to_terr_id = CACHE[self.map_name]['loc_to_ix'].get(to_loc_name, None) + from_terr_id = CACHE[self.map_name]['loc_to_ix'].get(from_loc_name, None) + + if to_loc_name is None: + LOGGER.error('[Convoy] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + if from_loc_name is None: + LOGGER.error('[Convoy] Received invalid from loc "%s" for map "%s".', from_terr_id, self.map_name) + LOGGER.error(order) + return + + # Finding at least one possible convoy path from src to dest + for nb_fleets in map_object.convoy_paths: + for start_loc, fleet_locs, dest_locs in map_object.convoy_paths[nb_fleets]: + if start_loc != from_loc_name or to_loc_name not in dest_locs: + continue + fleets_in_convoy |= fleet_locs + break + if fleets_in_convoy: + break + + # Converting to list of ints + if fleets_in_convoy: + convoy_path = [from_terr_id] + [CACHE[self.map_name]['loc_to_ix'][loc] for loc in fleets_in_convoy] + + self.order_str = '%s %s C A %s - %s' % (short_unit_type, loc_name, from_loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Convoy', + 'toTerrID': to_terr_id, + 'fromTerrID': from_terr_id, + 'viaConvoy': ''} + if convoy_path: + self.order_dict['convoyPath'] = [-1] + + # --- Retreat --- + # {"id": "152", "unitID": "18", "type": "Retreat", "toTerrID": "75", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'R': + if len(words) < 4: + LOGGER.error('[Retreat] Unable to parse the move order "%s". Require at least 4 words', order) + LOGGER.error(order) + return + + # Getting destination + to_loc_name = words[3] + to_terr_id = CACHE[self.map_name]['loc_to_ix'].get(to_loc_name, None) + + if to_loc_name is None: + return + + self.order_str = '%s %s R %s' % (short_unit_type, loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Retreat', + 'toTerrID': to_terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Disband (R phase) --- + # {"id": "152", "unitID": "18", "type": "Disband", "toTerrID": "", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'D' and self.phase_type == 'R': + loc_name = loc_name[:3] + terr_id = CACHE[self.map_name]['loc_to_ix'][loc_name] + self.order_str = '%s %s D' % (short_unit_type, loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Disband', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Build Army --- + # [{"id": "56", "unitID": null, "type": "Build Army", "toTerrID": "37", "fromTerrID": "", "viaConvoy": ""}] + elif order_type == 'B' and short_unit_type == 'A': + self.order_str = 'A %s B' % loc_name + self.order_dict = {'terrID': terr_id, + 'unitType': 'Army', + 'type': 'Build Army', + 'toTerrID': terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # -- Build Fleet --- + # [{"id": "56", "unitID": null, "type": "Build Fleet", "toTerrID": "37", "fromTerrID": "", "viaConvoy": ""}] + elif order_type == 'B' and short_unit_type == 'F': + self.order_str = 'F %s B' % loc_name + self.order_dict = {'terrID': terr_id, + 'unitType': 'Fleet', + 'type': 'Build Fleet', + 'toTerrID': terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # Disband (A phase) + # {"id": "152", "unitID": null, "type": "Destroy", "toTerrID": "18", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'D': + loc_name = loc_name[:3] + terr_id = CACHE[self.map_name]['loc_to_ix'][loc_name] + self.order_str = '%s %s D' % (short_unit_type, loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Destroy', + 'toTerrID': terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + def _build_from_dict(self, order): + """ Builds this object from a dictionary + :type order: dict + """ + # pylint: disable=too-many-return-statements + terr_id = order.get('terrID', None) + unit_type = order.get('unitType', None) + order_type = order.get('type', None) + to_terr_id = order.get('toTerrID', '') + from_terr_id = order.get('fromTerrID', '') + via_convoy = order.get('viaConvoy', '') + + # Using to_terr_id if terr_id is None + terr_id = terr_id if terr_id is not None else to_terr_id + + # Overriding unit type for builds + if order_type == 'Build Army': + unit_type = 'Army' + elif order_type == 'Build Fleet': + unit_type = 'Fleet' + elif order_type in ('Destroy', 'Wait') and unit_type not in ('Army', 'Fleet'): + unit_type = '?' + + # Validating order + if unit_type not in ('Army', 'Fleet', '?'): + LOGGER.error('Received invalid unit type "%s". Expected "Army" or "Fleet".', unit_type) + return + if order_type not in ('Hold', 'Move', 'Support hold', 'Support move', 'Convoy', 'Retreat', 'Disband', + 'Build Army', 'Build Fleet', 'Wait', 'Destroy'): + LOGGER.error('Received invalid order type "%s".', order_type) + return + if terr_id not in CACHE[self.map_name]['ix_to_loc'] and terr_id is not None: + LOGGER.error('Received invalid loc "%s" for map "%s".', terr_id, self.map_name) + return + if via_convoy not in ('Yes', 'No', '', None): + LOGGER.error('Received invalid via convoy "%s". Expected "Yes" or "No" or "".', via_convoy) + return + + # Extracting locations + loc_name = CACHE[self.map_name]['ix_to_loc'].get(terr_id, None) + to_loc_name = CACHE[self.map_name]['ix_to_loc'].get(to_terr_id, None) + from_loc_name = CACHE[self.map_name]['ix_to_loc'].get(from_terr_id, None) + + # Building order + # --- Hold --- + # {"id": "76", "unitID": "19", "type": "Hold", "toTerrID": "", "fromTerrID": "", "viaConvoy": ""} + if order_type == 'Hold': + self.order_str = '%s %s H' % (unit_type[0], loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Hold', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Move --- + # {"id": "73", "unitID": "16", "type": "Move", "toTerrID": "25", "fromTerrID": "", "viaConvoy": "Yes", + # "convoyPath": ["22", "69"]}, + # {"id": "74", "unitID": "17", "type": "Move", "toTerrID": "69", "fromTerrID": "", "viaConvoy": "No"} + elif order_type == 'Move': + if to_loc_name is None: + LOGGER.error('[Move] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + + # We don't need to set the "convoyPath" property if we are converting from an order_dict + via_flag = ' VIA' if via_convoy == 'Yes' else '' + self.order_str = '%s %s - %s%s' % (unit_type[0], loc_name, to_loc_name, via_flag) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Move', + 'toTerrID': to_terr_id, + 'fromTerrID': '', + 'viaConvoy': via_convoy} + + # --- Support hold --- + # {"id": "73", "unitID": "16", "type": "Support hold", "toTerrID": "24", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'Support hold': + if to_loc_name is None: + LOGGER.error('[Support H] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + self.order_str = '%s %s S %s' % (unit_type[0], loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Support hold', + 'toTerrID': to_terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Support move --- + # {"id": "73", "unitID": "16", "type": "Support move", "toTerrID": "24", "fromTerrID": "69", "viaConvoy": ""} + elif order_type == 'Support move': + if to_loc_name is None: + LOGGER.error('[Support M] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + if from_loc_name is None: + LOGGER.error('[Support M] Received invalid from loc "%s" for map "%s".', from_terr_id, self.map_name) + LOGGER.error(order) + return + self.order_str = '%s %s S %s - %s' % (unit_type[0], loc_name, from_loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Support move', + 'toTerrID': to_terr_id, + 'fromTerrID': from_terr_id, + 'viaConvoy': ''} + + # --- Convoy --- + # {"id": "79", "unitID": "22", "type": "Convoy", "toTerrID": "24", "fromTerrID": "20", "viaConvoy": "", + # "convoyPath": ["20", "69"]} + elif order_type == 'Convoy': + if to_loc_name is None: + LOGGER.error('[Convoy] Received invalid to loc "%s" for map "%s".', to_terr_id, self.map_name) + LOGGER.error(order) + return + if from_loc_name is None: + LOGGER.error('[Convoy] Received invalid from loc "%s" for map "%s".', from_terr_id, self.map_name) + LOGGER.error(order) + return + + # We don't need to set the "convoyPath" property if we are converting from an order_dict + self.order_str = '%s %s C A %s - %s' % (unit_type[0], loc_name, from_loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Convoy', + 'toTerrID': to_terr_id, + 'fromTerrID': from_terr_id, + 'viaConvoy': ''} + + # --- Retreat --- + # {"id": "152", "unitID": "18", "type": "Retreat", "toTerrID": "75", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'Retreat': + if to_loc_name is None: + return + self.order_str = '%s %s R %s' % (unit_type[0], loc_name, to_loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Retreat', + 'toTerrID': to_terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Disband --- + # {"id": "152", "unitID": "18", "type": "Disband", "toTerrID": "", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'Disband': + self.order_str = '%s %s D' % (unit_type[0], loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Disband', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Build Army --- + # [{"id": "56", "unitID": null, "type": "Build Army", "toTerrID": "37", "fromTerrID": "", "viaConvoy": ""}] + elif order_type == 'Build Army': + self.order_str = 'A %s B' % loc_name + self.order_dict = {'terrID': terr_id, + 'unitType': 'Army', + 'type': 'Build Army', + 'toTerrID': terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Build Fleet --- + # [{"id": "56", "unitID": null, "type": "Build Fleet", "toTerrID": "37", "fromTerrID": "", "viaConvoy": ""}] + elif order_type == 'Build Fleet': + self.order_str = 'F %s B' % loc_name + self.order_dict = {'terrID': terr_id, + 'unitType': 'Fleet', + 'type': 'Build Fleet', + 'toTerrID': terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + # --- Wait / Waive --- + # [{"id": "56", "unitID": null, "type": "Wait", "toTerrID": "", "fromTerrID": "", "viaConvoy": ""}] + elif order_type == 'Wait': + self.order_str = 'WAIVE' + self.order_dict = {'terrID': None, + 'unitType': '', + 'type': 'Wait', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + + # Disband (A phase) + # {"id": "152", "unitID": null, "type": "Destroy", "toTerrID": "18", "fromTerrID": "", "viaConvoy": ""} + elif order_type == 'Destroy': + self.order_str = '%s %s D' % (unit_type[0], loc_name) + self.order_dict = {'terrID': terr_id, + 'unitType': unit_type, + 'type': 'Destroy', + 'toTerrID': terr_id, + 'fromTerrID': '', + 'viaConvoy': ''} + + def __bool__(self): + """ Returns True if an order was parsed, False otherwise """ + return bool(self.order_str != '') + + def __str__(self): + """ Returns the string representation of the order """ + return self.order_str + + def to_string(self): + """ Returns the string representation of the order """ + return self.order_str + + def to_norm_string(self): + """ Returns a normalized order string """ + if self.order_str[-2:] == ' D': + order_str = '? ' + self.order_str[2:] + else: + order_str = self.order_str + return order_str\ + .replace(' S A ', ' S ')\ + .replace(' S F ', ' S ') \ + .replace(' C A ', ' C ') \ + .replace(' C F ', ' C ') \ + .replace(' VIA', '') + + def to_dict(self): + """ Returns the dictionary representation of the order """ + return self.order_dict diff --git a/diplomacy/integration/webdiplomacy_net/tests/__init__.py b/diplomacy/integration/webdiplomacy_net/tests/__init__.py new file mode 100644 index 0000000..4f2769f --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/tests/__init__.py @@ -0,0 +1,16 @@ +# ============================================================================== +# 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/>. +# ============================================================================== diff --git a/diplomacy/integration/webdiplomacy_net/tests/test_game.py b/diplomacy/integration/webdiplomacy_net/tests/test_game.py new file mode 100644 index 0000000..7a888d3 --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/tests/test_game.py @@ -0,0 +1,185 @@ +# ============================================================================== +# 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/>. +# ============================================================================== +""" Test order conversion. """ +from diplomacy.integration.webdiplomacy_net.game import turn_to_phase, unit_dict_to_str, center_dict_to_str, \ + order_dict_to_str + +# ------------------------------------ +# ---- Tests for turn_to_phase ---- +# ------------------------------------ +def test_phase_s1901m(): + """ Tests S1901M """ + phase = turn_to_phase(0, 'Diplomacy') + assert phase == 'S1901M' + +def test_phase_s1901r(): + """ Tests S1901R """ + phase = turn_to_phase(0, 'Retreats') + assert phase == 'S1901R' + +def test_phase_f1901m(): + """ Tests F1901M """ + phase = turn_to_phase(1, 'Diplomacy') + assert phase == 'F1901M' + +def test_phase_f1901r(): + """ Tests F1901R """ + phase = turn_to_phase(1, 'Retreats') + assert phase == 'F1901R' + +def test_phase_w1901a(): + """ Tests W1901A """ + phase = turn_to_phase(1, 'Builds') + assert phase == 'W1901A' + +def test_phase_s1902m(): + """ Tests S1902M """ + phase = turn_to_phase(2, 'Diplomacy') + assert phase == 'S1902M' + + +# ------------------------------------ +# ---- Tests for unit_dict_to_str ---- +# ------------------------------------ +def test_army_france(): + """ Tests Army France """ + unit_dict = {'unitType': 'Army', 'terrID': 47, 'countryID': 2, 'retreating': 'No'} + power_name, unit = unit_dict_to_str(unit_dict) + assert power_name == 'FRANCE' + assert unit == 'A PAR' + +def test_dis_fleet_england(): + """ Tests Dislodged Fleet England """ + unit_dict = {'unitType': 'Fleet', 'terrID': 6, 'countryID': 1, 'retreating': 'Yes'} + power_name, unit = unit_dict_to_str(unit_dict) + assert power_name == 'ENGLAND' + assert unit == '*F LON' + +def test_invalid_unit(): + """ Tests invalid unit """ + unit_dict = {'unitType': 'Fleet', 'terrID': 99, 'countryID': 0, 'retreating': 'No'} + power_name, unit = unit_dict_to_str(unit_dict) + assert power_name == '' + assert unit == '' + + +# -------------------------------------- +# ---- Tests for center_dict_to_str ---- +# -------------------------------------- +def test_center_dict(): + """ Tests parsing centers """ + power_name, center = center_dict_to_str({'countryID': 1, 'terrID': 6}, map_id=1) + assert power_name == 'ENGLAND' + assert center == 'LON' + + power_name, center = center_dict_to_str({'countryID': 2, 'terrID': 47}, map_id=1) + assert power_name == 'FRANCE' + assert center == 'PAR' + + +# ------------------------------------- +# ---- Tests for order_dict_to_str ---- +# ------------------------------------- +def test_s1901m_hold(): + """ Tests hold in S1901M """ + order_dict = {'turn': 0, + 'phase': 'Diplomacy', + 'countryID': 2, + 'terrID': 6, + 'unitType': 'Army', + 'type': 'Hold', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + power_name, order = order_dict_to_str(order_dict, phase='Diplomacy') + assert power_name == 'FRANCE' + assert order == 'A LON H' + +def test_s1901r_disband(): + """ Tests disband in S1901R """ + order_dict = {'turn': 0, + 'phase': 'Retreats', + 'countryID': 1, + 'terrID': 6, + 'unitType': 'Fleet', + 'type': 'Disband', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + power_name, order = order_dict_to_str(order_dict, phase='Retreats') + assert power_name == 'ENGLAND' + assert order == 'F LON D' + +def test_f1901m_move(): + """ Tests move in F1901M """ + order_dict = {'turn': 1, + 'phase': 'Diplomacy', + 'countryID': 2, + 'terrID': 6, + 'unitType': 'Army', + 'type': 'Move', + 'toTerrID': 47, + 'fromTerrID': '', + 'viaConvoy': 'Yes'} + power_name, order = order_dict_to_str(order_dict, phase='Diplomacy') + assert power_name == 'FRANCE' + assert order == 'A LON - PAR VIA' + +def test_f1901r_retreat(): + """ Tests retreat in F1901R """ + order_dict = {'turn': 1, + 'phase': 'Retreats', + 'countryID': 3, + 'terrID': 6, + 'unitType': 'Army', + 'type': 'Retreat', + 'toTerrID': 47, + 'fromTerrID': '', + 'viaConvoy': ''} + power_name, order = order_dict_to_str(order_dict, phase='Retreats') + assert power_name == 'ITALY' + assert order == 'A LON R PAR' + +def test_w1901a_build(): + """ Tests build army in W1901A """ + order_dict = {'turn': 1, + 'phase': 'Builds', + 'countryID': 2, + 'terrID': 6, + 'unitType': 'Army', + 'type': 'Build Army', + 'toTerrID': 6, + 'fromTerrID': '', + 'viaConvoy': ''} + power_name, order = order_dict_to_str(order_dict, phase='Builds') + assert power_name == 'FRANCE' + assert order == 'A LON B' + +def test_s1902m_hold(): + """ Tests hold in S1902M """ + order_dict = {'turn': 2, + 'phase': 'Diplomacy', + 'countryID': 2, + 'terrID': 6, + 'unitType': 'Army', + 'type': 'Hold', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + power_name, order = order_dict_to_str(order_dict, phase='Diplomacy') + assert power_name == 'FRANCE' + assert order == 'A LON H' diff --git a/diplomacy/integration/webdiplomacy_net/tests/test_orders.py b/diplomacy/integration/webdiplomacy_net/tests/test_orders.py new file mode 100644 index 0000000..aed58c6 --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/tests/test_orders.py @@ -0,0 +1,385 @@ +# ============================================================================== +# 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/>. +# ============================================================================== +""" Test order conversion. """ +from diplomacy.integration.webdiplomacy_net.orders import Order + +def compare_dicts(dict_1, dict_2): + """ Checks if two dictionaries are equal """ + keys_1 = set(dict_1.keys()) - {'convoyPath'} + keys_2 = set(dict_2.keys()) - {'convoyPath'} + if keys_1 != keys_2: + return False + for key in keys_1: + if dict_1[key] != dict_2[key]: + return False + return True + +def test_hold_army_001(): + """ Tests hold army """ + raw_order = 'A PAR H' + order_str = 'A PAR H' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Hold', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_hold_army_002(): + """ Tests hold army """ + raw_order = 'A ABC H' + order_str = '' + order_dict = {} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_hold_fleet_001(): + """ Tests hold fleet """ + raw_order = 'F LON H' + order_str = 'F LON H' + order_dict = {'terrID': 6, + 'unitType': 'Fleet', + 'type': 'Hold', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_move_army_001(): + """ Tests move army """ + raw_order = 'A YOR - LON' + order_str = 'A YOR - LON' + order_dict = {'terrID': 4, + 'unitType': 'Army', + 'type': 'Move', + 'toTerrID': 6, + 'fromTerrID': '', + 'viaConvoy': 'No'} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_move_army_002(): + """ Tests move army """ + raw_order = 'A PAR - LON VIA' + order_str = 'A PAR - LON VIA' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Move', + 'toTerrID': 6, + 'fromTerrID': '', + 'viaConvoy': 'Yes'} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_move_fleet_001(): + """ Tests move fleet """ + raw_order = 'F BRE - MAO' + order_str = 'F BRE - MAO' + order_dict = {'terrID': 46, + 'unitType': 'Fleet', + 'type': 'Move', + 'toTerrID': 61, + 'fromTerrID': '', + 'viaConvoy': 'No'} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_support_hold_001(): + """ Tests for support hold """ + raw_order = 'A PAR S F BRE' + order_str = 'A PAR S BRE' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Support hold', + 'toTerrID': 46, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_support_hold_002(): + """ Tests for support hold """ + raw_order = 'F MAO S F BRE' + order_str = 'F MAO S BRE' + order_dict = {'terrID': 61, + 'unitType': 'Fleet', + 'type': 'Support hold', + 'toTerrID': 46, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_support_move_001(): + """ Tests support move """ + raw_order = 'A PAR S F MAO - BRE' + order_str = 'A PAR S MAO - BRE' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Support move', + 'toTerrID': 46, + 'fromTerrID': 61, + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_support_move_002(): + """ Tests support move """ + raw_order = 'F MAO S A PAR - BRE' + order_str = 'F MAO S PAR - BRE' + order_dict = {'terrID': 61, + 'unitType': 'Fleet', + 'type': 'Support move', + 'toTerrID': 46, + 'fromTerrID': 47, + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_convoy_001(): + """ Tests convoy """ + raw_order = 'F MAO C A PAR - LON' + order_str = 'F MAO C A PAR - LON' + order_dict = {'terrID': 61, + 'unitType': 'Fleet', + 'type': 'Convoy', + 'toTerrID': 6, + 'fromTerrID': 47, + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_retreat_army_001(): + """ Tests retreat army """ + raw_order = 'A PAR R LON' + order_str = 'A PAR R LON' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Retreat', + 'toTerrID': 6, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_retreat_fleet_001(): + """ Tests retreat fleet """ + raw_order = 'F BRE R SPA/SC' + order_str = 'F BRE R SPA/SC' + order_dict = {'terrID': 46, + 'unitType': 'Fleet', + 'type': 'Retreat', + 'toTerrID': 77, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_disband_army_001(): + """ Tests disband army """ + raw_order = 'A PAR D' + order_str = 'A PAR D' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Disband', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order, phase_type='R') + order_from_dict = Order(order_dict, phase_type='R') + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_disband_fleet_001(): + """ Tests disband fleet """ + raw_order = 'F BRE D' + order_str = 'F BRE D' + order_dict = {'terrID': 46, + 'unitType': 'Fleet', + 'type': 'Disband', + 'toTerrID': '', + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order, phase_type='R') + order_from_dict = Order(order_dict, phase_type='R') + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_build_army_001(): + """ Tests build army """ + raw_order = 'A PAR B' + order_str = 'A PAR B' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Build Army', + 'toTerrID': 47, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_build_fleet_001(): + """ Tests build fleet """ + raw_order = 'F BRE B' + order_str = 'F BRE B' + order_dict = {'terrID': 46, + 'unitType': 'Fleet', + 'type': 'Build Fleet', + 'toTerrID': 46, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order) + order_from_dict = Order(order_dict) + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_disband_army_002(): + """ Tests disband army """ + raw_order = 'A PAR D' + order_str = 'A PAR D' + order_dict = {'terrID': 47, + 'unitType': 'Army', + 'type': 'Destroy', + 'toTerrID': 47, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order, phase_type='A') + order_from_dict = Order(order_dict, phase_type='A') + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) + +def test_disband_fleet_002(): + """ Tests disband fleet """ + raw_order = 'F BRE D' + order_str = 'F BRE D' + order_dict = {'terrID': 46, + 'unitType': 'Fleet', + 'type': 'Destroy', + 'toTerrID': 46, + 'fromTerrID': '', + 'viaConvoy': ''} + order_from_string = Order(raw_order, phase_type='A') + order_from_dict = Order(order_dict, phase_type='A') + + # Validating + assert order_from_string.to_string() == order_str + assert compare_dicts(order_from_string.to_dict(), order_dict) + assert order_from_dict.to_string() == order_str + assert compare_dicts(order_from_dict.to_dict(), order_dict) diff --git a/diplomacy/integration/webdiplomacy_net/utils.py b/diplomacy/integration/webdiplomacy_net/utils.py new file mode 100644 index 0000000..8c329aa --- /dev/null +++ b/diplomacy/integration/webdiplomacy_net/utils.py @@ -0,0 +1,75 @@ +# ============================================================================== +# Copyright 2019 - Philip Paquette. All Rights Reserved. +# +# NOTICE: All information contained herein is, and remains the property of the authors +# listed above. The intellectual and technical concepts contained herein are proprietary +# and may be covered by U.S. and Foreign Patents, patents in process, and are protected +# by trade secret or copyright law. Dissemination of this information or reproduction of +# this material is strictly forbidden unless prior written permission is obtained. +# ============================================================================== +""" Utilities - Builds a cache to query power_ix and loc_ix from webdiplomacy.net """ +import collections + +# Constants +CACHE = {'ix_to_map': {1: 'standard', 15: 'standard_france_austria', 23: 'standard_germany_italy'}, + 'map_to_ix': {'standard': 1, 'standard_france_austria': 15, 'standard_germany_italy': 23}} + +# Standard map +CACHE[1] = {'powers': ['GLOBAL', 'ENGLAND', 'FRANCE', 'ITALY', 'GERMANY', 'AUSTRIA', 'TURKEY', 'RUSSIA'], + 'locs': [None, 'CLY', 'EDI', 'LVP', 'YOR', 'WAL', 'LON', 'POR', 'SPA', 'NAF', 'TUN', 'NAP', 'ROM', 'TUS', + 'PIE', 'VEN', 'APU', 'GRE', 'ALB', 'SER', 'BUL', 'RUM', 'CON', 'SMY', 'ANK', 'ARM', 'SYR', 'SEV', + 'UKR', 'WAR', 'LVN', 'MOS', 'STP', 'FIN', 'SWE', 'NWY', 'DEN', 'KIE', 'BER', 'PRU', 'SIL', 'MUN', + 'RUH', 'HOL', 'BEL', 'PIC', 'BRE', 'PAR', 'BUR', 'MAR', 'GAS', 'BAR', 'NWG', 'NTH', 'SKA', 'HEL', + 'BAL', 'BOT', 'NAO', 'IRI', 'ENG', 'MAO', 'WES', 'LYO', 'TYS', 'ION', 'ADR', 'AEG', 'EAS', 'BLA', + 'TYR', 'BOH', 'VIE', 'TRI', 'BUD', 'GAL', 'SPA/NC', 'SPA/SC', 'STP/NC', 'STP/SC', 'BUL/EC', + 'BUL/SC']} +CACHE['standard'] = CACHE[1] + +# France-Austria Map +CACHE[15] = {'powers': ['GLOBAL', 'FRANCE', 'AUSTRIA'], + 'locs': [None, 'CLY', 'EDI', 'LVP', 'YOR', 'WAL', 'LON', 'POR', 'SPA', 'SPA/NC', 'SPA/SC', 'NAF', 'TUN', + 'NAP', 'ROM', 'TUS', 'PIE', 'VEN', 'APU', 'GRE', 'ALB', 'SER', 'BUL', 'BUL/EC', 'BUL/SC', 'RUM', + 'CON', 'SMY', 'ANK', 'ARM', 'SYR', 'SEV', 'UKR', 'WAR', 'LVN', 'MOS', 'STP', 'STP/NC', 'STP/SC', + 'FIN', 'SWE', 'NWY', 'DEN', 'KIE', 'BER', 'PRU', 'SIL', 'MUN', 'RUH', 'HOL', 'BEL', 'PIC', 'BRE', + 'PAR', 'BUR', 'MAR', 'GAS', 'BAR', 'NWG', 'NTH', 'SKA', 'HEL', 'BAL', 'BOT', 'NAO', 'IRI', 'ENG', + 'MAO', 'WES', 'LYO', 'TYS', 'ION', 'ADR', 'AEG', 'EAS', 'BLA', 'TYR', 'BOH', 'VIE', 'TRI', 'BUD', + 'GAL']} +CACHE['standard_france_austria'] = CACHE[15] + +# Germany-Italy Map +CACHE[23] = {'powers': ['GLOBAL', 'GERMANY', 'ITALY'], + 'locs': [None, 'CLY', 'EDI', 'LVP', 'YOR', 'WAL', 'LON', 'POR', 'SPA', 'SPA/NC', 'SPA/SC', 'NAF', 'TUN', + 'NAP', 'ROM', 'TUS', 'PIE', 'VEN', 'APU', 'GRE', 'ALB', 'SER', 'BUL', 'BUL/EC', 'BUL/SC', 'RUM', + 'CON', 'SMY', 'ANK', 'ARM', 'SYR', 'SEV', 'UKR', 'WAR', 'LVN', 'MOS', 'STP', 'STP/NC', 'STP/SC', + 'FIN', 'SWE', 'NWY', 'DEN', 'KIE', 'BER', 'PRU', 'SIL', 'MUN', 'RUH', 'HOL', 'BEL', 'PIC', 'BRE', + 'PAR', 'BUR', 'MAR', 'GAS', 'BAR', 'NWG', 'NTH', 'SKA', 'HEL', 'BAL', 'BOT', 'NAO', 'IRI', 'ENG', + 'MAO', 'WES', 'LYO', 'TYS', 'ION', 'ADR', 'AEG', 'EAS', 'BLA', 'TYR', 'BOH', 'VIE', 'TRI', 'BUD', + 'GAL']} +CACHE['standard_germany_italy'] = CACHE[23] + +# Named tuples +class GameIdCountryId( + collections.namedtuple('GameIdCountryId', ('game_id', 'country_id'))): + """ Tuple (game_id, country_id) """ + + +def build_cache(): + """ Computes a mapping from ix-to-power and ix-to-loc and vice-versa """ + for map_id in (1, 15, 23): + CACHE[map_id]['ix_to_power'] = {} + CACHE[map_id]['power_to_ix'] = {} + CACHE[map_id]['ix_to_loc'] = {} + CACHE[map_id]['loc_to_ix'] = {} + + for power_id, power_name in enumerate(CACHE[map_id]['powers']): + CACHE[map_id]['ix_to_power'][power_id] = power_name + CACHE[map_id]['power_to_ix'][power_name] = power_id + + for loc_id, loc_name in enumerate(CACHE[map_id]['locs']): + if loc_id == 0: + continue + CACHE[map_id]['ix_to_loc'][loc_id] = loc_name + CACHE[map_id]['loc_to_ix'][loc_name] = loc_id + +# Building cache +build_cache() |