diff --git a/.gitignore b/.gitignore index 894a44c..96905fd 100644 --- a/.gitignore +++ b/.gitignore @@ -90,6 +90,10 @@ ENV/ env.bak/ venv.bak/ +# Pipenv +Pipfile +Pipfile.lock + # Spyder project settings .spyderproject .spyproject @@ -102,3 +106,5 @@ venv.bak/ # mypy .mypy_cache/ + +test.toml \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..8cb2da2 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,5 @@ +{ + "python.analysis.diagnosticSeverityOverrides": { + "reportMissingImports": "none" + } +} diff --git a/README.md b/README.md index c9623c9..5770f1e 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ This is a small script that mutes, unmutes and toggles groups of channels on Beh - Download the repository files with git or the green `Code` button. Then in command prompt: -``` +```bash cd OBS-to-XAir pip install . ``` @@ -27,11 +27,13 @@ pip install . - Configure websocket settings within `OBS->Tools->obs-websocket Settings` -- Open the included `config.toml` and set OBS host, port and password as well as the xair mixers kind and ip. +- Open the included `config.toml`, use it to: + - Set the obs connection info `host`, `port` and `password` - - Mixer kind may be any one of (`XR12, XR16, XR18, MR18, X32`) + - Set the mixer's `kind_id` and `ip`. + - Mixer kind_id may be any one of (`XR12, XR16, XR18, MR18, X32`) -- Set the scene to channel mutes mapping in `mapping.toml`. + - Set the scene to channel mapping. ## Usage @@ -39,6 +41,15 @@ Simply run the script, there will be confirmation of mixer connection and OBS co Closing OBS will stop the script. +#### CLI options + +- `--config`: may be a full path to a config file or just a config name. + - If only a config name is passed the script will look in the following locations, returning the first config found: + - Current working directory (may be different from script location depending on how you launch the script) + - In the directory the script is located. + - `user home directory / .config / xair-obs` +- `--debug`, `--verbose`: separate logging levels. Debug will produce a lot of logging output. + ## Further notes Since this script relies upon two interfaces, `obsws-python` and `xair-api` this code can be readily modified to interact with any OBS events and set any xair parameters. Check the README files for each interface for further details. @@ -47,12 +58,14 @@ Since this script relies upon two interfaces, `obsws-python` and `xair-api` this This script was developed and tested with: -- OBS 28.01 -- obs-websocket 5.0.1 +- OBS 31.0.0 +- obs-websocket 5.5.4 - A Midas MR18 and an X32 emulator. ## Special Thanks -- OBS team and the obs-websocket developers + Behringer/Midas for the OSC protocol. -- [Adem](https://github.com/aatikturk) for contributions towards the obsws-python wrapper. -- [Onyx-and-Iris](https://github.com/onyx-and-iris) for contributions towards the obsws-python and xair-api wrappers. +- [Lebaston](https://github.com/lebaston100) for the initial implementation of this script. +- OBS team and the obs-websocket developers. +- Behringer/Midas for making their mixers programmable! +- [Adem](https://github.com/aatikturk) for contributions towards the obsws-python clients. +- [Onyx-and-Iris](https://github.com/onyx-and-iris) for contributions towards the obsws-python and xair-api interfaces. diff --git a/__main__.py b/__main__.py index 7f7d69e..a020789 100644 --- a/__main__.py +++ b/__main__.py @@ -1,6 +1,21 @@ +""" +This script connects to an XAir mixer and OBS (Open Broadcaster Software) to control the mixer based on OBS scene changes. +Classes: + Observer: Handles events from OBS and controls the XAir mixer. +Functions: + load_config(config: str) -> dict: Loads a configuration file in TOML format. + parse_args() -> argparse.Namespace: Parses command-line arguments. + main(): Main function to parse arguments, configure logging, load configuration, and start the XAir mixer observer. +Usage: + Run this script with optional arguments for configuration file path and logging level. + Example: python . --config path/to/config.toml --debug +""" + +import argparse import logging -import time +import threading from pathlib import Path +from typing import Callable, Mapping import obsws_python as obs import xair_api @@ -10,66 +25,177 @@ try: except ModuleNotFoundError: import tomli as tomllib -logging.basicConfig(level=logging.disable()) - - -def _get_mapping(): - filepath = Path.cwd() / "mapping.toml" - with open(filepath, "rb") as f: - return tomllib.load(f) - - -mapping = _get_mapping() - class Observer: - def __init__(self, mixer): + """ + Observer class to handle events from OBS (Open Broadcaster Software) and control an XAir mixer. + Attributes: + _mixer (xair_api.xair.XAirRemote): The XAir mixer remote control instance. + _stop_event (threading.Event): Event to signal stopping of the observer. + _mapping (dict): Mapping of OBS scenes to mixer actions. + _request (obs.ReqClient): OBS request client for sending requests. + _event (obs.EventClient): OBS event client for receiving events. + Methods: + __enter__(): Enter the runtime context related to this object. + __exit__(exc_type, exc_value, exc_traceback): Exit the runtime context related to this object. + on_current_program_scene_changed(data: type) -> None: Handles the event when the current program scene changes. + _mute_handler(i): Mutes the specified mixer strip. + _unmute_handler(i): Unmutes the specified mixer strip. + _toggle_handler(i): Toggles the mute state of the specified mixer strip. + on_exit_started(_): Handles the event when OBS is closing. + """ + + def __init__( + self, mixer: xair_api.xair.XAirRemote, stop_event: threading.Event, config: dict + ): self._mixer = mixer - self._request = obs.ReqClient() - self._event = obs.EventClient() + self._stop_event = stop_event + self._mapping = config["scene_mapping"] + self._request = obs.ReqClient(**config["obs"]) + self._event = obs.EventClient(**config["obs"]) self._event.callback.register( (self.on_current_program_scene_changed, self.on_exit_started) ) - self.running = True resp = self._request.get_version() - info = ( - f"Connected to OBS version:{resp.obs_version}", - f"with websocket version:{resp.obs_web_socket_version}", + print( + f"Connected to OBS version:{resp.obs_version} " + f"with websocket version:{resp.obs_web_socket_version}" ) - print(" ".join(info)) - def on_current_program_scene_changed(self, data): - def ftoggle(i): - self._mixer.strip[i - 1].mute = not self._mixer.strip[i - 1].mute + def __enter__(self): + return self - def fset(i, is_muted): - self._mixer.strip[i - 1].mute = is_muted + def __exit__(self, exc_type, exc_value, exc_traceback): + for client in (self._request, self._event): + client.disconnect() + + def on_current_program_scene_changed(self, data: type): + """ + Handles the event when the current program scene changes. + Args: + data: An object containing information about the scene change event. + It is expected to have an attribute `scene_name` which is the name of the new scene. + Returns: + None + """ scene = data.scene_name print(f"Switched to scene {scene}") - if map_ := mapping.get(scene): - for key in map_.keys(): - if key == "toggle": - [ftoggle(i) for i in map_[key]] - else: - [fset(i, key == "mute") for i in map_[key]] + + if not (map_ := self._mapping.get(scene)): + return + + actions: Mapping[str, Callable] = { + "mute": self._mute_handler, + "unmute": self._unmute_handler, + "toggle": self._toggle_handler, + } + + for action, indices in map_.items(): + if action in actions: + for i in indices: + actions[action](i - 1) + + def _mute_handler(self, i): + self._mixer.strip[i].mute = True + + def _unmute_handler(self, i): + self._mixer.strip[i].mute = False + + def _toggle_handler(self, i): + self._mixer.strip[i].mute = not self._mixer.strip[i].mute def on_exit_started(self, _): print("OBS closing") - self._event.unsubscribe() - self.running = False + self._stop_event.set() + + +def load_config(config: str) -> dict: + """ + Load a configuration file in TOML format. + Args: + config (str): The filepath/name of the configuration file to load. + Returns: + dict: The contents of the configuration file as a dictionary. + Raises: + FileNotFoundError: If the configuration file does not exist. + tomllib.TOMLDecodeError: If there is an error decoding the TOML file. + """ + + def get_filepath() -> Path | None: + for filepath in ( + Path(config), + Path.cwd() / config, + Path(__file__).parent / config, + Path.home() / ".config" / "xair-obs" / config, + ): + if filepath.is_file(): + return filepath + return None + + if not (filepath := get_filepath()): + raise FileNotFoundError(f"Config file {config} not found") + try: + with open(filepath, "rb") as f: + return tomllib.load(f) + except tomllib.TOMLDecodeError as e: + raise tomllib.TOMLDecodeError(f"Error decoding config file {filepath}") from e + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="OBS to Xair Controller") + parser.add_argument( + "-c", "--config", default="config.toml", help="Path to the config file" + ) + parser.add_argument( + "-d", + "--debug", + help="Debug output", + action="store_const", + dest="loglevel", + const=logging.DEBUG, + default=logging.WARNING, + ) + parser.add_argument( + "-v", + "--verbose", + help="Verbose output", + action="store_const", + dest="loglevel", + const=logging.INFO, + ) + args = parser.parse_args() + return args def main(): - filepath = Path.cwd() / "config.toml" - with open(filepath, "rb") as f: - kind_mixer = tomllib.load(f)["connection"].get("mixer") + """ + Main function to parse arguments, configure logging, load configuration, + and start the XAir mixer observer. + This function performs the following steps: + 1. Parses command-line arguments. + 2. Configures logging based on the provided log level. + 3. Loads the configuration file. + 4. Connects to the XAir mixer using the configuration. + 5. Starts an observer to monitor the mixer and waits for events. + The function blocks until a stop event is received. + Args: + None + Returns: + None + """ - with xair_api.connect(kind_mixer) as mixer: - o = Observer(mixer) + args = parse_args() + logging.basicConfig(level=args.loglevel) - while o.running: - time.sleep(0.5) + config = load_config(args.config) + + with xair_api.connect(**config["xair"]) as mixer: + print(f"Connected to {mixer.kind} mixer at {mixer.xair_ip}:{mixer.xair_port}") + stop_event = threading.Event() + + with Observer(mixer, stop_event, config): + stop_event.wait() if __name__ == "__main__": diff --git a/config.toml b/config.toml index 6b51dee..8f8275c 100644 --- a/config.toml +++ b/config.toml @@ -1,9 +1,22 @@ -[connection] +[obs] # OBS connection info host = "localhost" port = 4455 -password = "strongpassword" +password = "secretpassword" +[xair] # mixer kind and ip -mixer = "XR18" -ip = "mixer.local" \ No newline at end of file +kind_id = "XR18" +ip = "mixer.local" + +[scene_mapping] +# OBS scene name to mixer channel mapping +START.mute = [1, 3, 5] +START.unmute = [2, 7] + +BRB.mute = [2, 7] +BRB.unmute = [1, 3, 5] + +END.toggle = [12, 14] + +LIVE.toggle = [16] diff --git a/mapping.toml b/mapping.toml deleted file mode 100644 index 047afc9..0000000 --- a/mapping.toml +++ /dev/null @@ -1,9 +0,0 @@ -START.mute = [1, 3, 5] -START.unmute = [2, 7] - -BRB.mute = [2, 7] -BRB.unmute = [1, 3, 5] - -END.toggle = [12, 14] - -LIVE.toggle = [16] \ No newline at end of file diff --git a/setup.py b/setup.py index c3d008d..b09fc77 100644 --- a/setup.py +++ b/setup.py @@ -2,11 +2,11 @@ from setuptools import setup setup( name="xair-obs", - version="0.0.1", + version="1.0.0", description="Syncs Xair states to OBS scenes", install_requires=[ - "obsws-python", - "xair-api", + "obsws-python>=1.7.0", + "xair-api>=2.4.0", "tomli >= 2.0.1;python_version < '3.11'", ], python_requires=">=3.10",