binaryio.py

"""
@file: binaryio.py
Created on 01.04.2021
@project: CrazyAra
@author: queensgambit, maxalexger

Contains the main class to communicate with the C++ binary.
"""
import logging
import time

from subprocess import PIPE, Popen
from dataclasses import fields
from DeepCrazyhouse.configs.rl_config import UCIConfig, UCIConfigArena


class BinaryIO:
    """
    This class establishes a connection to the binary and handles
    the binary inputs and outputs.
    """
    def __init__(self, binary_path: str):
        """
        Open a process to the binary in order to send commands and read output
        :param binary_path: Path to the binary including the binary name
        """
        self.proc = Popen([binary_path], stdin=PIPE, stdout=PIPE, stderr=PIPE, shell=False)

    def compare_new_weights(self, nb_arena_games: int) -> bool:
        """
        Compares the old NN-weights with the newly acquired one.
        Internally, the binary uses the UCI Options 'Model_Directory' &
        'Model_Directory_contender' to read in the old and new NN-weights respectively.
        :param nb_arena_games: Number of non-standoff games btw old and new NN-weights
        :return: True - If current NN generator should be replaced
                 False - If current NN generator should be kept
        """
        self.proc.stdin.write(b"arena %d\n" % nb_arena_games)
        self.proc.stdin.flush()
        return self.read_output_arena(check_error=True)

    def generate_games(self):
        """
        Requests the binary to generate games. The number of games generated by the
        binary is determined by Selfplay_Number_Chunks * Selfplay_Chunk_Size.
        Both variables are UCI options within the binary.
        The output of the binary will be inside the binarz folder and contain a 'data.zarr'
        folder with the compressed games, as well as 'games.pgn' and 'gameIdx.txt'.
        :return:
        """
        logging.info(f'Generating games ...')
        # if 0, binary plays Selfplay_Number_Chunks * Selfplay_Chunk_Size games
        self.proc.stdin.write(b"selfplay 0\n")
        self.proc.stdin.flush()
        self.read_output(b"readyok\n", check_error=True)

    def get_uci_options(self):
        """
        Requests and reads uci options from a process.
        This is non UCI standard behavior! After sending 'activeuci' the
        process returns all the current options in the format
        'option name <name> value <value>' followed by 'readyok' at the end.

        :return: Dictionary with the option's names as keys and option's values as values
        """
        options = {}
        self.proc.stdin.write(b'activeuci\n')  # write bytes
        self.proc.stdin.flush()
        while True:
            line = self.proc.stdout.readline()
            line = line.decode("utf-8").strip()  # bytes to string
            if line == f'readyok':
                break
            elif line.startswith(f'option name ') and f' value' in line:
                idx = line.index(f' value')
                value = line[idx+7:]
                if value.replace('.','',1).isdigit():
                    value = float(value)
                options[line[12:idx]] = value
            else:
                raise ValueError(f'uci command activeuci returned wrong format')
        return options

    def load_network(self):
        """
        Tells the binary to load the network and waits until it's finished.
        :return:
        """
        logging.info(f'Loading network & creating backend files ...', )
        self.proc.stdin.write(b"isready\n")
        self.proc.stdin.flush()
        self.read_output(b"readyok\n", check_error=True)

    def read_output(self, last_line=b"readyok\n", check_error=True):
        """
        Reads the output of a process pip until the given last line has been reached.
        :param last_line Content when to stop reading (e.g. b'\n', b'', b"readyok\n")
        :param check_error: Listens to stdout for errors
        :return:
        """
        while True:
            line = self.proc.stdout.readline()
            if check_error and line == b'':
                error = self.proc.stderr.readline()
                if error != b'':
                    logging.error(error)
            if line == last_line:
                break

    def read_output_arena(self, check_error=True) -> bool:
        """
        Reads the output for arena matches and waits for the key-words "keep" or "replace"
        :param check_error: Listens to stdout for errors
        :return: True - If current NN generator should be replaced
                 False - If current NN generator should be kept
        """
        while True:
            line = self.proc.stdout.readline()
            if check_error and line == b'':
                error = self.proc.stderr.readline()
                if error != b'':
                    logging.error(error)
            elif line == b"keep\n":
                return False
            elif line == b"replace\n":
                return True

    def set_uci_options(self, uci_variant: str, context: str, device_id: str, precision: str,
                        model_dir: str, model_contender_dir: str, is_arena: bool = False):
        """
        Sets UCI options of the binary.
        :param uci_variant: The UCI variant that shall be trained.
        :param context: The context of the process (in ["cpu", "gpu"]).
        :param device_id: The id of the device we are using.
        :param precision: The precision of calculations.
        :param model_dir: The path to the model.
        :param model_contender_dir: Directory where the model contender dir will be saved.
        :param is_arena: Applies setting for the arena comparison
        :return:
        """
        self._set_uci_param(f'Model_Directory', model_dir)
        self._set_uci_param(f'Model_Directory_Contender', model_contender_dir)
        self._set_uci_param(f'UCI_Variant', uci_variant)
        self._set_uci_param(f'Context', context)
        self._set_uci_param(f'First_Device_ID', device_id)
        self._set_uci_param(f'Last_Device_ID', device_id)
        self._set_uci_param(f'Precision', precision)

        uci = UCIConfig()
        uci_arena = UCIConfigArena()

        # Send all UCI options (from basic UCIConfig class) that won't be sent further down
        for field in fields(uci):
            if not is_arena or (is_arena and field.name not in fields(uci_arena)):
                self._set_uci_param(field.name, getattr(uci, field.name))

        if is_arena:
            for field in fields(uci_arena):
                self._set_uci_param(field.name, getattr(uci_arena, field.name))

    def _set_uci_param(self, name: str, value):
        """
        Sets the value for a given UCI-parameter in the binary.
        :param name: Name of the UCI-parameter
        :param value: Value for the UCI-parameter (this value is always converted to a string)
        :return:
        """
        if type(value) == bool:
            value = f'true' if value else f'false'
        self.proc.stdin.write(b"setoption name %b value %b\n" % (bytes(name, encoding="utf-8"),
                                                                 bytes(str(value), encoding="utf-8")))
        self.proc.stdin.flush()

        # Log how many tablebases could be found
        if name == f'SyzygyPath' and value != '' and value != '<empty>':
            for _ in range(200):
                line = self.proc.stdout.readline().decode().rstrip('\n')
                if line.startswith('info string Found') and line.endswith('tablebases'):
                    logging.info(line[12:])
                    break

    def stop_process(self):
        """
        Kills the process that is attached to the binary.
        :return:
        """
        self.proc.kill()
        # sleep for 1 sec to ensure the process exited
        time.sleep(1)