6fc47abb35
* human-like model * kivy 2 * uv * window location fix * macosx package no thanks to jules who is a derp, but cloud is hapl and fix. * settings gui a bit clearer * better ci i hope --------- Co-authored-by: TuNanTang <tunantang@aliyun.com> Co-authored-by: jtkelm2 <feryllcuber@gmail.com>
1460 lines
77 KiB
Python
1460 lines
77 KiB
Python
from abc import ABC, abstractmethod
|
|
import heapq
|
|
import math
|
|
import random
|
|
import time
|
|
from typing import Dict, List, Optional, Tuple
|
|
|
|
from katrain.core.constants import (
|
|
AI_DEFAULT, AI_HANDICAP, AI_INFLUENCE, AI_INFLUENCE_ELO_GRID, AI_JIGO,
|
|
AI_ANTIMIRROR, AI_LOCAL, AI_LOCAL_ELO_GRID, AI_PICK, AI_PICK_ELO_GRID,
|
|
AI_POLICY, AI_RANK, AI_SCORELOSS, AI_SCORELOSS_ELO, AI_SETTLE_STONES,
|
|
AI_SIMPLE_OWNERSHIP, AI_STRENGTH,
|
|
AI_TENUKI, AI_TENUKI_ELO_GRID, AI_TERRITORY, AI_TERRITORY_ELO_GRID,
|
|
AI_WEIGHTED, AI_WEIGHTED_ELO, CALIBRATED_RANK_ELO, OUTPUT_DEBUG,
|
|
OUTPUT_ERROR, OUTPUT_INFO, PRIORITY_EXTRA_AI_QUERY, ADDITIONAL_MOVE_ORDER, AI_HUMAN, AI_PRO
|
|
)
|
|
from katrain.core.game import Game, GameNode, Move
|
|
from katrain.core.utils import var_to_grid, weighted_selection_without_replacement, evaluation_class
|
|
|
|
# Decorator pattern for adding classes to the registry
|
|
STRATEGY_REGISTRY = {}
|
|
|
|
def register_strategy(strategy_name):
|
|
def decorator(strategy_class):
|
|
STRATEGY_REGISTRY[strategy_name] = strategy_class
|
|
return strategy_class
|
|
return decorator
|
|
|
|
def interp_ix(lst, x):
|
|
i = 0
|
|
while i + 1 < len(lst) - 1 and lst[i + 1] < x:
|
|
i += 1
|
|
t = max(0, min(1, (x - lst[i]) / (lst[i + 1] - lst[i])))
|
|
return i, t
|
|
|
|
def interp1d(lst, x):
|
|
xs, ys = zip(*lst)
|
|
i, t = interp_ix(xs, x)
|
|
return (1 - t) * ys[i] + t * ys[i + 1]
|
|
|
|
def interp2d(gridspec, x, y):
|
|
xs, ys, matrix = gridspec
|
|
i, t = interp_ix(xs, x)
|
|
j, s = interp_ix(ys, y)
|
|
return (
|
|
matrix[j][i] * (1 - t) * (1 - s)
|
|
+ matrix[j][i + 1] * t * (1 - s)
|
|
+ matrix[j + 1][i] * (1 - t) * s
|
|
+ matrix[j + 1][i + 1] * t * s
|
|
)
|
|
|
|
def ai_rank_estimation(strategy, settings) -> int:
|
|
if strategy in [AI_DEFAULT, AI_HANDICAP, AI_JIGO, AI_PRO]:
|
|
return 9
|
|
if strategy == AI_RANK:
|
|
return 1 - settings["kyu_rank"]
|
|
if strategy == AI_HUMAN:
|
|
return 1 - settings["human_kyu_rank"]
|
|
|
|
if strategy in [AI_WEIGHTED, AI_SCORELOSS, AI_LOCAL, AI_TENUKI, AI_TERRITORY, AI_INFLUENCE, AI_PICK]:
|
|
if strategy == AI_WEIGHTED:
|
|
elo = interp1d(AI_WEIGHTED_ELO, settings["weaken_fac"])
|
|
if strategy == AI_SCORELOSS:
|
|
elo = interp1d(AI_SCORELOSS_ELO, settings["strength"])
|
|
if strategy == AI_PICK:
|
|
elo = interp2d(AI_PICK_ELO_GRID, settings["pick_frac"], settings["pick_n"])
|
|
if strategy == AI_LOCAL:
|
|
elo = interp2d(AI_LOCAL_ELO_GRID, settings["pick_frac"], settings["pick_n"])
|
|
if strategy == AI_TENUKI:
|
|
elo = interp2d(AI_TENUKI_ELO_GRID, settings["pick_frac"], settings["pick_n"])
|
|
if strategy == AI_TERRITORY:
|
|
elo = interp2d(AI_TERRITORY_ELO_GRID, settings["pick_frac"], settings["pick_n"])
|
|
if strategy == AI_INFLUENCE:
|
|
elo = interp2d(AI_INFLUENCE_ELO_GRID, settings["pick_frac"], settings["pick_n"])
|
|
|
|
kyu = interp1d(CALIBRATED_RANK_ELO, elo)
|
|
return 1 - kyu
|
|
else:
|
|
return AI_STRENGTH[strategy]
|
|
|
|
def game_report(game, thresholds, depth_filter=None):
|
|
cn = game.current_node
|
|
nodes = cn.nodes_from_root
|
|
while cn.children: # main branch
|
|
cn = cn.children[0]
|
|
nodes.append(cn)
|
|
|
|
x, y = game.board_size
|
|
depth_filter = [math.ceil(board_frac * x * y) for board_frac in depth_filter or (0, 1e9)]
|
|
nodes = [n for n in nodes if n.move and not n.is_root and depth_filter[0] <= n.depth < depth_filter[1]]
|
|
histogram = [{"B": 0, "W": 0} for _ in thresholds]
|
|
ai_top_move_count = {"B": 0, "W": 0}
|
|
ai_approved_move_count = {"B": 0, "W": 0}
|
|
player_ptloss = {"B": [], "W": []}
|
|
weights = {"B": [], "W": []}
|
|
|
|
for n in nodes:
|
|
points_lost = n.points_lost
|
|
if n.points_lost is None:
|
|
continue
|
|
else:
|
|
points_lost = max(0, points_lost)
|
|
bucket = len(thresholds) - 1 - evaluation_class(points_lost, thresholds)
|
|
player_ptloss[n.player].append(points_lost)
|
|
histogram[bucket][n.player] += 1
|
|
cands = n.parent.candidate_moves
|
|
filtered_cands = [d for d in cands if d["order"] < ADDITIONAL_MOVE_ORDER and "prior" in d]
|
|
weight = min(
|
|
1.0,
|
|
sum([max(d["pointsLost"], 0) * d["prior"] for d in filtered_cands])
|
|
/ (sum(d["prior"] for d in filtered_cands) or 1e-6),
|
|
) # complexity capped at 1
|
|
# adj_weight between 0.05 - 1, dependent on difficulty and points lost
|
|
adj_weight = max(0.05, min(1.0, max(weight, points_lost / 4)))
|
|
weights[n.player].append((weight, adj_weight))
|
|
if n.parent.analysis_complete:
|
|
ai_top_move_count[n.player] += int(cands[0]["move"] == n.move.gtp())
|
|
ai_approved_move_count[n.player] += int(
|
|
n.move.gtp()
|
|
in [d["move"] for d in filtered_cands if d["order"] == 0 or (d["pointsLost"] < 0.5 and d["order"] < 5)]
|
|
)
|
|
|
|
wt_loss = {
|
|
bw: sum(s * aw for s, (w, aw) in zip(player_ptloss[bw], weights[bw]))
|
|
/ (sum(aw for _, aw in weights[bw]) or 1e-6)
|
|
for bw in "BW"
|
|
}
|
|
sum_stats = {
|
|
bw: (
|
|
{
|
|
"accuracy": 100 * 0.75 ** wt_loss[bw],
|
|
"complexity": sum(w for w, aw in weights[bw]) / len(player_ptloss[bw]),
|
|
"mean_ptloss": sum(player_ptloss[bw]) / len(player_ptloss[bw]),
|
|
"weighted_ptloss": wt_loss[bw],
|
|
"ai_top_move": ai_top_move_count[bw] / len(player_ptloss[bw]),
|
|
"ai_top5_move": ai_approved_move_count[bw] / len(player_ptloss[bw]),
|
|
}
|
|
if len(player_ptloss[bw]) > 0
|
|
else {}
|
|
)
|
|
for bw in "BW"
|
|
}
|
|
return sum_stats, histogram, player_ptloss
|
|
|
|
def fmt_moves(moves: List[Tuple[float, Move]]):
|
|
return ", ".join(f"{mv.gtp()} ({p:.2%})" for p, mv in moves)
|
|
|
|
# Utility functions from the original code
|
|
def policy_weighted_move(policy_moves, lower_bound, weaken_fac):
|
|
lower_bound, weaken_fac = max(0, lower_bound), max(0.01, weaken_fac)
|
|
weighted_coords = [
|
|
(pv, pv ** (1 / weaken_fac), move) for pv, move in policy_moves if pv > lower_bound and not move.is_pass
|
|
]
|
|
if weighted_coords:
|
|
top = weighted_selection_without_replacement(weighted_coords, 1)[0]
|
|
move = top[2]
|
|
ai_thoughts = f"Playing policy-weighted random move {move.gtp()} ({top[0]:.1%}) from {len(weighted_coords)} moves above lower_bound of {lower_bound:.1%}."
|
|
else:
|
|
move = policy_moves[0][1]
|
|
ai_thoughts = f"Playing top policy move because no non-pass move > above lower_bound of {lower_bound:.1%}."
|
|
return move, ai_thoughts
|
|
|
|
def generate_influence_territory_weights(ai_mode, ai_settings, policy_grid, size):
|
|
thr_line = ai_settings["threshold"] - 1 # zero-based
|
|
if ai_mode == AI_INFLUENCE:
|
|
weight = lambda x, y: (1 / ai_settings["line_weight"]) ** ( # noqa E731
|
|
max(0, thr_line - min(size[0] - 1 - x, x)) + max(0, thr_line - min(size[1] - 1 - y, y))
|
|
) # noqa E731
|
|
else:
|
|
weight = lambda x, y: (1 / ai_settings["line_weight"]) ** ( # noqa E731
|
|
max(0, min(size[0] - 1 - x, x, size[1] - 1 - y, y) - thr_line)
|
|
)
|
|
weighted_coords = [
|
|
(policy_grid[y][x] * weight(x, y), weight(x, y), x, y)
|
|
for x in range(size[0])
|
|
for y in range(size[1])
|
|
if policy_grid[y][x] > 0
|
|
]
|
|
ai_thoughts = f"Generated weights for {ai_mode} according to weight factor {ai_settings['line_weight']} and distance from {thr_line + 1}th line. "
|
|
return weighted_coords, ai_thoughts
|
|
|
|
def generate_local_tenuki_weights(ai_mode, ai_settings, policy_grid, cn, size):
|
|
var = ai_settings["stddev"] ** 2
|
|
mx, my = cn.move.coords
|
|
weighted_coords = [
|
|
(policy_grid[y][x], math.exp(-0.5 * ((x - mx) ** 2 + (y - my) ** 2) / var), x, y)
|
|
for x in range(size[0])
|
|
for y in range(size[1])
|
|
if policy_grid[y][x] > 0
|
|
]
|
|
ai_thoughts = f"Generated weights based on one minus gaussian with variance {var} around coordinates {mx},{my}. "
|
|
if ai_mode == AI_TENUKI:
|
|
weighted_coords = [(p, 1 - w, x, y) for p, w, x, y in weighted_coords]
|
|
ai_thoughts = (
|
|
f"Generated weights based on one minus gaussian with variance {var} around coordinates {mx},{my}. "
|
|
)
|
|
return weighted_coords, ai_thoughts
|
|
|
|
class AIStrategy(ABC):
|
|
"""Base strategy class for AI move generation"""
|
|
|
|
def __init__(self, game: Game, ai_settings: Dict):
|
|
self.game = game
|
|
self.settings = ai_settings
|
|
self.cn = game.current_node
|
|
self.strategy_name = self.__class__.__name__
|
|
self.game.katrain.log(f"Initializing {self.strategy_name} with settings: {self.settings}", OUTPUT_DEBUG)
|
|
|
|
@abstractmethod
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
"""Generate a move and explanation"""
|
|
pass
|
|
|
|
def request_analysis(self, extra_settings: Dict) -> Optional[Dict]:
|
|
"""Helper to request additional analysis with custom settings"""
|
|
self.game.katrain.log(f"[{self.strategy_name}] Requesting analysis with settings: {extra_settings}", OUTPUT_DEBUG)
|
|
error = False
|
|
analysis = None
|
|
|
|
def set_analysis(a, partial_result):
|
|
nonlocal analysis
|
|
if not partial_result:
|
|
analysis = a
|
|
self.game.katrain.log(f"[{self.strategy_name}] Analysis received", OUTPUT_DEBUG)
|
|
|
|
def set_error(a):
|
|
nonlocal error
|
|
self.game.katrain.log(f"[{self.strategy_name}] Error in additional analysis query: {a}", OUTPUT_ERROR)
|
|
error = True
|
|
|
|
engine = self.game.engines[self.cn.player]
|
|
engine.request_analysis(
|
|
self.cn,
|
|
callback=set_analysis,
|
|
error_callback=set_error,
|
|
priority=PRIORITY_EXTRA_AI_QUERY,
|
|
ownership=False,
|
|
extra_settings=extra_settings,
|
|
)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Waiting for analysis to complete...", OUTPUT_DEBUG)
|
|
while not (error or analysis):
|
|
time.sleep(0.01) # TODO: prevent deadlock if esc, check node in queries?
|
|
engine.check_alive(exception_if_dead=True)
|
|
|
|
if analysis:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Analysis completed successfully", OUTPUT_DEBUG)
|
|
return analysis
|
|
|
|
def wait_for_analysis(self):
|
|
"""Wait for the analysis to complete"""
|
|
self.game.katrain.log(f"[{self.strategy_name}] Waiting for regular analysis to complete...", OUTPUT_DEBUG)
|
|
while not self.cn.analysis_complete:
|
|
time.sleep(0.01)
|
|
self.game.engines[self.cn.next_player].check_alive(exception_if_dead=True)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Regular analysis completed", OUTPUT_DEBUG)
|
|
|
|
def should_play_top_move(self, policy_moves, top_5_pass, override=0.0, overridetwo=1.0):
|
|
"""Check if we should play the top policy move, regardless of strategy"""
|
|
top_policy_move = policy_moves[0][1]
|
|
self.game.katrain.log(f"[{self.strategy_name}] Checking if should play top move. Top move: {top_policy_move.gtp()} ({policy_moves[0][0]:.2%})", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Override thresholds: single={override:.2%}, combined={overridetwo:.2%}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Top 5 pass: {top_5_pass}", OUTPUT_DEBUG)
|
|
|
|
if top_5_pass:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Playing top move because pass is in top 5", OUTPUT_DEBUG)
|
|
return top_policy_move, "Playing top one because one of them is pass."
|
|
|
|
if policy_moves[0][0] > override:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Playing top move because weight {policy_moves[0][0]:.2%} > override {override:.2%}", OUTPUT_DEBUG)
|
|
return top_policy_move, f"Top policy move has weight > {override:.1%}, so overriding other strategies."
|
|
|
|
if policy_moves[0][0] + policy_moves[1][0] > overridetwo:
|
|
combined = policy_moves[0][0] + policy_moves[1][0]
|
|
self.game.katrain.log(f"[{self.strategy_name}] Playing top move because combined weight {combined:.2%} > overridetwo {overridetwo:.2%}", OUTPUT_DEBUG)
|
|
return top_policy_move, f"Top two policy moves have cumulative weight > {overridetwo:.1%}, so overriding other strategies."
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] No override condition met, continuing with strategy", OUTPUT_DEBUG)
|
|
return None, ""
|
|
|
|
@register_strategy(AI_DEFAULT)
|
|
class DefaultStrategy(AIStrategy):
|
|
"""Default strategy - simply plays the top move from the engine"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[DefaultStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
candidate_moves = self.cn.candidate_moves
|
|
self.game.katrain.log(f"[DefaultStrategy] Analysis found {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
|
|
if not candidate_moves:
|
|
self.game.katrain.log(f"[DefaultStrategy] No candidate moves found, will play pass", OUTPUT_DEBUG)
|
|
top_cand = Move(is_pass=True, player=self.cn.next_player)
|
|
else:
|
|
top_move_data = candidate_moves[0]
|
|
top_cand = Move.from_gtp(top_move_data["move"], player=self.cn.next_player)
|
|
self.game.katrain.log(f"[DefaultStrategy] Top move: {top_cand.gtp()} with stats: {top_move_data}", OUTPUT_DEBUG)
|
|
|
|
ai_thoughts = f"Default strategy found {len(candidate_moves)} moves returned from the engine and chose {top_cand.gtp()} as top move"
|
|
self.game.katrain.log(f"[DefaultStrategy] Final decision: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
|
|
return top_cand, ai_thoughts
|
|
|
|
@register_strategy(AI_HANDICAP)
|
|
class HandicapStrategy(AIStrategy):
|
|
"""Handicap strategy - uses playoutDoublingAdvantage to analyze the position"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[HandicapStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
|
|
# Calculate PDA (Playout Doubling Advantage)
|
|
pda = self.settings["pda"]
|
|
self.game.katrain.log(f"[HandicapStrategy] Initial PDA from settings: {pda}", OUTPUT_DEBUG)
|
|
|
|
if self.settings["automatic"]:
|
|
n_handicaps = len(self.game.root.get_list_property("AB", []))
|
|
MOVE_VALUE = 14 # could be rules dependent
|
|
b_stones_advantage = max(n_handicaps - 1, 0) - (self.cn.komi - MOVE_VALUE / 2) / MOVE_VALUE
|
|
pda = min(3, max(-3, -b_stones_advantage * (3 / 8))) # max PDA at 8 stone adv, normal 9 stone game is 8.46
|
|
|
|
self.game.katrain.log(f"[HandicapStrategy] Automatic PDA calculation:", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HandicapStrategy] - Handicap stones: {n_handicaps}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HandicapStrategy] - Komi: {self.cn.komi}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HandicapStrategy] - Stone advantage: {b_stones_advantage}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HandicapStrategy] - Calculated PDA: {pda}", OUTPUT_DEBUG)
|
|
|
|
# Request additional analysis with PDA
|
|
self.game.katrain.log(f"[HandicapStrategy] Requesting analysis with PDA={pda}", OUTPUT_DEBUG)
|
|
handicap_analysis = self.request_analysis(
|
|
{"playoutDoublingAdvantage": pda, "playoutDoublingAdvantagePla": "BLACK"}
|
|
)
|
|
|
|
if not handicap_analysis:
|
|
self.game.katrain.log("[HandicapStrategy] Error getting handicap-based move, falling back to DefaultStrategy", OUTPUT_ERROR)
|
|
return DefaultStrategy(self.game, self.settings).generate_move()
|
|
|
|
self.wait_for_analysis()
|
|
|
|
candidate_moves = handicap_analysis["moveInfos"]
|
|
self.game.katrain.log(f"[HandicapStrategy] Analysis returned {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
|
|
# Get top candidate move
|
|
top_move_data = candidate_moves[0]
|
|
top_cand = Move.from_gtp(top_move_data["move"], player=self.cn.next_player)
|
|
|
|
# Log details about the top move
|
|
self.game.katrain.log(f"[HandicapStrategy] Top move: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HandicapStrategy] Score lead: {handicap_analysis['rootInfo']['scoreLead']}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HandicapStrategy] Win rate: {handicap_analysis['rootInfo']['winrate']}", OUTPUT_DEBUG)
|
|
|
|
ai_thoughts = f"Handicap strategy found {len(candidate_moves)} moves returned from the engine and chose {top_cand.gtp()} as top move. PDA based score {self.cn.format_score(handicap_analysis['rootInfo']['scoreLead'])} and win rate {self.cn.format_winrate(handicap_analysis['rootInfo']['winrate'])}"
|
|
|
|
self.game.katrain.log(f"[HandicapStrategy] Final decision: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
return top_cand, ai_thoughts
|
|
|
|
@register_strategy(AI_ANTIMIRROR)
|
|
class AntimirrorStrategy(AIStrategy):
|
|
"""Antimirror strategy - uses antiMirror to analyze the position"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
|
|
# Request analysis with antimirror option
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Requesting analysis with antiMirror=True", OUTPUT_DEBUG)
|
|
antimirror_analysis = self.request_analysis({"antiMirror": True})
|
|
|
|
if not antimirror_analysis:
|
|
self.game.katrain.log("[AntimirrorStrategy] Error getting antimirror move, falling back to DefaultStrategy", OUTPUT_ERROR)
|
|
return DefaultStrategy(self.game, self.settings).generate_move()
|
|
|
|
self.wait_for_analysis()
|
|
|
|
candidate_moves = antimirror_analysis["moveInfos"]
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Analysis returned {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
|
|
# Get top candidate move
|
|
top_move_data = candidate_moves[0]
|
|
top_cand = Move.from_gtp(top_move_data["move"], player=self.cn.next_player)
|
|
|
|
# Log details about the top move
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Top move: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Score lead: {antimirror_analysis['rootInfo']['scoreLead']}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Win rate: {antimirror_analysis['rootInfo']['winrate']}", OUTPUT_DEBUG)
|
|
|
|
# Log the top 3 moves for comparison
|
|
for i, move_data in enumerate(candidate_moves[:3]):
|
|
move = Move.from_gtp(move_data["move"], player=self.cn.next_player)
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Move #{i+1}: {move.gtp()} - visits: {move_data.get('visits', 'N/A')}, points lost: {move_data.get('pointsLost', 'N/A')}", OUTPUT_DEBUG)
|
|
|
|
ai_thoughts = f"AntiMirror strategy found {len(candidate_moves)} moves returned from the engine and chose {top_cand.gtp()} as top move. antiMirror based score {self.cn.format_score(antimirror_analysis['rootInfo']['scoreLead'])} and win rate {self.cn.format_winrate(antimirror_analysis['rootInfo']['winrate'])}"
|
|
|
|
self.game.katrain.log(f"[AntimirrorStrategy] Final decision: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
return top_cand, ai_thoughts
|
|
|
|
@register_strategy(AI_JIGO)
|
|
class JigoStrategy(AIStrategy):
|
|
"""Jigo strategy - aims for a specific score difference"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[JigoStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
candidate_moves = self.cn.candidate_moves
|
|
self.game.katrain.log(f"[JigoStrategy] Analysis found {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
|
|
if not candidate_moves:
|
|
self.game.katrain.log(f"[JigoStrategy] No candidate moves found, will play pass", OUTPUT_DEBUG)
|
|
return Move(is_pass=True, player=self.cn.next_player), "No candidate moves found, passing"
|
|
|
|
# Get top engine move for reference
|
|
top_cand = Move.from_gtp(candidate_moves[0]["move"], player=self.cn.next_player)
|
|
self.game.katrain.log(f"[JigoStrategy] Top engine move would be: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
|
|
# Calculate player sign (1 for black, -1 for white)
|
|
sign = self.cn.player_sign(self.cn.next_player)
|
|
self.game.katrain.log(f"[JigoStrategy] Player sign: {sign}", OUTPUT_DEBUG)
|
|
|
|
# Get target score from settings
|
|
target_score = self.settings["target_score"]
|
|
self.game.katrain.log(f"[JigoStrategy] Target score: {target_score}", OUTPUT_DEBUG)
|
|
|
|
# Log score leads before selecting jigo move
|
|
self.game.katrain.log("[JigoStrategy] Candidate move score leads:", OUTPUT_DEBUG)
|
|
for i, move_data in enumerate(candidate_moves[:5]):
|
|
move = Move.from_gtp(move_data["move"], player=self.cn.next_player)
|
|
score_diff = abs(sign * move_data["scoreLead"] - target_score)
|
|
self.game.katrain.log(f"[JigoStrategy] - {move.gtp()}: scoreLead={move_data['scoreLead']}, diff from target={score_diff}", OUTPUT_DEBUG)
|
|
|
|
# Find the move that gives a score closest to the target
|
|
jigo_move = min(
|
|
candidate_moves,
|
|
key=lambda move: abs(sign * move["scoreLead"] - target_score)
|
|
)
|
|
|
|
aimove = Move.from_gtp(jigo_move["move"], player=self.cn.next_player)
|
|
jigo_score_diff = abs(sign * jigo_move["scoreLead"] - target_score)
|
|
|
|
self.game.katrain.log(f"[JigoStrategy] Selected move: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[JigoStrategy] Selected move score lead: {jigo_move['scoreLead']}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[JigoStrategy] Distance from target: {jigo_score_diff}", OUTPUT_DEBUG)
|
|
|
|
ai_thoughts = f"Jigo strategy found {len(candidate_moves)} candidate moves (best {top_cand.gtp()}) and chose {aimove.gtp()} as closest to 0.5 point win"
|
|
|
|
self.game.katrain.log(f"[JigoStrategy] Final decision: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
return aimove, ai_thoughts
|
|
|
|
@register_strategy(AI_SCORELOSS)
|
|
class ScoreLossStrategy(AIStrategy):
|
|
"""ScoreLoss strategy - weights moves based on point loss"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
candidate_moves = self.cn.candidate_moves
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Analysis found {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
|
|
if not candidate_moves:
|
|
self.game.katrain.log(f"[ScoreLossStrategy] No candidate moves found, will play pass", OUTPUT_DEBUG)
|
|
return Move(is_pass=True, player=self.cn.next_player), "No candidate moves found, passing"
|
|
|
|
top_cand = Move.from_gtp(candidate_moves[0]["move"], player=self.cn.next_player)
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Top engine move would be: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
|
|
# Check if top move is pass
|
|
if top_cand.is_pass:
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Top move is pass, so passing regardless of strategy", OUTPUT_DEBUG)
|
|
return top_cand, "Top move is pass, so passing regardless of strategy."
|
|
|
|
# Get strength parameter
|
|
c = self.settings["strength"]
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Strength parameter: {c}", OUTPUT_DEBUG)
|
|
|
|
# Calculate weights for moves based on point loss
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Calculating weights for candidate moves", OUTPUT_DEBUG)
|
|
|
|
moves = []
|
|
for i, d in enumerate(candidate_moves):
|
|
move = Move.from_gtp(d["move"], player=self.cn.next_player)
|
|
points_lost = d["pointsLost"]
|
|
weight = math.exp(min(200, -c * max(0, points_lost)))
|
|
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Move {i+1}: {move.gtp()} - Points lost: {points_lost:.2f}, Weight: {weight:.6f}", OUTPUT_DEBUG)
|
|
moves.append((points_lost, weight, move))
|
|
|
|
# Select move based on weights
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Selecting move with weighted selection", OUTPUT_DEBUG)
|
|
topmove = weighted_selection_without_replacement(moves, 1)[0]
|
|
aimove = topmove[2]
|
|
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Selected move: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Selected move points lost: {topmove[0]:.2f}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Selected move weight: {topmove[1]:.6f}", OUTPUT_DEBUG)
|
|
|
|
ai_thoughts = f"ScoreLoss strategy found {len(candidate_moves)} candidate moves (best {top_cand.gtp()}) and chose {aimove.gtp()} (weight {topmove[1]:.3f}, point loss {topmove[0]:.1f}) based on score weights."
|
|
|
|
self.game.katrain.log(f"[ScoreLossStrategy] Final decision: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
return aimove, ai_thoughts
|
|
|
|
class OwnershipBaseStrategy(AIStrategy):
|
|
"""Base class for ownership-based strategies"""
|
|
|
|
def settledness(self, d, player_sign, player):
|
|
"""Calculate settledness for Simple Ownership strategy"""
|
|
ownership_sum = sum([abs(o) for o in d["ownership"] if player_sign * o > 0])
|
|
self.game.katrain.log(f"[{self.strategy_name}] Calculating settledness for {player}, sign={player_sign}: {ownership_sum:.2f}", OUTPUT_DEBUG)
|
|
return ownership_sum
|
|
|
|
def is_attachment(self, move):
|
|
"""Check if a move is an attachment"""
|
|
if move.is_pass:
|
|
return False
|
|
|
|
stones_with_player = {(*s.coords, s.player) for s in self.game.stones}
|
|
|
|
attach_opponent_stones = sum(
|
|
(move.coords[0] + dx, move.coords[1] + dy, self.cn.player) in stones_with_player
|
|
for dx in [-1, 0, 1]
|
|
for dy in [-1, 0, 1]
|
|
if abs(dx) + abs(dy) == 1
|
|
)
|
|
|
|
nearby_own_stones = sum(
|
|
(move.coords[0] + dx, move.coords[1] + dy, self.cn.next_player) in stones_with_player
|
|
for dx in [-2, 0, 1, 2]
|
|
for dy in [-2 - 1, 0, 1, 2]
|
|
if abs(dx) + abs(dy) <= 2 # allows clamps/jumps
|
|
)
|
|
|
|
is_attach = attach_opponent_stones >= 1 and nearby_own_stones == 0
|
|
self.game.katrain.log(f"[{self.strategy_name}] Is move {move.gtp()} an attachment? {is_attach} (opponent stones: {attach_opponent_stones}, own stones: {nearby_own_stones})", OUTPUT_DEBUG)
|
|
return is_attach
|
|
|
|
def is_tenuki(self, move):
|
|
"""Check if a move is a tenuki (far from previous moves)"""
|
|
if move.is_pass:
|
|
return False
|
|
|
|
result = not any(
|
|
not node
|
|
or not node.move
|
|
or node.move.is_pass
|
|
or max(abs(last_c - cand_c) for last_c, cand_c in zip(node.move.coords, move.coords)) < 5
|
|
for node in [self.cn, self.cn.parent]
|
|
)
|
|
|
|
distances = []
|
|
for node in [self.cn, self.cn.parent]:
|
|
if node and node.move and not node.move.is_pass:
|
|
dist = max(abs(last_c - cand_c) for last_c, cand_c in zip(node.move.coords, move.coords))
|
|
distances.append(dist)
|
|
|
|
if distances:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Is move {move.gtp()} a tenuki? {result} (distances: {distances})", OUTPUT_DEBUG)
|
|
else:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Is move {move.gtp()} a tenuki? {result} (no valid previous moves)", OUTPUT_DEBUG)
|
|
|
|
return result
|
|
|
|
def get_moves_with_settledness(self):
|
|
"""Get moves with ownership and settledness information"""
|
|
self.game.katrain.log(f"[{self.strategy_name}] Getting moves with settledness information", OUTPUT_DEBUG)
|
|
|
|
next_player_sign = self.cn.player_sign(self.cn.next_player)
|
|
candidate_moves = self.cn.candidate_moves
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Processing {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Settings: max_points_lost={self.settings['max_points_lost']}, min_visits={self.settings.get('min_visits', 1)}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Penalties: attach={self.settings['attach_penalty']}, tenuki={self.settings['tenuki_penalty']}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Weights: settled={self.settings['settled_weight']}, opponent_fac={self.settings['opponent_fac']}", OUTPUT_DEBUG)
|
|
|
|
moves_data = []
|
|
for d in candidate_moves:
|
|
# Check basic filtering conditions
|
|
if "pointsLost" not in d:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Move {d['move']} has no pointsLost, skipping", OUTPUT_DEBUG)
|
|
continue
|
|
|
|
if d["pointsLost"] >= self.settings["max_points_lost"]:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Move {d['move']} has pointsLost={d['pointsLost']}, which exceeds max_points_lost={self.settings['max_points_lost']}, skipping", OUTPUT_DEBUG)
|
|
continue
|
|
|
|
if "ownership" not in d:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Move {d['move']} has no ownership data, skipping", OUTPUT_DEBUG)
|
|
continue
|
|
|
|
if not (d["order"] <= 1 or d["visits"] >= self.settings.get("min_visits", 1)):
|
|
self.game.katrain.log(f"[{self.strategy_name}] Move {d['move']} has order={d['order']} and visits={d.get('visits', 'N/A')}, doesn't meet criteria, skipping", OUTPUT_DEBUG)
|
|
continue
|
|
|
|
move = Move.from_gtp(d["move"], player=self.cn.next_player)
|
|
if move.is_pass and d["pointsLost"] > 0.75:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Move {move.gtp()} is pass with high point loss ({d['pointsLost']}), skipping", OUTPUT_DEBUG)
|
|
continue
|
|
|
|
# Calculate metrics
|
|
own_settledness = self.settledness(d, next_player_sign, self.cn.next_player)
|
|
opp_settledness = self.settledness(d, -next_player_sign, self.cn.player)
|
|
is_attach = self.is_attachment(move)
|
|
is_tenuki = self.is_tenuki(move)
|
|
|
|
# Calculate total score for sorting
|
|
score = (d["pointsLost"]
|
|
+ self.settings["attach_penalty"] * is_attach
|
|
+ self.settings["tenuki_penalty"] * is_tenuki
|
|
- self.settings["settled_weight"] * (own_settledness + self.settings["opponent_fac"] * opp_settledness))
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Move {move.gtp()}: points_lost={d['pointsLost']:.2f}, own_settled={own_settledness:.2f}, opp_settled={opp_settledness:.2f}, attach={is_attach}, tenuki={is_tenuki}, score={score:.2f}", OUTPUT_DEBUG)
|
|
|
|
moves_data.append((
|
|
move,
|
|
own_settledness,
|
|
opp_settledness,
|
|
is_attach,
|
|
is_tenuki,
|
|
d,
|
|
score # Store the score for debugging
|
|
))
|
|
|
|
# Sort moves by score
|
|
sorted_moves = sorted(
|
|
moves_data,
|
|
key=lambda t: t[6] # Sort by the precalculated score
|
|
)
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Found {len(sorted_moves)} valid moves with settledness data", OUTPUT_DEBUG)
|
|
if sorted_moves:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Top move after sorting: {sorted_moves[0][0].gtp()} with score {sorted_moves[0][6]:.2f}", OUTPUT_DEBUG)
|
|
|
|
# Return all data except the score which was just for debugging
|
|
return [(move, own_settled, opp_settled, is_attach, is_tenuki, d) for move, own_settled, opp_settled, is_attach, is_tenuki, d, _ in sorted_moves]
|
|
|
|
@register_strategy(AI_SIMPLE_OWNERSHIP)
|
|
class SimpleOwnershipStrategy(OwnershipBaseStrategy):
|
|
"""Simple Ownership strategy - weights moves based on territory control"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
candidate_moves = self.cn.candidate_moves
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Analysis found {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
|
|
if not candidate_moves:
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] No candidate moves found, will play pass", OUTPUT_DEBUG)
|
|
return Move(is_pass=True, player=self.cn.next_player), "No candidate moves found, passing"
|
|
|
|
top_cand = Move.from_gtp(candidate_moves[0]["move"], player=self.cn.next_player)
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Top engine move would be: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
|
|
# Check if top move is pass
|
|
if top_cand.is_pass:
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Top move is pass, so passing regardless of strategy", OUTPUT_DEBUG)
|
|
return top_cand, "Top move is pass, so passing regardless of strategy."
|
|
|
|
# Get moves sorted by settledness criteria
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Getting moves with settledness info", OUTPUT_DEBUG)
|
|
moves_with_settledness = self.get_moves_with_settledness()
|
|
|
|
if moves_with_settledness:
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Found {len(moves_with_settledness)} moves with settledness info", OUTPUT_DEBUG)
|
|
|
|
# Log top 5 candidates in detail
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Top 5 candidates:", OUTPUT_DEBUG)
|
|
for i, (move, settled, oppsettled, isattach, istenuki, d) in enumerate(moves_with_settledness[:5]):
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] #{i+1}: {move.gtp()} - pt_lost: {d['pointsLost']:.1f}, visits: {d.get('visits', 'N/A')}, settledness: {settled:.1f}, opp_settled: {oppsettled:.1f}, attach: {isattach}, tenuki: {istenuki}", OUTPUT_DEBUG)
|
|
|
|
# Format candidate moves for ai_thoughts
|
|
cands = [
|
|
f"{move.gtp()} ({d['pointsLost']:.1f} pt lost, {d.get('visits', 'N/A')} visits, {settled:.1f} settledness, {oppsettled:.1f} opponent settledness{', attachment' if isattach else ''}{', tenuki' if istenuki else ''})"
|
|
for move, settled, oppsettled, isattach, istenuki, d in moves_with_settledness[:5]
|
|
]
|
|
|
|
ai_thoughts = f"{AI_SIMPLE_OWNERSHIP} strategy. Top 5 Candidates {', '.join(cands)} "
|
|
aimove = moves_with_settledness[0][0]
|
|
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Selected move: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
else:
|
|
error_msg = "No moves found - are you using an older KataGo with no per-move ownership info?"
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Error: {error_msg}", OUTPUT_ERROR)
|
|
raise Exception(error_msg)
|
|
|
|
self.game.katrain.log(f"[SimpleOwnershipStrategy] Final decision: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
return aimove, ai_thoughts
|
|
|
|
@register_strategy(AI_SETTLE_STONES)
|
|
class SettleStonesStrategy(OwnershipBaseStrategy):
|
|
"""Settle Stones strategy - focuses on settled stones"""
|
|
|
|
def settledness(self, d, player_sign, player):
|
|
"""Calculate settledness for Settle Stones strategy"""
|
|
board_size_x, board_size_y = self.game.board_size
|
|
ownership_grid = var_to_grid(d["ownership"], (board_size_x, board_size_y))
|
|
|
|
# Sum the absolute ownership values of existing stones
|
|
stone_ownership_values = [abs(ownership_grid[s.coords[0]][s.coords[1]]) for s in self.game.stones if s.player == player]
|
|
total_settledness = sum(stone_ownership_values)
|
|
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Calculating settledness for {player}, sign={player_sign}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Number of stones considered: {len(stone_ownership_values)}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Total settledness: {total_settledness:.2f}", OUTPUT_DEBUG)
|
|
|
|
if stone_ownership_values:
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Min stone ownership: {min(stone_ownership_values):.2f}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Max stone ownership: {max(stone_ownership_values):.2f}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Avg stone ownership: {total_settledness / len(stone_ownership_values):.2f}", OUTPUT_DEBUG)
|
|
|
|
return total_settledness
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
candidate_moves = self.cn.candidate_moves
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Analysis found {len(candidate_moves)} candidate moves", OUTPUT_DEBUG)
|
|
|
|
if not candidate_moves:
|
|
self.game.katrain.log(f"[SettleStonesStrategy] No candidate moves found, will play pass", OUTPUT_DEBUG)
|
|
return Move(is_pass=True, player=self.cn.next_player), "No candidate moves found, passing"
|
|
|
|
top_cand = Move.from_gtp(candidate_moves[0]["move"], player=self.cn.next_player)
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Top engine move would be: {top_cand.gtp()}", OUTPUT_DEBUG)
|
|
|
|
# Check if top move is pass
|
|
if top_cand.is_pass:
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Top move is pass, so passing regardless of strategy", OUTPUT_DEBUG)
|
|
return top_cand, "Top move is pass, so passing regardless of strategy."
|
|
|
|
# Log the number of stones on the board
|
|
black_stones = sum(1 for s in self.game.stones if s.player == "B")
|
|
white_stones = sum(1 for s in self.game.stones if s.player == "W")
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Stones on board: B={black_stones}, W={white_stones}", OUTPUT_DEBUG)
|
|
|
|
# Get moves sorted by settledness criteria
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Getting moves with settledness info", OUTPUT_DEBUG)
|
|
moves_with_settledness = self.get_moves_with_settledness()
|
|
|
|
if moves_with_settledness:
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Found {len(moves_with_settledness)} moves with settledness info", OUTPUT_DEBUG)
|
|
|
|
# Log top 5 candidates in detail
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Top 5 candidates:", OUTPUT_DEBUG)
|
|
for i, (move, settled, oppsettled, isattach, istenuki, d) in enumerate(moves_with_settledness[:5]):
|
|
self.game.katrain.log(f"[SettleStonesStrategy] #{i+1}: {move.gtp()} - pt_lost: {d['pointsLost']:.1f}, visits: {d.get('visits', 'N/A')}, settledness: {settled:.1f}, opp_settled: {oppsettled:.1f}, attach: {isattach}, tenuki: {istenuki}", OUTPUT_DEBUG)
|
|
|
|
# Format candidate moves for ai_thoughts
|
|
cands = [
|
|
f"{move.gtp()} ({d['pointsLost']:.1f} pt lost, {d.get('visits', 'N/A')} visits, {settled:.1f} settledness, {oppsettled:.1f} opponent settledness{', attachment' if isattach else ''}{', tenuki' if istenuki else ''})"
|
|
for move, settled, oppsettled, isattach, istenuki, d in moves_with_settledness[:5]
|
|
]
|
|
|
|
ai_thoughts = f"{AI_SETTLE_STONES} strategy. Top 5 Candidates {', '.join(cands)} "
|
|
aimove = moves_with_settledness[0][0]
|
|
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Selected move: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
else:
|
|
error_msg = "No moves found - are you using an older KataGo with no per-move ownership info?"
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Error: {error_msg}", OUTPUT_ERROR)
|
|
raise Exception(error_msg)
|
|
|
|
self.game.katrain.log(f"[SettleStonesStrategy] Final decision: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
return aimove, ai_thoughts
|
|
|
|
@register_strategy(AI_POLICY)
|
|
class PolicyStrategy(AIStrategy):
|
|
"""Policy strategy - plays the top move suggested by policy network"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[PolicyStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
# Ensure policy is available
|
|
if not self.cn.policy:
|
|
self.game.katrain.log(f"[PolicyStrategy] No policy data available, falling back to DefaultStrategy", OUTPUT_DEBUG)
|
|
return DefaultStrategy(self.game, self.settings).generate_move()
|
|
|
|
policy_moves = self.cn.policy_ranking
|
|
pass_policy = self.cn.policy[-1]
|
|
|
|
self.game.katrain.log(f"[PolicyStrategy] Got {len(policy_moves)} policy moves", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[PolicyStrategy] Current move depth: {self.cn.depth}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[PolicyStrategy] Opening moves setting: {self.settings.get('opening_moves', 0)}", OUTPUT_DEBUG)
|
|
|
|
# Log top 5 policy moves
|
|
self.game.katrain.log(f"[PolicyStrategy] Top 5 policy moves:", OUTPUT_DEBUG)
|
|
for i, (prob, move) in enumerate(policy_moves[:5]):
|
|
self.game.katrain.log(f"[PolicyStrategy] #{i+1}: {move.gtp()} - {prob:.2%}", OUTPUT_DEBUG)
|
|
|
|
self.game.katrain.log(f"[PolicyStrategy] Pass policy: {pass_policy:.2%}", OUTPUT_DEBUG)
|
|
|
|
# Check for pass in top 5
|
|
top_5_pass = any([polmove[1].is_pass for polmove in policy_moves[:5]])
|
|
self.game.katrain.log(f"[PolicyStrategy] Pass in top 5: {top_5_pass}", OUTPUT_DEBUG)
|
|
|
|
# Handle opening moves override
|
|
if self.cn.depth <= self.settings.get("opening_moves", 0):
|
|
self.game.katrain.log(f"[PolicyStrategy] In opening phase, using WeightedStrategy instead", OUTPUT_DEBUG)
|
|
weighted_settings = {
|
|
"pick_override": 0.9,
|
|
"weaken_fac": 1,
|
|
"lower_bound": 0.02
|
|
}
|
|
self.game.katrain.log(f"[PolicyStrategy] Weighted settings: {weighted_settings}", OUTPUT_DEBUG)
|
|
return WeightedStrategy(self.game, weighted_settings).generate_move()
|
|
|
|
# Check for pass in top 5
|
|
if top_5_pass:
|
|
aimove = policy_moves[0][1]
|
|
self.game.katrain.log(f"[PolicyStrategy] Playing top move {aimove.gtp()} because pass in top 5", OUTPUT_DEBUG)
|
|
ai_thoughts = "Playing top one because one of them is pass."
|
|
return aimove, ai_thoughts
|
|
|
|
# Otherwise play top policy move
|
|
aimove = policy_moves[0][1]
|
|
self.game.katrain.log(f"[PolicyStrategy] Playing top policy move {aimove.gtp()} with probability {policy_moves[0][0]:.2%}", OUTPUT_DEBUG)
|
|
ai_thoughts = f"Playing top policy move {aimove.gtp()}."
|
|
|
|
self.game.katrain.log(f"[PolicyStrategy] Final decision: {aimove.gtp()}", OUTPUT_DEBUG)
|
|
return aimove, ai_thoughts
|
|
|
|
@register_strategy(AI_WEIGHTED)
|
|
class WeightedStrategy(AIStrategy):
|
|
"""Weighted strategy - weights moves based on policy and a weakening factor"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[WeightedStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
# Ensure policy is available
|
|
if not self.cn.policy:
|
|
self.game.katrain.log(f"[WeightedStrategy] No policy data available, falling back to DefaultStrategy", OUTPUT_DEBUG)
|
|
return DefaultStrategy(self.game, self.settings).generate_move()
|
|
|
|
policy_moves = self.cn.policy_ranking
|
|
pass_policy = self.cn.policy[-1]
|
|
|
|
self.game.katrain.log(f"[WeightedStrategy] Got {len(policy_moves)} policy moves", OUTPUT_DEBUG)
|
|
|
|
# Log top 5 policy moves
|
|
self.game.katrain.log(f"[WeightedStrategy] Top 5 policy moves:", OUTPUT_DEBUG)
|
|
for i, (prob, move) in enumerate(policy_moves[:5]):
|
|
self.game.katrain.log(f"[WeightedStrategy] #{i+1}: {move.gtp()} - {prob:.2%}", OUTPUT_DEBUG)
|
|
|
|
self.game.katrain.log(f"[WeightedStrategy] Pass policy: {pass_policy:.2%}", OUTPUT_DEBUG)
|
|
|
|
# Check for pass in top 5
|
|
top_5_pass = any([polmove[1].is_pass for polmove in policy_moves[:5]])
|
|
self.game.katrain.log(f"[WeightedStrategy] Pass in top 5: {top_5_pass}", OUTPUT_DEBUG)
|
|
|
|
# Get override threshold
|
|
override = self.settings.get("pick_override", 0.0)
|
|
self.game.katrain.log(f"[WeightedStrategy] Override threshold: {override:.2%}", OUTPUT_DEBUG)
|
|
|
|
# Check if we should override with top move
|
|
override_move, override_thoughts = self.should_play_top_move(
|
|
policy_moves,
|
|
top_5_pass,
|
|
override=override
|
|
)
|
|
|
|
if override_move:
|
|
self.game.katrain.log(f"[WeightedStrategy] Using override move: {override_move.gtp()}", OUTPUT_DEBUG)
|
|
return override_move, override_thoughts
|
|
|
|
# Apply weighted policy move selection
|
|
lower_bound = self.settings.get("lower_bound", 0.02)
|
|
weaken_fac = self.settings.get("weaken_fac", 1.0)
|
|
|
|
self.game.katrain.log(f"[WeightedStrategy] Using weighted selection with lower_bound={lower_bound:.2%}, weaken_fac={weaken_fac}", OUTPUT_DEBUG)
|
|
|
|
# Generate list of weighted coordinates
|
|
weighted_coords = [
|
|
(pv, pv ** (1 / weaken_fac), move) for pv, move in policy_moves if pv > lower_bound and not move.is_pass
|
|
]
|
|
|
|
self.game.katrain.log(f"[WeightedStrategy] Found {len(weighted_coords)} moves above lower bound", OUTPUT_DEBUG)
|
|
|
|
if weighted_coords:
|
|
self.game.katrain.log(f"[WeightedStrategy] Performing weighted selection", OUTPUT_DEBUG)
|
|
top = weighted_selection_without_replacement(weighted_coords, 1)[0]
|
|
move = top[2]
|
|
prob = top[0]
|
|
|
|
self.game.katrain.log(f"[WeightedStrategy] Selected move {move.gtp()} with probability {prob:.2%}", OUTPUT_DEBUG)
|
|
ai_thoughts = f"Playing policy-weighted random move {move.gtp()} ({prob:.1%}) from {len(weighted_coords)} moves above lower_bound of {lower_bound:.1%}."
|
|
else:
|
|
move = policy_moves[0][1]
|
|
self.game.katrain.log(f"[WeightedStrategy] No moves above lower bound, playing top policy move {move.gtp()}", OUTPUT_DEBUG)
|
|
ai_thoughts = f"Playing top policy move because no non-pass move > above lower_bound of {lower_bound:.1%}."
|
|
|
|
self.game.katrain.log(f"[WeightedStrategy] Final decision: {move.gtp()}", OUTPUT_DEBUG)
|
|
return move, ai_thoughts
|
|
|
|
class PickBasedStrategy(AIStrategy):
|
|
"""Base class for pick-based strategies"""
|
|
|
|
def get_n_moves(self, legal_policy_moves):
|
|
"""Calculate the number of moves to consider"""
|
|
board_squares = self.game.board_size[0] * self.game.board_size[1]
|
|
|
|
if self.settings.get("pick_frac") is not None:
|
|
n_moves = max(1, int(self.settings["pick_frac"] * len(legal_policy_moves) + self.settings["pick_n"]))
|
|
self.game.katrain.log(f"[{self.strategy_name}] Calculated n_moves={n_moves} from pick_frac={self.settings['pick_frac']}, pick_n={self.settings['pick_n']}, legal_moves={len(legal_policy_moves)}", OUTPUT_DEBUG)
|
|
else:
|
|
n_moves = 1 # Default
|
|
self.game.katrain.log(f"[{self.strategy_name}] Using default n_moves={n_moves} (no pick_frac in settings)", OUTPUT_DEBUG)
|
|
|
|
return n_moves
|
|
|
|
def generate_weighted_coords(self, legal_policy_moves, policy_grid, size):
|
|
"""Generate weighted coordinates for selection"""
|
|
self.game.katrain.log(f"[{self.strategy_name}] Generating weighted coordinates (default equal weights implementation)", OUTPUT_DEBUG)
|
|
|
|
# Default implementation for AI_PICK - equal weights
|
|
weighted_coords = [
|
|
(policy_grid[y][x], 1, x, y)
|
|
for x in range(size[0])
|
|
for y in range(size[1])
|
|
if policy_grid[y][x] > 0
|
|
]
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Generated {len(weighted_coords)} weighted coordinates", OUTPUT_DEBUG)
|
|
|
|
if weighted_coords:
|
|
top5 = heapq.nlargest(5, weighted_coords, key=lambda t: t[0])
|
|
self.game.katrain.log(f"[{self.strategy_name}] Top 5 weighted coordinates by policy value:", OUTPUT_DEBUG)
|
|
for i, (pol, wt, x, y) in enumerate(top5):
|
|
self.game.katrain.log(f"[{self.strategy_name}] #{i+1}: ({x},{y}) - policy={pol:.2%}, weight={wt}", OUTPUT_DEBUG)
|
|
|
|
return weighted_coords, "Generated equal weights for all moves. "
|
|
|
|
def handle_endgame(self, legal_policy_moves, policy_grid, size):
|
|
"""Handle special endgame case"""
|
|
board_squares = size[0] * size[1]
|
|
endgame_threshold = self.settings.get("endgame", 0.75) * board_squares
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Checking endgame condition: move depth {self.cn.depth} vs threshold {endgame_threshold}", OUTPUT_DEBUG)
|
|
|
|
if self.cn.depth > endgame_threshold:
|
|
self.game.katrain.log(f"[{self.strategy_name}] In endgame phase (move {self.cn.depth} > {endgame_threshold})", OUTPUT_DEBUG)
|
|
|
|
weighted_coords = [(pol, 1, *mv.coords) for pol, mv in legal_policy_moves]
|
|
ai_thoughts = f"Generated equal weights as move number >= {self.settings['endgame'] * size[0] * size[1]}. "
|
|
|
|
n_moves = int(max(self.get_n_moves(legal_policy_moves), len(legal_policy_moves) // 2))
|
|
self.game.katrain.log(f"[{self.strategy_name}] Using endgame n_moves={n_moves}", OUTPUT_DEBUG)
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Generated {len(weighted_coords)} weighted coordinates for endgame", OUTPUT_DEBUG)
|
|
|
|
return weighted_coords, ai_thoughts, n_moves, True
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Not in endgame phase yet", OUTPUT_DEBUG)
|
|
return None, "", None, False
|
|
|
|
def select_from_weighted_coords(self, weighted_coords, n_moves, pass_policy):
|
|
"""Select moves from weighted coordinates"""
|
|
self.game.katrain.log(f"[{self.strategy_name}] Selecting from {len(weighted_coords)} weighted coordinates, n_moves={n_moves}", OUTPUT_DEBUG)
|
|
|
|
# Perform weighted selection
|
|
pick_moves = weighted_selection_without_replacement(weighted_coords, n_moves)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Picked {len(pick_moves)} moves", OUTPUT_DEBUG)
|
|
|
|
if pick_moves:
|
|
# Get top 5 from picked moves
|
|
top_picked = heapq.nlargest(5, pick_moves)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Top 5 after selection:", OUTPUT_DEBUG)
|
|
for i, (p, wt, x, y) in enumerate(top_picked):
|
|
self.game.katrain.log(f"[{self.strategy_name}] #{i+1}: ({x},{y}) - policy={p:.2%}, weight={wt}", OUTPUT_DEBUG)
|
|
|
|
# Convert to move objects
|
|
new_top = [
|
|
(p, Move((x, y), player=self.cn.next_player)) for p, wt, x, y in top_picked
|
|
]
|
|
|
|
aimove = new_top[0][1]
|
|
ai_thoughts = f"Top 5 among these were {fmt_moves(new_top)} and picked top {aimove.gtp()}. "
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Top picked move: {aimove.gtp()} ({new_top[0][0]:.2%})", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Pass policy: {pass_policy:.2%}", OUTPUT_DEBUG)
|
|
|
|
# Check if pass is better
|
|
if new_top[0][0] < pass_policy:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Pass policy {pass_policy:.2%} is better than top move {aimove.gtp()} ({new_top[0][0]:.2%}), switching to top policy move", OUTPUT_DEBUG)
|
|
|
|
policy_moves = self.cn.policy_ranking
|
|
top_policy_move = policy_moves[0][1]
|
|
|
|
ai_thoughts += f"But found pass ({pass_policy:.2%} to be higher rated than {aimove.gtp()} ({new_top[0][0]:.2%}) so will play top policy move instead."
|
|
aimove = top_policy_move
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Final move (after pass check): {aimove.gtp()}", OUTPUT_DEBUG)
|
|
else:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Top move is better than pass, keeping it", OUTPUT_DEBUG)
|
|
else:
|
|
self.game.katrain.log(f"[{self.strategy_name}] No moves selected, falling back to top policy move", OUTPUT_DEBUG)
|
|
|
|
policy_moves = self.cn.policy_ranking
|
|
top_policy_move = policy_moves[0][1]
|
|
aimove = top_policy_move
|
|
|
|
ai_thoughts = f"Pick policy strategy failed to find legal moves, so is playing top policy move {aimove.gtp()}."
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Final move (fallback): {aimove.gtp()}", OUTPUT_DEBUG)
|
|
|
|
return aimove, ai_thoughts
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Starting move generation", OUTPUT_DEBUG)
|
|
self.wait_for_analysis()
|
|
|
|
# Ensure policy is available
|
|
if not self.cn.policy:
|
|
self.game.katrain.log(f"[{self.strategy_name}] No policy data available, falling back to DefaultStrategy", OUTPUT_DEBUG)
|
|
return DefaultStrategy(self.game, self.settings).generate_move()
|
|
|
|
policy_moves = self.cn.policy_ranking
|
|
pass_policy = self.cn.policy[-1]
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Got {len(policy_moves)} policy moves", OUTPUT_DEBUG)
|
|
|
|
# Log top 5 policy moves
|
|
self.game.katrain.log(f"[{self.strategy_name}] Top 5 policy moves:", OUTPUT_DEBUG)
|
|
for i, (prob, move) in enumerate(policy_moves[:5]):
|
|
self.game.katrain.log(f"[{self.strategy_name}] #{i+1}: {move.gtp()} - {prob:.2%}", OUTPUT_DEBUG)
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Pass policy: {pass_policy:.2%}", OUTPUT_DEBUG)
|
|
|
|
# Check for pass in top 5
|
|
top_5_pass = any([polmove[1].is_pass for polmove in policy_moves[:5]])
|
|
self.game.katrain.log(f"[{self.strategy_name}] Pass in top 5: {top_5_pass}", OUTPUT_DEBUG)
|
|
|
|
# Get override settings
|
|
override = self.settings.get("pick_override", 0.0)
|
|
overridetwo = self.settings.get("pick_override_two", 1.0)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Override settings: single={override:.2%}, combined={overridetwo:.2%}", OUTPUT_DEBUG)
|
|
|
|
# Check if we should override with top move
|
|
override_move, override_thoughts = self.should_play_top_move(
|
|
policy_moves,
|
|
top_5_pass,
|
|
override=override,
|
|
overridetwo=overridetwo
|
|
)
|
|
|
|
if override_move:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Using override move: {override_move.gtp()}", OUTPUT_DEBUG)
|
|
return override_move, override_thoughts
|
|
|
|
# Get legal policy moves
|
|
legal_policy_moves = [(pol, mv) for pol, mv in policy_moves if not mv.is_pass and pol > 0]
|
|
self.game.katrain.log(f"[{self.strategy_name}] Found {len(legal_policy_moves)} legal non-pass policy moves", OUTPUT_DEBUG)
|
|
|
|
# Create policy grid
|
|
# Create policy grid
|
|
size = self.game.board_size
|
|
self.game.katrain.log(f"[{self.strategy_name}] Board size: {size}", OUTPUT_DEBUG)
|
|
policy_grid = var_to_grid(self.cn.policy, size)
|
|
|
|
# Check for endgame
|
|
end_coords, end_thoughts, end_n_moves, is_endgame = self.handle_endgame(legal_policy_moves, policy_grid, size)
|
|
|
|
if is_endgame:
|
|
self.game.katrain.log(f"[{self.strategy_name}] Using endgame logic", OUTPUT_DEBUG)
|
|
return self.select_from_weighted_coords(end_coords, end_n_moves, pass_policy)
|
|
|
|
# Get weighted coordinates
|
|
self.game.katrain.log(f"[{self.strategy_name}] Generating weighted coordinates", OUTPUT_DEBUG)
|
|
weighted_coords, weight_thoughts = self.generate_weighted_coords(legal_policy_moves, policy_grid, size)
|
|
|
|
# Get number of moves to consider
|
|
n_moves = self.get_n_moves(legal_policy_moves)
|
|
self.game.katrain.log(f"[{self.strategy_name}] Using n_moves={n_moves}", OUTPUT_DEBUG)
|
|
|
|
ai_thoughts = weight_thoughts + f"Picked {min(n_moves, len(weighted_coords))} random moves according to weights. "
|
|
|
|
# Select and return move
|
|
self.game.katrain.log(f"[{self.strategy_name}] Selecting move from weighted coordinates", OUTPUT_DEBUG)
|
|
move, thoughts = self.select_from_weighted_coords(weighted_coords, n_moves, pass_policy)
|
|
|
|
self.game.katrain.log(f"[{self.strategy_name}] Final decision: {move.gtp()}", OUTPUT_DEBUG)
|
|
return move, ai_thoughts + thoughts
|
|
|
|
@register_strategy(AI_PICK)
|
|
class PickStrategy(PickBasedStrategy):
|
|
"""Pick strategy - picks a move from a subset of legal moves"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[PickStrategy] Starting move generation using base PickBasedStrategy implementation", OUTPUT_DEBUG)
|
|
return super().generate_move()
|
|
|
|
def handle_endgame(self, legal_policy_moves, policy_grid, size):
|
|
return None, "", None, False
|
|
|
|
@register_strategy(AI_RANK)
|
|
class RankStrategy(PickBasedStrategy):
|
|
"""Rank strategy - similar to Pick but calibrated based on rank"""
|
|
|
|
def get_n_moves(self, legal_policy_moves):
|
|
"""Calculate n_moves based on rank"""
|
|
self.game.katrain.log(f"[RankStrategy] Calculating n_moves based on rank", OUTPUT_DEBUG)
|
|
|
|
size = self.game.board_size
|
|
board_squares = size[0] * size[1]
|
|
norm_leg_moves = len(legal_policy_moves) / board_squares
|
|
|
|
self.game.katrain.log(f"[RankStrategy] Board squares: {board_squares}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[RankStrategy] Legal moves: {len(legal_policy_moves)}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[RankStrategy] Normalized legal moves: {norm_leg_moves:.4f}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[RankStrategy] Kyu rank: {self.settings['kyu_rank']}", OUTPUT_DEBUG)
|
|
|
|
# Calculate n_moves using the rank formula
|
|
orig_calib_avemodrank = 0.063015 + 0.7624 * board_squares / (
|
|
10 ** (-0.05737 * self.settings["kyu_rank"] + 1.9482)
|
|
)
|
|
|
|
self.game.katrain.log(f"[RankStrategy] Original calibrated average mod rank: {orig_calib_avemodrank:.4f}", OUTPUT_DEBUG)
|
|
|
|
exponent_term = (
|
|
3.002 * norm_leg_moves * norm_leg_moves
|
|
- norm_leg_moves
|
|
- 0.034889 * self.settings["kyu_rank"]
|
|
- 0.5097
|
|
)
|
|
self.game.katrain.log(f"[RankStrategy] Exponent term: {exponent_term:.4f}", OUTPUT_DEBUG)
|
|
|
|
modified_calib_avemodrank = (
|
|
0.3931
|
|
+ 0.6559
|
|
* norm_leg_moves
|
|
* math.exp(-1 * exponent_term ** 2)
|
|
- 0.01093 * self.settings["kyu_rank"]
|
|
) * orig_calib_avemodrank
|
|
|
|
self.game.katrain.log(f"[RankStrategy] Modified calibrated average mod rank: {modified_calib_avemodrank:.4f}", OUTPUT_DEBUG)
|
|
|
|
denominator = 1.31165 * (modified_calib_avemodrank + 1) - 0.082653
|
|
self.game.katrain.log(f"[RankStrategy] Denominator: {denominator:.4f}", OUTPUT_DEBUG)
|
|
|
|
n_moves = board_squares * norm_leg_moves / denominator
|
|
n_moves = max(1, round(n_moves))
|
|
|
|
self.game.katrain.log(f"[RankStrategy] Calculated n_moves: {n_moves}", OUTPUT_DEBUG)
|
|
|
|
return n_moves
|
|
|
|
def should_play_top_move(self, policy_moves, top_5_pass, override=0.0, overridetwo=1.0):
|
|
"""Special override logic for rank-based"""
|
|
self.game.katrain.log(f"[RankStrategy] Calculating special override thresholds based on rank", OUTPUT_DEBUG)
|
|
|
|
size = self.game.board_size
|
|
board_squares = size[0] * size[1]
|
|
legal_policy_moves = [(pol, mv) for pol, mv in policy_moves if not mv.is_pass and pol > 0]
|
|
|
|
# Parameters for calculating the overrides
|
|
self.game.katrain.log(f"[RankStrategy] Board squares: {board_squares}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[RankStrategy] Legal non-pass moves: {len(legal_policy_moves)}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[RankStrategy] Kyu rank: {self.settings['kyu_rank']}", OUTPUT_DEBUG)
|
|
|
|
# Calibrated override based on board filling
|
|
ratio = (board_squares - len(legal_policy_moves)) / board_squares
|
|
override = 0.8 * (1 - 0.5 * ratio)
|
|
self.game.katrain.log(f"[RankStrategy] Calculated override: {override:.2%} (from board filling ratio {ratio:.2f})", OUTPUT_DEBUG)
|
|
|
|
overridetwo = 0.85 + max(0, 0.02 * (self.settings["kyu_rank"] - 8))
|
|
self.game.katrain.log(f"[RankStrategy] Calculated overridetwo: {overridetwo:.2%} (from kyu rank adjustment)", OUTPUT_DEBUG)
|
|
|
|
# Call the parent class method with calculated overrides
|
|
return super().should_play_top_move(policy_moves, top_5_pass, override, overridetwo)
|
|
|
|
def handle_endgame(self, legal_policy_moves, policy_grid, size):
|
|
return None, "", None, False
|
|
|
|
@register_strategy(AI_INFLUENCE)
|
|
class InfluenceStrategy(PickBasedStrategy):
|
|
"""Influence strategy - weights moves based on influence (distance from edge)"""
|
|
|
|
def generate_weighted_coords(self, legal_policy_moves, policy_grid, size):
|
|
"""Generate influence-based weights"""
|
|
self.game.katrain.log(f"[InfluenceStrategy] Generating influence-based weights", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[InfluenceStrategy] Settings: threshold={self.settings['threshold']}, line_weight={self.settings['line_weight']}", OUTPUT_DEBUG)
|
|
weighted_coords, ai_thoughts = generate_influence_territory_weights(
|
|
AI_INFLUENCE,
|
|
self.settings,
|
|
policy_grid,
|
|
size
|
|
)
|
|
self.game.katrain.log(f"[InfluenceStrategy] Generated {len(weighted_coords)} weighted coordinates", OUTPUT_DEBUG)
|
|
if weighted_coords:
|
|
top5 = heapq.nlargest(5, weighted_coords, key=lambda t: t[0] * t[1])
|
|
self.game.katrain.log(f"[InfluenceStrategy] Top 5 weighted coordinates (by policy*weight):", OUTPUT_DEBUG)
|
|
for i, (pol, wt, x, y) in enumerate(top5):
|
|
self.game.katrain.log(f"[InfluenceStrategy] #{i+1}: ({x},{y}) - policy={pol:.2%}, weight={wt}, combined={pol*wt:.2%}", OUTPUT_DEBUG)
|
|
return weighted_coords, ai_thoughts
|
|
|
|
@register_strategy(AI_TERRITORY)
|
|
class TerritoryStrategy(PickBasedStrategy):
|
|
"""Territory strategy - weights moves based on territory (distance from center)"""
|
|
|
|
def generate_weighted_coords(self, legal_policy_moves, policy_grid, size):
|
|
"""Generate territory-based weights"""
|
|
self.game.katrain.log(f"[TerritoryStrategy] Generating territory-based weights", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[TerritoryStrategy] Settings: threshold={self.settings['threshold']}, line_weight={self.settings['line_weight']}", OUTPUT_DEBUG)
|
|
weighted_coords, ai_thoughts = generate_influence_territory_weights(
|
|
AI_TERRITORY,
|
|
self.settings,
|
|
policy_grid,
|
|
size
|
|
)
|
|
self.game.katrain.log(f"[TerritoryStrategy] Generated {len(weighted_coords)} weighted coordinates", OUTPUT_DEBUG)
|
|
if weighted_coords:
|
|
top5 = heapq.nlargest(5, weighted_coords, key=lambda t: t[0] * t[1])
|
|
self.game.katrain.log(f"[TerritoryStrategy] Top 5 weighted coordinates (by policy*weight):", OUTPUT_DEBUG)
|
|
for i, (pol, wt, x, y) in enumerate(top5):
|
|
self.game.katrain.log(f"[TerritoryStrategy] #{i+1}: ({x},{y}) - policy={pol:.2%}, weight={wt}, combined={pol*wt:.2%}", OUTPUT_DEBUG)
|
|
return weighted_coords, ai_thoughts
|
|
|
|
@register_strategy(AI_LOCAL)
|
|
class LocalStrategy(PickBasedStrategy):
|
|
"""Local strategy - weights moves based on proximity to the last move"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
# Handle the case where there's no previous move
|
|
if not (self.cn.move and self.cn.move.coords):
|
|
self.game.katrain.log(f"[LocalStrategy] No previous move with valid coordinates found, falling back to WeightedStrategy", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[LocalStrategy] Using default weighted settings: pick_override=0.9, weaken_fac=1, lower_bound=0.02", OUTPUT_DEBUG)
|
|
return WeightedStrategy(self.game, {
|
|
"pick_override": 0.9,
|
|
"weaken_fac": 1,
|
|
"lower_bound": 0.02
|
|
}).generate_move()
|
|
|
|
return super().generate_move()
|
|
|
|
def generate_weighted_coords(self, legal_policy_moves, policy_grid, size):
|
|
"""Generate local-based weights"""
|
|
self.game.katrain.log(f"[LocalStrategy] Generating local-based weights around previous move", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[LocalStrategy] Previous move: {self.cn.move.gtp()}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[LocalStrategy] Variance setting: {self.settings['stddev']}", OUTPUT_DEBUG)
|
|
weighted_coords, ai_thoughts = generate_local_tenuki_weights(
|
|
AI_LOCAL,
|
|
self.settings,
|
|
policy_grid,
|
|
self.cn,
|
|
size
|
|
)
|
|
self.game.katrain.log(f"[LocalStrategy] Generated {len(weighted_coords)} weighted coordinates", OUTPUT_DEBUG)
|
|
if weighted_coords:
|
|
top5 = heapq.nlargest(5, weighted_coords, key=lambda t: t[0] * t[1])
|
|
self.game.katrain.log(f"[LocalStrategy] Top 5 weighted coordinates (by policy*weight):", OUTPUT_DEBUG)
|
|
for i, (pol, wt, x, y) in enumerate(top5):
|
|
self.game.katrain.log(f"[LocalStrategy] #{i+1}: ({x},{y}) - policy={pol:.2%}, weight={wt}, combined={pol*wt:.2%}", OUTPUT_DEBUG)
|
|
return weighted_coords, ai_thoughts
|
|
|
|
@register_strategy(AI_TENUKI)
|
|
class TenukiStrategy(PickBasedStrategy):
|
|
"""Tenuki strategy - weights moves based on distance from the last move"""
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
# Handle the case where there's no previous move
|
|
if not (self.cn.move and self.cn.move.coords):
|
|
self.game.katrain.log(f"[TenukiStrategy] No previous move with valid coordinates found, falling back to WeightedStrategy", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[TenukiStrategy] Using default weighted settings: pick_override=0.9, weaken_fac=1, lower_bound=0.02", OUTPUT_DEBUG)
|
|
return WeightedStrategy(self.game, {
|
|
"pick_override": 0.9,
|
|
"weaken_fac": 1,
|
|
"lower_bound": 0.02
|
|
}).generate_move()
|
|
|
|
return super().generate_move()
|
|
|
|
def generate_weighted_coords(self, legal_policy_moves, policy_grid, size):
|
|
"""Generate tenuki-based weights"""
|
|
self.game.katrain.log(f"[TenukiStrategy] Generating tenuki-based weights (far from previous move)", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[TenukiStrategy] Previous move: {self.cn.move.gtp()}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[TenukiStrategy] Variance setting: {self.settings['stddev']}", OUTPUT_DEBUG)
|
|
weighted_coords, ai_thoughts = generate_local_tenuki_weights(
|
|
AI_TENUKI,
|
|
self.settings,
|
|
policy_grid,
|
|
self.cn,
|
|
size
|
|
)
|
|
self.game.katrain.log(f"[TenukiStrategy] Generated {len(weighted_coords)} weighted coordinates", OUTPUT_DEBUG)
|
|
if weighted_coords:
|
|
top5 = heapq.nlargest(5, weighted_coords, key=lambda t: t[0] * t[1])
|
|
self.game.katrain.log(f"[TenukiStrategy] Top 5 weighted coordinates (by policy*weight):", OUTPUT_DEBUG)
|
|
for i, (pol, wt, x, y) in enumerate(top5):
|
|
self.game.katrain.log(f"[TenukiStrategy] #{i+1}: ({x},{y}) - policy={pol:.2%}, weight={wt}, combined={pol*wt:.2%}", OUTPUT_DEBUG)
|
|
return weighted_coords, ai_thoughts
|
|
|
|
@register_strategy(AI_HUMAN)
|
|
@register_strategy(AI_PRO)
|
|
class HumanStyleStrategy(AIStrategy):
|
|
"""Strategy that imitates human play at various skill levels"""
|
|
|
|
def __init__(self, game: Game, ai_settings: Dict):
|
|
super().__init__(game, ai_settings)
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Initializing HumanStyleStrategy", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HumanStyleStrategy] AI settings: {ai_settings}", OUTPUT_DEBUG)
|
|
|
|
def generate_move(self) -> Tuple[Move, str]:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Starting move generation", OUTPUT_DEBUG)
|
|
|
|
if "human_kyu_rank" in self.settings:
|
|
human_kyu_rank = round(self.settings["human_kyu_rank"])
|
|
human_style = "rank" if self.settings["modern_style"] else "preaz"
|
|
|
|
if human_kyu_rank <= 0: # dan ranks
|
|
rank_text = f"{1-human_kyu_rank}d"
|
|
else: # kyu ranks
|
|
rank_text = f"{human_kyu_rank}k"
|
|
|
|
human_profile = f"{human_style}_{rank_text}"
|
|
else:
|
|
pro_year = round(self.settings["pro_year"])
|
|
human_profile = f"proyear_{pro_year}"
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Human profile string: {human_profile}", OUTPUT_DEBUG)
|
|
|
|
# Define override settings (separate from includePolicy)
|
|
override_settings = {
|
|
"humanSLProfile": human_profile,
|
|
"ignorePreRootHistory": False,
|
|
}
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Override settings for engine: {override_settings}", OUTPUT_DEBUG)
|
|
|
|
# Request analysis from engine - note includePolicy is a direct parameter
|
|
analysis = None
|
|
|
|
def set_analysis(a, partial_result):
|
|
nonlocal analysis
|
|
if not partial_result:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Full analysis results received", OUTPUT_DEBUG)
|
|
analysis = a
|
|
# Log some analysis stats for debugging
|
|
if a:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Analysis contains humanPolicy: {'humanPolicy' in a}", OUTPUT_DEBUG)
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Analysis contains moveInfos: {len(a.get('moveInfos', []))} moves", OUTPUT_DEBUG)
|
|
if 'humanPolicy' in a:
|
|
policy_sum = sum(a['humanPolicy'])
|
|
policy_max = max(a['humanPolicy'])
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Human policy sum: {policy_sum}, max: {policy_max}", OUTPUT_DEBUG)
|
|
else:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Received partial analysis results - ignoring", OUTPUT_DEBUG)
|
|
|
|
def set_error(a):
|
|
nonlocal error
|
|
error = True
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Error in human analysis query: {a}", OUTPUT_ERROR)
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Will attempt to fall back to policy move", OUTPUT_DEBUG)
|
|
|
|
error = False
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Getting engine for player", OUTPUT_DEBUG)
|
|
engine = self.game.engines[self.cn.player]
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Using engine for player {self.cn.player}", OUTPUT_DEBUG)
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Requesting analysis with human profile settings", OUTPUT_DEBUG)
|
|
engine.request_analysis(
|
|
self.cn,
|
|
callback=set_analysis,
|
|
error_callback=set_error,
|
|
priority=PRIORITY_EXTRA_AI_QUERY,
|
|
include_policy=True,
|
|
extra_settings=override_settings
|
|
)
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Analysis request sent, waiting for results", OUTPUT_DEBUG)
|
|
|
|
# Wait for analysis to complete
|
|
wait_count = 0
|
|
while not (error or analysis):
|
|
import time
|
|
time.sleep(0.01)
|
|
wait_count += 1
|
|
if wait_count % 100 == 0: # Log every 1 second
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Still waiting for analysis results ({wait_count/100:.1f}s)", OUTPUT_DEBUG)
|
|
engine.check_alive(exception_if_dead=True)
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Finished waiting for analysis, error={error}, analysis received={analysis is not None}", OUTPUT_DEBUG)
|
|
|
|
if error or not analysis:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Analysis failed or returned empty", OUTPUT_DEBUG)
|
|
# Fall back to policy
|
|
policy_move = self.cn.policy_ranking[0][1] if self.cn.policy_ranking else None
|
|
if policy_move:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Falling back to top policy move: {policy_move.gtp()}", OUTPUT_DEBUG)
|
|
return policy_move, "Falling back to policy move due to error in human analysis."
|
|
else:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] No policy moves available for fallback - will return pass", OUTPUT_DEBUG)
|
|
return Move(None, player=self.cn.next_player), "No valid moves found."
|
|
|
|
# Check if human policy is available
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Processing analysis results", OUTPUT_DEBUG)
|
|
if "humanPolicy" not in analysis:
|
|
error_msg = "humanPolicy not found in analysis—have you downloaded and configured your human model yet?"
|
|
raise Exception(error_msg)
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Human policy found in analysis", OUTPUT_DEBUG)
|
|
board_size = self.game.board_size
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Board size: {board_size}", OUTPUT_DEBUG)
|
|
human_policy = analysis["humanPolicy"]
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Human policy length: {len(human_policy)}", OUTPUT_DEBUG)
|
|
if len(human_policy) != 362:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] WARNING: Human policy length {len(human_policy)} != 362", OUTPUT_ERROR)
|
|
|
|
# Create a list of moves with their human policy weights
|
|
moves = []
|
|
for x in range(board_size[0]):
|
|
for y in range(board_size[1]):
|
|
idx = (board_size[1] - y - 1) * board_size[0] + x
|
|
if idx < len(human_policy) and human_policy[idx] > 0:
|
|
moves.append((Move((x, y), player=self.cn.next_player), human_policy[idx]))
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Generated {len(moves)} candidate moves from human policy", OUTPUT_DEBUG)
|
|
|
|
# Add pass move if it has positive probability
|
|
if len(human_policy) > board_size[0] * board_size[1] and human_policy[-1] > 0:
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Adding pass move with probability {human_policy[-1]}", OUTPUT_DEBUG)
|
|
moves.append((Move(None, player=self.cn.next_player), human_policy[-1]))
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Performing weighted selection from {len(moves)} moves", OUTPUT_DEBUG)
|
|
top_moves = sorted(moves, key=lambda x: -x[1])
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Top 5 moves by probability:", OUTPUT_DEBUG)
|
|
|
|
# Create a formatted string of top 5 moves for ai_thoughts
|
|
top_moves_str = "\n".join([f"#{i+1}: {move.gtp()} - {prob:.1%}" for i, (move, prob) in enumerate(top_moves[:5])])
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy]\n{top_moves_str}", OUTPUT_DEBUG)
|
|
|
|
selected = weighted_selection_without_replacement(moves, 1)[0]
|
|
move = selected[0]
|
|
prob = selected[1]
|
|
|
|
# Find the rank of the selected move
|
|
selected_rank = next((i+1 for i, (m, _) in enumerate(top_moves) if m.gtp() == move.gtp()), "ERROR: move not found in ranking")
|
|
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Selected move {move.gtp()} with probability {prob:.4f}", OUTPUT_DEBUG)
|
|
ai_thoughts = f"\n{top_moves_str}\n\nPlayed move {move.gtp()} ({prob:.1%}) as the #{selected_rank} top move."
|
|
self.game.katrain.log(f"[HumanStyleStrategy] Final decision: {move.gtp()}", OUTPUT_DEBUG)
|
|
return move, ai_thoughts
|
|
|
|
def generate_ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode]:
|
|
"""Generate a move using the selected AI strategy"""
|
|
game.katrain.log(f"Generate AI move called with mode: {ai_mode}", OUTPUT_DEBUG)
|
|
|
|
# Create the appropriate strategy based on mode
|
|
|
|
strategy = STRATEGY_REGISTRY[ai_mode](game, ai_settings)
|
|
|
|
# Generate the move
|
|
game.katrain.log(f"Generating move using {strategy.__class__.__name__}", OUTPUT_DEBUG)
|
|
move, ai_thoughts = strategy.generate_move()
|
|
|
|
# Play the move and return
|
|
game.katrain.log(f"Playing move {move.gtp()} and creating game node", OUTPUT_DEBUG)
|
|
played_node = game.play(move)
|
|
game.katrain.log(f"AI thoughts: {ai_thoughts}", OUTPUT_DEBUG)
|
|
played_node.ai_thoughts = ai_thoughts
|
|
|
|
game.katrain.log(f"Move generation complete: {move.gtp()}", OUTPUT_DEBUG)
|
|
return move, played_node |