"""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"]) # -------------------------------------------------------------------------- # Analyses (analysis/.json, voir README.md, section « Analyses »). # Seuils de gnubg (arSkillLevel et arLuckLevel) : une appréciation ou une # chance n'est comparée strictement que loin de ces seuils. SKILL_THRESHOLDS = (0.03, 0.06, 0.12) LUCK_THRESHOLDS = (0.3, 0.6) def run_cli(cli, args): command = shlex.split(cli) + args 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 near_threshold(value, thresholds, tolerance): return any(abs(abs(value) - t) <= tolerance for t in thresholds) def same_label(report, where, field, expected, actual, value, thresholds): """Même appréciation ou chance, sauf si la valeur de gnubg est à moins de la tolérance d'un seuil : remarque sans échec.""" if expected == actual: return if near_threshold(value, thresholds, report.tolerance): report.notes.append("%s : %s %s pour gnubg, %s pour le moteur ; valeur %.4f proche " "d'un seuil" % (where, field, expected, actual, value)) else: report.same(where, field, expected, actual) def compare_move_analysis(report, cli, case): dice = "%d%d" % tuple(case["dice"]) args = [case["position_id"], case["match_id"], dice] luck = run_cli(cli, ["luck"] + args) where = "%s chance" % case["id"] report.number(where, "luck", case["luck"]["value"], luck["luck"]) same_label(report, where, "rating", case["luck"]["rating"], luck["rating"], case["luck"]["value"], LUCK_THRESHOLDS) for ply, expected in sorted(case["plies"].items()): actual = run_cli(cli, ["--ply", ply, "analysemove"] + args + [expected["played_position_id"]]) where = "%s à %s pli(s)" % (case["id"], ply) report.same(where, "move_count", expected["move_count"], actual["move_count"]) report.same(where, "coup joué", expected["played_move"], actual["played"]["move"]) # Rang : le même, sauf si les coups entre les deux rangs ont, pour # gnubg, la même équité que le coup joué à la tolérance près. if expected["rank"] != actual["rank"]: low, high = sorted((expected["rank"], actual["rank"])) between = expected["ranking"][low - 1:high] tied = len(between) == high - low + 1 and all( abs(m["equity"] - expected["played_equity"]) <= report.tolerance for m in between) if tied: report.notes.append("%s : rang %d pour gnubg, %d pour le moteur ; égalité" % (where, expected["rank"], actual["rank"])) else: report.same(where, "rank", expected["rank"], actual["rank"]) if expected["best_move"] != actual["best"]["move"]: ranking = expected["ranking"] tied = len(ranking) > 1 and ranking[0]["equity"] - ranking[1]["equity"] <= report.tolerance if tied: report.notes.append("%s : meilleur coup %s pour gnubg, %s pour le moteur ; égalité" % (where, expected["best_move"], actual["best"]["move"])) else: report.same(where, "meilleur coup", expected["best_move"], actual["best"]["move"]) report.number(where, "played_equity", expected["played_equity"], actual["played"]["equity"]) report.number(where, "best_equity", expected["best_equity"], actual["best"]["equity"]) report.number(where, "equity_loss", expected["equity_loss"], actual["equity_loss"]) same_label(report, where, "skill", expected["skill"], actual["skill"], expected["equity_loss"], SKILL_THRESHOLDS) def compare_cube_analysis(report, cli, case): for ply, expected in sorted(case["plies"].items()): actual = run_cli(cli, ["--ply", ply, "analysecube", case["position_id"], case["match_id"], case["action"]]) where = "%s à %s pli(s)" % (case["id"], ply) report.same(where, "analysed", expected["analysed"], actual["analysed"]) if expected["analysed"] and actual["analysed"]: compare_cube(report, where, expected["cube"], actual["cube"]) report.number(where, "equity_loss", expected["equity_loss"], actual["equity_loss"]) same_label(report, where, "skill", expected["skill"], actual["skill"], expected["equity_loss"], SKILL_THRESHOLDS) def compare_levels(report, cli, case, settings): for name, expected in sorted(case["levels"].items()): where = "%s niveau %s" % (case["id"], name) report.same(where, "réglages", (expected["plies"], expected["noise"]), (settings[name]["plies"], round(settings[name]["noise"], 6))) if case["dice"]: actual = run_cli(cli, ["--level", name, "bestmove", case["position_id"], case["match_id"], "%d%d" % tuple(case["dice"])]) report.same(where, "coup choisi", expected["move"], actual["best_move"]["move"]) else: actual = run_cli(cli, ["--level", name, "cube", case["position_id"], case["match_id"]]) compare_cube(report, where, expected, actual) def compare_analyses(report, cli, folder): """Compare toutes les analyses ; renvoie leur nombre.""" settings = {level["name"]: level for level in run_cli(cli, ["levels"])["levels"]} paths = sorted(glob.glob(os.path.join(folder, "*.json"))) for path in paths: with open(path, encoding="utf-8") as f: case = json.load(f) if case["type"] == "move": compare_move_analysis(report, cli, case) elif case["type"] == "cube": compare_cube_analysis(report, cli, case) else: compare_levels(report, cli, case, settings) return len(paths) 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("--analysis", default=os.path.join(here, "analysis")) 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) analyses = compare_analyses(report, args.cli, args.analysis) if args.details: print_table(report.lines) for note in report.notes: print(" Remarque : " + note) print(" %d positions, %d niveaux, %d analyses, %d nombres comparés ; écart maximal %.2e " "(%s), tolérance %.0e." % (len(paths), levels, analyses, 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()