feat: add extensive logging to python code

Author: S1M0N38

bots/example.py

@@ -4,12 +4,13 @@
 a predefined sequence of actions.
 """
 
+import argparse
 import itertools
 from typing import Any, Iterator
 
-from balatrobot import Actions, ActionSchema, Bot, Decks, Stakes
+from balatrobot import Actions, ActionSchema, Bot, Decks, Stakes, configure_bot_logging
 
-# Predefined sequence of actions using the ActionSchema format
+# Predefined sequence of actions using the ActionCall format
 plays: Iterator[ActionSchema] = itertools.cycle(
     [
         # This sequence of plays is winning for the first round
@@ -22,8 +23,8 @@
 )
 
 
-class MyFirstBot(Bot):
-    """Example bot implementation using the ActionSchema API.
+class ExampleBot(Bot):
+    """Example bot implementation using the ActionCall API.
 
     This bot demonstrates a simple strategy using predefined actions.
     It always selects blinds, uses a fixed sequence of plays, and
@@ -42,9 +43,9 @@ def __init__(
         """Initialize the bot with default settings.
 
         Args:
-            deck (Decks): The deck type to use.
-            stake (Stakes): The stake level to play at.
-            seed (str): The random seed for the game.
+            deck: The deck type to use.
+            stake: The stake level to play at.
+            seed: The random seed for the game.
         """
         super().__init__(deck=deck, stake=stake, seed=seed)
         self.round_count: int = 0
@@ -56,7 +57,7 @@ def skip_or_select_blind(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to select blind.
+            ActionCall: Action to select blind.
         """
         return {"action": Actions.SELECT_BLIND, "args": None}
 
