Add CLI options (#8)

* add cli args --config, --debug and --verbose

load_config() will search for a relevant config file in several dirs

add separate handler methods for mute,unmute and toggle

mapping.toml removed, everything merged into config.toml

* add cli options subsection

* rename event variable to make it clear we're waiting for a stop event (and not some other obs event)

* fix argparse description

* add return None (for type checker)

* fix example in top level docstring

* add shorthand `-c` cli option

.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

.vscode/settings.json

@@ -0,0 +1,5 @@
+{
+    "python.analysis.diagnosticSeverityOverrides": {
+        "reportMissingImports": "none"
+    }    
+}

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.

__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__":

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"
+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]

mapping.toml

@@ -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]

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",