Add hand info, speedup, flushbot, and configuration

Author: giewev

.gitignore

@@ -1 +1,3 @@
-__pycache__
+__pycache__
+Balatro
+gamestate_cache

bot.py

@@ -5,6 +5,10 @@
 import socket
 import time
 from enum import Enum
+from gamestates import cache_state
+import subprocess
+import random
+
 
 class State(Enum):
     SELECTING_HAND = 1
@@ -27,6 +31,7 @@ class State(Enum):
     BUFFOON_PACK = 18
     NEW_ROUND = 19
 
+
 class Actions(Enum):
     SELECT_BLIND = 1
     SKIP_BLIND = 2
@@ -49,137 +54,192 @@ class Actions(Enum):
     START_RUN = 19
     SEND_GAMESTATE = 20
 
-class Bot:
 
-    def __init__(self, deck: str, stake: int = 1, seed: str = None, challenge: str = None):
+class Bot:
+    def __init__(
+        self,
+        deck: str,
+        stake: int = 1,
+        seed: str = None,
+        challenge: str = None,
+        bot_port: int = 12346,
+    ):
         self.G = None
         self.deck = deck
         self.stake = stake
         self.seed = seed
         self.challenge = challenge
 
-        self.addr = ("127.0.0.1", 12345)
+        self.bot_port = bot_port
+
+        self.addr = ("localhost", self.bot_port)
         self.running = False
+        self.balatro_instance = None
+
+        self.sock = None
+
+        self.state = {}
 
-        self.state = { }
-    
     def skip_or_select_blind(self):
-        raise NotImplementedError("Error: Bot.skip_or_select_blind must be implemented.")
+        raise NotImplementedError(
+            "Error: Bot.skip_or_select_blind must be implemented."
+        )
 
     def select_cards_from_hand(self):
-        raise NotImplementedError("Error: Bot.select_cards_from_hand must be implemented.")
+        raise NotImplementedError(
+            "Error: Bot.select_cards_from_hand must be implemented."
+        )
 
     def select_shop_action(self):
         raise NotImplementedError("Error: Bot.select_shop_action must be implemented.")
-    
+
     def select_booster_action(self):
-        raise NotImplementedError("Error: Bot.select_booster_action must be implemented.")
-    
+        raise NotImplementedError(
+            "Error: Bot.select_booster_action must be implemented."
+        )
+
     def sell_jokers(self):
         raise NotImplementedError("Error: Bot.sell_jokers must be implemented.")
 
     def rearrange_jokers(self):
         raise NotImplementedError("Error: Bot.rearrange_jokers must be implemented.")
-    
+
     def use_or_sell_consumables(self):
-        raise NotImplementedError("Error: Bot.use_or_sell_consumables must be implemented.")
-    
+        raise NotImplementedError(
+            "Error: Bot.use_or_sell_consumables must be implemented."
+        )
+
     def rearrange_consumables(self):
-        raise NotImplementedError("Error: Bot.rearrange_consumables must be implemented.")
-    
+        raise NotImplementedError(
+            "Error: Bot.rearrange_consumables must be implemented."
+        )
+
     def rearrange_hand(self):
         raise NotImplementedError("Error: Bot.rearrange_hand must be implemented.")
 
+    def start_balatro_instance(self):
+        balatro_exec_path = (
+            r"C:\Program Files (x86)\Steam\steamapps\common\Balatro\Balatro.exe"
+        )
+        self.balatro_instance = subprocess.Popen(
+            [balatro_exec_path, str(self.bot_port)]
+        )
+
+    def stop_balatro_instance(self):
+        if self.balatro_instance:
+            self.balatro_instance.kill()
+
     def sendcmd(self, cmd, **kwargs):
-        msg = bytes(cmd, 'utf-8')
+        msg = bytes(cmd, "utf-8")
         self.sock.sendto(msg, self.addr)
 
     def actionToCmd(self, action):
-        result = [ ]
+        result = []
 
         for x in action:
             if isinstance(x, Actions):
                 result.append(x.name)
             elif type(x) is list:
-                result.append(','.join([str(y) for y in x]))
+                result.append(",".join([str(y) for y in x]))
             else:
-                result.append(str(x))                
+                result.append(str(x))
 
-        return '|'.join(result)
+        return "|".join(result)
 
     def verifyimplemented(self):
         try:
