diff --git a/board.py b/board.py index b5eda42..b498305 100644 --- a/board.py +++ b/board.py @@ -1,9 +1,122 @@ -from move import Move - class IllegalMoveException(Exception): pass +class Move: + GTP_COORD = "ABCDEFGHJKLMNOPQRSTUVWYXYZ" + PLAYERS = "BW" + SGF_COORD = [chr(i) for i in range(97, 123)] + + def __init__(self, player, coords=None, gtpcoords=None, sgfcoords=None, robot=False): + self.id = None + self.player = player + self.coords = coords or (gtpcoords and self.gtp2ix(gtpcoords)) or self.sgf2ix(sgfcoords) + self.children = [] + self.parent = None + self.robot = robot + self.analysis = None + self.pass_analysis = None + self.outdated_evaluation = None + self.evaluation = None + self.ownership = None + self.points_lost = 0 + self.previous_temperature = None + self.comment = "" + + def __repr__(self): + return f"{Move.PLAYERS[self.player]}{self.gtp()}" + + def __eq__(self, other): + return self.coords == other.coords and self.player == other.player + + def __hash__(self): + return self.gtp().__hash__() + + def play(self, move): + try: + return self.children[self.children.index(move)] + except ValueError: + move.parent = self + self.children.append(move) + return move + + def temperature(self): + if self.analysis: + best_score = float(self.analysis[0]["scoreLead"]) + worst_score = -float(self.pass_analysis[0]["scoreLead"]) + return best_score - worst_score + else: + return 0 + + def set_analysis(self,analysis_blob,is_pass): + if is_pass: + self.pass_analysis = analysis_blob['moveInfos'] + else: + self.analysis = analysis_blob['moveInfos'] + self.ownership = analysis_blob['ownership'] + if self.analysis and self.pass_analysis: + if self.parent.analysis: + self.evaluate() + for cm in self.children: + cm.evaluate() + + def evaluate(self): + previous_move = self.parent + best_score = float(previous_move.analysis[0]["scoreLead"]) + worst_score = -float(previous_move.pass_analysis[0]["scoreLead"]) + last_move_score = -float(self.analysis[0]["scoreLead"]) + self.previous_temperature = best_score - worst_score + self.points_lost = best_score - last_move_score + prev_analysis_current_move = [d for d in previous_move.analysis if d["move"] == self.gtp()] + + if abs(self.previous_temperature) > 0.5: + self.evaluation = (last_move_score - worst_score) / (best_score - worst_score) + self.move_options = [previous_move.analysis[0]["scoreLead"]] + else: + self.evaluation = None + if self.evaluation: + self.comment = f"Evaluation: {100*self.evaluation:.1f}%{' (AI Move)' if self.robot else ''}\n" + if prev_analysis_current_move: + self.outdated_evaluation = (prev_analysis_current_move[0]["scoreLead"] - worst_score) / ( + best_score - worst_score + ) + self.comment += f"(Was considered last move as: {100 * self.outdated_evaluation:.1f}%)\n" + else: + self.comment = "Temperature too low for evaluation\n" + self.comment += f"Estimate point loss: {self.points_lost:.1f}\n" + self.comment += f"Last move score was {last_move_score:.1f}\n" + self.comment += f"Score of top move was {previous_move.analysis[0]['scoreLead']:.1f} @ {previous_move.analysis[0]['move']}\n" + self.comment += f"Pass score was {worst_score:.1f}\n" + + @property + def is_pass(self): + return self.coords[0] is None + + def gtp2ix(self, gtpmove): + if "pass" in gtpmove: + return (None, None) + return Move.GTP_COORD.index(gtpmove[0]), int(gtpmove[1:]) - 1 + + def sgf2ix(self, sgfmove_with_board_size): + sgfmove, board_size = sgfmove_with_board_size + if sgfmove == "": + return (None, None) + return Move.SGF_COORD.index(sgfmove[0]), board_size - Move.SGF_COORD.index(sgfmove[1]) - 1 + + def gtp(self): + if self.is_pass: + return "pass" + return Move.GTP_COORD[self.coords[0]] + str(self.coords[1] + 1) + + def sgfcoords(self, board_size): + return f"{Move.SGF_COORD[self.coords[0]]}{Move.SGF_COORD[board_size - self.coords[1] - 1]}" + + def sgf(self, board_size): + if self.is_pass: + return f"{Move.PLAYERS[self.player]}[]" + else: + return f"{Move.PLAYERS[self.player]}[{self.sgfcoords(board_size)}]" + class Board: _move_id_counter = 0 # used to make a map to all moves across all games @@ -13,7 +126,7 @@ class Board: self.root = Move(1, (None, None)) # root is 1=white so black is first self.root.id = -1 self.current_move = self.root - self.all_moves = {} + self.all_moves = {-1: self.root} self._init_chains() # -- move tree functions -- @@ -107,18 +220,20 @@ class Board: @property def current_player(self): - return self.current_move.player + return 1 - self.current_move.player # --analysis def store_analysis(self, json): - if json["id"].starts_with("PASS_"): + if json["id"].startswith("PASS_"): id = int(json["id"].lstrip("PASS_")) + is_pass = True else: id = int(json["id"]) + is_pass = False move = self.all_moves.get(id) if move: # else this should be old - move.set_analysis(json) + move.set_analysis(json,is_pass) else: print("WARNING: ORPHANED ANALYSIS FOUND - RECENT NEW GAME?") diff --git a/controller.py b/controller.py index 58ad0ff..8b718d2 100644 --- a/controller.py +++ b/controller.py @@ -1,101 +1,316 @@ from kivy.storage.jsonstore import JsonStore from kivy.uix.gridlayout import GridLayout +import json +import copy +import random +import re +import shlex +import subprocess +import threading +import time +from queue import Queue -from engine import KataEngine -from move import Move +from board import Board, Move, IllegalMoveException -Config = JsonStore("config.json") +Config = JsonStore("Config.json") class EngineControls(GridLayout): + def __init__(self, **kwargs): super(EngineControls, self).__init__(**kwargs) - self.engine = KataEngine(self, Config) + self.command = shlex.split(Config.get("engine")["command"]) - def restart(self, boardsize=None): - self.engine.restart(boardsize) - - def action(self, message, *args): - self.engine.action(message, *args) - - @property - def board(self): - return self.engine.board - - @property - def ready(self): - return self.engine.ready - - @property - def boardsize(self): - return self.engine.boardsize - - @property - def stones(self): - return self.engine.board.stones - - @property - def moves(self): - return self.engine.board.moves + analysis_settings = Config.get("analysis") + self.visits = [ + [analysis_settings["pass_visits"], analysis_settings["visits"]], + [analysis_settings["pass_visits_fast"], analysis_settings["visits_fast"]], + ] + self.min_nopass_visits = analysis_settings["nopass_visits"] + self.train_settings = Config.get("trainer") + self.debug = Config.get("debug")["level"] + self.board_size = Config.get("board")["size"] + self.komi = Config.get("board")["komi"] + self.ready = False + self.message_queue = None + self.board = Board(self.board_size) + self.outstanding_analysis_queries = [] # allows faster interaction while kata is starting + self.kata = None @property def current_player(self): - return self.engine.current_player + return self.board.current_player + + @property + def stones(self): + return self.board.stones + + @property + def moves(self): + return self.board.moves def redraw(self, include_board=False): if include_board: self.parent.board.draw_board() self.parent.board.redraw() - def update_analysis(self, analysis, mode, ownership): - for d in analysis: - d["scoreMean"] = float(d["scoreMean"]) - - if mode == 0: - pm = [d for d in analysis if d["move"] == "pass"] - npm = [d for d in analysis if d["move"] != "pass"] - if pm: - pv = sum([int(d["visits"]) for d in pm], 0) - npv = sum([int(d["visits"]) for d in npm], 0) - print("pass visits", pv, "other", npv) - if pv > npv: - print(analysis) - self.moves[-1].pass_analysis = [d for d in analysis if d["move"] != "pass"] + def restart(self, board_size=None): + self.ready = False + if not self.message_queue: + self.message_queue = Queue() + self.thread = threading.Thread(target=self._engine_thread, daemon=True).start() else: - if ownership: - self.moves[-1].ownership = [float(p) for p in ownership[0].strip().split(" ")] - best = analysis[0]["scoreMean"] - worst = -self.moves[-1].pass_analysis[0]["scoreMean"] - for d in analysis: - d["evaluation"] = (d["scoreMean"] - worst) / (best - worst) - self.moves[-1].analysis = analysis + with self.message_queue.mutex: + self.message_queue.queue.clear() + self.action("init", board_size or self.board_size) + + def action(self, message, *args): + self.message_queue.put([message, *args]) + + # engine main loop + def _engine_thread(self): + self.kata = subprocess.Popen(self.command, stdin=subprocess.PIPE, stdout=subprocess.PIPE) + threading.Thread(target=self._analysis_read_thread, daemon=True).start() + + msg, *args = self.message_queue.get() + while True: + try: + if self.debug: + print("MESSAGE", msg, args) + getattr(self, f"_do_{msg.replace('-','_')}")(*args) + except Exception as e: + self.info.text = f"Exception in Engine thread: {e}" + raise + msg, *args = self.message_queue.get() + + def play(self, move): + try: + mr = self.board.play(move) + except IllegalMoveException as e: + print(str(e)) + self.info.text = f"Illegal move: {str(e)}" + return + print("PLAYED",move,self.board.stones) + self._request_analysis(mr) + + # engine action functions + def _do_play(self, *args): + print("CURRENT PLAYER",self.current_player) + move = Move(player=self.current_player, coords=args[0]) + self.play(move) + + self.undo.disabled = True # undo while waiting for this does weird things + undid = False + self.info.text = "" + if self.auto_undo.active(1 - self.current_player): + undid = self._auto_undo(move) + if self.ai_auto.active and not undid: + self._do_aimove(move,True) + self.undo.disabled = False + self.redraw() + + def _evaluate_move(self, move, show=True): + while not move.analysis: + time.sleep(0.01) # wait for analysis + if self.board.current_move.evaluation and show: + self.info.text = f"Your move {self.moves[-1].gtp()} was {100 * self.moves[-1].evaluation:.1f}% efficient and lost {self.moves[-1].points_lost:.1f} point(s).\n" + + def _auto_undo(self, move): + ts = self.train_settings + self.info.text = "Evaluating..." + self._evaluate_move() + if ( + move.evaluation + and move.evaluation < ts["undo_eval_threshold"] + and move.points_lost >= ts["undo_point_threshold"] + and ts["num_undo_prompts"] > 0 + ): + if move.outdated_evaluation: + outdated_points_lost = (1 - move.outdated_evaluation) * move.points_lost / (1 - move.evaluation) + # so if the move was not that far off (>undo_outdated_eval_threshold) and according to last move's analysis it was fine, don't undo. + if ( + move.outdated_evaluation + and ( + move.outdated_evaluation >= ts["undo_eval_threshold"] + or outdated_points_lost < ts["undo_point_threshold"] + ) + and ( + move.evaluation > ts["undo_outdated_eval_threshold"] + or outdated_points_lost < ts["undo_point_threshold"] + ) + ): + self.info.text += f"\nBut according to my previous evaluation it was {move.outdated_evaluation*100:.1f}% effective and lost {outdated_points_lost:.1f} point(s), so let's continue anyway.\n" + else: + if len(self.board.current_move.parent.children) <= ts["num_undo_prompts"]: + self.info.text += f"\nLet's try again.\n" + self.board.undo() + return True + else: + evaled_moves = sorted( + [m for m in self.board.current_move.parent.children if m.evaluation], key=lambda m: -m.evaluation + ) + if evaled_moves and evaled_moves[0].coords != move.coords: + self.board.undo() + self.board.play(evaled_moves[0]) + summary = "\n".join(f"{m.gtp()}: {100*m.evaluation:.1f}% effective" for m in evaled_moves) + self.info.text += ( + f"\nYour moves:\n{summary}.\nLet's continue with {evaled_moves[0].gtp()}.\n" + ) + return False + + def _do_aimove(self, move, auto=False): + ts = self.train_settings + if not auto: + self.info.text = "Thinking..." + self._evaluate_move(auto and not self.auto_undo.active(1 - self.current_player)) + # select move + pos_moves = [ + (d["move"], float(d["scoreMean"]), d["evaluation"]) + for d in move.analysis + if int(d["visits"]) >= ts["balance_play_min_visits"] + ] + if ts["show_ai_options"]: + self.info.text += "AI Options: " + " ".join( + [f"{move}({100*eval:.0f}%,{score:.1f}pt)" for move, score, eval in pos_moves] + ) + selmove = pos_moves[0][0] + if ( + self.ai_balance.active and pos_moves[0][0] != "pass" + ): # don't play suicidal to balance score - pass when it's best + selmoves = [ + move + for move, score, eval in pos_moves + if eval > ts["balance_play_randomize_eval"] + or eval > ts["balance_play_min_eval"] + and score > ts["balance_play_target_score"] + ] + selmove = random.choice(selmoves) # some kind of when further ahead play worse? + self.board.play(Move(player=self.current_player, gtpcoords=selmove, robot=True)) + + def _do_undo(self): + if self.ai_auto.active and self.board.current_move.robot: + self.board.undo() + if ( + self.ai_lock.active + and self.auto_undo.active(self.board.current_move.parent.player) + and len(self.board.current_move.parent.player.children) > self.train_settings["num_undo_prompts"] + ): + self.info.text = ( + f"Can't undo more than {self.train_settings['num_undo_prompts']} time(s) when locked" + ) + return + self.board.undo() + + def _do_init(self, board_size, komi=None): + self.board_size = board_size + self.board = Board(board_size) + self._request_analysis(self.board.root) + self.redraw(include_board=True) + self.ready = True + + def _do_analyze_sgf(self, sgf): + self._do_init(self.board_size, self.komi) + sgfmoves = re.findall(r"([BW])\[([a-z]{2})\]", sgf) + moves = [Move(player=Move.PLAYERS.index(p.upper()), sgfcoords=(mv, self.board_size)) for p, mv in sgfmoves] + for move in moves: + self.board.play(move) + while not all(m.analysis for m in moves): + time.sleep(0.01) + self.info.text = f"{sum([1 if m.analysis else 0 for m in moves])}/{len(moves)} analyzed" + + # analysis thread + def _analysis_read_thread(self): + while True: + while self.outstanding_analysis_queries: + print("processing outstanding query") + self._send_analysis_query(self.outstanding_analysis_queries.pop(0)) + print('reading kata line') + line = self.kata.stdout.readline() + print("KATA LINE", line) + self.board.store_analysis(json.loads(line)) + + def _send_analysis_query(self,query): + if self.kata: + self.kata.stdin.write((json.dumps(query) + "\n").encode()) + self.kata.stdin.flush() + else: # early on / root / etc + self.outstanding_analysis_queries.append(copy.copy(query)) + + def _request_analysis(self, move): + move_id = move.id + moves = self.board.moves + fast = self.ai_fast.active + query = { + "id": str(move_id), + "moves": [str(m) for m in moves], + "rules": "japanese", + "komi": self.komi, + "boardXSize": self.board_size, + "boardYSize": self.board_size, + "analyzeTurns": [len(moves)], + "includeOwnership": True, + "maxVisits": self.visits[fast][1], + } + print('query',query) + self._send_analysis_query(query) + query.update({"id": f"PASS_{move_id}", "maxVisits": self.visits[fast][0], "includeOwnership": False}) # TODO: merge? + query["moves"] += ["pass"] + query["analyzeTurns"][0] += 1 + print("pass-query", query) + self._send_analysis_query(query) + + + +# def update_analysis(self, analysis, mode, ownership): +# for d in analysis: +# d["scoreMean"] = float(d["scoreMean"]) +# +# if mode == 0: +# pm = [d for d in analysis if d["move"] == "pass"] +# npm = [d for d in analysis if d["move"] != "pass"] +# if pm: +# pv = sum([int(d["visits"]) for d in pm], 0) +# npv = sum([int(d["visits"]) for d in npm], 0) +# print("pass visits", pv, "other", npv) +# if pv > npv: +# print(analysis) +# self.moves[-1].pass_analysis = [d for d in analysis if d["move"] != "pass"] +# else: +# if ownership: +# self.moves[-1].ownership = [float(p) for p in ownership[0].strip().split(" ")] +# best = analysis[0]["scoreMean"] +# worst = -self.moves[-1].pass_analysis[0]["scoreMean"] +# for d in analysis: +# d["evaluation"] = (d["scoreMean"] - worst) / (best - worst) +# self.moves[-1].analysis = analysis +# if self.eval.active(1 - self.current_player): +# self.temperature.text = f"{self.moves[-1].temperature():.1f}" +# self.score.text = f"{Move.PLAYERS[self.current_player]}{float(analysis[0]['scoreMean']):+.1f}".replace("-", "\u2013") # en dash +# if len(self.moves) >= 2 and self.moves[-2].analysis: +# self.moves[-1].evaluate(self.moves[-2]) +# if self.eval.active(1 - self.current_player): +# if self.moves[-1].evaluation: +# self.evaluation.text = f"{100 * self.moves[-1].evaluation:.1f}%" +# else: +# self.evaluation.text = "N/A" +# self.redraw(include_board=False) # for dots and stuff + - if self.eval.active(1 - self.current_player): - self.temperature.text = f"{self.moves[-1].temperature():.1f}" - self.score.text = f"{Move.PLAYERS[self.current_player]}{float(analysis[0]['scoreMean']):+.1f}".replace("-", "\u2013") # en dash - if len(self.moves) >= 2 and self.moves[-2].analysis: - self.moves[-1].evaluate(self.moves[-2]) - if self.eval.active(1 - self.current_player): - if self.moves[-1].evaluation: - self.evaluation.text = f"{100 * self.moves[-1].evaluation:.1f}%" - else: - self.evaluation.text = "N/A" - self.redraw(include_board=False) # for dots and stuff def sgf(self): def sgfify(mvs): - return f"(;GM[1]FF[4]SZ[{self.boardsize}]KM[{self.engine.komi}]RU[CN];" + ";".join(mvs) + ")" + return f"(;GM[1]FF[4]SZ[{self.board_size}]KM[{self.komi}]RU[CN];" + ";".join(mvs) + ")" def format_move(m, pm): undo_comment = "".join(f"\nUndo: {u.gtp()} was {100*u.evaluation:.1f}%" for u in pm.undos if u.evaluation) - undo_cr = "".join(f"MA[{u.sgfcoords(self.boardsize)}]" for u in pm.undos if u.coords[0]) + undo_cr = "".join(f"MA[{u.sgfcoords(self.board_size)}]" for u in pm.undos if u.coords[0]) if pm.analysis and pm.analysis[0]["move"] != "pass": - best_sq = f"SQ[{Move(gtpcoords=pm.analysis[0]['move'], player=0).sgfcoords(self.boardsize)}]" + best_sq = f"SQ[{Move(gtpcoords=pm.analysis[0]['move'], player=0).sgfcoords(self.board_size)}]" else: best_sq = "" - return m.sgf(self.boardsize) + f"C[{m.comment}{undo_comment}]{undo_cr}{best_sq}" + return m.sgf(self.board_size) + f"C[{m.comment}{undo_comment}]{undo_cr}{best_sq}" - sgfmoves_small = [mv.sgf(self.boardsize) for mv in self.moves[1:]] + sgfmoves_small = [mv.sgf(self.board_size) for mv in self.moves[1:]] sgfmoves = [format_move(mv, pmv) for mv, pmv in zip(self.moves[1:], self.moves[:-1])] with open("out.sgf", "w") as f: diff --git a/engine.py b/engine.py deleted file mode 100644 index 96d5b6c..0000000 --- a/engine.py +++ /dev/null @@ -1,238 +0,0 @@ -import json -import random -import re -import shlex -import subprocess -import threading -import time -from queue import Queue - -from board import Board, IllegalMoveException -from move import Move - - -class KataEngine: - def __init__(self, controls, config): - self.controls = controls - self.command = shlex.split(config.get("engine")["command"]) - - analysis_settings = config.get("analysis") - self.visits = [ - [analysis_settings["pass_visits"], analysis_settings["visits"]], - [analysis_settings["pass_visits_fast"], analysis_settings["visits_fast"]], - ] - self.min_nopass_visits = analysis_settings["nopass_visits"] - self.train_settings = config.get("trainer") - self.debug = config.get("debug")["level"] - self.boardsize = config.get("board")["size"] - self.komi = config.get("board")["komi"] - self.ready = False - self.stones = [] - self.message_queue = None - self.board = Board(self.boardsize) - - self.kata = None - - @property - def current_player(self): - return self.board.current_player - - def restart(self, boardsize): - self.ready = False - if not self.message_queue: - self.message_queue = Queue() - self.thread = threading.Thread(target=self._engine_thread, daemon=True).start() - else: - with self.message_queue.mutex: - self.message_queue.queue.clear() - self.stones = [] - self.action("init", boardsize or self.boardsize) - - def action(self, message, *args): - self.message_queue.put([message, *args]) - - # engine main loop - def _engine_thread(self): - self.kata = subprocess.Popen(self.command, stdin=subprocess.PIPE, stdout=subprocess.PIPE) - print("STARTING KATAGO", self.command, self.kata) - analysis_thread = threading.Thread(target=self._analyze_thread, daemon=True).start() - - msg, *args = self.message_queue.get() - while True: - try: - if self.debug: - print("MESSAGE", msg, args) - getattr(self, f"_do_{msg.replace('-','_')}")(*args) - except Exception as e: - self.controls.info.text = f"Exception in Engine thread: {e}" - raise - msg, *args = self.message_queue.get() - - def play(self, move): - try: - mr = self.board.play(move) - except IllegalMoveException as e: - print(str(e)) - self.controls.info.text = f"Illegal move: {str(e)}" - return - print("PLAYED",move,self.board.stones) - self._request_analysis(mr) - - def _request_analysis(self, move): - while not self.kata: - print("waiting for kata to start") - time.sleep(0.05) - move_id = move.id - moves = self.board.moves - fast = self.controls.ai_fast.active - query = { - "id": str(move_id), - "moves": [str(m) for m in moves], - "rules": "japanese", - "komi": self.komi, - "boardXSize": self.boardsize, - "boardYSize": self.boardsize, - "analyzeTurns": [len(moves) - 1], - "includeOwnership": True, - "maxVisits": self.visits[fast][1], - } - print("query", query) - self.kata.stdin.write(json.dumps(query).encode()) - query.update({"id": f"PASS_{move_id}", "maxVisits": self.visits[fast][0], "includeOwnership": True}) - query["moves"] += ["pass"] - query["analyzeTurns"][0] += 1 - - print("pass-query", query) - self.kata.stdin.write(json.dumps(query).encode()) - - # engine action functions - def _do_play(self, *args): - move = Move(player=self.current_player, coords=args[0]) - self.play(move) - - self.controls.undo.disabled = True # undo while waiting for this does weird things - undid = False - self.controls.info.text = "" - if self.controls.auto_undo.active(1 - self.current_player): - undid = self._auto_undo(move) - if self.controls.ai_auto.active and not undid: - self._do_aimove(move,True) - self.controls.undo.disabled = False - self.controls.redraw() - - def _evaluate_move(self, move, show=True): - while not move.analysis: - time.sleep(0.01) # wait for analysis - if self.board.current_move.evaluation and show: - self.controls.info.text = f"Your move {self.moves[-1].gtp()} was {100 * self.moves[-1].evaluation:.1f}% efficient and lost {self.moves[-1].points_lost:.1f} point(s).\n" - - def _auto_undo(self, move): - ts = self.train_settings - self.controls.info.text = "Evaluating..." - self._evaluate_move() - if ( - move.evaluation - and move.evaluation < ts["undo_eval_threshold"] - and move.points_lost >= ts["undo_point_threshold"] - and ts["num_undo_prompts"] > 0 - ): - if move.outdated_evaluation: - outdated_points_lost = (1 - move.outdated_evaluation) * move.points_lost / (1 - move.evaluation) - # so if the move was not that far off (>undo_outdated_eval_threshold) and according to last move's analysis it was fine, don't undo. - if ( - move.outdated_evaluation - and ( - move.outdated_evaluation >= ts["undo_eval_threshold"] - or outdated_points_lost < ts["undo_point_threshold"] - ) - and ( - move.evaluation > ts["undo_outdated_eval_threshold"] - or outdated_points_lost < ts["undo_point_threshold"] - ) - ): - self.controls.info.text += f"\nBut according to my previous evaluation it was {move.outdated_evaluation*100:.1f}% effective and lost {outdated_points_lost:.1f} point(s), so let's continue anyway.\n" - else: - if len(self.board.current_move.parent.children) <= ts["num_undo_prompts"]: - self.controls.info.text += f"\nLet's try again.\n" - self.board.undo() - return True - else: - evaled_moves = sorted( - [m for m in self.board.current_move.parent.children if m.evaluation], key=lambda m: -m.evaluation - ) - if evaled_moves and evaled_moves[0].coords != move.coords: - self.board.undo() - self.board.play(evaled_moves[0]) - summary = "\n".join(f"{m.gtp()}: {100*m.evaluation:.1f}% effective" for m in evaled_moves) - self.controls.info.text += ( - f"\nYour moves:\n{summary}.\nLet's continue with {evaled_moves[0].gtp()}.\n" - ) - return False - - def _do_aimove(self, move, auto=False): - ts = self.train_settings - if not auto: - self.controls.info.text = "Thinking..." - self._evaluate_move(auto and not self.controls.auto_undo.active(1 - self.current_player)) - # select move - pos_moves = [ - (d["move"], float(d["scoreMean"]), d["evaluation"]) - for d in move.analysis - if int(d["visits"]) >= ts["balance_play_min_visits"] - ] - if ts["show_ai_options"]: - self.controls.info.text += "AI Options: " + " ".join( - [f"{move}({100*eval:.0f}%,{score:.1f}pt)" for move, score, eval in pos_moves] - ) - selmove = pos_moves[0][0] - if ( - self.controls.ai_balance.active and pos_moves[0][0] != "pass" - ): # don't play suicidal to balance score - pass when it's best - selmoves = [ - move - for move, score, eval in pos_moves - if eval > ts["balance_play_randomize_eval"] - or eval > ts["balance_play_min_eval"] - and score > ts["balance_play_target_score"] - ] - selmove = random.choice(selmoves) # some kind of when further ahead play worse? - self.board.play(Move(player=self.current_player, gtpcoords=selmove, robot=True)) - - def _do_undo(self): - if self.controls.ai_auto.active and self.board.current_move.robot: - self.board.undo() - if ( - self.controls.ai_lock.active - and self.controls.auto_undo.active(self.board.current_move.parent.player) - and len(self.board.current_move.parent.player.children) > self.train_settings["num_undo_prompts"] - ): - self.controls.info.text = ( - f"Can't undo more than {self.train_settings['num_undo_prompts']} time(s) when locked" - ) - return - self.board.undo() - - def _do_init(self, boardsize, komi=None): - self.boardsize = boardsize - self.stones = [] - self.board = Board(boardsize) - self._request_analysis(self.board.root) - self.controls.redraw(include_board=True) - self.ready = True - - def _do_analyze_sgf(self, sgf): - self._do_init(self.boardsize, self.komi) - sgfmoves = re.findall(r"([BW])\[([a-z]{2})\]", sgf) - moves = [Move(player=Move.PLAYERS.index(p.upper()), sgfcoords=(mv, self.boardsize)) for p, mv in sgfmoves] - for move in moves: - self.board.play(move) - while not all(m.analysis for m in moves): - time.sleep(0.01) - self.controls.info.text = f"{sum([1 if m.analysis else 0 for m in moves])}/{len(moves)} analyzed" - - # analysis thread - def _analyze_thread(self): - while True: - line = self.kata.stdout.readline() - print("KATA LINE", line) - self.board.store_analysis(json.loads(line)) diff --git a/katrain.py b/katrain.py index 8753019..04c5b71 100644 --- a/katrain.py +++ b/katrain.py @@ -7,12 +7,9 @@ from kivy.uix.widget import Widget import math from controller import Config -from move import Move +from board import Move from kivyutils import * -# (;GM[1]SZ[9]KM[7.5]RU[JP];B[fe];W[de];B[ec];W[dc];B[eg];W[dg];B[dh];W[ed];B[fd];W[ef];B[ff];W[eb];B[fc];W[eh];B[fg];W[ch];B[ee];W[df];B[dd];W[cd];B[db];W[cc];B[cb];W[fb];B[gb];W[bb];B[ea];W[ca];B[fa];W[fh];B[gh];W[ba];B[fi];W[di];B[da];W[ed];B[bc];W[bd];B[dd];W[ei];B[gi];W[ed]) -# (;GM[1]SZ[19]KM[7.5]RU[JP];B[qd];W[pp];B[cd];W[cp];B[ec];W[od];B[oc];W[nc];B[pc];W[nd];B[qf];W[jc];B[eq];W[do];B[hq];W[jq];B[cr];W[qn];B[cj];W[cl];B[nq];W[oq];B[np];W[lp];B[cg];W[nn];B[lr];W[kq];B[mo];W[kn];B[qi];W[mn];B[hc];W[qk];B[lc];W[je];B[jb];W[kb];B[kc];W[ib];B[jd];W[ja];B[lb];W[id];B[kd];W[ic];B[oj];W[pd];B[qe];W[qc];B[qb];W[rc];B[rb];W[pb];B[ob];W[nb];B[pa];W[lf];B[la];W[ma];B[mj];W[nk];B[ge];W[hd];B[hg];W[fc];B[fb];W[gc];B[ed];W[og];B[of];W[nf];B[pg];W[fq];B[fp];W[er];B[fr];W[dq];B[gq];W[dr];B[eo];W[en];B[fn];W[fm];B[gn];W[gm];B[dn];W[em];B[co];W[dp];B[hn];W[ep];B[fq];W[fj];B[jo];W[jn];B[jr];W[kr];B[hr];W[fo];B[js];W[ks];B[iq];W[io];B[ng];W[mg];B[oh];W[ne];B[hm];W[hl];B[fg];W[ip];B[go];W[gs];B[fs];W[ok];B[mh];W[lh];B[li];W[hp];B[im];W[hs];B[il];W[ir];B[ke];W[kf];B[gp];W[bk];B[kl];W[bj];B[lo];W[ko];B[ll];W[ml];B[rj];W[rk];B[kh];W[ci];B[lg];W[if];B[hk];W[ei];B[gi];W[bg];B[bh];W[ch];B[bf];W[dg];B[cf];W[df];B[pj];W[pk];B[sk];W[sl];B[sj];W[rl];B[gb];W[hb];B[gl];W[fl];B[gj];W[de];B[bi];W[ai];B[ag];W[dd];B[dc];W[ce];B[be];W[cc];B[bd];W[qj];B[ri];W[eh];B[lm];W[ln];B[jg];W[jf];B[mf];W[me];B[ig];W[mk];B[lk];W[nj];B[ni];W[eo];B[mm];W[nm];B[fd];W[gd];B[hf];W[ga];B[ea];W[jm];B[jl];W[pe];B[fh];W[fi];B[es];W[ds];B[fk];W[ek];B[gk];W[mb];B[eg];W[oa];B[na];W[fe];B[he];W[oa];B[pb];W[ee];B[cb];W[ie];B[ff];W[dh];B[pf];W[mg];B[ej];W[dj];B[kg];W[mf];B[in];W[jp];B[na];W[aj];B[ah];W[oa];B[pq];W[or];B[na];W[ha];B[oa];W[fa];B[eb];W[];B[gr];W[is];B[oe];W[ho];B[km];W[ef]) - COLORS = Config.get("ui")["stones"] GHOST_ALPHA = Config.get("ui")["ghost_alpha"] @@ -36,12 +33,12 @@ class BadukPanWidget(Widget): def on_touch_down(self, touch): xd, xp = self._find_closest(touch.x) yd, yp = self._find_closest(touch.y) - prevghost = self.ghost_stone + prev_ghost = self.ghost_stone if self.engine.ready and max(yd, xd) < self.grid_size / 2 and (xp, yp) not in [m.coords for m in self.engine.stones]: self.ghost_stone = (xp, yp) else: self.ghost_stone = None - if prevghost != self.ghost_stone: + if prev_ghost != self.ghost_stone: self.redraw() def on_touch_move(self, touch): # on_motion on_touch_move @@ -85,27 +82,27 @@ class BadukPanWidget(Widget): # grid lines margin = Config.get("ui")["board_margin"] - self.grid_size = board.size[0] / (self.engine.boardsize - 1 + 1.5 * margin) + self.grid_size = board.size[0] / (self.engine.board_size - 1 + 1.5 * margin) self.stone_size = self.grid_size * Config.get("ui")["stone_size"] - self.gridpos = [math.floor((margin + i) * self.grid_size + 0.5) for i in range(self.engine.boardsize)] + self.gridpos = [math.floor((margin + i) * self.grid_size + 0.5) for i in range(self.engine.board_size)] line_color = Config.get("ui")["line_color"] Color(*line_color) lo, hi = self.gridpos[0], self.gridpos[-1] - for i in range(self.engine.boardsize): + for i in range(self.engine.board_size): Line(points=[(self.gridpos[i], lo), (self.gridpos[i], hi)]) Line(points=[(lo, self.gridpos[i]), (hi, self.gridpos[i])]) # star points - star_point_pos = 3 if self.engine.boardsize <= 11 else 4 + star_point_pos = 3 if self.engine.board_size <= 11 else 4 starpt_size = self.grid_size * Config.get("ui")["starpoint_size"] - for x in [star_point_pos - 1, self.engine.boardsize - star_point_pos, int(self.engine.boardsize / 2)]: - for y in [star_point_pos - 1, self.engine.boardsize - star_point_pos, int(self.engine.boardsize / 2)]: + for x in [star_point_pos - 1, self.engine.board_size - star_point_pos, int(self.engine.board_size / 2)]: + for y in [star_point_pos - 1, self.engine.board_size - star_point_pos, int(self.engine.board_size / 2)]: draw_circle((self.gridpos[x], self.gridpos[y]), starpt_size, line_color) # coordinates Color(0.25, 0.25, 0.25) - for i in range(self.engine.boardsize): + for i in range(self.engine.board_size): draw_text(pos=(self.gridpos[i], lo / 2), text=Move.GTP_COORD[i], font_size=self.grid_size / 1.5) draw_text(pos=(lo / 2, self.gridpos[i]), text=str(i + 1), font_size=self.grid_size / 1.5) @@ -131,8 +128,8 @@ class BadukPanWidget(Widget): rsz = self.grid_size * 0.2 ix = 0 cp = self.engine.current_player - for y in range(self.engine.boardsize - 1, -1, -1): - for x in range(self.engine.boardsize): + for y in range(self.engine.board_size - 1, -1, -1): + for x in range(self.engine.board_size): ix_owner = cp if ownership[ix] > 0 else 1 - cp if ix_owner != (has_stone.get((x, y), -1)): Color(*COLORS[ix_owner], abs(ownership[ix])) @@ -168,7 +165,7 @@ class BadukPanWidget(Widget): else: text = "pass" Color(0.45, 0.05, 0.45, 0.5) - center = self.gridpos[int(self.engine.boardsize / 2)] + center = self.gridpos[int(self.engine.board_size / 2)] Ellipse(pos=(center - self.grid_size * 1.5, center - self.grid_size * 1.5), size=(self.grid_size * 3, self.grid_size * 3)) Color(0.15, 0.15, 0.15) draw_text(pos=(center, center), text=text, font_size=self.grid_size * 0.66, halign="center", outline_color=[0.95, 0.95, 0.95]) diff --git a/move.py b/move.py deleted file mode 100644 index 8a0ef1f..0000000 --- a/move.py +++ /dev/null @@ -1,109 +0,0 @@ - - -class Move: - GTP_COORD = "ABCDEFGHJKLMNOPQRSTUVWYXYZ" - PLAYERS = "BW" - SGF_COORD = [chr(i) for i in range(97, 123)] - - def __init__(self, player, coords=None, gtpcoords=None, sgfcoords=None, robot=False): - self.id = None - self.player = player - self.coords = coords or (gtpcoords and self.gtp2ix(gtpcoords)) or self.sgf2ix(sgfcoords) - self.children = [] - self.parent = None - self.robot = robot - self.analysis = None - self.pass_analysis = None - self.outdated_evaluation = None - self.evaluation = None - self.ownership = None - self.points_lost = 0 - self.previous_temperature = None - self.comment = "" - - def __repr__(self): - return f"{Move.PLAYERS[self.player]}{self.gtp()}" - - def __eq__(self, other): - return self.coords == other.coords and self.player == other.player - - def __hash__(self): - return self.gtp().__hash__() - - def play(self, move): - try: - return self.children[self.children.index(move)] - except ValueError: - move.parent = self - self.children.append(move) - return move - - def temperature(self): - if self.analysis: - best_score = float(self.analysis[0]["scoreLead"]) - worst_score = -float(self.pass_analysis[0]["scoreLead"]) - return best_score - worst_score - else: - return 0 - - def evaluate(self,analysis_blob): - self.analysis = analysis_blob['moveInfos'] - self.ownership = analysis_blob['ownership'] - previous_move = self.parent - if not self.analysis and self.pass_analysis and previous_move.analysis: - return - # TODO: update children? - best_score = float(previous_move.analysis[0]["scoreLead"]) - worst_score = -float(previous_move.pass_analysis[0]["scoreLead"]) - last_move_score = -float(self.analysis[0]["scoreLead"]) - self.previous_temperature = best_score - worst_score - self.points_lost = best_score - last_move_score - prev_analysis_current_move = [d for d in previous_move.analysis if d["move"] == self.gtp()] - - if abs(self.previous_temperature) > 0.5: - self.evaluation = (last_move_score - worst_score) / (best_score - worst_score) - self.move_options = [previous_move.analysis[0]["scoreLead"]] - else: - self.evaluation = None - if self.evaluation: - self.comment = f"Evaluation: {100*self.evaluation:.1f}%{' (AI Move)' if self.robot else ''}\n" - if prev_analysis_current_move: - self.outdated_evaluation = (prev_analysis_current_move[0]["scoreLead"] - worst_score) / ( - best_score - worst_score - ) - self.comment += f"(Was considered last move as: {100 * self.outdated_evaluation:.1f}%)\n" - else: - self.comment = "Temperature too low for evaluation\n" - self.comment += f"Estimate point loss: {self.points_lost:.1f}\n" - self.comment += f"Last move score was {last_move_score:.1f}\n" - self.comment += f"Score of top move was {previous_move.analysis[0]['scoreLead']:.1f} @ {previous_move.analysis[0]['move']}\n" - self.comment += f"Pass score was {worst_score:.1f}\n" - - @property - def is_pass(self): - return self.coords[0] is None - - def gtp2ix(self, gtpmove): - if "pass" in gtpmove: - return (None, None) - return Move.GTP_COORD.index(gtpmove[0]), int(gtpmove[1:]) - 1 - - def sgf2ix(self, sgfmove_with_boardsize): - sgfmove, boardsize = sgfmove_with_boardsize - if sgfmove == "": - return (None, None) - return Move.SGF_COORD.index(sgfmove[0]), boardsize - Move.SGF_COORD.index(sgfmove[1]) - 1 - - def gtp(self): - if self.is_pass: - return "pass" - return Move.GTP_COORD[self.coords[0]] + str(self.coords[1] + 1) - - def sgfcoords(self, boardsize): - return f"{Move.SGF_COORD[self.coords[0]]}{Move.SGF_COORD[boardsize - self.coords[1] - 1]}" - - def sgf(self, boardsize): - if self.is_pass: - return f"{Move.PLAYERS[self.player]}[]" - else: - return f"{Move.PLAYERS[self.player]}[{self.sgfcoords(boardsize)}]"