ai2gtp and fixes

TODO

@@ -7,7 +7,7 @@
 [x] graph wonky on branch switch -> check in children &c.
 [x] Scrolling add a move on the board instead of navigating through the game. This was already the case in the 0.9 version and it's quite annoying as scrolling seemed only natural and I kept forgetting not to do it :p
 [x] show PV on hint hover?  Although Katrain wasn't meant to be like Lizzie to begin with, it would be really neat if we could visualize the expected variations when hovering over the top moves.
-[x Self-play tournaments in separate script.
+[x] Self-play tournaments in separate script.
 [x] ai thoughts in sgf
 [x] more AI modes?
 
@@ -21,9 +21,9 @@
 [] pol value override > 0.9 ?
 [] clarify score change vs score
 [] pv with overlap?
-
-- dots: SPINNER! off last few / white black / >x pt  (multi select?)
-
+[] config player to sep. row/popups?
+[] dots: SPINNER! off last few / white black / >x pt  (multi select?)
+[] ogs bot connector
 
 Low priority
 [] more teaching / groups in danger? hard

ai.py

@@ -11,7 +11,7 @@ from engine import EngineDiedException
 from game import Move, Game, IllegalMoveException
 
 
-def weighted_selection_without_replacement(items: List[Tuple[float,float,int,int]], pick_n: int) -> List[Tuple[float,float,int,int]]:
+def weighted_selection_without_replacement(items: List[Tuple[float, float, int, int]], pick_n: int) -> List[Tuple[float, float, int, int]]:
     """For a list of tuples where the second element is a weight, returns random items with those weights, without replacement."""
     elt = [(math.log(random.random()) / item[1], item) for item in items]  # magic
     return [e[1] for e in heapq.nlargest(pick_n, elt)]  # NB fine if too small
@@ -20,8 +20,9 @@ def weighted_selection_without_replacement(items: List[Tuple[float,float,int,int
 def dirichlet_noise(num, dir_alpha=0.3):
     return np.random.dirichlet([dir_alpha] * num)
 
-def fmt_moves(moves: List[Tuple[float,Move]]):
-    return ', '.join(f"{mv.gtp()} ({p:.2%})" for p, mv in moves)
+
+def fmt_moves(moves: List[Tuple[float, Move]]):
+    return ", ".join(f"{mv.gtp()} ({p:.2%})" for p, mv in moves)
 
 
 def ai_move(game: Game, ai_mode: str, ai_settings: Dict):
@@ -32,7 +33,7 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict):
         if engine.katago_process.poll() is not None:  # TODO: clean up
             raise EngineDiedException(f"Engine for {cn.next_player} ({engine.config}) died")
     ai_mode = ai_mode.lower()
-    ai_thoughts = ''
+    ai_thoughts = ""
     candidate_ai_moves = cn.candidate_moves
     if ("policy" in ai_mode or "p+" in ai_mode) and cn.policy:
         policy_moves = cn.policy_ranking
@@ -40,17 +41,17 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict):
         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]]
-        legal_policy_moves = [(pol,mv) for pol, mv in policy_moves if not mv.is_pass if pol > 0]
+        policy_grid = var_to_grid(cn.policy, size)  # type: List[List[float]]
+        legal_policy_moves = [(pol, mv) for pol, mv in policy_moves if not mv.is_pass if pol > 0]
         top_policy_move = policy_moves[0][1]
         ai_thoughts += f"Using policy based strategy, base top 5 moves are {fmt_moves(policy_moves[:5])}. "
         if top_policy_move.is_pass:
             aimove = top_policy_move
-            ai_thoughts += 'Playing top one because it is pass.'
+            ai_thoughts += "Playing top one because it is pass."
         elif "policy" in ai_mode:
             aimove = top_policy_move
             ai_thoughts += f"Playing top policy move {aimove.gtp()} due to mode chosen."
-        elif policy_moves[0][0] > ai_settings['pick_override']:
+        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."
         elif top_5_pass:
