diff --git a/katrain/__main__.py b/katrain/__main__.py index 6c752d0..a7117cb 100644 --- a/katrain/__main__.py +++ b/katrain/__main__.py @@ -1,5 +1,8 @@ # first, logging level lower import os + +from kivy.base import ExceptionHandler, ExceptionManager + os.environ["KCFG_KIVY_LOG_LEVEL"] = os.environ.get("KCFG_KIVY_LOG_LEVEL", "warning") os.environ['KIVY_AUDIO'] = "sdl2" # force working audio @@ -481,6 +484,8 @@ class KaTrainApp(MDApp): sys.exit(0) + + def run_app(): app = KaTrainApp() signal.signal(signal.SIGINT, app.signal_handler) @@ -488,9 +493,24 @@ def run_app(): try: app.run() except Exception as e: - print(e) app.on_request_close() - raise + print(f"FATAL ERROR: {e}") + traceback.print_exc() + if __name__ == "__main__": + class CrashHandler(ExceptionHandler): + def handle_exception(self, inst): + args = list(inst.args) + trace = traceback.format_exc() + message = args[0] + app = MDApp.get_running_app() + if app and app.gui: + app.gui.log("Exception: " + message + "\n" + trace,OUTPUT_ERROR) + else: + print("Exception: " + message + "\n" + trace) + return ExceptionManager.PASS + + ExceptionManager.add_handler(CrashHandler()) + run_app() diff --git a/katrain/core/ai.py b/katrain/core/ai.py index 464cfe9..422480b 100644 --- a/katrain/core/ai.py +++ b/katrain/core/ai.py @@ -43,7 +43,9 @@ def fmt_moves(moves: List[Tuple[float, Move]]): def policy_weighted_move(policy_moves, lower_bound, weaken_fac): lower_bound, weaken_fac = max(0, lower_bound), max(0.01, weaken_fac) - weighted_coords = [(pv, pv ** (1 / weaken_fac), move) for pv, move in policy_moves if pv > lower_bound and not move.is_pass] + weighted_coords = [ + (pv, pv ** (1 / weaken_fac), move) for pv, move in policy_moves if pv > lower_bound and not move.is_pass + ] if weighted_coords: top = weighted_selection_without_replacement(weighted_coords, 1)[0] ai_thoughts = f"Playing policy-weighted random move {top[2].gtp()} ({top[0]:.1%}) from {len(weighted_coords)} moves above lower_bound of {lower_bound:.1%}." @@ -56,10 +58,19 @@ def policy_weighted_move(policy_moves, lower_bound, weaken_fac): def generate_influence_territory_weights(ai_mode, ai_settings, policy_grid, size): thr_line = ai_settings["threshold"] - 1 # zero-based if ai_mode == AI_INFLUENCE: - weight = lambda x, y: (1 / ai_settings["line_weight"]) ** (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. " return weighted_coords, ai_thoughts @@ -67,11 +78,18 @@ def generate_influence_territory_weights(ai_mode, ai_settings, policy_grid, size def generate_local_tenuki_weights(ai_mode, ai_settings, policy_grid, cn, size): var = ai_settings["stddev"] ** 2 mx, my = cn.move.coords - weighted_coords = [(policy_grid[y][x], math.exp(-0.5 * ((x - mx) ** 2 + (y - my) ** 2) / var), x, y) for x in range(size[0]) for y in range(size[1]) if policy_grid[y][x] > 0] + weighted_coords = [ + (policy_grid[y][x], math.exp(-0.5 * ((x - mx) ** 2 + (y - my) ** 2) / var), x, y) + for x in range(size[0]) + for y in range(size[1]) + if policy_grid[y][x] > 0 + ] ai_thoughts = f"Generated weights based on one minus gaussian with variance {var} around coordinates {mx},{my}. " if ai_mode == AI_TENUKI: weighted_coords = [(p, 1 - w, x, y) for p, w, x, y in weighted_coords] - ai_thoughts = f"Generated weights based on one minus gaussian with variance {var} around coordinates {mx},{my}. " + ai_thoughts = ( + f"Generated weights based on one minus gaussian with variance {var} around coordinates {mx},{my}. " + ) return weighted_coords, ai_thoughts @@ -92,7 +110,9 @@ def generate_ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, policy_grid = var_to_grid(cn.policy, size) # type: List[List[float]] 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 (ai_mode == AI_POLICY and cn.depth <= ai_settings["opening_moves"]) or (ai_mode in [AI_LOCAL, AI_TENUKI] and not cn.move or cn.move.coords is None): + if (ai_mode == AI_POLICY and cn.depth <= ai_settings["opening_moves"]) or ( + ai_mode in [AI_LOCAL, AI_TENUKI] and not (cn.move and cn.move.coords) + ): ai_mode = AI_WEIGHTED ai_thoughts += f"Strategy override, using policy-weighted strategy instead. " ai_settings = {"pick_override": 0.9, "weaken_fac": 1, "lower_bound": 0.02} @@ -103,7 +123,7 @@ def generate_ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, elif ai_mode == AI_POLICY: aimove = top_policy_move ai_thoughts += f"Playing top policy move {aimove.gtp()}." - else: # weighted or pick-based + else: # weighted or pick-based legal_policy_moves = [(pol, mv) for pol, mv in policy_moves if not mv.is_pass and pol > 0] board_squares = size[0] * size[1] if ai_mode == AI_RANK: # calibrated, override from 0.8 at start to ~0.4 at full board @@ -115,7 +135,9 @@ def generate_ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, aimove = top_policy_move ai_thoughts += f"Top policy move has weight > {override:.1%}, so overriding other strategies." elif ai_mode == AI_WEIGHTED: - aimove, ai_thoughts = policy_weighted_move(policy_moves, ai_settings["lower_bound"], ai_settings["weaken_fac"]) + aimove, ai_thoughts = policy_weighted_move( + policy_moves, ai_settings["lower_bound"], ai_settings["weaken_fac"] + ) elif ai_mode in AI_STRATEGIES_PICK: if ai_mode != AI_RANK: @@ -126,20 +148,33 @@ def generate_ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, if ai_mode in [AI_INFLUENCE, AI_TERRITORY, AI_LOCAL, AI_TENUKI]: if cn.depth > ai_settings["endgame"] * board_squares: weighted_coords = [(pol, 1, *mv.coords) for pol, mv in legal_policy_moves] - x_ai_thoughts = f"Generated equal weights as move number >= {ai_settings['endgame'] * size[0] * size[1]}. " + x_ai_thoughts = ( + f"Generated equal weights as move number >= {ai_settings['endgame'] * size[0] * size[1]}. " + ) elif ai_mode in [AI_INFLUENCE, AI_TERRITORY]: - weighted_coords, x_ai_thoughts = generate_influence_territory_weights(ai_mode, ai_settings, policy_grid, size) + weighted_coords, x_ai_thoughts = generate_influence_territory_weights( + ai_mode, ai_settings, policy_grid, size + ) else: # ai_mode in [AI_LOCAL, AI_TENUKI] - weighted_coords, x_ai_thoughts = generate_local_tenuki_weights(ai_mode, ai_settings, policy_grid, cn, size) + weighted_coords, x_ai_thoughts = generate_local_tenuki_weights( + ai_mode, ai_settings, policy_grid, cn, size + ) ai_thoughts += x_ai_thoughts else: # ai_mode in [AI_PICK, AI_RANK]: - 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 + ] 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. " 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: @@ -159,12 +194,21 @@ def generate_ai_move(game: Game, ai_mode: str, ai_settings: Dict) -> Tuple[Move, 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." diff --git a/katrain/core/base_katrain.py b/katrain/core/base_katrain.py index 2d04655..b24be45 100644 --- a/katrain/core/base_katrain.py +++ b/katrain/core/base_katrain.py @@ -64,7 +64,7 @@ class KaTrainBase: def log(self, message, level=OUTPUT_INFO): if level == OUTPUT_ERROR: - print(f"ERROR: {message}", file=sys.stderr) + print(f"ERROR: {message}") elif self.debug_level >= level: print(message) diff --git a/katrain/core/engine.py b/katrain/core/engine.py index 5897002..f2b1132 100644 --- a/katrain/core/engine.py +++ b/katrain/core/engine.py @@ -1,7 +1,7 @@ import copy import json import subprocess -import sys +import traceback import threading import time from typing import Callable, Optional @@ -157,7 +157,8 @@ class KataGoEngine: if getattr(self.katrain, "update_state", None): # easier mocking etc self.katrain.update_state() except Exception as e: - self.katrain.log(f"Unexpected exception while processing KataGo output {line}", OUTPUT_ERROR) + traceback.print_exc(e) + self.katrain.log(f"Unexpected exception {e} while processing KataGo output {line}", OUTPUT_ERROR) def send_query(self, query, callback, error_callback, next_move=None): with self._lock: diff --git a/katrain/gui/popups.py b/katrain/gui/popups.py index 5b91caf..91e9e41 100644 --- a/katrain/gui/popups.py +++ b/katrain/gui/popups.py @@ -213,6 +213,9 @@ class ConfigTimerPopup(QuickConfigGui): class NewGamePopup(QuickConfigGui): def __init__(self, katrain): super().__init__(katrain) + for bw, info in katrain.players_info.items(): + self.player_setup.update_players(bw, info) + self.rules_spinner.value_refs = [name for abbr, name in katrain.engine.RULESETS_ABBR] def update_config(self, save_to_file=True): diff --git a/tests/test_ai.py b/tests/test_ai.py index d17938a..235890d 100644 --- a/tests/test_ai.py +++ b/tests/test_ai.py @@ -12,12 +12,13 @@ class TestAI: def test_order(self): assert set(AI_STRATEGIES_RECOMMENDED_ORDER) == set(AI_STRATEGIES) - @pytest.mark.skipif(os.environ.get('CI').lower() == 'true', reason='GH actions has no OpenCL') + @pytest.mark.skipif(os.environ.get('CI','').lower() == 'true', reason='GH actions has no OpenCL') def test_ai_strategies(self): katrain = KaTrainBase(force_package_config=True, debug_level=0) engine = KataGoEngine(katrain, katrain.config("engine")) - game = Game(katrain, engine) + + game = Game(katrain, engine) n_rounds = 3 for _ in range(n_rounds): for strategy in AI_STRATEGIES: @@ -28,3 +29,11 @@ class TestAI: assert played_node == game.current_node assert game.current_node.depth == len(AI_STRATEGIES) * n_rounds + + for strategy in AI_STRATEGIES: + game = Game(katrain, engine) + settings = katrain.config(f"ai/{strategy}") + move, played_node = generate_ai_move(game, strategy, settings) + katrain.log(f"Testing strategy on first move {strategy} -> {move}", OUTPUT_INFO) + assert game.current_node.depth == 1 +