Source code for diplomacy.utils.export

# ==============================================================================
# Copyright (C) 2019 - Philip Paquette, Steven Bocco
#
#  This program is free software: you can redistribute it and/or modify it under
#  the terms of the GNU Affero General Public License as published by the Free
#  Software Foundation, either version 3 of the License, or (at your option) any
#  later version.
#
#  This program is distributed in the hope that it will be useful, but WITHOUT
#  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
#  FOR A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more
#  details.
#
#  You should have received a copy of the GNU Affero General Public License along
#  with this program.  If not, see <https://www.gnu.org/licenses/>.
# ==============================================================================
""" Exporter
    - Responsible for exporting games in a standardized format to disk
"""
import logging
import os
import ujson as json
from diplomacy.engine.game import Game
from diplomacy.engine.map import Map
from diplomacy.utils import strings
from diplomacy.utils.game_phase_data import GamePhaseData

# Constants
LOGGER = logging.getLogger(__name__)
RULES_TO_SKIP = ['SOLITAIRE', 'NO_DEADLINE', 'CD_DUMMIES', 'ALWAYS_WAIT', 'IGNORE_ERRORS']

def to_saved_game_format(game, output_path=None, output_mode='a'):
    """ Converts a game to a standardized JSON format

        :param game: game to convert.
        :param output_path: Optional path to file. If set, the json.dumps() of the saved_game is written to that file.
        :param output_mode: Optional. The mode to use to write to the output_path (if provided). Defaults to 'a'
        :return: A game in the standard format used to saved game, that can be converted to JSON for serialization
        :type game: diplomacy.engine.game.Game
        :type output_path: str | None, optional
        :type output_mode: str, optional
        :rtype: Dict
    """
    phases = Game.get_phase_history(game)                                       # Get phase history.
    phases.append(Game.get_phase_data(game))                                    # Add current game phase.
    rules = [rule for rule in game.rules if rule not in RULES_TO_SKIP]          # Filter rules.

    # Extend states fields.
    phases_to_dict = [phase.to_dict() for phase in phases]
    for phase_dct in phases_to_dict:
        phase_dct['state']['game_id'] = game.game_id
        phase_dct['state']['map'] = game.map_name
        phase_dct['state']['rules'] = rules

    # Building saved game
    saved_game = {'id': game.game_id,
                  'map': game.map_name,
                  'rules': rules,
                  'phases': phases_to_dict}

    # Writing to disk
    if output_path:
        with open(output_path, output_mode) as output_file:
            output_file.write(json.dumps(saved_game) + '\n')

    # Returning
    return saved_game

def from_saved_game_format(saved_game):
    """ Rebuilds a :class:`diplomacy.engine.game.Game` object from the saved game (python :class:`Dict`)
        saved_game is the dictionary. It can be built by calling json.loads(json_line).

        :param saved_game: The saved game exported from :meth:`to_saved_game_format`
        :type saved_game: Dict
        :rtype: diplomacy.engine.game.Game
        :return: The game object restored from the saved game
    """
    game_id = saved_game.get('id', None)
    kwargs = {strings.MAP_NAME: saved_game.get('map', 'standard'),
              strings.RULES: saved_game.get('rules', [])}

    # Building game
    game = Game(game_id=game_id, **kwargs)
    phase_history = []

    # Restoring every phase
    for phase_dct in saved_game.get('phases', []):
        phase_history.append(GamePhaseData.from_dict(phase_dct))
    game.set_phase_data(phase_history, clear_history=True)

    # Returning game
    return game

def load_saved_games_from_disk(input_path, on_error='raise'):
    """ Rebuids multiple :class:`diplomacy.engine.game.Game` from each line in a .jsonl file

        :param input_path: The path to the input file. Expected content is one saved_game json per line.
        :param on_error: Optional. What to do if a game conversion fails. Either 'raise', 'warn', 'ignore'
        :type input_path: str
        :rtype: List[diplomacy.Game]
        :return: A list of :class:`diplomacy.engine.game.Game` objects.
    """
    loaded_games = []
    assert on_error in ('raise', 'warn', 'ignore'), 'Expected values for on_error are "raise", "warn", "ignore".'

    # File does not exist
    if not os.path.exists(input_path):
        LOGGER.warning('File %s does not exist. Aborting.', input_path)
        return loaded_games

    # Importing each game
    with open(input_path, 'r') as file:
        for line in file:
            try:
                saved_game = json.loads(line.rstrip('\n'))
                game = from_saved_game_format(saved_game)
                loaded_games.append(game)
            except Exception as exc:                    # pylint: disable=broad-except
                if on_error == 'raise':
                    raise exc
                if on_error == 'warn':
                    LOGGER.warning(exc)

    # Returning
    return loaded_games

