Add more OBS WS events

manuelseeger · Jul 23, 2024 · 72b874b · 72b874b
1 parent d63e21f
commit 72b874b
Show file tree

Hide file tree

Showing 2 changed files with 74 additions and 41 deletions.
diff --git a/obs_client.py b/obs_client.py
@@ -1,47 +1,79 @@
 from time import sleep
 
 import click
-import obsws_python as obs
+import obsws_python as obsws
 from rich import print
 
 from config import config
 from obs_tools.sc2client import sc2client
-from obs_tools.types import Screen, UIInfo
+from obs_tools.types import Screen
 
 
+# we set this up as a standalone process so that OBS can run and react to SC2 UI changes without the need
+# to run the rest of the project.
+# This will send the currently visible screen(s) in SC2 menus to OBS via the AdvancedSceneSwitcher plugin
+# If ingame, it will send "In game" to OBS
+# AdvancedSceneSwitcher can use these messages in macro conditions
 @click.command()
-def main():
+@click.option("--verbose", is_flag=True)
+def main(verbose):
     """Monitor SC2 UI through client API and let OBS know when loading screen is active"""
-    with obs.ReqClient(
+
+    menu_screens = set([Screen.background, Screen.foreground, Screen.navigation])
+
+    with obsws.ReqClient(
         host="localhost", port=4455, password=config.obs_ws_pw, timeout=3
-    ) as cl:
-        resp = cl.get_version()
+    ) as obs:
+        resp = obs.get_version()
         print(f"OBS Version: {resp.obs_version}")
 
+        last_ui = None
         while True:
-            ui = sc2client.get_screens()
+            ui = sc2client.get_uiinfo()
 
             if ui is None:
-                print("SC2 not running?")
+                print(":warning: SC2 not running?")
                 sleep(5)
                 continue
 
-            if len(ui.activeScreens) and Screen.loading in ui.activeScreens:
+            if ui == last_ui:
+                # only notify OBS on changes
+                sleep(0.5)
+                continue
+
+            if verbose:
+                print(ui.activeScreens)
+
+            if len(ui.activeScreens) == 0:
+                print("In game")
+                data = {"message": "In game"}
+                obs.call_vendor_request(
+                    vendor_name="AdvancedSceneSwitcher",
+                    request_type="AdvancedSceneSwitcherMessage",
+                    request_data=data,
+                )
+            elif Screen.loading in ui.activeScreens:
                 print(Screen.loading)
 
                 data = {"message": Screen.loading}
-
-                cl.call_vendor_request(
+                obs.call_vendor_request(
+                    vendor_name="AdvancedSceneSwitcher",
+                    request_type="AdvancedSceneSwitcherMessage",
+                    request_data=data,
+                )
+            elif menu_screens < ui.activeScreens:
+                menues = ui.activeScreens - menu_screens
+                print("In menues " + str(menues))
+                data = {"message": "\n".join(sorted(menues))}
+                obs.call_vendor_request(
                     vendor_name="AdvancedSceneSwitcher",
                     request_type="AdvancedSceneSwitcherMessage",
                     request_data=data,
                 )
-                sleep(5)
-            elif len(ui.activeScreens) == 0:
-                print("In game")
-                sleep(10)
             else:
-                sleep(0.25)
+                pass
+
+            last_ui = ui
 
 
 if __name__ == "__main__":

diff --git a/obs_tools/sc2client.py b/obs_tools/sc2client.py
@@ -25,15 +25,20 @@
 class SC2Client:
     def get_gameinfo(self) -> GameInfo:
         try:
-            response = requests.get(urljoin(config.sc2_client_url, "/game"))
-            if response.status_code == 200:
-                try:
-                    game = GameInfo.model_validate_json(response.text)
-                    return game
-                except ValidationError as e:
-                    log.warn(f"Invalid game data: {e}")
-        except ConnectionError as e:
-            log.warn("Could not connect to SC2 game client, is SC2 running?")
+            game = self._get_info("/game")
+            gameinfo = GameInfo.model_validate_json(game)
+            return gameinfo
+        except ValidationError as e:
+            log.warn(f"Invalid game data: {e}")
+        return None
+
+    def get_uiinfo(self) -> UIInfo:
+        try:
+            ui = self._get_info("/ui")
+            uiinfo = UIInfo.model_validate_json(ui)
+            return uiinfo
+        except ValidationError as e:
+            log.warn(f"Invalid UI data: {e}")
         return None
 
     def get_opponent_name(self, gameinfo=None) -> str:
@@ -45,15 +50,11 @@ def get_opponent_name(self, gameinfo=None) -> str:
                     return player.name
         return None
 
-    def get_screens(self) -> UIInfo:
+    def _get_info(self, path) -> str:
         try:
-            response = requests.get(urljoin(config.sc2_client_url, "/ui"))
+            response = requests.get(urljoin(config.sc2_client_url, path))
             if response.status_code == 200:
-                try:
-                    ui = UIInfo.model_validate_json(response.text)
-                    return ui
-                except ValidationError as e:
-                    log.warn(f"Invalid UI data: {e}")
+                return response.text
         except ConnectionError as e:
             log.warn("Could not connect to SC2 game client, is SC2 running?")
         return None
@@ -63,7 +64,6 @@ def wait_for_gameinfo(
     ) -> GameInfo:
         start_time = time()
         while time() - start_time < timeout:
-            gameinfo = self.get_gameinfo()
             if ongoing:
                 gameinfo = self.get_ongoing_gameinfo()
             else:
@@ -116,7 +116,7 @@ def scan_client_api(self):
 
             gameinfo = sc2client.get_ongoing_gameinfo()
 
-            if self.is_live_game(gameinfo):
+            if is_live_game(gameinfo):
                 if gameinfo == self.last_gameinfo:
                     # same ongoing game, just later in time
                     if gameinfo.displayTime >= self.last_gameinfo.displayTime:
@@ -131,13 +131,14 @@ def scan_client_api(self):
                 loading_screen.send(self, scanresult=scanresult)
             sleep(1)
 
-    def is_live_game(self, gameinfo):
-        return (
-            gameinfo
-            and gameinfo.displayTime > 0
-            and gameinfo.players[0].result == Result.undecided
-            and not gameinfo.isReplay
-        )
+
+def is_live_game(gameinfo: GameInfo) -> bool:
+    return (
+        gameinfo
+        and gameinfo.displayTime > 0
+        and gameinfo.players[0].result == Result.undecided
+        and not gameinfo.isReplay
+    )
 
 
 if __name__ == "__main__":