455 lines
18 KiB
Python
455 lines
18 KiB
Python
# 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")
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# Analyses (analyses.json -> analysis/<id>.json) : coup joué, décision de
|
|
# videau, chance d'un jet et niveaux de jeu. Les valeurs viennent de l'analyse
|
|
# de partie de gnubg (« analyse game ») sur une courte partie rejouée ici :
|
|
# la position, puis le coup ou la décision. Voir README.md.
|
|
|
|
# Appréciations et chances : noms de gnubg (skillString et luckString de
|
|
# gnubgmodule.c ; absents du dictionnaire quand rien n'est à signaler) vers
|
|
# ceux d'engine-cli.
|
|
SKILLS = {None: "none", "doubtful": "doubtful", "bad": "bad", "very bad": "very_bad"}
|
|
LUCKS = {None: "none", "verybad": "very_unlucky", "bad": "unlucky", "good": "lucky",
|
|
"verygood": "very_lucky"}
|
|
|
|
# Niveaux de jeu : réglages prédéfinis de gnubg (aecSettings, eval.c
|
|
# l. 318-328), en (plis, bruit) ; l'élagage est actif à partir d'1 pli, le
|
|
# bruit est déterministe. Ceux qui sont vérifiés : 3 et 4 plis sont trop
|
|
# lents pour la suite de tests.
|
|
LEVEL_SETTINGS = {
|
|
"beginner": (0, 0.06), "casual": (0, 0.05), "intermediate": (0, 0.04),
|
|
"advanced": (0, 0.015), "0ply": (0, 0.0), "1ply": (1, 0.0), "2ply": (2, 0.0),
|
|
"3ply": (3, 0.0), "4ply": (4, 0.0),
|
|
}
|
|
CHECKED_LEVELS = ("beginner", "casual", "intermediate", "advanced", "0ply", "1ply", "2ply")
|
|
RANKING_KEPT = 10 # coups gardés dans le classement d'une analyse (plus le coup joué)
|
|
|
|
|
|
def level_context(name):
|
|
plies, noise = LEVEL_SETTINGS[name]
|
|
return {"cubeful": 1, "plies": plies, "deterministic": 1,
|
|
"prune": 1 if plies > 0 else 0, "noise": noise}
|
|
|
|
|
|
def start_game(case):
|
|
"""Nouvelle partie entre deux joueurs humains (gnubg ne joue rien de
|
|
lui-même), placée sur la position du cas, cache vidé."""
|
|
gnubg.command("set player 0 human")
|
|
gnubg.command("set player 1 human")
|
|
gnubg.command("new session")
|
|
gnubg.command("clear cache")
|
|
gnubg.command("set gnubgid %s:%s" % (case["position_id"], case["match_id"]))
|
|
read_back = (gnubg.positionid(), gnubg.matchid())
|
|
if read_back != (case["position_id"], case["match_id"]):
|
|
raise SystemExit("%s : gnubg relit %s" % (case["id"], read_back))
|
|
|
|
|
|
def set_analysis(ply):
|
|
"""Réglages de l'analyse de partie : niveau ply sans bruit, comme le
|
|
réglage prédéfini de gnubg ; filtre de coups et analyse de la chance par
|
|
défaut (« Normal », ecLuck)."""
|
|
for kind in ("chequerplay", "cubedecision"):
|
|
prefix = "set analysis %s evaluation " % kind
|
|
gnubg.command("set analysis %s type evaluation" % kind)
|
|
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 analyse_game():
|
|
"""Analyse la partie en cours ; renvoie ses enregistrements et ses
|
|
statistiques par joueur (« X », « O »). Le cache est vidé avant : les
|
|
évaluations à 2 plis faites pour choisir un coup y laissent des
|
|
résultats à 0 pli des réseaux d'élagage, que l'analyse à 0 pli
|
|
reprendrait (voir analyse_move)."""
|
|
gnubg.command("clear cache")
|
|
gnubg.command("analyse game")
|
|
game = gnubg.match(analysis=1, verbose=1, statistics=1, boards=1)["games"][-1]
|
|
return game["game"], game["stats"]
|
|
|
|
|
|
def last_record(records, action):
|
|
found = [r for r in records if r.get("action") == action]
|
|
if not found:
|
|
raise SystemExit("aucun enregistrement « %s » dans la partie analysée" % action)
|
|
return found[-1]
|
|
|
|
|
|
def swapped(board):
|
|
"""La même position vue de l'autre joueur (gnubg.board() donne le joueur
|
|
au trait en second)."""
|
|
return (board[1], board[0])
|
|
|
|
|
|
def set_dice(case):
|
|
"""Les dés du cas, qui priment sur ceux du Match ID."""
|
|
gnubg.command("set dice %d %d" % tuple(case["dice"]))
|
|
|
|
|
|
def best_move():
|
|
"""Le meilleur coup selon gnubg (findbestmove, réglages d'évaluation
|
|
courants), en notation de gnubg. Pas « hint » : gnubg garderait son
|
|
analyse dans l'enregistrement du coup, et « analyse game » la
|
|
reprendrait au lieu de calculer au niveau demandé."""
|
|
board = gnubg.board()
|
|
return gnubg.movetupletostring(gnubg.findbestmove(), board)
|
|
|
|
|
|
def play_move(case):
|
|
"""Joue le coup du cas, ou le meilleur selon gnubg s'il n'est pas donné
|
|
(cas qui ne servent qu'à la chance du jet)."""
|
|
gnubg.command("move " + (case.get("played") or best_move()))
|
|
|
|
|
|
def analyse_move(case, ply):
|
|
start_game(case)
|
|
set_dice(case)
|
|
before = gnubg.board()
|
|
play_move(case)
|
|
# Sans analyse du videau : gnubg la ferait juste avant la chance du jet,
|
|
# et à 2 plis elle laisse dans son cache d'évaluation des résultats à
|
|
# 0 pli calculés avec les réseaux d'élagage, que l'analyse de la chance
|
|
# (0 pli, sans élagage) reprend ensuite : la chance dépendrait alors de ce
|
|
# qui a été calculé avant (écart observé : 5e-3 sur move-middlegame-close).
|
|
gnubg.command("set analysis cube off")
|
|
# Après le coup, l'adversaire est au trait : la position vue de celui qui
|
|
# vient de jouer est la position retournée.
|
|
played_id = gnubg.positionid(swapped(gnubg.board()))
|
|
set_analysis(ply)
|
|
records, stats = analyse_game()
|
|
record = last_record(records, "move")
|
|
analysis = record["analysis"]
|
|
moves, index = analysis["moves"], analysis["imove"]
|
|
for m in moves:
|
|
if m["type"] != "eval":
|
|
raise SystemExit("%s : coup évalué autrement que prévu : %r" % (case["id"], m))
|
|
|
|
def notation(move):
|
|
return gnubg.movetupletostring(move, before)
|
|
|
|
kept = moves[:max(RANKING_KEPT, index + 1)]
|
|
level = {
|
|
"played_move": notation(record["move"]),
|
|
"played_position_id": played_id,
|
|
"move_count": len(moves),
|
|
"rank": index + 1,
|
|
"played_equity": f32(moves[index]["score"]),
|
|
"best_move": notation(moves[0]["move"]),
|
|
"best_equity": f32(moves[0]["score"]),
|
|
# Perte calculée par gnubg pour ses statistiques (rChequerSkill),
|
|
# comptée en positif.
|
|
"equity_loss": f32(-stats[record["player"]]["moves"]["error-skill"]),
|
|
"skill": SKILLS[analysis.get("skill")],
|
|
"ranking": [{"move": notation(m["move"]), "equity": f32(m["score"])} for m in kept],
|
|
}
|
|
luck = {"value": f32(analysis["luck-value"]), "rating": LUCKS[analysis.get("luck")]}
|
|
return level, luck
|
|
|
|
|
|
def analyse_cube(case, ply):
|
|
start_game(case)
|
|
gnubg.command("set analysis cube on")
|
|
set_analysis(ply)
|
|
# Décision de videau de gnubg au même niveau, avant toute action.
|
|
decision = cube_decision(ply)
|
|
action = case["action"]
|
|
if action == "no_double":
|
|
# Ne pas doubler, c'est jeter les dés : n'importe quel jet et le
|
|
# meilleur coup, l'analyse du videau portant sur la position avant.
|
|
gnubg.command("set dice 2 1")
|
|
gnubg.command("move " + best_move())
|
|
records, stats = analyse_game()
|
|
record = last_record(records, "move")
|
|
analysed = "nd-cubeful-eq" in record["analysis"]
|
|
skill = record["analysis"].get("cube-skill")
|
|
else:
|
|
# gnubg n'analyse un double qu'une fois la réponse donnée : après
|
|
# « double », l'adversaire prend, ce qui ne change rien à l'analyse
|
|
# du double.
|
|
gnubg.command("double")
|
|
gnubg.command("drop" if action == "pass" else "take")
|
|
records, stats = analyse_game()
|
|
analysed = "nd-cubeful-eq" in last_record(records, "double").get("analysis", {})
|
|
record = last_record(records, {"double": "double", "take": "take", "pass": "drop"}[action])
|
|
skill = record.get("analysis", {}).get("skill")
|
|
return {
|
|
"analysed": analysed,
|
|
"cube": decision if analysed else None,
|
|
"equity_loss": f32(-stats[record["player"]]["cube"]["error-skill"]),
|
|
"skill": SKILLS[skill],
|
|
}
|
|
|
|
|
|
def level_move(case, name):
|
|
"""Le coup que joue l'ordinateur de gnubg réglé à ce niveau."""
|
|
start_game(case)
|
|
set_dice(case)
|
|
before = gnubg.board()
|
|
turn = gnubg.posinfo()["turn"]
|
|
plies, noise = LEVEL_SETTINGS[name]
|
|
prefix = "set player %d chequerplay evaluation " % turn
|
|
gnubg.command("set player %d gnubg" % turn)
|
|
gnubg.command(prefix + "plies %d" % plies)
|
|
gnubg.command(prefix + "noise %g" % noise)
|
|
gnubg.command(prefix + "deterministic on")
|
|
gnubg.command(prefix + "cubeful on")
|
|
gnubg.command(prefix + "prune " + ("on" if plies > 0 else "off"))
|
|
gnubg.command("play")
|
|
record = last_record(gnubg.match(analysis=0, boards=1)["games"][-1]["game"], "move")
|
|
gnubg.command("set player %d human" % turn)
|
|
return {"move": gnubg.movetupletostring(record["move"], before)}
|
|
|
|
|
|
def level_cube(case, name):
|
|
"""Décision de videau de gnubg avec les réglages de ce niveau."""
|
|
start_game(case)
|
|
r = gnubg.cfevaluate(gnubg.board(), cube_info(), level_context(name))
|
|
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 write_analysis(out_dir, result):
|
|
with open(os.path.join(out_dir, result["id"] + ".json"), "w", encoding="utf-8") as f:
|
|
json.dump(result, f, ensure_ascii=False, indent=2)
|
|
f.write("\n")
|
|
print("analyse écrite :", result["id"])
|
|
|
|
|
|
def generate_analyses(commit):
|
|
with open(os.path.join(FOLDER, "analyses.json"), encoding="utf-8") as f:
|
|
cases = json.load(f)
|
|
out_dir = os.path.join(FOLDER, "analysis")
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
# Pas d'affichage du plateau ; pas de nouvelle partie ni de jet de dés
|
|
# automatiques, qui ajouteraient des enregistrements après le dernier.
|
|
gnubg.command("set display off")
|
|
gnubg.command("set automatic game off")
|
|
gnubg.command("set automatic roll off")
|
|
for kind in ("moves", "cubes", "levels"):
|
|
for case in cases[kind]:
|
|
result = dict(case, type=kind[:-1], gnubg_commit=commit, met=MET)
|
|
if kind == "moves":
|
|
result["plies"] = {}
|
|
for ply in PLIES:
|
|
result["plies"][str(ply)], result["luck"] = analyse_move(case, ply)
|
|
elif kind == "cubes":
|
|
result["plies"] = {str(ply): analyse_cube(case, ply) for ply in PLIES}
|
|
else:
|
|
result["levels"] = {}
|
|
for name in CHECKED_LEVELS:
|
|
plies, noise = LEVEL_SETTINGS[name]
|
|
level = {"plies": plies, "noise": noise}
|
|
level.update(level_move(case, name) if case["dice"] else level_cube(case, name))
|
|
result["levels"][name] = level
|
|
write_analysis(out_dir, result)
|
|
|
|
|
|
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"])
|
|
generate_analyses(commit)
|
|
|
|
|
|
main()
|