def is_valid_saved_game(saved_game):
    """ Checks if the saved game is valid.
        This is an expensive operation because it replays the game.

        :param saved_game: The saved game (from to_saved_game_format)
        :return: A boolean that indicates if the game is valid
    """
    # pylint: disable=too-many-return-statements, too-many-nested-blocks, too-many-branches
    nb_forced_phases = 0
    max_nb_forced_phases = 1 if 'DIFFERENT_ADJUDICATION' in saved_game.get('rules', []) else 0

    # Validating default fields
    if 'id' not in saved_game or not saved_game['id']:
        return False
    if 'map' not in saved_game:
        return False
    map_object = Map(saved_game['map'])
    if map_object.name != saved_game['map']:
        return False
    if 'rules' not in saved_game:
        return False
    if 'phases' not in saved_game:
        return False

    # Validating each phase
    nb_messages = 0
    nb_phases = len(saved_game['phases'])
    last_time_sent = -1
    for phase_ix in range(nb_phases):
        current_phase = saved_game['phases'][phase_ix]
        state = current_phase['state']
        phase_orders = current_phase['orders']
        previous_phase_name = 'FORMING' if phase_ix == 0 else saved_game['phases'][phase_ix - 1]['name']
        next_phase_name = 'COMPLETED' if phase_ix == nb_phases - 1 else saved_game['phases'][phase_ix + 1]['name']
        power_names = list(state['units'].keys())

        # Validating messages
        for message in saved_game['phases'][phase_ix]['messages']:
            nb_messages += 1
            if map_object.compare_phases(previous_phase_name, message['phase']) >= 0:
                return False
            if map_object.compare_phases(message['phase'], next_phase_name) > 0:
                return False
            if message['sender'] not in power_names + ['SYSTEM']:
                return False
            if message['recipient'] not in power_names + ['GLOBAL']:
                return False
            if message['time_sent'] < last_time_sent:
                return False
            last_time_sent = message['time_sent']

        # Validating phase
        if phase_ix < (nb_phases - 1):
            is_forced_phase = False

            # Setting game state
            game = Game(saved_game['id'], map_name=saved_game['map'], rules=['SOLITAIRE'] + saved_game['rules'])
            game.set_phase_data(GamePhaseData.from_dict(current_phase))

            # Determining what phase we should expect from the dataset.
            next_state = saved_game['phases'][phase_ix + 1]['state']

            # Setting orders
            game.clear_orders()
            for power_name in phase_orders:
                game.set_orders(power_name, phase_orders[power_name])

            # Validating orders
            orders = game.get_orders()
            possible_orders = game.get_all_possible_orders()
            for power_name in orders:
                if sorted(orders[power_name]) != sorted(current_phase['orders'][power_name]):
                    return False
                if 'NO_CHECK' not in game.rules:
                    for order in orders[power_name]:
                        loc = order.split()[1]
                        if order not in possible_orders[loc]:
                            return False

            # Validating resulting state
            game.process()

            # Checking phase name
            if game.get_current_phase() != next_state['name']:
                is_forced_phase = True

            # Checking zobrist hash
            if game.get_hash() != next_state['zobrist_hash']:
                is_forced_phase = True

            # Checking units
            units = game.get_units()
            for power_name in units:
                if sorted(units[power_name]) != sorted(next_state['units'][power_name]):
                    is_forced_phase = True

            # Checking centers
            centers = game.get_centers()
            for power_name in centers:
                if sorted(centers[power_name]) != sorted(next_state['centers'][power_name]):
                    is_forced_phase = True

            # Allowing 1 forced phase if DIFFERENT_ADJUDICATION is in rule
            if is_forced_phase:
                nb_forced_phases += 1
            if nb_forced_phases > max_nb_forced_phases:
                return False

    # Making sure NO_PRESS is not set
    if 'NO_PRESS' in saved_game['rules'] and nb_messages > 0:
        return False

    # The data is valid
    return True