ai selfplay, black 120

This commit is contained in:
Sander Land committed 2020-05-30 23:33:10 +02:00
1 parent 06cc316326
commit 9c6d521a88
16 files changed
+565 -133

No files matched your search

+10 -2
View File
@@ -37,8 +37,16 @@ for lang in locales:
for lang in locales:
if lang != DEFAULT_LANG:
for msgid in lang_to_strings[lang]:
if DEFAULT_LANG in strings_to_keys[msgid] and strings_to_keys[msgid][lang] != strings_to_keys[msgid][DEFAULT_LANG]:
print(f"{msgid} has inconstent formatting keys for {lang}: ", strings_to_keys[msgid][lang], "is different from default", strings_to_keys[msgid][DEFAULT_LANG])
if (
DEFAULT_LANG in strings_to_keys[msgid]
and strings_to_keys[msgid][lang] != strings_to_keys[msgid][DEFAULT_LANG]
):
print(
f"{msgid} has inconstent formatting keys for {lang}: ",
strings_to_keys[msgid][lang],
"is different from default",
strings_to_keys[msgid][DEFAULT_LANG],
)
errors = True
for msgid in strings_to_langs.keys() - lang_to_strings[lang]:
+55 -14
View File
@@ -27,7 +27,16 @@ from katrain.core.utils import (
DEFAULT_LANGUAGE,
find_package_resource,
)
from katrain.core.constants import OUTPUT_ERROR, OUTPUT_KATAGO_STDERR, OUTPUT_INFO, OUTPUT_DEBUG, OUTPUT_EXTRA_DEBUG, MODE_PLAY, HOMEPAGE, VERSION
from katrain.core.constants import (
OUTPUT_ERROR,
OUTPUT_KATAGO_STDERR,
OUTPUT_INFO,
OUTPUT_DEBUG,
OUTPUT_EXTRA_DEBUG,
MODE_PLAY,
HOMEPAGE,
VERSION,
)
from katrain.gui.popups import ConfigTeacherPopup, ConfigTimerPopup, I18NPopup
from katrain.core.base_katrain import KaTrainBase
from katrain.core.engine import KataGoEngine
@@ -42,7 +51,6 @@ from katrain.gui.popups import ConfigPopup, LoadSGFPopup, NewGamePopup, AIPopup
from katrain.gui.style import ENGINE_BUSY_COL, ENGINE_DOWN_COL, ENGINE_READY_COL
class KaTrainGui(Screen, KaTrainBase):
"""Top level class responsible for tying everything together"""
@@ -75,7 +83,9 @@ class KaTrainGui(Screen, KaTrainBase):
return
if "ready" in message.lower():
self.controls.set_status(f"KataGo engine ready.")
if (level == OUTPUT_ERROR or (level == OUTPUT_KATAGO_STDERR and "error" in message.lower())) and getattr(self, "controls", None):
if (level == OUTPUT_ERROR or (level == OUTPUT_KATAGO_STDERR and "error" in message.lower())) and getattr(
self, "controls", None
):
self.controls.set_status(f"ERROR: {message}")
@property
@@ -90,16 +100,33 @@ class KaTrainGui(Screen, KaTrainBase):
threading.Thread(target=self._message_loop_thread, daemon=True).start()
self._do_new_game()
def update_state(self, redraw_board=False): # is called after every message and on receiving analyses and config changes
def update_state(
self, redraw_board=False
): # is called after every message and on receiving analyses and config changes
# AI and Trainer/auto-undo handlers
cn = self.game.current_node
last_player, next_player = self.players_info[cn.player], self.players_info[cn.next_player]
if self.play_analyze_mode == MODE_PLAY:
teaching_undo = cn.player and last_player.being_taught
if teaching_undo and cn.analysis_ready and cn.parent and cn.parent.analysis_ready and not cn.children and not self.game.ended:
if (
teaching_undo
and cn.analysis_ready
and cn.parent
and cn.parent.analysis_ready
and not cn.children
and not self.game.ended
):
self.game.analyze_undo(cn) # not via message loop
if cn.analysis_ready and next_player.ai and not cn.children and not self.game.ended and not (teaching_undo and cn.auto_undo is None):
self._do_ai_move(cn) # cn mismatch stops this if undo fired. avoid message loop here or fires repeatedly.
if (
cn.analysis_ready
and next_player.ai
and not cn.children
and not self.game.ended
and not (teaching_undo and cn.auto_undo is None)
):
self._do_ai_move(
cn
) # cn mismatch stops this if undo fired. avoid message loop here or fires repeatedly.
# Handle prisoners and next player display
prisoners = self.game.prisoner_count
@@ -149,7 +176,9 @@ class KaTrainGui(Screen, KaTrainBase):
try:
self.log(f"Message Loop Received {msg}: {args} for Game {game}", OUTPUT_EXTRA_DEBUG)
if game != self.game.game_id:
self.log(f"Message skipped as it is outdated (current game is {self.game.game_id}", OUTPUT_EXTRA_DEBUG)
self.log(
f"Message skipped as it is outdated (current game is {self.game.game_id}", OUTPUT_EXTRA_DEBUG
)
continue
fn = getattr(self, f"_do_{msg.replace('-','_')}")
fn(*args)
@@ -211,28 +240,36 @@ class KaTrainGui(Screen, KaTrainBase):
def _do_new_game_popup(self):
self.controls.timer.paused = True
if not self.new_game_popup:
self.new_game_popup = I18NPopup(title_key="New Game title", size=[800, 800], content=NewGamePopup(self)).__self__
self.new_game_popup = I18NPopup(
title_key="New Game title", size=[800, 800], content=NewGamePopup(self)
).__self__
self.new_game_popup.content.popup = self.new_game_popup
self.new_game_popup.open()
def _do_timer_popup(self):
self.controls.timer.paused = True
if not self.timer_settings_popup:
self.timer_settings_popup = I18NPopup(title_key="timer settings", size=[350, 350], content=ConfigTimerPopup(self)).__self__
self.timer_settings_popup = I18NPopup(
title_key="timer settings", size=[350, 350], content=ConfigTimerPopup(self)
).__self__
self.timer_settings_popup.content.popup = self.timer_settings_popup
self.timer_settings_popup.open()
def _do_teacher_popup(self):
self.controls.timer.paused = True
if not self.teacher_settings_popup:
self.teacher_settings_popup = I18NPopup(title_key="teacher settings", size=[800, 800], content=ConfigTeacherPopup(self)).__self__
self.teacher_settings_popup = I18NPopup(
title_key="teacher settings", size=[800, 800], content=ConfigTeacherPopup(self)
).__self__
self.teacher_settings_popup.content.popup = self.teacher_settings_popup
self.teacher_settings_popup.open()
def _do_config_popup(self):
self.controls.timer.paused = True
if not self.config_popup:
self.config_popup = I18NPopup(title_key="general settings title", size=[1200, 800], content=ConfigPopup(self)).__self__
self.config_popup = I18NPopup(
title_key="general settings title", size=[1200, 800], content=ConfigPopup(self)
).__self__
self.config_popup.content.popup = self.config_popup
self.config_popup.open()
@@ -247,7 +284,9 @@ class KaTrainGui(Screen, KaTrainBase):
if not self.fileselect_popup:
popup_contents = LoadSGFPopup()
popup_contents.filesel.path = os.path.abspath(os.path.expanduser(self.config("general/sgf_load", ".")))
self.fileselect_popup = I18NPopup(title_key="load sgf title", size=[1200, 800], content=popup_contents).__self__
self.fileselect_popup = I18NPopup(
title_key="load sgf title", size=[1200, 800], content=popup_contents
).__self__
def readfile(*args):
files = popup_contents.filesel.selection
@@ -280,7 +319,9 @@ class KaTrainGui(Screen, KaTrainBase):
except Exception as e:
self.controls.set_status(i18n._("Failed to import from clipboard").format(error=e, contents=clipboard[:50]))
return
move_tree.nodes_in_tree[-1].analyze(self.engine, analyze_fast=False) # speed up result for looking at end of game
move_tree.nodes_in_tree[-1].analyze(
self.engine, analyze_fast=False
) # speed up result for looking at end of game
self._do_new_game(move_tree=move_tree, analyze_fast=True)
self("redo", 999)
self.log("Imported game from clipboard.", OUTPUT_INFO)
+69 -15
View File
@@ -5,7 +5,17 @@ import time
from typing import Dict, List, Tuple
from katrain.core.utils import var_to_grid
from katrain.core.constants import OUTPUT_INFO, OUTPUT_DEBUG, AI_STRATEGIES_POLICY, AI_POLICY, AI_WEIGHTED, AI_STRATEGIES_PICK, AI_JIGO, AI_SCORELOSS, AI_DEFAULT
from katrain.core.constants import (
OUTPUT_INFO,
OUTPUT_DEBUG,
AI_STRATEGIES_POLICY,
AI_POLICY,
AI_WEIGHTED,
AI_STRATEGIES_PICK,
AI_JIGO,
AI_SCORELOSS,
AI_DEFAULT,
)
from katrain.core.engine import EngineDiedException
from katrain.core.game import Game, GameNode, Move
@@ -37,7 +47,9 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
if (ai_mode in AI_STRATEGIES_POLICY) and cn.policy: # pure policy based move
policy_moves = cn.policy_ranking
pass_policy = cn.policy[-1]
top_5_pass = any([polmove[1].is_pass for polmove in policy_moves[:5]]) # dont make it jump around for the last few sensible non pass moves
top_5_pass = any(
[polmove[1].is_pass for polmove in policy_moves[:5]]
) # dont make it jump around for the last few sensible non pass moves
size = game.board_size
policy_grid = var_to_grid(cn.policy, size) # type: List[List[float]]
@@ -55,7 +67,9 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
ai_thoughts += f"Playing top policy move {aimove.gtp()}."
elif policy_moves[0][0] > ai_settings["pick_override"]:
aimove = top_policy_move
ai_thoughts += f"Top policy move has weight > {ai_settings['pick_override']:.1%}, so overriding other strategies."
ai_thoughts += (
f"Top policy move has weight > {ai_settings['pick_override']:.1%}, so overriding other strategies."
)
elif ai_mode == AI_WEIGHTED:
lower_bound = max(0, ai_settings["lower_bound"]) * 2 # compensate for first halving in loop
weaken_fac = max(0.01, ai_settings["weaken_fac"])
@@ -63,7 +77,10 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
while not weighted_coords and lower_bound > 1e-6: # fix edge case where no moves are > lb
lower_bound /= 2
weighted_coords = [
(policy_grid[y][x], policy_grid[y][x] ** (1 / weaken_fac), x, y) for x in range(size[0]) for y in range(size[1]) if policy_grid[y][x] > lower_bound
(policy_grid[y][x], policy_grid[y][x] ** (1 / weaken_fac), x, y)
for x in range(size[0])
for y in range(size[1])
if policy_grid[y][x] > lower_bound
]
top = weighted_selection_without_replacement(weighted_coords, 1)
if top:
@@ -75,7 +92,9 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
coords = None
aimove = Move(coords, player=cn.next_player) # just take a random move by policy w/o noise
ai_thoughts += f"Playing policy-weighted random move {aimove.gtp()} ({policy_value:.1%})" + (
" because no other moves were found." if not top else f" because strategy is weighted (lower bound={lower_bound:.2%}, num moves > lb={len(weighted_coords)})."
" because no other moves were found."
if not top
else f" because strategy is weighted (lower bound={lower_bound:.2%}, num moves > lb={len(weighted_coords)})."
)
elif ai_mode in AI_STRATEGIES_PICK:
legal_policy_moves = [(pol, mv) for pol, mv in policy_moves if not mv.is_pass if pol > 0]
@@ -84,14 +103,30 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
thr_line = ai_settings["threshold"] - 1 # zero-based
if cn.depth >= ai_settings["endgame"] * size[0] * size[1]:
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]
ai_thoughts += f"Generated equal weights as move number >= {ai_settings['endgame'] * size[0] * size[1]}. "
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
]
ai_thoughts += (
f"Generated equal weights as move number >= {ai_settings['endgame'] * size[0] * size[1]}. "
)
else:
if "influence" in ai_mode:
weight = lambda x, y: (1 / ai_settings["line_weight"]) ** (max(0, thr_line - min(size[0] - 1 - x, x)) + max(0, thr_line - min(size[1] - 1 - y, y)))
weight = lambda x, y: (1 / ai_settings["line_weight"]) ** (
max(0, thr_line - min(size[0] - 1 - x, x)) + max(0, thr_line - min(size[1] - 1 - y, y))
)
else:
weight = lambda x, y: (1 / ai_settings["line_weight"]) ** (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]
weight = lambda x, y: (1 / ai_settings["line_weight"]) ** (
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. "
elif "local" in ai_mode or "tenuki" in ai_mode:
var = ai_settings["stddev"] ** 2
@@ -101,7 +136,10 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
else:
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
(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
]
if "tenuki" in ai_mode:
if cn.depth < ai_settings["endgame"] * size[0] * size[1]:
@@ -111,9 +149,16 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
weighted_coords = [(p, 1, x, y) for p, w, x, y in weighted_coords]
ai_thoughts += f"Generated equal weights as move number >= {ai_settings['endgame'] * size[0] * size[1]}. "
else:
ai_thoughts += f"Generated weights based on gaussian with variance {var} around coordinates {mx},{my}. "
ai_thoughts += (
f"Generated weights based on gaussian with variance {var} around coordinates {mx},{my}. "
)
elif "pick" in ai_mode:
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]
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
]
else:
raise ValueError(f"Unknown AI mode {ai_mode}")
pick_moves = weighted_selection_without_replacement(weighted_coords, n_moves)
@@ -139,12 +184,21 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, GameNode
else:
if ai_mode == AI_JIGO:
sign = cn.player_sign(cn.next_player)
jigo_move = min(candidate_ai_moves, key=lambda move: abs(sign * move["scoreLead"] - ai_settings["target_score"]))
jigo_move = min(
candidate_ai_moves, key=lambda move: abs(sign * move["scoreLead"] - ai_settings["target_score"])
)
aimove = Move.from_gtp(jigo_move["move"], player=cn.next_player)
ai_thoughts += f"Jigo strategy found {len(candidate_ai_moves)} candidate moves (best {top_cand.gtp()}) and chose {aimove.gtp()} as closest to 0.5 point win"
elif ai_mode == AI_SCORELOSS:
c = ai_settings["strength"]
moves = [(d["pointsLost"], math.exp(min(200, -c * max(0, d["pointsLost"]))), Move.from_gtp(d["move"], player=cn.next_player)) for d in candidate_ai_moves]
moves = [
(
d["pointsLost"],
math.exp(min(200, -c * max(0, d["pointsLost"]))),
Move.from_gtp(d["move"], player=cn.next_player),
)
for d in candidate_ai_moves
]
topmove = weighted_selection_without_replacement(moves, 1)[0]
aimove = topmove[2]
ai_thoughts += f"ScoreLoss strategy found {len(candidate_ai_moves)} candidate moves (best {top_cand.gtp()}) and chose {aimove.gtp()} (weight {topmove[1]:.3f}, point loss {topmove[0]:.1f}) based on score weights."
+11 -6
View File
@@ -76,21 +76,27 @@ class KaTrainBase:
try:
if not os.path.exists(user_config_file):
os.makedirs(os.path.split(user_config_file)[0], exist_ok=True)
shutil.copyfile(package_config_file,user_config_file)
shutil.copyfile(package_config_file, user_config_file)
config_file = user_config_file
self.log(f"Copied package config to local file {config_file}", OUTPUT_INFO)
else: # user file exists
version = JsonStore(user_config_file, indent=4).get("general")["version"]
if version != VERSION:
backup = user_config_file+f".{version}.backup"
shutil.copyfile(user_config_file,backup)
backup = user_config_file + f".{version}.backup"
shutil.copyfile(user_config_file, backup)
shutil.copyfile(package_config_file, user_config_file)
self.log(f"Copied package config file to {user_config_file} as user file is outdated (<{VERSION}). Old version stored as {backup}", OUTPUT_INFO)
self.log(
f"Copied package config file to {user_config_file} as user file is outdated (<{VERSION}). Old version stored as {backup}",
OUTPUT_INFO,
)
config_file = user_config_file
self.log(f"Using user config file {config_file}", OUTPUT_INFO)
except Exception as e:
config_file = package_config_file
self.log(f"Using package config file {config_file} (exception {e} occurred when finding or creating user config)", OUTPUT_INFO)
self.log(
f"Using package config file {config_file} (exception {e} occurred when finding or creating user config)",
OUTPUT_INFO,
)
try:
self._config_store = JsonStore(config_file, indent=4)
except Exception as e:
@@ -99,7 +105,6 @@ class KaTrainBase:
self._config = dict(self._config_store)
return config_file
def save_config(self):
for k, v in self._config.items():
self._config_store.put(k, **v)
+12 -1
View File
@@ -26,7 +26,18 @@ AI_STRATEGIES_ENGINE = [AI_DEFAULT, AI_SCORELOSS, AI_JIGO]
AI_STRATEGIES_PICK = [AI_PICK, AI_LOCAL, AI_TENUKI, AI_INFLUENCE, AI_TERRITORY]
AI_STRATEGIES_POLICY = [AI_WEIGHTED, AI_POLICY] + AI_STRATEGIES_PICK
AI_STRATEGIES = AI_STRATEGIES_ENGINE + AI_STRATEGIES_POLICY
AI_STRATEGIES_RECOMMENDED_ORDER = [AI_DEFAULT, AI_SCORELOSS, AI_POLICY, AI_WEIGHTED, AI_PICK, AI_LOCAL, AI_TENUKI, AI_TERRITORY, AI_INFLUENCE, AI_JIGO]
AI_STRATEGIES_RECOMMENDED_ORDER = [
AI_DEFAULT,
AI_SCORELOSS,
AI_POLICY,
AI_WEIGHTED,
AI_PICK,
AI_LOCAL,
AI_TENUKI,
AI_TERRITORY,
AI_INFLUENCE,
AI_JIGO,
]
AI_STRENGTH = {
+14 -6
View File
@@ -37,8 +37,14 @@ class KataGoEngine:
else: # e.g. MacOS after brewing
executable = "katago"
modelfile, configfile, exefile = find_package_resource(config["model"]), find_package_resource(config["config"]), find_package_resource(executable)
self.command = f'{exefile} analysis -model "{modelfile}" -config "{configfile}" -analysis-threads {config["threads"]}'
modelfile, configfile, exefile = (
find_package_resource(config["model"]),
find_package_resource(config["config"]),
find_package_resource(executable),
)
self.command = (
f'{exefile} analysis -model "{modelfile}" -config "{configfile}" -analysis-threads {config["threads"]}'
)
if not sys.platform.startswith("win"):
self.command = shlex.split(self.command)
self.queries = {} # outstanding query id -> start time and callback
@@ -56,9 +62,9 @@ class KataGoEngine:
try:
self.katrain.log(f"Starting KataGo with {self.command}", OUTPUT_DEBUG)
self.katago_process = subprocess.Popen(self.command, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
shell=True)
self.katago_process = subprocess.Popen(
self.command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True
)
except (FileNotFoundError, PermissionError, OSError) as e:
if not self.config["katago"].strip():
self.katrain.log(
@@ -133,7 +139,9 @@ class KataGoEngine:
else:
del self.queries[query_id]
time_taken = time.time() - start_time
self.katrain.log(f"[{time_taken:.1f}][{analysis['id']}] KataGo Analysis Received: {analysis.keys()}", OUTPUT_DEBUG)
self.katrain.log(
f"[{time_taken:.1f}][{analysis['id']}] KataGo Analysis Received: {analysis.keys()}", OUTPUT_DEBUG
)
self.katrain.log(line, OUTPUT_EXTRA_DEBUG)
try:
callback(analysis)
+80 -20
View File
@@ -41,7 +41,9 @@ class Game:
else:
board_size = katrain.config("game/size")
self.komi = katrain.config("game/komi")
self.root = GameNode(properties={**Game.DEFAULT_PROPERTIES, **{"SZ": board_size, "KM": self.komi, "DT": self.game_id}})
self.root = GameNode(
properties={**Game.DEFAULT_PROPERTIES, **{"SZ": board_size, "KM": self.komi, "DT": self.game_id}}
)
handicap = katrain.config("game/handicap")
if handicap:
self.place_handicap_stones(handicap)
@@ -50,7 +52,9 @@ class Game:
self.root.set_property("RU", katrain.config("game/rules"))
self.set_current_node(self.root)
threading.Thread(target=lambda: self.analyze_all_nodes(-1_000_000, analyze_fast=analyze_fast), daemon=True).start() # return faster, but bypass Kivy Clock
threading.Thread(
target=lambda: self.analyze_all_nodes(-1_000_000, analyze_fast=analyze_fast), daemon=True
).start() # return faster, but bypass Kivy Clock
def analyze_all_nodes(self, priority=0, analyze_fast=False):
for node in self.root.nodes_in_tree:
@@ -59,7 +63,9 @@ class Game:
# -- move tree functions --
def _calculate_groups(self):
board_size_x, board_size_y = self.board_size
self.board = [[-1 for _x in range(board_size_x)] for _y in range(board_size_y)] # type: List[List[int]] # board pos -> chain id
self.board = [
[-1 for _x in range(board_size_x)] for _y in range(board_size_y)
] # type: List[List[int]] # board pos -> chain id
self.chains = [] # type: List[List[Move]] # chain id -> chain
self.prisoners = [] # type: List[Move]
self.last_capture = [] # type: List[Move]
@@ -94,7 +100,9 @@ class Game:
nb_chains = list({c for c in neighbours([move]) if c >= 0 and self.chains[c][0].player == move.player})
if nb_chains:
this_chain = nb_chains[0]
self.board = [[nb_chains[0] if sq in nb_chains else sq for sq in line] for line in self.board] # merge chains connected by this move
self.board = [
[nb_chains[0] if sq in nb_chains else sq for sq in line] for line in self.board
] # merge chains connected by this move
for oc in nb_chains[1:]:
self.chains[nb_chains[0]] += self.chains[oc]
self.chains[oc] = []
@@ -175,13 +183,18 @@ class Game:
near_x -= 1
spacing = (far_x - near_x) / (stones_per_row - 1)
coords = list({math.floor(0.5 + near_x + i * spacing) for i in range(stones_per_row)})
stones = sorted([(x, y) for x in coords for y in coords], key=lambda xy: -((xy[0] - (board_size_x - 1) / 2) ** 2 + (xy[1] - (board_size_y - 1) / 2) ** 2))
stones = sorted(
[(x, y) for x in coords for y in coords],
key=lambda xy: -((xy[0] - (board_size_x - 1) / 2) ** 2 + (xy[1] - (board_size_y - 1) / 2) ** 2),
)
else: # max 9
stones = [(far_x, far_y), (near_x, near_y), (far_x, near_y), (near_x, far_y)]
if n_handicaps % 2 == 1:
stones.append((middle_x, middle_y))
stones += [(near_x, middle_y), (far_x, middle_y), (middle_x, near_y), (middle_x, far_y)]
self.root.set_property("AB", list({Move(stone).sgf(board_size=(board_size_x, board_size_y)) for stone in stones[:n_handicaps]}))
self.root.set_property(
"AB", list({Move(stone).sgf(board_size=(board_size_x, board_size_y)) for stone in stones[:n_handicaps]})
)
@property
def board_size(self):
@@ -196,7 +209,9 @@ class Game:
return self.current_node.parent and self.current_node.is_pass and self.current_node.parent.is_pass
@property
def prisoner_count(self) -> Dict: # returns prisoners that are of a certain colour as {B: black stones captures, W: white stones captures}
def prisoner_count(
self,
) -> Dict: # returns prisoners that are of a certain colour as {B: black stones captures, W: white stones captures}
return {player: sum([m.player == player for m in self.prisoners]) for player in Move.PLAYERS}
@property
@@ -205,7 +220,10 @@ class Game:
if not self.current_node.ownership or rules != "japanese":
if not self.current_node.score:
return None
self.katrain.log(f"rules '{rules}' are not japanese, or no ownership available ({not self.current_node.ownership}) -> no manual score available", OUTPUT_DEBUG)
self.katrain.log(
f"rules '{rules}' are not japanese, or no ownership available ({not self.current_node.ownership}) -> no manual score available",
OUTPUT_DEBUG,
)
return self.current_node.format_score(round(2 * self.current_node.score) / 2) + "?"
board_size_x, board_size_y = self.board_size
ownership_grid = var_to_grid(self.current_node.ownership, (board_size_x, board_size_y))
@@ -217,7 +235,11 @@ class Game:
def japanese_score_square(square, owner):
player = stones.get(square, None)
if (player == "B" and owner > hi_threshold) or (player == "W" and owner < -hi_threshold) or abs(owner) < lo_threshold:
if (
(player == "B" and owner > hi_threshold)
or (player == "W" and owner < -hi_threshold)
or abs(owner) < lo_threshold
):
return 0 # dame or own stones
if player is None and abs(owner) >= hi_threshold:
return round(owner) # surrounded empty intersection
@@ -225,20 +247,36 @@ class Game:
return 2 * round(owner) # captured stone
return math.nan # unknown!
scored_squares = [japanese_score_square((x, y), ownership_grid[y][x]) for y in range(board_size_y) for x in range(board_size_x)]
scored_squares = [
japanese_score_square((x, y), ownership_grid[y][x])
for y in range(board_size_y)
for x in range(board_size_x)
]
num_sq = {t: sum([s == t for s in scored_squares]) for t in [-2, -1, 0, 1, 2]}
num_unkn = sum(math.isnan(s) for s in scored_squares)
prisoners = self.prisoner_count
score = sum([t * n for t, n in num_sq.items()]) + prisoners["W"] - prisoners["B"] - self.komi
self.katrain.log(f"Manual Scoring: {num_sq} score by square with {num_unkn} unknown, {prisoners} captures, and {self.komi} komi -> score = {score}", OUTPUT_INFO)
self.katrain.log(
f"Manual Scoring: {num_sq} score by square with {num_unkn} unknown, {prisoners} captures, and {self.komi} komi -> score = {score}",
OUTPUT_DEBUG,
)
if num_unkn > max_unknown or (num_sq[0] - len(stones)) > max_dame:
return None
return self.current_node.format_score(score)
def __repr__(self):
return "\n".join("".join(self.chains[c][0].player if c >= 0 else "-" for c in line) for line in self.board) + f"\ncaptures: {self.prisoner_count}"
return (
"\n".join("".join(self.chains[c][0].player if c >= 0 else "-" for c in line) for line in self.board)
+ f"\ncaptures: {self.prisoner_count}"
)
def write_sgf(self, path: str, trainer_config: Optional[Dict] = None, save_feedback: Optional[List] = None, eval_thresholds: Optional[List] = None):
def write_sgf(
self,
path: str,
trainer_config: Optional[Dict] = None,
save_feedback: Optional[List] = None,
eval_thresholds: Optional[List] = None,
):
if trainer_config is None:
trainer_config = self.katrain.config("trainer")
if save_feedback is None:
@@ -249,13 +287,22 @@ class Game:
def player_name(player_info):
return f"{i18n._(player_info.player_type)} ({i18n._(player_info.player_subtype)})"
player_names = {bw: re.sub(r"['<>:\"/\\|?*]", "", self.root.get_property("P" + bw) or player_name(self.katrain.players_info[bw])) for bw in "BW"}
player_names = {
bw: re.sub(
r"['<>:\"/\\|?*]", "", self.root.get_property("P" + bw) or player_name(self.katrain.players_info[bw])
)
for bw in "BW"
}
game_name = f"katrain_{player_names['B']} vs {player_names['W']} {self.game_id}"
file_name = os.path.abspath(os.path.join(path, f"{game_name}.sgf"))
os.makedirs(os.path.dirname(file_name), exist_ok=True)
show_dots_for = {bw: trainer_config.get("eval_show_ai", True) or pl.human for bw, pl in self.katrain.players_info.items()}
sgf = self.root.sgf(save_comments_player=show_dots_for, save_comments_class=save_feedback, eval_thresholds=eval_thresholds)
show_dots_for = {
bw: trainer_config.get("eval_show_ai", True) or pl.human for bw, pl in self.katrain.players_info.items()
}
sgf = self.root.sgf(
save_comments_player=show_dots_for, save_comments_class=save_feedback, eval_thresholds=eval_thresholds
)
with open(file_name, "w") as f:
f.write(sgf)
return i18n._("sgf written").format(file_name=file_name)
@@ -273,7 +320,11 @@ class Game:
elif mode == "sweep":
board_size_x, board_size_y = self.board_size
if cn.analysis_ready:
policy_grid = var_to_grid(self.current_node.policy, size=(board_size_x, board_size_y)) if self.current_node.policy else None
policy_grid = (
var_to_grid(self.current_node.policy, size=(board_size_x, board_size_y))
if self.current_node.policy
else None
)
analyze_moves = sorted(
[
Move(coords=(x, y), player=cn.next_player)
@@ -284,7 +335,12 @@ class Game:
key=lambda mv: -policy_grid[mv.coords[1]][mv.coords[0]],
)
else:
analyze_moves = [Move(coords=(x, y), player=cn.next_player) for x in range(board_size_x) for y in range(board_size_y) if (x, y) not in stones]
analyze_moves = [
Move(coords=(x, y), player=cn.next_player)
for x in range(board_size_x)
for y in range(board_size_y)
if (x, y) not in stones
]
visits = engine.config["fast_visits"]
self.katrain.controls.set_status(f"Refining analysis of entire board to {visits} visits")
priority = -1_000_000_000
@@ -298,7 +354,9 @@ class Game:
self.katrain.controls.set_status(f"Equalizing analysis of candidate moves to {visits} visits")
priority = -1_000
for move in analyze_moves:
cn.analyze(engine, priority, visits=visits, refine_move=move, time_limit=False) # explicitly requested so take as long as you need
cn.analyze(
engine, priority, visits=visits, refine_move=move, time_limit=False
) # explicitly requested so take as long as you need
def analyze_undo(self, node):
train_config = self.katrain.config("trainer")
@@ -325,5 +383,7 @@ class Game:
node.auto_undo = undo
if undo:
self.undo(1)
self.katrain.controls.set_status(f"Undid move {move.gtp()} as it lost {points_lost:.1f} points{xmsg}. Hover over the move to see expected refutation.")
self.katrain.controls.set_status(
f"Undid move {move.gtp()} as it lost {points_lost:.1f} points{xmsg}. Hover over the move to see expected refutation."
)
self.katrain.update_state()
+90 -18
View File
@@ -21,7 +21,9 @@ class GameNode(SGFNode):
self.move_number = 0
self.time_used = 0
self.analysis_visits_requested = 0
self.undo_threshold = random.random() # for fractional undos, store the random threshold in the move itself for consistency
self.undo_threshold = (
random.random()
) # for fractional undos, store the random threshold in the move itself for consistency
self._favourite_child = None
def sgf_properties(self, save_comments_player=None, save_comments_class=None, eval_thresholds=None):
@@ -30,10 +32,18 @@ class GameNode(SGFNode):
show_class = save_comments_class[evaluation_class(self.points_lost, eval_thresholds)]
else:
show_class = False
if (save_comments_player or {}).get(self.player, False) and show_class and self.analysis_ready and self.parent and self.parent.analysis_ready:
if (
(save_comments_player or {}).get(self.player, False)
and show_class
and self.analysis_ready
and self.parent
and self.parent.analysis_ready
):
candidate_moves = self.parent.candidate_moves
top_x = Move.from_gtp(candidate_moves[0]["move"]).sgf(self.board_size)
best_sq = [Move.from_gtp(d["move"]).sgf(self.board_size) for d in candidate_moves[1:] if d["pointsLost"] <= 0.5]
best_sq = [
Move.from_gtp(d["move"]).sgf(self.board_size) for d in candidate_moves[1:] if d["pointsLost"] <= 0.5
]
if best_sq and "SQ" not in properties:
properties["SQ"] = best_sq
if top_x and "MA" not in properties:
@@ -42,14 +52,21 @@ class GameNode(SGFNode):
if comment:
properties["C"] = ["\n".join(properties.get("C", "")) + comment]
if self.is_root:
properties["C"] = [i18n._("SGF start message") + "\n" + "\n".join(properties.get("C", "")) + "\nSGF with review generated by KaTrain."]
properties["C"] = [
i18n._("SGF start message")
+ "\n"
+ "\n".join(properties.get("C", ""))
+ "\nSGF with review generated by KaTrain."
]
if self.note.strip():
properties["C"] = ["\n".join(properties.get("C", "")) + f"\nNote: {self.note}"]
return properties
@staticmethod
def order_children(children):
return sorted(children, key=lambda c: 0.5 if c.auto_undo is None else int(c.auto_undo)) # analyzed/not undone main, non-teach second, undone last
return sorted(
children, key=lambda c: 0.5 if c.auto_undo is None else int(c.auto_undo)
) # analyzed/not undone main, non-teach second, undone last
def set_favourite_child(self, c):
self._favourite_child = c
@@ -66,20 +83,32 @@ class GameNode(SGFNode):
if visits and not refine_move:
self.analysis_visits_requested = max(self.analysis_visits_requested, engine.config["max_visits"])
engine.request_analysis(
self, lambda result: self.set_analysis(result, refine_move), priority=priority, visits=visits, analyze_fast=analyze_fast, time_limit=time_limit, next_move=refine_move
self,
lambda result: self.set_analysis(result, refine_move),
priority=priority,
visits=visits,
analyze_fast=analyze_fast,
time_limit=time_limit,
next_move=refine_move,
)
def update_move_analysis(self, move_analysis, move_gtp):
cur = self.analysis["moves"].get(move_gtp)
if cur is None:
self.analysis["moves"][move_gtp] = {"move": move_gtp, "order": 999, **move_analysis} # some default values for keys missing in rootInfo
self.analysis["moves"][move_gtp] = {
"move": move_gtp,
"order": 999,
**move_analysis,
} # some default values for keys missing in rootInfo
elif cur["visits"] < move_analysis["visits"]:
cur.update(move_analysis)
def set_analysis(self, analysis_json, refine_move):
if refine_move:
pvtail = analysis_json["moveInfos"][0]["pv"] if analysis_json["moveInfos"] else []
self.update_move_analysis({"pv": [refine_move.gtp()] + pvtail, **analysis_json["rootInfo"]}, refine_move.gtp())
self.update_move_analysis(
{"pv": [refine_move.gtp()] + pvtail, **analysis_json["rootInfo"]}, refine_move.gtp()
)
else:
for move_analysis in analysis_json["moveInfos"]:
self.update_move_analysis(move_analysis, move_analysis["move"])
@@ -87,8 +116,12 @@ class GameNode(SGFNode):
self.policy = analysis_json.get("policy")
self.analysis["root"] = analysis_json["rootInfo"]
if self.parent and self.move:
analysis_json["rootInfo"]["pv"] = [self.move.gtp()] + (analysis_json["moveInfos"][0]["pv"] if analysis_json["moveInfos"] else [])
self.parent.update_move_analysis(analysis_json["rootInfo"], self.move.gtp()) # update analysis in parent for consistency
analysis_json["rootInfo"]["pv"] = [self.move.gtp()] + (
analysis_json["moveInfos"][0]["pv"] if analysis_json["moveInfos"] else []
)
self.parent.update_move_analysis(
analysis_json["rootInfo"], self.move.gtp()
) # update analysis in parent for consistency
@property
def analysis_ready(self):
@@ -138,19 +171,44 @@ class GameNode(SGFNode):
points_lost = self.points_lost
if sgf and points_lost > 0.5:
text += i18n._("Info:point loss").format(points_lost=points_lost) + "\n"
text += i18n._("Info:top move").format(top_move=previous_top_move["move"], score=self.format_score(previous_top_move["scoreLead"])) + "\n"
text += (
i18n._("Info:top move").format(
top_move=previous_top_move["move"],
score=self.format_score(previous_top_move["scoreLead"]),
)
+ "\n"
)
else:
text += i18n._("Info:best move") + "\n"
if previous_top_move.get("pv") and (sgf or hints):
text += i18n._("Info:PV").format(pv=self.make_pv(single_move.player, previous_top_move["pv"], interactive)) + "\n"
text += (
i18n._("Info:PV").format(
pv=self.make_pv(single_move.player, previous_top_move["pv"], interactive)
)
+ "\n"
)
if sgf or hints or teach:
policy_ranking = self.parent.policy_ranking
currmove_policy_with_ix = [(ix + 1, p) for (p, m), ix in zip(policy_ranking, range(len(policy_ranking))) if m == single_move]
currmove_policy_with_ix = [
(ix + 1, p)
for (p, m), ix in zip(policy_ranking, range(len(policy_ranking)))
if m == single_move
]
if currmove_policy_with_ix:
text += i18n._("Info:policy rank").format(rank=currmove_policy_with_ix[0][0], probability=currmove_policy_with_ix[0][1]) + "\n"
text += (
i18n._("Info:policy rank").format(
rank=currmove_policy_with_ix[0][0], probability=currmove_policy_with_ix[0][1]
)
+ "\n"
)
if not currmove_policy_with_ix or currmove_policy_with_ix[0] != 1 and (sgf or hints):
text += i18n._("Info:policy best").format(move=policy_ranking[0][1].gtp(), probability=policy_ranking[0][0]) + "\n"
text += (
i18n._("Info:policy best").format(
move=policy_ranking[0][1].gtp(), probability=policy_ranking[0][0]
)
+ "\n"
)
if self.auto_undo and sgf:
text += i18n._("Info:teaching undo") + "\n"
top_pv = self.analysis_ready and self.candidate_moves[0].get("pv")
@@ -173,7 +231,13 @@ class GameNode(SGFNode):
@property
def parent_realized_points_lost(self) -> Optional[float]:
single_move = self.move
if single_move and self.parent and self.parent.parent and self.analysis_ready and self.parent.parent.analysis_ready:
if (
single_move
and self.parent
and self.parent.parent
and self.analysis_ready
and self.parent.parent.analysis_ready
):
parent_parent_score = self.parent.parent.score
score = self.score
return self.player_sign(single_move.player) * (score - parent_parent_score)
@@ -189,11 +253,19 @@ class GameNode(SGFNode):
if not self.analysis["moves"]:
polmoves = self.policy_ranking
top_polmove = polmoves[0][1] if polmoves else Move(None) # if no info at all, pass
return [{**self.analysis["root"], "pointsLost": 0, "order": 0, "move": top_polmove.gtp()}] # single visit -> go by policy/root
return [
{**self.analysis["root"], "pointsLost": 0, "order": 0, "move": top_polmove.gtp()}
] # single visit -> go by policy/root
root_score = self.analysis["root"]["scoreLead"]
move_dicts = list(self.analysis["moves"].values()) # prevent incoming analysis from causing crash
return sorted([{"pointsLost": self.player_sign(self.next_player) * (root_score - d["scoreLead"]), **d} for d in move_dicts], key=lambda d: (d["order"], d["pointsLost"]))
return sorted(
[
{"pointsLost": self.player_sign(self.next_player) * (root_score - d["scoreLead"]), **d}
for d in move_dicts
],
key=lambda d: (d["order"], d["pointsLost"]),
)
@property
def policy_ranking(self) -> Optional[List[Tuple[float, Move]]]: # return moves from highest policy value to lowest
+24 -5
View File
@@ -11,7 +11,9 @@ class ParseError(Exception):
class Move:
GTP_COORD = list("ABCDEFGHJKLMNOPQRSTUVWXYZ") + [xa + c for xa in "AB" for c in "ABCDEFGHJKLMNOPQRSTUVWXYZ"] # board size 52+ support
GTP_COORD = list("ABCDEFGHJKLMNOPQRSTUVWXYZ") + [
xa + c for xa in "AB" for c in "ABCDEFGHJKLMNOPQRSTUVWXYZ"
] # board size 52+ support
PLAYERS = "BW"
SGF_COORD = list("ABCDEFGHIJKLMNOPQRSTUVWXYZ".lower()) + list("ABCDEFGHIJKLMNOPQRSTUVWXYZ") # sgf goes to 52
@@ -28,7 +30,10 @@ class Move:
"""Initialize a move from SGF coordinates and player"""
if sgf_coords == "" or Move.SGF_COORD.index(sgf_coords[0]) == board_size[0]: # some servers use [tt] for pass
return cls(coords=None, player=player)
return cls(coords=(Move.SGF_COORD.index(sgf_coords[0]), board_size[1] - Move.SGF_COORD.index(sgf_coords[1]) - 1), player=player)
return cls(
coords=(Move.SGF_COORD.index(sgf_coords[0]), board_size[1] - Move.SGF_COORD.index(sgf_coords[1]) - 1),
player=player,
)
def __init__(self, coords: Optional[Tuple[int, int]] = None, player: str = "B"):
"""Initialize a move from zero-based coordinates and player"""
@@ -90,7 +95,13 @@ class SGFNode:
"""Generates an SGF, calling sgf_properties on each node with the given xargs, so it can filter relevant properties if needed."""
def node_sgf_str(node):
return ";" + "".join([prop + "".join(f"[{v}]" for v in values) for prop, values in node.sgf_properties(**xargs).items() if values])
return ";" + "".join(
[
prop + "".join(f"[{v}]" for v in values)
for prop, values in node.sgf_properties(**xargs).items()
if values
]
)
stack = [")", self, "("]
sgf_str = ""
@@ -177,12 +188,20 @@ class SGFNode:
@property
def moves(self) -> List[Move]:
"""Returns all moves in the node - typically 'move' will be better."""
return [Move.from_sgf(move, player=pl, board_size=self.board_size) for pl in Move.PLAYERS for move in self.get_list_property(pl, [])]
return [
Move.from_sgf(move, player=pl, board_size=self.board_size)
for pl in Move.PLAYERS
for move in self.get_list_property(pl, [])
]
@property
def placements(self) -> List[Move]:
"""Returns all placements (AB/AW) in the node."""
return [Move.from_sgf(sgf_coords, player=pl, board_size=self.board_size) for pl in Move.PLAYERS for sgf_coords in self.get_list_property("A" + pl, [])]
return [
Move.from_sgf(sgf_coords, player=pl, board_size=self.board_size)
for pl in Move.PLAYERS
for sgf_coords in self.get_list_property("A" + pl, [])
]
@property
def move_with_placements(self) -> List[Move]:
+91 -21
View File
@@ -96,7 +96,8 @@ class BadukPanWidget(Widget):
near_move = [
(pv, node)
for move, pv, node in self.active_pv_moves
if abs(rel_pos[0] - self.gridpos_x[move[0]]) < self.grid_size / 2 and abs(rel_pos[1] - self.gridpos_y[move[1]]) < self.grid_size / 2
if abs(rel_pos[0] - self.gridpos_x[move[0]]) < self.grid_size / 2
and abs(rel_pos[1] - self.gridpos_y[move[1]]) < self.grid_size / 2
]
if near_move:
self.set_animating_pv(near_move[0][0], near_move[0][1])
@@ -120,7 +121,9 @@ class BadukPanWidget(Widget):
xd, xp = self._find_closest(touch.x, self.gridpos_x)
yd, yp = self._find_closest(touch.y, self.gridpos_y)
nodes_here = [node for node in katrain.game.current_node.nodes_from_root if node.move and node.move.coords == (xp, yp)]
nodes_here = [
node for node in katrain.game.current_node.nodes_from_root if node.move and node.move.coords == (xp, yp)
]
if nodes_here and max(yd, xd) < self.grid_size / 2: # load old comment
if touch.is_double_tap: # navigate to move
katrain.game.set_current_node(nodes_here[-1])
@@ -181,12 +184,19 @@ class BadukPanWidget(Widget):
extra_px_margin_y = (self.height - board_height_with_margins) / 2
self.stone_size = self.grid_size * STONE_SIZE
self.gridpos_x = [self.pos[0] + extra_px_margin_x + math.floor((grid_spaces_margin_x[0] + i) * self.grid_size + 0.5) for i in range(board_size_x)]
self.gridpos_y = [self.pos[1] + extra_px_margin_y + math.floor((grid_spaces_margin_y[0] + i) * self.grid_size + 0.5) for i in range(board_size_y)]
self.gridpos_x = [
self.pos[0] + extra_px_margin_x + math.floor((grid_spaces_margin_x[0] + i) * self.grid_size + 0.5)
for i in range(board_size_x)
]
self.gridpos_y = [
self.pos[1] + extra_px_margin_y + math.floor((grid_spaces_margin_y[0] + i) * self.grid_size + 0.5)
for i in range(board_size_y)
]
Color(*BOARD_COLOR)
Rectangle(
pos=(self.gridpos_x[0] - self.grid_size * 1.5, self.gridpos_y[0] - self.grid_size * 1.5), size=(self.grid_size * x_grid_spaces, self.grid_size * y_grid_spaces)
pos=(self.gridpos_x[0] - self.grid_size * 1.5, self.gridpos_y[0] - self.grid_size * 1.5),
size=(self.grid_size * x_grid_spaces, self.grid_size * y_grid_spaces),
)
Color(*LINE_COLOR)
@@ -200,7 +210,9 @@ class BadukPanWidget(Widget):
star_point_pos = 3 if size <= 11 else 4
if size < 7:
return []
return [star_point_pos - 1, size - star_point_pos] + ([int(size / 2)] if size % 2 == 1 and size > 7 else [])
return [star_point_pos - 1, size - star_point_pos] + (
[int(size / 2)] if size % 2 == 1 and size > 7 else []
)
starpt_size = self.grid_size * STARPOINT_SIZE
for x in star_point_coords(board_size_x):
@@ -211,9 +223,19 @@ class BadukPanWidget(Widget):
Color(0.25, 0.25, 0.25)
coord_offset = self.grid_size * 1.5 / 2
for i in range(board_size_x):
draw_text(pos=(self.gridpos_x[i], self.gridpos_y[0] - coord_offset), text=Move.GTP_COORD[i], font_size=self.grid_size / 1.5, font_name="Roboto")
draw_text(
pos=(self.gridpos_x[i], self.gridpos_y[0] - coord_offset),
text=Move.GTP_COORD[i],
font_size=self.grid_size / 1.5,
font_name="Roboto",
)
for i in range(board_size_y):
draw_text(pos=(self.gridpos_x[0] - coord_offset, self.gridpos_y[i]), text=str(i + 1), font_size=self.grid_size / 1.5, font_name="Roboto")
draw_text(
pos=(self.gridpos_x[0] - coord_offset, self.gridpos_y[i]),
text=str(i + 1),
font_size=self.grid_size / 1.5,
font_name="Roboto",
)
def draw_board_contents(self, *_args):
if not (self.katrain and self.katrain.game):
@@ -236,7 +258,9 @@ class BadukPanWidget(Widget):
for m in katrain.game.stones:
has_stone[m.coords] = m.player
show_dots_for = {p: self.trainer_config["eval_show_ai"] or katrain.players_info[p].human for p in Move.PLAYERS}
show_dots_for = {
p: self.trainer_config["eval_show_ai"] or katrain.players_info[p].human for p in Move.PLAYERS
}
show_dots_for_class = self.trainer_config["show_dots"]
nodes = katrain.game.current_node.nodes_from_root
realized_points_lost = None
@@ -259,7 +283,15 @@ class BadukPanWidget(Widget):
evalcol = None
inner = stone_color[m.opponent] if i == 0 else None
drawn_stone[m.coords] = m.player
self.draw_stone(m.coords[0], m.coords[1], stone_color[m.player], outline_color[m.player], inner, evalcol, evalsize)
self.draw_stone(
m.coords[0],
m.coords[1],
stone_color[m.player],
outline_color[m.player],
inner,
evalcol,
evalsize,
)
realized_points_lost = node.parent_realized_points_lost
if katrain.game.current_node.is_root and katrain.debug_level >= 3: # secret ;)
@@ -284,8 +316,16 @@ class BadukPanWidget(Widget):
Rectangle(pos=(self.gridpos_x[x] - rsz / 2, self.gridpos_y[y] - rsz / 2), size=(rsz, rsz))
policy = current_node.policy
if not policy and current_node.parent and current_node.parent.policy and katrain.last_player_info.ai and katrain.next_player_info.ai:
policy = current_node.parent.policy # in the case of AI self-play we allow the policy to be one step out of date
if (
not policy
and current_node.parent
and current_node.parent.policy
and katrain.last_player_info.ai
and katrain.next_player_info.ai
):
policy = (
current_node.parent.policy
) # in the case of AI self-play we allow the policy to be one step out of date
pass_btn = katrain.board_controls.pass_btn
pass_btn.canvas.after.clear()
@@ -303,7 +343,11 @@ class BadukPanWidget(Widget):
self.draw_stone(x, y, policy_circle_color, scale=polsize)
polsize = math.sqrt(policy[-1])
with pass_btn.canvas.after:
draw_circle((pass_btn.pos[0] + pass_btn.width / 2, pass_btn.pos[1] + pass_btn.height / 2), polsize * pass_btn.height / 2, POLICY_COLOR)
draw_circle(
(pass_btn.pos[0] + pass_btn.width / 2, pass_btn.pos[1] + pass_btn.height / 2),
polsize * pass_btn.height / 2,
POLICY_COLOR,
)
# pass circle
passed = len(nodes) > 1 and current_node.is_pass
@@ -317,7 +361,9 @@ class BadukPanWidget(Widget):
size = min(self.width, self.height) * 0.22
Ellipse(pos=(center[0] - size / 2, center[1] - size / 2), size=(size, size))
Color(0.85, 0.85, 0.85)
draw_text(pos=center, text=text, font_size=size * 0.25, halign="center", outline_color=[0.95, 0.95, 0.95])
draw_text(
pos=center, text=text, font_size=size * 0.25, halign="center", outline_color=[0.95, 0.95, 0.95]
)
self.draw_hover_contents()
@@ -347,9 +393,20 @@ class BadukPanWidget(Widget):
evalcol = copy.copy(self.eval_color(points_lost))
evalcol[3] = alpha
if child_node.analysis_ready:
self.active_pv_moves.append((move.coords, [move.gtp()] + child_node.candidate_moves[0]["pv"], current_node))
self.active_pv_moves.append(
(move.coords, [move.gtp()] + child_node.candidate_moves[0]["pv"], current_node)
)
scale = CHILD_SCALE
self.draw_stone(move.coords[0], move.coords[1], (*stone_color[move.player][:3], alpha), None, None, evalcol, evalscale=scale, scale=scale)
self.draw_stone(
move.coords[0],
move.coords[1],
(*stone_color[move.player][:3], alpha),
None,
None,
evalcol,
evalscale=scale,
scale=scale,
)
# hints or PV
if katrain.analysis_controls.hints.active and not game_ended and not lock_ai:
@@ -366,7 +423,12 @@ class BadukPanWidget(Widget):
self.active_pv_moves.append((move.coords, move_dict["pv"], current_node))
else:
katrain.log(f"PV missing for move_dict {move_dict}", OUTPUT_DEBUG)
self.draw_stone(move.coords[0], move.coords[1], [*self.eval_color(move_dict["pointsLost"])[:3], alpha], scale=scale)
self.draw_stone(
move.coords[0],
move.coords[1],
[*self.eval_color(move_dict["pointsLost"])[:3], alpha],
scale=scale,
)
# hover next move ghost stone
if self.ghost_stone:
@@ -403,7 +465,9 @@ class BadukPanWidget(Widget):
if coords is None: # tee-hee
sizefac = katrain.board_controls.pass_btn.size[1] / 2 / self.stone_size
board_coords = [
katrain.board_controls.pass_btn.pos[0] + katrain.board_controls.pass_btn.size[0] + self.stone_size * sizefac,
katrain.board_controls.pass_btn.pos[0]
+ katrain.board_controls.pass_btn.size[0]
+ self.stone_size * sizefac,
katrain.board_controls.pass_btn.pos[1] + katrain.board_controls.pass_btn.size[1] / 2,
]
else:
@@ -414,7 +478,9 @@ class BadukPanWidget(Widget):
draw_text(pos=board_coords, text=str(i + 1), font_size=self.grid_size / 1.45, font_name="Roboto")
def set_animating_pv(self, pv, node):
if node is not None and (not self.animating_pv or not (self.animating_pv[0] == pv and self.animating_pv[1] == node)):
if node is not None and (
not self.animating_pv or not (self.animating_pv[0] == pv and self.animating_pv[1] == node)
):
self.animating_pv = (pv, node, time.time(), self.last_mouse_pos)
def show_pv_from_comments(self, pv_str):
@@ -441,10 +507,14 @@ class AnalysisControls(MDFloatLayout):
def build_menu(self, _dt):
menu_items = [
{"text": i18n._(text) + f" ({shortcut})"} # , "right_content_cls": AnalysisDropdownMenuRightContent(text=shortcut)}
{
"text": i18n._(text) + f" ({shortcut})"
} # , "right_content_cls": AnalysisDropdownMenuRightContent(text=shortcut)}
for text, shortcut in zip(self.ANALYSIS_OPTIONS, self.ANALYSIS_SHORTCUTS)
]
self.analysis_menu = AnalysisDropdownMenu(caller=self.analysis_button, items=menu_items, width_mult=5, use_icon_item=False, callback=self.action)
self.analysis_menu = AnalysisDropdownMenu(
caller=self.analysis_button, items=menu_items, width_mult=5, use_icon_item=False, callback=self.action
)
def action(self, item):
katrain = MDApp.get_running_app().gui
+9 -2
View File
@@ -56,11 +56,14 @@ class ControlsPanel(BoxLayout):
if not game:
return
current_node, move = game.current_node, game.current_node.move
if game.current_node is not self.status_node and not (self.status is not None and self.status_node is None and game.current_node.is_root): # startup errors on root
if game.current_node is not self.status_node and not (
self.status is not None and self.status_node is None and game.current_node.is_root
): # startup errors on root
self.status.text = ""
self.status_node = None
last_player_was_ai_playing_human = katrain.last_player_info.ai and katrain.next_player_info.human
both_players_are_robots = katrain.last_player_info.ai and katrain.next_player_info.ai
self.active_comment_node = current_node
if katrain.play_analyze_mode == MODE_PLAY and last_player_was_ai_playing_human:
@@ -68,11 +71,15 @@ class ControlsPanel(BoxLayout):
self.active_comment_node = current_node.children[-1]
elif current_node.parent:
self.active_comment_node = current_node.parent
elif both_players_are_robots and not current_node.analysis_ready and current_node.parent:
self.active_comment_node = current_node.parent
hints = katrain.analysis_controls.hints.active
info = ""
if current_node.move and not current_node.is_root:
info = self.active_comment_node.comment(teach=katrain.players_info[self.active_comment_node.player].being_taught, hints=hints)
info = self.active_comment_node.comment(
teach=katrain.players_info[self.active_comment_node.player].being_taught, hints=hints
)
if self.active_comment_node.analysis_ready:
self.stats.score = self.active_comment_node.format_score() or ""
+30 -7
View File
@@ -77,7 +77,9 @@ class LeftButtonBehavior(ButtonBehavior): # stops buttons etc activating on rig
# -- resizeable buttons
class SizedButton(LeftButtonBehavior, RectangularRippleBehavior, BasePressedButton, BaseFlatButton, BackgroundMixin): # avoid baserectangular for sizing
class SizedButton(
LeftButtonBehavior, RectangularRippleBehavior, BasePressedButton, BaseFlatButton, BackgroundMixin
): # avoid baserectangular for sizing
text = StringProperty("")
text_color = ListProperty(WHITE)
text_size = ListProperty([100, 100])
@@ -263,7 +265,10 @@ class PlayerSetup(MDBoxLayout):
if self.mode == PLAYER_AI:
return {"player_type": self.player_type.selected[1], "player_subtype": self.player_subtype_ai.selected[1]}
else:
return {"player_type": self.player_type.selected[1], "player_subtype": self.player_subtype_human.selected[1]}
return {
"player_type": self.player_type.selected[1],
"player_subtype": self.player_subtype_human.selected[1],
}
def update_widget(self, player_type, player_subtype):
self.player_type.select_key(player_type) # should trigger setup options
@@ -373,18 +378,30 @@ class CollapsablePanel(MDBoxLayout):
super().__init__(**kwargs)
self.orientation = "vertical"
self.bind(
options=self.build_options, option_colors=self.build_options, options_height=self.build_options, option_active=self.build_options, options_spacing=self.build_options,
options=self.build_options,
option_colors=self.build_options,
options_height=self.build_options,
option_active=self.build_options,
options_spacing=self.build_options,
)
self.bind(state=self.build, size_hint_y_open=self.build, height_open=self.build)
MDApp.get_running_app().bind(language=lambda *_: Clock.schedule_once(self.build_options, 0))
self.build_options()
def build_options(self, *args, **kwargs):
self.header = CollapsablePanelHeader(height=self.options_height, size_hint_y=None, spacing=self.options_spacing, padding=[1, 0, 0, 0])
self.header = CollapsablePanelHeader(
height=self.options_height, size_hint_y=None, spacing=self.options_spacing, padding=[1, 0, 0, 0]
)
self.option_buttons = []
option_labels = self.option_labels or [i18n._(f"tab:{opt}") for opt in self.options]
for ix, (lbl, opt_col, active) in enumerate(zip(option_labels, self.option_colors, self.option_active)):
button = CollapsablePanelTab(text=lbl, font_name=i18n.font_name, active_outline_color=opt_col, height=self.options_height, state="down" if active else "normal")
button = CollapsablePanelTab(
text=lbl,
font_name=i18n.font_name,
active_outline_color=opt_col,
height=self.options_height,
state="down" if active else "normal",
)
self.option_buttons.append(button)
button.bind(state=lambda *_args, _ix=ix: self.trigger_select(_ix))
self.open_close_button = TransparentIconButton( # << / >> collapse button
@@ -405,7 +422,11 @@ class CollapsablePanel(MDBoxLayout):
self.header.add_widget(Label()) # spacer
self.trigger_select(ix=None)
else:
self.header.add_widget(Label(text=i18n._(self.closed_label), font_name=i18n.font_name, halign="right", height=self.options_height))
self.header.add_widget(
Label(
text=i18n._(self.closed_label), font_name=i18n.font_name, halign="right", height=self.options_height
)
)
self.header.add_widget(self.open_close_button)
super().clear_widgets()
@@ -474,7 +495,9 @@ def draw_text(pos, text, font_name=None, **kw):
label = CoreLabel(text=text, bold=True, font_name=font_name or i18n.font_name, **kw) #
label.refresh()
Rectangle(
texture=label.texture, pos=(pos[0] - label.texture.size[0] / 2, pos[1] - label.texture.size[1] / 2), size=label.texture.size,
texture=label.texture,
pos=(pos[0] - label.texture.size[0] / 2, pos[1] - label.texture.size[1] / 2),
size=label.texture.size,
)
+31 -8
View File
@@ -12,7 +12,14 @@ from kivymd.uix.boxlayout import MDBoxLayout
from kivymd.uix.selectioncontrol import MDCheckbox
from kivymd.uix.textfield import MDTextField
from katrain.core.constants import OUTPUT_ERROR, OUTPUT_DEBUG, OUTPUT_INFO, AI_DEFAULT, AI_CONFIG_DEFAULT, AI_STRATEGIES_RECOMMENDED_ORDER
from katrain.core.constants import (
OUTPUT_ERROR,
OUTPUT_DEBUG,
OUTPUT_INFO,
AI_DEFAULT,
AI_CONFIG_DEFAULT,
AI_STRATEGIES_RECOMMENDED_ORDER,
)
from katrain.core.engine import KataGoEngine
from katrain.core.utils import i18n, find_package_resource
from katrain.gui.kivyutils import I18NSpinner, BackgroundMixin
@@ -116,11 +123,15 @@ class QuickConfigGui(MDBoxLayout):
Clock.schedule_once(lambda _dt: self.set_properties(self))
def collect_properties(self, widget) -> Dict:
if isinstance(widget, (LabelledTextInput, LabelledSpinner, LabelledCheckBox)) and getattr(widget, "input_property", None):
if isinstance(widget, (LabelledTextInput, LabelledSpinner, LabelledCheckBox)) and getattr(
widget, "input_property", None
):
try:
ret = {widget.input_property: widget.input_value}
except Exception as e:
raise InputParseError(f"Could not parse value for {widget.input_property} ({widget.__class__}): {e}") # TODO : on widget!
raise InputParseError(
f"Could not parse value for {widget.input_property} ({widget.__class__}): {e}"
) # TODO : on widget!
else:
ret = {}
for c in widget.children:
@@ -144,11 +155,16 @@ class QuickConfigGui(MDBoxLayout):
else:
if keys[-1] not in config:
config[keys[-1]] = ""
self.katrain.log(f"Configuration setting {repr(key)} was missing, created it, but this likely indicates a broken config file.", OUTPUT_ERROR)
self.katrain.log(
f"Configuration setting {repr(key)} was missing, created it, but this likely indicates a broken config file.",
OUTPUT_ERROR,
)
return config[keys[-1]], config, keys[-1]
def set_properties(self, widget):
if isinstance(widget, (LabelledTextInput, LabelledSpinner, LabelledCheckBox)) and getattr(widget, "input_property", None):
if isinstance(widget, (LabelledTextInput, LabelledSpinner, LabelledCheckBox)) and getattr(
widget, "input_property", None
):
value = self.get_setting(widget.input_property)[0]
if isinstance(widget, LabelledCheckBox):
widget.active = value is True
@@ -228,7 +244,9 @@ class ConfigTeacherPopup(QuickConfigGui):
savesgfs = self.katrain.config("trainer/save_feedback")
show_dots = self.katrain.config("trainer/show_dots")
for i, (color, threshold, undo, show_dot, savesgf) in enumerate(zip(EVAL_COLORS, thresholds, undos, show_dots, savesgfs)):
for i, (color, threshold, undo, show_dot, savesgf) in enumerate(
zip(EVAL_COLORS, thresholds, undos, show_dots, savesgfs)
):
self.add_option_widgets(
[
BackgroundMixin(background_color=color, size_hint=[0.9, 0.9]),
@@ -264,7 +282,9 @@ class AIPopup(QuickConfigGui):
self.help_label.text = i18n._(strategy.replace("ai:", "aihelp:"))
for k, v in sorted(mode_settings.items(), key=lambda kv: kv[0]):
self.options_grid.add_widget(DescriptionLabel(text=k))
self.options_grid.add_widget(wrap_anchor(LabelledFloatInput(text=str(v), input_property=f"ai/{strategy}/{k}")))
self.options_grid.add_widget(
wrap_anchor(LabelledFloatInput(text=str(v), input_property=f"ai/{strategy}/{k}"))
)
for _ in range((self.max_options - len(mode_settings)) * 2):
self.options_grid.add_widget(Label())
@@ -299,7 +319,10 @@ class LoadSGFPopup(BoxLayout):
def __init__(self, **kwargs):
super().__init__(**kwargs)
app = MDApp.get_running_app()
self.filesel.favorites = [(os.path.abspath(app.gui.config("general/sgf_load")), "SGF Load Dir"), (os.path.abspath(app.gui.config("general/sgf_save")), "SGF Save Dir")]
self.filesel.favorites = [
(os.path.abspath(app.gui.config("general/sgf_load")), "SGF Load Dir"),
(os.path.abspath(app.gui.config("general/sgf_save")), "SGF Save Dir"),
]
self.filesel.path = os.path.abspath(os.path.expanduser(app.gui.config("general/sgf_load")))
print(os.path.abspath(os.path.expanduser(app.gui.config("general/sgf_load"))))
self.filesel.select_string = i18n._("Load File")
+3 -1
View File
@@ -58,7 +58,9 @@ if platform == "win":
def last_modified_first(files, filesystem):
return sorted(f for f in files if filesystem.is_dir(f)) + sorted([f for f in files if not filesystem.is_dir(f)], key=lambda f: -getmtime(f))
return sorted(f for f in files if filesystem.is_dir(f)) + sorted(
[f for f in files if not filesystem.is_dir(f)], key=lambda f: -getmtime(f)
)
def get_home_directory():
+30 -6
View File
@@ -50,14 +50,30 @@ class ScoreGraph(BackgroundMixin):
score_granularity = 5
winrate_granularity = 10
self.score_scale = max(math.ceil(max(-score_values_range[0], score_values_range[1]) / score_granularity), 1) * score_granularity
self.winrate_scale = max(math.ceil(max(-winrate_values_range[0], winrate_values_range[1]) / winrate_granularity), 1) * winrate_granularity
self.score_scale = (
max(math.ceil(max(-score_values_range[0], score_values_range[1]) / score_granularity), 1)
* score_granularity
)
self.winrate_scale = (
max(math.ceil(max(-winrate_values_range[0], winrate_values_range[1]) / winrate_granularity), 1)
* winrate_granularity
)
xscale = self.width / max(len(score_values) - 1, 15)
available_height = self.height
score_line_points = [[self.pos[0] + i * xscale, self.pos[1] + self.height / 2 + available_height / 2 * (val / self.score_scale),] for i, val in enumerate(score_values)]
score_line_points = [
[
self.pos[0] + i * xscale,
self.pos[1] + self.height / 2 + available_height / 2 * (val / self.score_scale),
]
for i, val in enumerate(score_values)
]
winrate_line_points = [
[self.pos[0] + i * xscale, self.pos[1] + self.height / 2 + available_height / 2 * (val / self.winrate_scale),] for i, val in enumerate(winrate_values)
[
self.pos[0] + i * xscale,
self.pos[1] + self.height / 2 + available_height / 2 * (val / self.winrate_scale),
]
for i, val in enumerate(winrate_values)
]
self.score_points = sum(score_line_points, [])
self.winrate_points = sum(winrate_line_points, [])
@@ -67,10 +83,18 @@ class ScoreGraph(BackgroundMixin):
score_dot_point = score_line_points[self.highlighted_index]
winrate_dot_point = winrate_line_points[self.highlighted_index]
if math.isnan(score_dot_point[1]):
score_dot_point[1] = self.pos[1] + self.height / 2 + available_height / 2 * ((score_nn_values or [0])[-1] / self.score_scale)
score_dot_point[1] = (
self.pos[1]
+ self.height / 2
+ available_height / 2 * ((score_nn_values or [0])[-1] / self.score_scale)
)
self.score_dot_pos = [c - self.highlight_size / 2 for c in score_dot_point]
if math.isnan(winrate_dot_point[1]):
winrate_dot_point[1] = self.pos[1] + self.height / 2 + available_height / 2 * ((winrate_nn_values or [0])[-1] / self.winrate_scale)
winrate_dot_point[1] = (
self.pos[1]
+ self.height / 2
+ available_height / 2 * ((winrate_nn_values or [0])[-1] / self.winrate_scale)
)
self.winrate_dot_pos = [c - self.highlight_size / 2 for c in winrate_dot_point]
def update_value(self, node):
+6 -1
View File
@@ -51,7 +51,12 @@ setup(
],
python_requires=">=3.6, <4",
entry_points={"console_scripts": ["katrain=katrain.__main__:run_app"]},
classifiers=["Development Status :: 4 - Beta", "Operating System :: Microsoft :: Windows", "Operating System :: POSIX :: Linux", "Programming Language :: Python :: 3",],
classifiers=[
"Development Status :: 4 - Beta",
"Operating System :: Microsoft :: Windows",
"Operating System :: POSIX :: Linux",
"Programming Language :: Python :: 3",
],
packages=packages,
package_data=package_data,
)