@@ -60,10 +61,10 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict):
         elif "noise" in ai_mode:
             noise_str = ai_settings["noise_strength"]
             d_noise = dirichlet_noise(len(legal_policy_moves))
-            noisy_policy_moves = [(((1 - noise_str) * pol + noise_str * noise), mv)  for ((pol, mv), noise) in zip(legal_policy_moves, d_noise)]
-            new_top = heapq.nlargest(5,noisy_policy_moves)
+            noisy_policy_moves = [(((1 - noise_str) * pol + noise_str * noise), mv) for ((pol, mv), noise) in zip(legal_policy_moves, d_noise)]
+            new_top = heapq.nlargest(5, noisy_policy_moves)
             aimove = new_top[0][1]
-            ai_thoughts +=f"Noisy policy strategy (strength={noise_str:.2f}) generated 5 moves {fmt_moves(new_top)} so picked {aimove.gtp()}. "
+            ai_thoughts += f"Noisy policy strategy (strength={noise_str:.2f}) generated 5 moves {fmt_moves(new_top)} so picked {aimove.gtp()}. "
         elif any(keyword in ai_mode for keyword in ["influence", "territory", "local", "tenuki", "pick"]):
             n_moves = int(ai_settings["pick_frac"] * len(legal_policy_moves) + ai_settings["pick_n"])
             if "influence" in ai_mode or "territory" in ai_mode:
@@ -81,25 +82,25 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict):
                 else:
                     mx, my = cn.single_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:
-                        weighted_coords = [(p,1 - w, x, y) for p, w, x, y in weighted_coords]
+                        weighted_coords = [(p, 1 - w, x, y) for p, w, x, y in weighted_coords]
                         ai_thoughts += f"Generated weights based on one minus gaussian with variance {var} around coordinates {mx},{my}. "
                     else:
                         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)
-            ai_thoughts +=f"Picked {min(n_moves,len(weighted_coords))} random moves according to weights. "
+            ai_thoughts += f"Picked {min(n_moves,len(weighted_coords))} random moves according to weights. "
             if pick_moves:
-                new_top = [(p,Move((x,y),player=cn.next_player)) for p,wt,x,y in heapq.nlargest(5,pick_moves)]
+                new_top = [(p, Move((x, y), player=cn.next_player)) for p, wt, x, y in heapq.nlargest(5, pick_moves)]
                 aimove = new_top[0][1]
                 ai_thoughts += f"Top 5 among these were {fmt_moves(new_top)} and picked top {aimove.gtp()}. "
                 if new_top[0][0] < pass_policy:
-                    ai_thoughts += f"But found pass ({pass_policy:.1%} to be higher rated than {aimove.gtp()} ({new_top[0][0]:.1%}) so will pass instead."
+                    ai_thoughts += f"But found pass ({pass_policy:.2%} to be higher rated than {aimove.gtp()} ({new_top[0][0]:.2%}) so will pass instead."
                     aimove = Move(None, player=cn.next_player)
             else:
                 aimove = top_policy_move
@@ -132,11 +133,11 @@ def ai_move(game: Game, ai_mode: str, ai_settings: Dict):
             ai_thoughts += f"Strategy {ai_mode} not found or unexpected fallback."
         aimove = Move.from_gtp(candidate_ai_moves[0]["move"], player=cn.next_player)
         ai_thoughts += f"Default strategy found {len(candidate_ai_moves)} moves returned from the engine and chose {aimove.gtp()} as top move"
-    game.katrain.log(f"AI thoughts: {ai_thoughts}",OUTPUT_DEBUG)
+    game.katrain.log(f"AI thoughts: {ai_thoughts}", OUTPUT_DEBUG)
     try:
-        played_move = game.play(aimove)
-        played_move.ai_thoughts = ai_thoughts
+        played_node = game.play(aimove)
+        played_node.ai_thoughts = ai_thoughts
     except IllegalMoveException as e:
         game.katrain.log(f"AI Strategy {ai_mode} generated illegal move {aimove.gtp()}:  {e}", OUTPUT_ERROR)
 
