Ustagammon/tests/reference/compare.py

"""Compare le moteur embarqué aux résultats de gnubg de bureau.

Pour chaque fichier de résultats/ et chaque niveau, lance
`engine-cli --ply N reference POSITION_ID MATCH_ID [DÉS]` et vérifie les
critères de concordance décrits dans README.md (section « Comparaison avec
le moteur embarqué »). Affiche un rapport et sort en erreur si un critère
n'est pas tenu.

Usage (dans l'image Docker ustagammon/build-c, voir
tests/test_engine_reference.sh) :

    python3 tests/reference/compare.py [--cli build/engine-cli]
        [--results tests/reference/results] [--tolerance 0.002] [--details]

--cli choisit le programme interrogé, qui doit imiter engine-cli ; ce peut
être une commande avec ses arguments. Pour le moteur WebAssembly
(tests/test_wasm_reference.sh, dans l'image Emscripten qui contient Node) :

    python3 tests/reference/compare.py --cli "node engine-wasm/engine-cli.mjs"
"""

import argparse
import glob
import json
import os
import shlex
import subprocess
import sys

# Écart maximal accepté sur une probabilité ou une équité : voir la
# justification dans README.md.
TOLERANCE = 0.002

PROBABILITIES = ("win", "win_gammon", "win_backgammon", "lose_gammon", "lose_backgammon")
CUBE_EQUITIES = ("no_double", "double_take", "double_pass", "optimal")


class Report:
    """Collecte les écarts d'une comparaison."""

    def __init__(self, tolerance):
        self.tolerance = tolerance
        self.failures = []      # critères non tenus
        self.notes = []         # remarques qui ne font pas échouer
        self.lines = []         # tous les nombres comparés (--details)
        self.max_gap = 0.0
        self.max_gap_where = ""

    def number(self, where, field, expected, actual):
        gap = abs(actual - expected)
        line = (where, field, "%.9g" % expected, "%.9g" % actual, "%.2e" % gap)
        self.lines.append(line)
        if gap > self.max_gap:
            self.max_gap = gap
            self.max_gap_where = "%s %s" % (where, field)
        if gap > self.tolerance:
            self.failures.append(line)

    def same(self, where, field, expected, actual):
        if expected != actual:
            self.failures.append((where, field, str(expected), str(actual), "différent"))


def run_engine(cli, position, ply):
    command = shlex.split(cli) + ["--ply", str(ply), "reference", position["position_id"],
               position["match_id"]]
    dice = position["context"]["dice"]
    if dice:
        command.append("%d%d" % tuple(dice))
    output = subprocess.run(command, capture_output=True, text=True)
    if output.returncode != 0:
        raise SystemExit("%s a échoué : %s" % (" ".join(command), output.stdout + output.stderr))
    return json.loads(output.stdout)


def compare_probabilities(report, where, expected, actual):
    for name in PROBABILITIES:
        report.number(where, name, expected[name], actual[name])


def compare_moves(report, where, expected, actual):
    report.same(where, "legal_move_count", expected["legal_move_count"],
                actual["legal_move_count"])
    reference = expected["ranking"]
    engine = actual["ranking"]
    if not reference or not engine:
        report.same(where, "ranking", [m["move"] for m in reference], [m["move"] for m in engine])
        return

    # Meilleur coup : le même, sauf si gnubg lui-même ne sépare ses deux
    # premiers coups que de moins que la tolérance (égalité).
    best, engine_best = reference[0]["move"], engine[0]["move"]
    if best != engine_best:
        tied = len(reference) > 1 and -reference[1]["equity_diff"] <= report.tolerance
        if tied:
            report.notes.append("%s : meilleur coup %s pour gnubg, %s pour le moteur ; "
                                "égalité, gnubg ne les sépare que de %.2e"
                                % (where, best, engine_best, -reference[1]["equity_diff"]))
        else:
            report.same(where, "meilleur coup", best, engine_best)

    # Coups présents dans les deux classements : même pli pour le meilleur
    # coup de gnubg, équités et probabilités à la tolérance près pour tous.
    by_move = {m["move"]: m for m in engine}
    for m in reference:
        other = by_move.get(m["move"])
        if other is None:
            continue
        here = "%s %s" % (where, m["move"])
        if m is reference[0]:
            report.same(here, "ply", m["ply"], other["ply"])
        report.number(here, "equity", m["equity"], other["equity"])
        compare_probabilities(report, here, m["probabilities"], other["probabilities"])


def compare_cube(report, where, expected, actual):
    report.same(where, "decision", expected["decision"], actual["decision"])
    report.same(where, "decision_text", expected["decision_text"], actual["decision_text"])
    for name in CUBE_EQUITIES:
        report.number(where, name, expected[name], actual[name])


def compare_position(report, cli, path):
    with open(path, encoding="utf-8") as f:
        position = json.load(f)
    for ply, expected in sorted(position["plies"].items()):
        actual = run_engine(cli, position, int(ply))
        where = "%s à %s pli(s)" % (position["id"], ply)
        compare_probabilities(report, where + " évaluation",
                              expected["evaluation"]["probabilities"],
                              actual["evaluation"]["probabilities"])
        report.number(where + " évaluation", "equity", expected["evaluation"]["equity"],
                      actual["evaluation"]["equity"])
        if "moves" in expected:
            compare_moves(report, where, expected["moves"], actual["moves"])
        else:
            compare_cube(report, where, expected["cube"], actual["cube"])
    return len(position["plies"])


def print_table(lines):
    header = ("position et niveau", "champ", "attendu", "obtenu", "écart")
    rows = [header] + lines
    widths = [max(len(row[i]) for row in rows) for i in range(len(header))]
    for row in rows:
        print("  " + "  ".join(cell.ljust(width) for cell, width in zip(row, widths)).rstrip())


def main():
    here = os.path.dirname(os.path.abspath(__file__))
    parser = argparse.ArgumentParser(description="Compare engine-cli aux références de gnubg.")
    parser.add_argument("--cli", default="build/engine-cli",
                        help="programme interrogé, éventuellement avec ses arguments")
    parser.add_argument("--results", default=os.path.join(here, "results"))
    parser.add_argument("--tolerance", type=float, default=TOLERANCE)
    parser.add_argument("--details", action="store_true",
                        help="affiche tous les nombres comparés")
    args = parser.parse_args()

    report = Report(args.tolerance)
    paths = sorted(glob.glob(os.path.join(args.results, "*.json")))
    levels = sum(compare_position(report, args.cli, path) for path in paths)

    if args.details:
        print_table(report.lines)
    for note in report.notes:
        print("  Remarque : " + note)
    print("  %d positions, %d niveaux, %d nombres comparés ; écart maximal %.2e (%s), "
          "tolérance %.0e." % (len(paths), levels, len(report.lines), report.max_gap,
                               report.max_gap_where, args.tolerance))
    if report.failures:
        print("  %d écart(s) hors critères :" % len(report.failures))
        print_table(report.failures)
        sys.exit(1)


main()