aboutsummaryrefslogtreecommitdiff
path: root/diplomacy/integration/webdiplomacy_net/api.py
diff options
context:
space:
mode:
Diffstat (limited to 'diplomacy/integration/webdiplomacy_net/api.py')
-rw-r--r--diplomacy/integration/webdiplomacy_net/api.py228
1 files changed, 228 insertions, 0 deletions
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
generated by cgit v1.2.3 (git 2.39.1) at 2025-06-08 14:51:14 +0000