-    return aimove
+    return aimove, played_node

ai2gtp.py

@@ -0,0 +1,97 @@
+# This is a script that turns a KaTrain AI into a sort-of GTP compatible bot
+import time, sys
+import traceback
+from game import Game, Move
+from ai import ai_move
+from engine import KataGoEngine
+from common import OUTPUT_ERROR, OUTPUT_INFO, OUTPUT_DEBUG
+from sgf_parser import Move
+
+DB_FILENAME = "ai_performance.pickle"
+
+
+class Logger:
+    def log(self, msg, level):
+        if level <= OUTPUT_DEBUG:
+            print(msg, file=sys.stderr)
+
+
+logger = Logger()
+
+ENGINE_SETTINGS = {
+    "katago": "../KataGo/cpp/katago",
+    "model": " models/b15-1.3.2.txt.gz",
+    "config": "KataGo/analysis_config.cfg",
+    "max_visits": 5,
+    "max_time": 5.0,
+    "enable_ownership": False,
+    "threads": 1,
+}
+ai_settings = {
+    "balance_target_score": 2,
+    "balance_random_loss": 1,
+    "balance_max_loss": 5,
+    "balance_min_visits": 20,
+    "noise_strength": 0.8,
+    "pick_n": 10,
+    "pick_frac": 0.2,
+    "local_stddev": 10,
+    "influence_weight": 0.1,
+    "pick_override": 0.95,
+}
+
+engine = KataGoEngine(logger, ENGINE_SETTINGS)
+ai_strategy = "P+Pick"
+ai_strategy = "P+Local"
+ai_settings["pick_frac"] = 0.0
+ai_settings["pick_n"] = 10
+ai_settings["local_stddev"] = 1.0
+
+ai_strategy = "P+Influence"
+ai_settings["pick_frac"] = 0.5
+ai_settings["influence_weight"] = 0.1
+
+
+logger.log(f"STARTED ENGINE", OUTPUT_ERROR)
+
+game = Game(Logger(), engine, {})
+
+while not game.ended:
+    p = game.current_node.next_player
+    line = input()
+    logger.log(f"GOT INPUT {line}", OUTPUT_ERROR)
+    if "boardsize" in line:
+        _, size = line.split(" ")
+        game = Game(Logger(), engine, {"init_size": int(size)})
+        logger.log(f"Init game {game.root.properties}", OUTPUT_ERROR)
+    if "komi" in line:
+        _, komi = line.split(" ")
+        game.root.properties["KM"] = [komi.strip()]
+        logger.log(f"Setting komi {game.root.properties}", OUTPUT_ERROR)
+    elif "genmove" in line:
+        game.current_node.analyze(engine)
+        move, node = ai_move(game, ai_strategy, ai_settings)
+        logger.log(f"SENT TO GTP: = {move.gtp()}", OUTPUT_ERROR)
+        print(f"= {move.gtp()}\n")
+        sys.stdout.flush()
+        cn = game.current_node
+        while not cn.analysis_ready:
+            time.sleep(0.001)
+        pv = ""
+        moves = sorted(list(cn.analysis["moves"].values()), key=lambda d: d["order"])
+        if moves:
+            pv = " ".join(moves[0]["pv"])
+        # print(cn.analysis,cn.analysis.get('root'),file=sys.stderr)
+        print(
+            f"CHAT:Visits {cn.ai_thoughts} Winrate {cn.analysis['root']['winrate']:.2%} ScoreLead {cn.analysis['root']['scoreLead']:.1f} ScoreStdev 0.0 PV {move.gtp()} {pv}",
+            file=sys.stderr,
+        )  #
+        continue
+    elif "play" in line:
+        _, player, move = line.split(" ")
+        node = game.play(Move.from_gtp(move.upper(), player=player[0].upper()), analyze=False)
+        logger.log(f"played {player} {move}", OUTPUT_ERROR)
+    print(f"= \n")
+
+game.game_id += f"_{game.current_node.format_score()}"
+game.write_sgf("sgf_ogs/")

