Add different colours for concurrent games .

game.py

@@ -6,14 +6,16 @@
 from chatter import Chatter
 from config import Config
 from lichess_game import Lichess_Game
+from utils import ColorLogger
 
 
 class Game:
-    def __init__(self, api: API, config: Config, username: str, game_id: str) -> None:
+    def __init__(self, api: API, config: Config, username: str, game_id: str, color_logger: ColorLogger) -> None:
         self.api = api
         self.config = config
         self.username = username
         self.game_id = game_id
+        self.color_logger = color_logger
 
         self.takeback_count = 0
         self.was_aborted = False
@@ -26,7 +28,7 @@ async def run(self) -> None:
         game_stream_queue: asyncio.Queue[dict[str, Any]] = asyncio.Queue()
         self._task = asyncio.create_task(self.api.get_game_stream(self.game_id, game_stream_queue))
         info = Game_Information.from_gameFull_event(await game_stream_queue.get())
-        lichess_game = await Lichess_Game.acreate(self.api, self.config, self.username, info)
+        lichess_game = await Lichess_Game.acreate(self.api, self.config, self.username, info, self.color_logger)
         chatter = Chatter(self.api, self.config, self.username, info, lichess_game)
 
         self._print_game_information(info)
@@ -113,12 +115,11 @@ async def _abortion_task(self, lichess_game: Lichess_Game, chatter: Chatter, abo
 
         self.abortion_task = None
 
-    @staticmethod
-    def _print_game_information(info: Game_Information) -> None:
+    def _print_game_information(self, info: Game_Information) -> None:
         opponents_str = f"{info.white_str}   -   {info.black_str}"
         message = " • ".join([info.id_str, opponents_str, info.tc_format, info.rated_str, info.variant_str])
 
-        print(f"\n{message}\n{128 * '‾'}")
+        self.color_logger.print(f"\n{message}\n{128 * '‾'}", self.game_id)
 
     def _print_result_message(
         self, game_state: dict[str, Any], lichess_game: Lichess_Game, info: Game_Information
@@ -185,4 +186,4 @@ def _print_result_message(
         opponents_str = f"{info.white_str} {white_result} - {black_result} {info.black_str}"
         message = " • ".join([info.id_str, opponents_str, message])
 
-        print(f"{message}\n{128 * '‾'}")
+        self.color_logger.print(f"{message}\n{128 * '‾'}", self.game_id)

game_manager.py

@@ -9,14 +9,15 @@
 from config import Config
 from game import Game
 from matchmaking import Matchmaking
-from utils import get_future_timestamp
+from utils import ColorLogger, get_future_timestamp
 
 
 class Game_Manager:
     def __init__(self, api: API, config: Config, username: str) -> None:
         self.api = api
         self.config = config
         self.username = username
+        self.color_logger = ColorLogger()
 
         self.challenger = Challenger(api)
         self.changed_event = Event()
@@ -216,6 +217,7 @@ def _set_next_matchmaking(self, delay: int) -> None:
 
     def _task_callback(self, task: Task[None]) -> None:
         game = self.tasks.pop(task)
+        self.color_logger.remove_color(game.game_id)
 
         if game.game_id == self.current_matchmaking_game_id:
             self.matchmaking.on_game_finished(game.was_aborted)
@@ -240,7 +242,7 @@ async def _start_game(self, game_event: dict[str, Any]) -> None:
             self.tournaments[tournament.id_] = tournament
             print(f'External joined tournament "{tournament.name}" detected.')
 
-        game = Game(self.api, self.config, self.username, game_event["id"])
+        game = Game(self.api, self.config, self.username, game_event["id"], self.color_logger)
         task = asyncio.create_task(game.run())
         task.add_done_callback(self._task_callback)
         self.tasks[task] = game

lichess_game.py

@@ -28,6 +28,7 @@
 from configs import Engine_Config, Syzygy_Config
 from engine import Engine
 from enums import Variant
+from utils import ColorLogger
 
 
 class Lichess_Game:
@@ -41,10 +42,12 @@ def __init__(
         syzygy_config: Syzygy_Config,
         engine_key: str,
         engine: Engine,
+        color_logger: ColorLogger,
     ) -> None:
         self.api = api
         self.config = config
         self.game_info = game_info
+        self.color_logger = color_logger
         self.board = board
         self.syzygy_config = syzygy_config
         self.white_time: float = self.game_info.state["wtime"] / 1000
@@ -71,7 +74,9 @@ def __init__(
         self.last_pv: list[chess.Move] = []
 
     @classmethod
-    async def acreate(cls, api: API, config: Config, username: str, game_info: Game_Information) -> "Lichess_Game":
+    async def acreate(
+        cls, api: API, config: Config, username: str, game_info: Game_Information, color_logger: ColorLogger
+    ) -> "Lichess_Game":
         board = cls._get_board(game_info)
         is_white = game_info.white_name == username
         engine_key = cls._get_engine_key(config, board, is_white, game_info)
@@ -81,7 +86,7 @@ async def acreate(cls, api: API, config: Config, username: str, game_info: Game_
             syzygy_config,
             game_info.black_opponent if is_white else game_info.white_opponent,
         )
-        return cls(api, config, username, game_info, board, syzygy_config, engine_key, engine)
+        return cls(api, config, username, game_info, board, syzygy_config, engine_key, engine, color_logger)
 
     @staticmethod
     def _get_board(game_info: Game_Information) -> chess.Board:
@@ -159,8 +164,9 @@ async def make_move(self) -> Lichess_Move:
                 self.board.push(move_response.move)
                 await self.engine.start_pondering(self.board)
 
-                print(f"{move_response.public_message} {move_response.private_message}".strip())
+                message = f"{move_response.public_message} {move_response.private_message}".strip()
                 self.last_message = move_response.public_message
+                self.color_logger.print(message, self.game_info.id_)
                 self.last_pv = move_response.pv
                 return Lichess_Move(
                     move_response.move.uci(),
@@ -174,7 +180,7 @@ async def make_move(self) -> Lichess_Move:
             self.scores.append(info["score"])
 
         message = f"Engine:  {self._format_move(move):14} {self._format_engine_info(info)}"
-        print(message)
+        self.color_logger.print(message, self.game_info.id_)
         self.last_message = message
         self.last_pv = info.get("pv", [])
 
@@ -336,7 +342,7 @@ async def _make_book_move(self) -> Move_Response | None:
             try:
                 entries = list(book_reader.find_all(self.board))
             except struct.error:
-                print(f'Skipping book "{name}" due to error.')
+                self.color_logger.print(f'Skipping book "{name}" due to error.', self.game_info.id_)
                 continue
 
             if not entries:
@@ -595,7 +601,7 @@ async def _make_chessdb_move(self) -> Move_Response | None:
 
         if response["status"] != "ok":
             if response["status"] != "unknown":
-                print(f"ChessDB: {response['status']}")
+                self.color_logger.print(f"ChessDB: {response['status']}", self.game_info.id_)
             self.out_of_chessdb_counter += 1
             return
 

requirements.txt

@@ -2,4 +2,5 @@ aiohttp[speedups] == 3.13.0
 chess == 1.11.2
 psutil == 7.1.0
 PyYAML == 6.0.3
-tenacity == 9.1.2
+tenacity == 9.1.2
+rich == 14.1.0

utils.py

@@ -1,6 +1,9 @@
 import textwrap
 from datetime import datetime, timedelta
 
+from rich.console import Console
+from rich.style import Style
+
 from enums import Variant
 
 ALIASES = {
@@ -47,3 +50,40 @@ def parse_time_control(time_control: str) -> tuple[int, int]:
     initial_time = int(float(initial_time_str) * 60)
     increment = int(increment_str)
     return initial_time, increment
+
+
+class ColorLogger:
+    def __init__(self):
+        self.console = Console()
+        self.game_colors = {}
+        self.available_colors = [
+            "red",
+            "green",
+            "yellow",
+            "blue",
+            "magenta",
+            "cyan",
+            "bright_red",
+            "bright_green",
+            "bright_yellow",
+            "bright_blue",
+            "bright_magenta",
+            "bright_cyan",
+        ]
+        self.color_index = 0
+
+    def assign_color(self, game_id: str) -> str:
+        if game_id not in self.game_colors:
+            self.game_colors[game_id] = self.available_colors[self.color_index % len(self.available_colors)]
+            self.color_index += 1
+        return self.game_colors[game_id]
+
+    def print(self, message: str, game_id: str | None = None):
+        if game_id:
+            color = self.assign_color(game_id)
+            self.console.print(message, style=Style(color=color))
+        else:
+            self.console.print(message)
+
+    def remove_color(self, game_id: str):
+        self.game_colors.pop(game_id, None)