Ustagammon/tests/reference/generate.py

# Générateur des résultats de référence d'Ustagammon.
#
# Ce script s'exécute DANS gnubg (scripting Python intégré), dans l'image
# Docker ustagammon/gnubg-reference : voir `make reference` et README.md.
# Il lit positions.json et écrit un fichier results/<id>.json par position.
#
# Rien n'est calculé ici : toutes les valeurs viennent de gnubg. Le script ne
# fait que piloter gnubg, vérifier ses réglages et mettre en forme.

import json
import os
import struct

import gnubg

FOLDER = os.environ.get("REFERENCE_DIR", "/reference")
PLIES = (0, 2)  # niveaux en plis gnubg : 0 = évaluation statique du réseau
MAX_MOVES = 10  # coups gardés dans le classement (tous s'il y en a moins)
MET = "Kazaross-XG2"  # table d'équité de match par défaut de gnubg

# Noms des décisions de videau : énumération cubedecision de eval.h, dans
# l'ordre (gnubg renvoie l'indice).
CUBE_DECISIONS = [
    "DOUBLE_TAKE", "DOUBLE_PASS", "NODOUBLE_TAKE", "TOOGOOD_TAKE",
    "TOOGOOD_PASS", "DOUBLE_BEAVER", "NODOUBLE_BEAVER", "REDOUBLE_TAKE",
    "REDOUBLE_PASS", "NO_REDOUBLE_TAKE", "TOOGOODRE_TAKE", "TOOGOODRE_PASS",
    "NO_REDOUBLE_BEAVER", "NODOUBLE_DEADCUBE", "NO_REDOUBLE_DEADCUBE",
    "NOT_AVAILABLE", "OPTIONAL_DOUBLE_TAKE", "OPTIONAL_REDOUBLE_TAKE",
    "OPTIONAL_DOUBLE_BEAVER", "OPTIONAL_DOUBLE_PASS", "OPTIONAL_REDOUBLE_PASS",
]


def f32(x):
    """Écriture décimale la plus courte qui redonne exactement le float32
    calculé par gnubg (gnubg travaille en simple précision)."""
    target = struct.pack("<f", x)
    for digits in range(1, 10):
        text = "%.*g" % (digits, x)
        if struct.pack("<f", float(text)) == target:
            return float(text)
    raise ValueError(x)


def probabilities(p):
    return {
        "win": f32(p[0]),
        "win_gammon": f32(p[1]),
        "win_backgammon": f32(p[2]),
        "lose_gammon": f32(p[3]),
        "lose_backgammon": f32(p[4]),
    }


def eval_context(ply):
    """Réglage prédéfini « 0ply » ou « 2ply » de gnubg (aecSettings, eval.c)."""
    return {"cubeful": 1, "plies": ply, "deterministic": 1,
            "prune": 1 if ply > 0 else 0, "noise": 0.0}


def set_chequer_settings(ply):
    """Règle l'évaluation utilisée par la commande hint de gnubg."""
    prefix = "set evaluation chequerplay evaluation "
    gnubg.command("set evaluation chequerplay type evaluation")
    gnubg.command(prefix + "plies %d" % ply)
    gnubg.command(prefix + "cubeful on")
    gnubg.command(prefix + "prune " + ("on" if ply > 0 else "off"))
    gnubg.command(prefix + "deterministic on")
    gnubg.command(prefix + "noise 0")


def load_position(pos):
    """Installe la position et vide le cache d'évaluation : chaque calcul se
    fait comme sur un moteur neuf. Vérifie que gnubg relit les mêmes IDs."""
    gnubg.command("clear cache")
    gnubg.command("set gnubgid %s:%s" % (pos["position_id"], pos["match_id"]))
    read_back = (gnubg.positionid(), gnubg.matchid())
    if read_back != (pos["position_id"], pos["match_id"]):
        raise SystemExit("%s : gnubg relit %s" % (pos["id"], read_back))


def cube_info():
    """Contexte du videau de la position courante, tel que gnubg l'utilise
    pour ses propres analyses (GetMatchStateCubeInfo). gnubg.cubeinfo()
    déduit les castors du nombre de castors déjà joués dans la partie (0) ;
    gnubg, lui, les autorise en partie libre (réglage par défaut nBeavers = 3)
    et les ignore en match."""
    ci = gnubg.cubeinfo()
    ci["beavers"] = 1 if ci["matchto"] == 0 else 0
    return ci