-            self.skip_or_select_blind(self, { })
-            self.select_cards_from_hand(self, { })
-            self.select_shop_action(self, { })
-            self.select_booster_action(self, { })
-            self.sell_jokers(self, { })
-            self.rearrange_jokers(self, { })
-            self.use_or_sell_consumables(self, { })
-            self.rearrange_consumables(self, { })
-            self.rearrange_hand(self, { })
+            self.skip_or_select_blind(self, {})
+            self.select_cards_from_hand(self, {})
+            self.select_shop_action(self, {})
+            self.select_booster_action(self, {})
+            self.sell_jokers(self, {})
+            self.rearrange_jokers(self, {})
+            self.use_or_sell_consumables(self, {})
+            self.rearrange_consumables(self, {})
+            self.rearrange_hand(self, {})
         except NotImplementedError as e:
             print(e)
             sys.exit(0)
         except:
             pass
 
+    def random_seed(self):
+        # e.g. 1OGB5WO
+        return "".join(random.choices("0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ", k=7))
+
     def chooseaction(self):
-        if self.G['state'] == State.GAME_OVER:
+        if self.G["state"] == State.GAME_OVER:
             self.running = False
 
-        match self.G['waitingFor']:
-            case 'start_run':
-                return [ Actions.START_RUN, self.stake, self.deck, self.seed, self.challenge ]
-            case 'skip_or_select_blind':
+        match self.G["waitingFor"]:
+            case "start_run":
+                seed = self.seed
+                if seed is None:
+                    seed = self.random_seed()
+                return [
+                    Actions.START_RUN,
+                    self.stake,
+                    self.deck,
+                    seed,
+                    self.challenge,
+                ]
+            case "skip_or_select_blind":
                 return self.skip_or_select_blind(self, self.G)
-            case 'select_cards_from_hand':
+            case "select_cards_from_hand":
                 return self.select_cards_from_hand(self, self.G)
-            case 'select_shop_action':
+            case "select_shop_action":
                 return self.select_shop_action(self, self.G)
-            case 'select_booster_action':
+            case "select_booster_action":
                 return self.select_booster_action(self, self.G)
-            case 'sell_jokers':
+            case "sell_jokers":
                 return self.sell_jokers(self, self.G)
-            case 'rearrange_jokers':
+            case "rearrange_jokers":
                 return self.rearrange_jokers(self, self.G)
-            case 'use_or_sell_consumables':
+            case "use_or_sell_consumables":
                 return self.use_or_sell_consumables(self, self.G)
-            case 'rearrange_consumables':
+            case "rearrange_consumables":
                 return self.rearrange_consumables(self, self.G)
-            case 'rearrange_hand':
+            case "rearrange_hand":
                 return self.rearrange_hand(self, self.G)
 
-    def run(self):
-        self.verifyimplemented()
-        self.state = { }
+    def run_step(self):
+        if self.sock is None:
+            self.verifyimplemented()
+            self.state = {}
+            self.G = None
 
-        self.running = True
+            self.running = True
+            self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+            self.sock.settimeout(1)
+            self.sock.connect(self.addr)
 
-        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-        self.sock.settimeout(1.0)
+        if self.running:
+            self.sendcmd("HELLO")
 
-        while self.running:
-            self.sendcmd('HELLO')
-
-            jsondata = { }
+            jsondata = {}
             try:
-                data = self.sock.recv(4096)
+                data = self.sock.recv(65536)
                 jsondata = json.loads(data)
 
-                if 'response' in jsondata:
-                    print(jsondata['response'])
+                if "response" in jsondata:
+                    print(jsondata["response"])
                 else:
                     self.G = jsondata
-                    if self.G['waitingForAction']:
-                        # Choose next action
+                    if self.G["waitingForAction"]:
+                        cache_state(self.G["waitingFor"], self.G)
                         action = self.chooseaction()
                         if action == None:
                             raise ValueError("All actions must return a value!")
-                        
+
                         cmdstr = self.actionToCmd(action)
-                        print(f'CMD: {cmdstr}')
                         self.sendcmd(cmdstr)
-            except socket.error:
+            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.0)
+                self.sock.settimeout(1)
+                self.sock.connect(self.addr)
+
+    def run(self):
+        while self.running:
+            self.run_step()