From 72b874b49e67e1314f685a513fffe94f7a4b1535 Mon Sep 17 00:00:00 2001 From: manuelseeger Date: Tue, 23 Jul 2024 12:48:00 +0200 Subject: [PATCH] Add more OBS WS events --- obs_client.py | 64 +++++++++++++++++++++++++++++++----------- obs_tools/sc2client.py | 51 ++++++++++++++++----------------- 2 files changed, 74 insertions(+), 41 deletions(-) diff --git a/obs_client.py b/obs_client.py index 125fb53..3999fca 100644 --- 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 index 2adcff3..02fe017 100644 --- 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__":