def game_context(pos):
    ci = cube_info()
    dice = gnubg.posinfo()["dice"]
    dice = list(dice) if dice[0] else None
    if dice != pos["dice"]:
        raise SystemExit("%s : dés %s dans le Match ID, %s dans positions.json"
                         % (pos["id"], dice, pos["dice"]))
    me = ci["move"]
    owner = ci["cubeowner"]
    return {
        "dice": dice,
        "match_length": ci["matchto"],
        "score_player": ci["score"][me],
        "score_opponent": ci["score"][1 - me],
        "crawford": bool(ci["crawford"]),
        "jacoby": bool(ci["jacoby"]),
        "beavers": bool(ci["beavers"]),
        "cube_value": ci["cube"],
        "cube_owner": "centered" if owner < 0 else "player" if owner == me else "opponent",
    }


def evaluate(ply):
    r = gnubg.evaluate(gnubg.board(), cube_info(), eval_context(ply))
    return {"probabilities": probabilities(r[:5]), "equity": f32(r[5])}


def rank_moves(pos, ply):
    set_chequer_settings(ply)
    load_position(pos)  # efface aussi l'analyse que gnubg garde en mémoire
    hint = gnubg.hint()
    moves = hint["hint"] if hint else []
    ranking = []
    for m in moves[:MAX_MOVES]:
        ctx = m.get("context", {})
        if m["type"] != "eval" or ctx["noise"] != 0 or not ctx["cubeful"]:
            raise SystemExit("%s : coup évalué autrement que prévu : %r" % (pos["id"], m))
        ranking.append({
            "rank": m["movenum"],
            "move": m["move"],
            "ply": ctx["plies"],
            "equity": f32(m["equity"]),
            "equity_diff": f32(m["eqdiff"]),
            "probabilities": probabilities(m["details"]["probs"]),
        })
    return {"legal_move_count": len(moves), "ranking": ranking}


def cube_decision(ply):
    r = gnubg.cfevaluate(gnubg.board(), cube_info(), eval_context(ply))
    return {
        "no_double": f32(r[1]),
        "double_take": f32(r[2]),
        "double_pass": f32(r[3]),
        "optimal": f32(r[0]),
        "decision": CUBE_DECISIONS[r[4]],
        "decision_text": r[5],
    }


def check_settings():
    """Vérifie les réglages par défaut dont dépendent les résultats."""
    # Filtre « Normal » pour 2 plis : (niveau, coups acceptés, en plus, écart).
    expected = [(0, 0, 12, 0.24), (1, 0, 6, 0.12)]
    two_ply = [(f["level"], f["acceptmoves"], f["extramoves"], round(f["threshold"], 6))
               for f in gnubg.getevalhintfilter() if f["ply"] == 2]
    if two_ply != expected:
        raise SystemExit("filtre de coups inattendu : %r" % (two_ply,))
    if MET not in gnubg.show("matchequitytable"):
        raise SystemExit("table d'équité de match inattendue")


def main():
    with open(os.path.join(FOLDER, "positions.json"), encoding="utf-8") as f:
        positions = json.load(f)["positions"]
    with open("/usr/local/share/gnubg/commit") as f:
        commit = f.read().strip()
    check_settings()
    out_dir = os.path.join(FOLDER, "results")
    os.makedirs(out_dir, exist_ok=True)
    for pos in positions:
        load_position(pos)
        result = {
            "id": pos["id"],
            "type": pos["type"],
            "categories": pos["categories"],
            "description": pos["description"],
            "position_id": pos["position_id"],
            "match_id": pos["match_id"],
            "gnubg_commit": commit,
            "met": MET,
            "context": game_context(pos),
            "plies": {},
        }
        for ply in PLIES:
            load_position(pos)
            level = {"evaluation": evaluate(ply)}
            if pos["type"] == "move":
                level["moves"] = rank_moves(pos, ply)
            else:
                level["cube"] = cube_decision(ply)
            result["plies"][str(ply)] = level
        with open(os.path.join(out_dir, pos["id"] + ".json"), "w", encoding="utf-8") as f:
            json.dump(result, f, ensure_ascii=False, indent=2)
            f.write("\n")
        print("référence écrite :", pos["id"])


main()