common.py

@@ -6,7 +6,7 @@ OUTPUT_DEBUG = 1
 OUTPUT_EXTRA_DEBUG = 2
 
 
-def var_to_grid(array_var: List[Any], size: Tuple[int,int]) -> List[List[Any]]:
+def var_to_grid(array_var: List[Any], size: Tuple[int, int]) -> List[List[Any]]:
     """convert ownership/policy to grid format such that grid[y][x] is for move with coords x,y"""
     ix = 0
     grid = [[]] * size[1]

engine.py

@@ -86,7 +86,8 @@ class KataGoEngine:
             else:
                 callback, start_time, next_move = self.queries[analysis["id"]]
                 time_taken = time.time() - start_time
-                self.katrain.log(f"[{time_taken:.1f}][{analysis['id']}] KataGo Analysis Received: {analysis.keys()}   {line[:80]}...", OUTPUT_EXTRA_DEBUG)
+                self.katrain.log(f"[{time_taken:.1f}][{analysis['id']}] KataGo Analysis Received: {analysis.keys()}", OUTPUT_DEBUG)
+                self.katrain.log(line, OUTPUT_EXTRA_DEBUG)
                 callback(analysis)
                 del self.queries[analysis["id"]]
                 if getattr(self.katrain, "update_state", None):  # easier mocking etc
@@ -99,7 +100,7 @@ class KataGoEngine:
                 query["id"] = f"QUERY:{str(self.query_counter)}"
             self.queries[query["id"]] = (callback, time.time(), next_move)
         if self.katago_process:
-            self.katrain.log(f"Sending query {query['id']}: {str(query)}", OUTPUT_EXTRA_DEBUG)
+            self.katrain.log(f"Sending query {query['id']}: {str(query)}", OUTPUT_DEBUG)
             try:
                 self.katago_process.stdin.write((json.dumps(query) + "\n").encode())
                 self.katago_process.stdin.flush()

game.py

@@ -34,7 +34,7 @@ class Game:
         if move_tree:
             self.root = move_tree
             self.komi = self.root.komi
-            handicap = self.root.get_first("HA")
+            handicap = int(self.root.get_first("HA",0))
             if handicap and not self.root.placements:
                 self.place_handicap_stones(handicap)
         else:
@@ -113,7 +113,7 @@ class Game:
             raise IllegalMoveException("Suicide")
 
     # Play a Move from the current position, raise IllegalMoveException if invalid.
-    def play(self, move: Move, ignore_ko: bool = False):
+    def play(self, move: Move, ignore_ko: bool = False, analyze=True):
         board_size_x, board_size_y = self.board_size
         if not move.is_pass and not (0 <= move.coords[0] < board_size_x and 0 <= move.coords[1] < board_size_y):
             raise IllegalMoveException(f"Move {move} outside of board coordinates")
@@ -124,7 +124,8 @@ class Game:
             raise
         played_node = self.current_node.play(move)
         self.current_node = played_node
-        played_node.analyze(self.engines[played_node.next_player])
+        if analyze:
+            played_node.analyze(self.engines[played_node.next_player])
         return played_node
 
     def undo(self, n_times=1):

game_node.py

@@ -136,5 +136,5 @@ class GameNode(SGFNode):
             szx, szy = self.board_size
             policy_grid = var_to_grid(self.policy, size=[szx, szy])
             moves = [(policy_grid[y][x], Move((x, y), player=self.next_player)) for x in range(szx) for y in range(szy)]
-            moves.append((self.policy[-1],Move(None, player=self.next_player)))
+            moves.append((self.policy[-1], Move(None, player=self.next_player)))
             return sorted(moves, key=lambda mp: -mp[0])

gui/kivyutils.py

@@ -114,8 +114,7 @@ class LabelledCheckBox(CheckBox):
 
     def __init__(self, text=None, **kwargs):
         if text is not None:
-            kwargs["active"] = bool(text)
-        print("CB", text, kwargs)
+            kwargs["active"] = text.lower() == "true"
         super().__init__(**kwargs)
 
     @property