@@ -67,7 +68,7 @@ def select_cards_from_hand(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action with card selection from predefined sequence.
+            ActionCall: Action with card selection from predefined sequence.
         """
         return next(plays)
 
@@ -78,7 +79,7 @@ def select_shop_action(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to end shop.
+            ActionCall: Action to end shop.
         """
         return {"action": Actions.END_SHOP, "args": None}
 
@@ -89,7 +90,7 @@ def select_booster_action(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to skip booster pack.
+            ActionCall: Action to skip booster pack.
         """
         return {"action": Actions.SKIP_BOOSTER_PACK, "args": None}
 
@@ -100,7 +101,7 @@ def sell_jokers(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to sell jokers with empty list.
+            ActionCall: Action to sell jokers with empty list.
         """
         return {"action": Actions.SELL_JOKER, "args": []}
 
@@ -111,7 +112,7 @@ def rearrange_jokers(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to rearrange jokers with empty list.
+            ActionCall: Action to rearrange jokers with empty list.
         """
         return {"action": Actions.REARRANGE_JOKERS, "args": []}
 
@@ -122,7 +123,7 @@ def use_or_sell_consumables(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to use consumables with empty list.
+            ActionCall: Action to use consumables with empty list.
         """
         return {"action": Actions.USE_CONSUMABLE, "args": []}
 
@@ -133,7 +134,7 @@ def rearrange_consumables(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to rearrange consumables with empty list.
+            ActionCall: Action to rearrange consumables with empty list.
         """
         return {"action": Actions.REARRANGE_CONSUMABLES, "args": []}
 
@@ -144,13 +145,27 @@ def rearrange_hand(self, env: dict[str, Any]) -> ActionSchema:
             env (dict[str, Any]): The current game environment state.
 
         Returns:
-            ActionSchema: Action to rearrange hand with empty list.
+            ActionCall: Action to rearrange hand with empty list.
         """
         return {"action": Actions.REARRANGE_HAND, "args": []}
 
 
 def main() -> None:
-    bot = MyFirstBot()
+    """Main function to run the example bot with command-line argument support."""
+    parser = argparse.ArgumentParser(description="Run the example Balatro bot")
+    parser.add_argument(
+        "--log",
+        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
+        default="DEBUG",
+        help="Set the console logging level (default: DEBUG)",
+    )
+
+    args = parser.parse_args()
+
+    # Configure logging with the specified level
+    configure_bot_logging(args.log)
+
+    bot = ExampleBot()
     bot.running = True
     bot.run()
 

src/balatrobot/__init__.py

@@ -1,5 +1,6 @@
 from .base import ActionSchema, Bot
 from .enums import Actions, Decks, Stakes, State
+from .utils import configure_bot_logging, get_logger, setup_logging
 
 __all__ = [
     "ActionSchema",
@@ -8,6 +9,9 @@
     "Decks",
     "Stakes",
     "State",
+    "configure_bot_logging",
+    "get_logger",
+    "setup_logging",
 ]
 
 __version__ = "0.1.0"

src/balatrobot/base.py

@@ -6,6 +6,7 @@
 from typing import Any, TypedDict
 
 from .enums import Actions, Decks, Stakes, State
+from .utils import get_logger
 
 
 class ActionSchema(TypedDict):
@@ -67,6 +68,23 @@ def __init__(
         self.sock: socket.socket | None = None
 
         self.state: dict[str, Any] = {}
+        self.logger = get_logger("bot")
+
+        # Initialize socket connection
+        self._initialize_socket()
+
+    def _initialize_socket(self) -> None:
+        """Initialize the socket connection for bot communication.
+
+        Sets up the UDP socket with timeout and connects to the game instance.
+        """
+        self.state = {}
+        self.G = None
+        self.running = True
+        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.sock.settimeout(1)
+        self.sock.connect(self.addr)
+        self.logger.debug("Socket initialized and connected to %s:%d", *self.addr)
 
     @staticmethod
     def random_seed() -> str:
@@ -214,44 +232,48 @@ def chooseaction(self, env: dict[str, Any]) -> ActionSchema:
         Returns:
             ActionSchema: The action to perform with 'action' and optional 'args'.
         """
-        print("Choosing action based on game state:", env["state"])
+        try:
+            state_name = State(env["state"]).name
+        except ValueError:
+            state_name = f"UNKNOWN({env['state']})"
+        self.logger.debug("Choosing action based on game state: %s", state_name)
         if env["state"] == State.GAME_OVER:
             self.running = False
 
         match env["waitingFor"]:
             case "start_run":
-                print("Starting run with deck:", self.deck)
+                self.logger.info("Starting run with deck: %s", self.deck)
                 seed = self.seed or self.random_seed()
                 return {
                     "action": Actions.START_RUN,
                     "args": [self.stake.value, self.deck.value, seed, self.challenge],
                 }
             case "skip_or_select_blind":
-                print("Choosing action: skip_or_select_blind")
+                self.logger.debug("Choosing action: skip_or_select_blind")
                 return self.skip_or_select_blind(env)
             case "select_cards_from_hand":
-                print("Choosing action: select_cards_from_hand")
+                self.logger.debug("Choosing action: select_cards_from_hand")
                 return self.select_cards_from_hand(env)
             case "select_shop_action":
-                print("Choosing action: select_shop_action")
+                self.logger.debug("Choosing action: select_shop_action")
                 return self.select_shop_action(env)
             case "select_booster_action":
-                print("Choosing action: select_booster_action")
+                self.logger.debug("Choosing action: select_booster_action")
                 return self.select_booster_action(env)
             case "sell_jokers":
-                print("Choosing action: sell_jokers")
+                self.logger.debug("Choosing action: sell_jokers")
                 return self.sell_jokers(env)
             case "rearrange_jokers":
-                print("Choosing action: rearrange_jokers")
+                self.logger.debug("Choosing action: rearrange_jokers")
                 return self.rearrange_jokers(env)
             case "use_or_sell_consumables":
-                print("Choosing action: use_or_sell_consumables")
+                self.logger.debug("Choosing action: use_or_sell_consumables")
                 return self.use_or_sell_consumables(env)
             case "rearrange_consumables":
-                print("Choosing action: rearrange_consumables")
+                self.logger.debug("Choosing action: rearrange_consumables")
                 return self.rearrange_consumables(env)
             case "rearrange_hand":
-                print("Choosing action: rearrange_hand")
+                self.logger.debug("Choosing action: rearrange_hand")
                 return self.rearrange_hand(env)
             case _:
                 raise ValueError(f"Unhandled waitingFor state: {env['waitingFor']}")
@@ -262,14 +284,8 @@ def run_step(self) -> None:
         This method handles socket communication, receives game state updates,
         and sends actions back to the game.
         """
-        if self.sock is None:
-            self.state = {}
-            self.G = None
-
-            self.running = True
-            self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-            self.sock.settimeout(1)
-            self.sock.connect(self.addr)
+        if not self.running or self.sock is None:
+            return
 
         if self.running:
             self.sock.send(bytes("HELLO", "utf-8"))
@@ -279,7 +295,7 @@ def run_step(self) -> None:
                 env = json.loads(data)
 
                 if "response" in env:
-                    print(env["response"])
+                    self.logger.info("Game response: %s", env["response"])
                 else:
                     if env["waitingForAction"]:
                         action = self.chooseaction(env)
@@ -290,11 +306,9 @@ def run_step(self) -> None:
                             raise ValueError("All actions must return a value!")
 
             except socket.error as e:
-                print(e)
-                print("Socket error, reconnecting...")
-                self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-                self.sock.settimeout(1)
-                self.sock.connect(self.addr)
+                self.logger.error("Socket error: %s", e)
+                self.logger.warning("Socket error, reconnecting...")
+                self._initialize_socket()
 
     def run(self) -> None:
         """Run the bot's main game loop.