gui/popups.py

@@ -24,7 +24,7 @@ class QuickConfigGui(BoxLayout):
             self.set_properties(self, initial_values)
 
     def collect_properties(self, widget):
-        if isinstance(widget, (LabelledTextInput, LabelledSpinner)):
+        if isinstance(widget, (LabelledTextInput, LabelledSpinner, LabelledCheckBox)):
             try:
                 ret = {widget.input_property: widget.input_value}
             except Exception as e:
@@ -143,7 +143,7 @@ class ConfigPopup(QuickConfigGui):
         engine_updates = updated_cat["engine"]
         if "visits" in engine_updates:
             self.katrain.engine.visits = engine_updates["visits"]
-        if {key for key in engine_updates if key not in {"max_visits", "max_time"}}:
+        if {key for key in engine_updates if key not in {"max_visits", "max_time", "enable_ownership"}}:
             self.katrain.log(f"Restarting Engine after {engine_updates} settings change")
             self.katrain.controls.set_status(f"Restarting Engine after {engine_updates} settings change")
             old_engine = self.katrain.engine
@@ -155,4 +155,5 @@ class ConfigPopup(QuickConfigGui):
             else:
                 self.katrain.game.analyze_all_nodes()  # old engine was broken, so make sure we redo any failures
 
+        self.katrain.debug_level = self.config["debug"]["level"]
         self.katrain.update_state(redraw_board=True)

selfplay.py

@@ -142,17 +142,17 @@ test_ais = [
     AI("P+Local", {"local_stddev": 10}),
     AI("P+Local", {"local_stddev": 5}),
     AI("P+Local", {"local_stddev": 1}),
-    AI("P+Local", {"local_stddev": 1,"pick_frac": 0.0, "pick_n": 20}),
+    AI("P+Local", {"local_stddev": 1, "pick_frac": 0.0, "pick_n": 20}),
 ]
 
-#test_ais = [
+# test_ais = [
 #    AI("Policy", {}),
 #    AI("P+Noise", {"noise_strength": 0.4}),
 #    AI("P+Noise", {"noise_strength": 0.5}),
 #    AI("P+Noise", {"noise_strength": 0.6}),
 #    AI("P+Noise", {"noise_strength": 0.7}),
 #    AI("P+Noise", {"noise_strength": 0.8}),
-#]
+# ]
 
 # ai_database = [ai for ai in ai_database if "Territory" not in ai.name and "Influence" not in ai.name]
 for ai in test_ais:
@@ -195,7 +195,7 @@ def play_games(black: AI, white: AI, n: int = N_GAMES):
             results[tag].append(score)
             all_results.append((black.name, white.name, score))
 
-        with open('tmp.pickle', "wb") as f:
+        with open("tmp.pickle", "wb") as f:
             pickle.dump((ai_database, all_results), f)
     except Exception as e:
         print(f"Exception in playing {tag}: {e}")

sgf_parser.py

@@ -18,7 +18,7 @@ class Move:
 
     @classmethod
     def from_gtp(cls, gtp_coords, player="B"):
-        if "pass" in gtp_coords:
+        if "pass" in gtp_coords.lower():
             return cls(coords=None, player=player)
         match = re.match(r"([A-Z]+)(\d+)", gtp_coords)
         return cls(coords=(Move.GTP_COORD.index(match[1]), int(match[2]) - 1), player=player)

start_bot

@@ -0,0 +1,3 @@
+GREETING="Hello, welcome to an experimental version of KaTrain AIs - These are based on weakened policy nets of KataGo. Current mode is: Play an influential style."
+MAXGAMES=3
+gtp2ogs --apikey $(cat my/apikey) --username katrain-dev --greeting "$GREETING" --debug --ogspv katago --noclock --maxconnectedgames $MAXGAMES --persist --minrank 15k --noautohandicap --maxhandicap 0 --fakerank 3k --boardsizes 9,13,19 --komis automatic,6.5 -- python ai2gtp.py