Add generic CLI utility

Stanislav Ochotnicky · Stanislav Ochotnicky · commit c7f1ba414ebe · 2020-12-06T20:43:53.000+01:00
This adds a basic CLI utility that uses introspection to get all possible commands and enables running them from console. It also provides way to specify complex JSON arguments if needed. Fixes #46
diff --git a/README.md b/README.md
@@ -1,5 +1,5 @@
 # obs-websocket-py
-Python library to communicate with an [obs-websocket](https://github.com/Palakis/obs-websocket) server.
+Python library & CLI to communicate with an [obs-websocket](https://github.com/Palakis/obs-websocket) server.
 
 _Licensed under the MIT License_
 
@@ -95,6 +95,52 @@ obswebsocket.core.obsws = class obsws
  |      :return: Nothing
 ```
 
+There is also a simple CLI provided with the installation. It can be used in variety of ways, but is not meant to cover all use cases.
+
+```
+$ obs-web-cli --help
+OBS Studio CLI using OBS Websocket Plugin
+
+optional arguments:
+  -h, --help            show this help message and exit
+  --host HOST           Hostname to connect to (default: localhost)
+  --port PORT           Port to connect to (default: 4444)
+  --password PASSWORD   Password to use. Defaults to OBS_WEBSOCKET_PASS env
+                        var (default: None)
+  --debug               Enable debugging output (default: False)
+
+Recognized commands:
+  {GetStreamingStatus,StartStopStreaming,StartStreaming,StopStreaming,SetStreamSettings,GetStreamSettings,SaveStreamSettings,SendCaptions,GetStudioModeStatus,GetPreviewScene,SetPreviewScene,TransitionToProgram,EnableStudioMode,DisableStudioMode,ToggleStudioMode,ListOutputs,GetOutputInfo,StartOutput,StopOutput,StartStopReplayBuffer,StartReplayBuffer,StopReplayBuffer,SaveReplayBuffer,SetCurrentScene,GetCurrentScene,GetSceneList,ReorderSceneItems,SetCurrentProfile,GetCurrentProfile,ListProfiles,GetVersion,GetAuthRequired,Authenticate,SetHeartbeat,SetFilenameFormatting,GetFilenameFormatting,GetStats,BroadcastCustomMessage,GetVideoInfo,StartStopRecording,StartRecording,StopRecording,PauseRecording,ResumeRecording,SetRecordingFolder,GetRecordingFolder,GetSourcesList,GetSourceTypesList,GetVolume,SetVolume,GetMute,SetMute,ToggleMute,SetSyncOffset,GetSyncOffset,GetSourceSettings,SetSourceSettings,GetTextGDIPlusProperties,SetTextGDIPlusProperties,GetTextFreetype2Properties,SetTextFreetype2Properties,GetBrowserSourceProperties,SetBrowserSourceProperties,GetSpecialSources,GetSourceFilters,GetSourceFilterInfo,AddFilterToSource,RemoveFilterFromSource,ReorderSourceFilter,MoveSourceFilter,SetSourceFilterSettings,SetSourceFilterVisibility,TakeSourceScreenshot,SetCurrentSceneCollection,GetCurrentSceneCollection,ListSceneCollections,GetTransitionList,GetCurrentTransition,SetCurrentTransition,SetTransitionDuration,GetTransitionDuration,GetSceneItemProperties,SetSceneItemProperties,ResetSceneItem,SetSceneItemRender,SetSceneItemPosition,SetSceneItemTransform,SetSceneItemCrop,DeleteSceneItem,DuplicateSceneItem}
+```
+
+Simple arguments can be provided directly on the command line:
+
+```
+$ obs-web-cli SetCurrentSceneCollection "Untitled"
+INFO:obswebsocket.core:Connecting...
+INFO:obswebsocket.core:Connected!
+{}
+INFO:obswebsocket.core:Disconnecting...
+```
+
+More complex arguments might require passing in a JSON string `json:` prefix. For example:
+
+```
+$ obs-web-cli SetSourceSettings "gif_source1" 'json:{"looping": true}' ffmpeg_source
+INFO:obswebsocket.core:Connecting...
+INFO:obswebsocket.core:Connected!
+{
+    "sourceName": "gif_source1",
+    "sourceSettings": {
+        "hw_decode": true,
+        "local_file": "/images/demo.gif",
+        "looping": true
+    },
+    "sourceType": "ffmpeg_source"
+}
+INFO:obswebsocket.core:Disconnecting...
+```
+
 ## Problems?
 
 Please check on [Github project issues](https://github.com/Elektordi/obs-websocket-py/issues), and if nobody else have experienced it before, you can [file a new issue](https://github.com/Elektordi/obs-websocket-py/issues/new).
diff --git a/obswebsocket/cli.py b/obswebsocket/cli.py
@@ -0,0 +1,101 @@
+import argparse
+import inspect
+import json
+import logging
+import os
+import sys
+
+import obswebsocket
+import obswebsocket.requests
+
+
+def setup_parser():
+    parser = argparse.ArgumentParser(
+        description="OBS Studio CLI using OBS Websocket Plugin",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+    subparsers = parser.add_subparsers(
+        help="OBS Commands", title="Recognized commands", dest="command"
+    )
+
+    parser.add_argument("--host", default="localhost", help="Hostname to connect to")
+    parser.add_argument("--port", default=4444, type=int, help="Port to connect to")
+    parser.add_argument(
+        "--password",
+        default=os.environ.get("OBS_WEBSOCKET_PASS", None),
+        help="Password to use. Defaults to OBS_WEBSOCKET_PASS env var",
+    )
+    parser.add_argument(
+        "--debug", default=False, action="store_true", help="Enable debugging output"
+    )
+
+    for subclass in obswebsocket.base_classes.Baserequests.__subclasses__():
+        # Generate a subcommand for each subclass with argument
+        subparser = subparsers.add_parser(subclass.__name__)
+        for arg in inspect.getargspec(subclass.__init__).args:
+            if arg == "self":
+                continue
+            # TODO: add defaults and maybe deal with optional args?
+            subparser.add_argument(arg)
+
+    return parser
+
+
+def main():
+    parser = setup_parser()
+    args = parser.parse_args()
+    if not args.command:
+        parser.error("No command specified")
+
+    log_level = logging.INFO
+    if args.debug:
+        log_level = logging.DEBUG
+
+    logging.basicConfig(level=log_level)
+
+    client = obswebsocket.obsws(args.host, args.port, args.password or "")
+    try:
+        client.connect()
+    except obswebsocket.exceptions.ConnectionFailure as e:
+        logging.error(e)
+        sys.exit(1)
+
+    for subclass in obswebsocket.base_classes.Baserequests.__subclasses__():
+        if subclass.__name__ != args.command:
+            continue
+
+        # OK, found which request class we need to instantiate.
+        # Now let's populate the arguments if we can
+        command_args = []
+        for arg in inspect.getfullargspec(subclass.__init__).args:
+            if arg == "self":
+                continue
+            val = args.__dict__.get(arg)
+            if val.startswith("json:"):
+                # Support "objects" by parsing JSON strings if the argument is
+                # prefixed with "json:"
+                val = json.loads(val[5:])
+            else:
+                # Try to convert numbers from string
+                try:
+                    val = int(val)
+                except ValueError:
+                    pass
+            command_args.append(val)
+
+        # Instantiate the request class based on collected args
+        instance = subclass(*command_args)
+        ret = client.call(instance)
+        if not ret.status:
+            # Call failed let's report and exit
+            logging.error("Call to OBS failed: %s", ret.datain["error"])
+            client.disconnect()
+            sys.exit(1)
+
+        print(json.dumps(ret.datain, indent=4))
+
+    client.disconnect()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/setup.py b/setup.py
@@ -44,6 +44,7 @@ def run(self):
     author_email='elektordi@elektordi.net',
     url='https://github.com/Elektordi/obs-websocket-py',
     keywords=['obs', 'obs-studio', 'websocket'],
+    entry_points={"console_scripts": ["obs-web-cli=obswebsocket.cli:main"]},
     classifiers=[
         'License :: OSI Approved :: MIT License',
         'Environment :: Plugins',
diff --git a/test_ci.py b/test_ci.py
@@ -1,4 +1,5 @@
-from obswebsocket import obsws, requests, events
+import os
+from obswebsocket import obsws, requests, events, cli
 
 host = "127.0.0.1"
 port = 4444
@@ -18,3 +19,154 @@ def test_build_ok_requests():
 def test_build_ok_events():
     e = events.Heartbeat()
     assert e.name == "Heartbeat"
+
+
+def test_cli_parser():
+    os.environ["OBS_WEBSOCKET_PASS"] = "obs-password"
+    parser = cli.setup_parser()
+    assert parser
+    subparsers = parser._get_positional_actions()[0]
+    # We expect at least these subcommands. Ideally would be kept up
+    # to date so we remove commands
+    expected_subcommands = {
+        "PlayPauseMedia",
+        "RestartMedia",
+        "StopMedia",
+        "NextMedia",
+        "PreviousMedia",
+        "GetMediaDuration",
+        "GetMediaTime",
+        "SetMediaTime",
+        "ScrubMedia",
+        "GetMediaState",
+        "GetStreamingStatus",
+        "StartStopStreaming",
+        "StartStreaming",
+        "StopStreaming",
+        "SetStreamSettings",
+        "GetStreamSettings",
+        "SaveStreamSettings",
+        "SendCaptions",
+        "GetStudioModeStatus",
+        "GetPreviewScene",
+        "SetPreviewScene",
+        "TransitionToProgram",
+        "EnableStudioMode",
+        "DisableStudioMode",
+        "ToggleStudioMode",
+        "ListOutputs",
+        "GetOutputInfo",
+        "StartOutput",
+        "StopOutput",
+        "GetReplayBufferStatus",
+        "StartStopReplayBuffer",
+        "StartReplayBuffer",
+        "StopReplayBuffer",
+        "SaveReplayBuffer",
+        "SetCurrentScene",
+        "GetCurrentScene",
+        "GetSceneList",
+        "CreateScene",
+        "ReorderSceneItems",
+        "SetSceneTransitionOverride",
+        "RemoveSceneTransitionOverride",
+        "GetSceneTransitionOverride",
+        "SetCurrentProfile",
+        "GetCurrentProfile",
+        "ListProfiles",
+        "GetVersion",
+        "GetAuthRequired",
+        "Authenticate",
+        "SetHeartbeat",
+        "SetFilenameFormatting",
+        "GetFilenameFormatting",
+        "GetStats",
+        "BroadcastCustomMessage",
+        "GetVideoInfo",
+        "OpenProjector",
+        "TriggerHotkeyByName",
+        "TriggerHotkeyBySequence",
+        "GetRecordingStatus",
+        "StartStopRecording",
+        "StartRecording",
+        "StopRecording",
+        "PauseRecording",
+        "ResumeRecording",
+        "SetRecordingFolder",
+        "GetRecordingFolder",
+        "GetMediaSourcesList",
+        "CreateSource",
+        "GetSourcesList",
+        "GetSourceTypesList",
+        "GetVolume",
+        "SetVolume",
+        "GetMute",
+        "SetMute",
+        "ToggleMute",
+        "GetAudioActive",
+        "SetSourceName",
+        "SetSyncOffset",
+        "GetSyncOffset",
+        "GetSourceSettings",
+        "SetSourceSettings",
+        "GetTextGDIPlusProperties",
+        "SetTextGDIPlusProperties",
+        "GetTextFreetype2Properties",
+        "SetTextFreetype2Properties",
+        "GetBrowserSourceProperties",
+        "SetBrowserSourceProperties",
+        "GetSpecialSources",
+        "GetSourceFilters",
+        "GetSourceFilterInfo",
+        "AddFilterToSource",
+        "RemoveFilterFromSource",
+        "ReorderSourceFilter",
+        "MoveSourceFilter",
+        "SetSourceFilterSettings",
+        "SetSourceFilterVisibility",
+        "GetAudioMonitorType",
+        "SetAudioMonitorType",
+        "TakeSourceScreenshot",
+        "SetCurrentSceneCollection",
+        "GetCurrentSceneCollection",
+        "ListSceneCollections",
+        "GetTransitionList",
+        "GetCurrentTransition",
+        "SetCurrentTransition",
+        "SetTransitionDuration",
+        "GetTransitionDuration",
+        "GetTransitionPosition",
+        "GetTransitionSettings",
+        "SetTransitionSettings",
+        "ReleaseTBar",
+        "SetTBarPosition",
+        "GetSceneItemList",
+        "GetSceneItemProperties",
+        "SetSceneItemProperties",
+        "ResetSceneItem",
+        "SetSceneItemRender",
+        "SetSceneItemPosition",
+        "SetSceneItemTransform",
+        "SetSceneItemCrop",
+        "DeleteSceneItem",
+        "AddSceneItem",
+        "DuplicateSceneItem",
+    }
+    assert not expected_subcommands - set(subparsers.choices.keys())
+
+    # Basic arguments + environ parsing
+    args = parser.parse_args(["--host", "hostname", "--port", "1234", "GetVideoInfo"])
+    assert args.host == "hostname"
+    assert args.port == 1234
+    assert args.password == "obs-password"
+    assert args.command == "GetVideoInfo"
+
+    # More complex command parsing
+    args = parser.parse_args(
+        ["SetSourceSettings", "SourceName", 'json:{"looping": true}', "ffmpeg_source"]
+    )
+    assert args.command == "SetSourceSettings"
+    # These will change depending on the sub-command
+    assert args.sourceName == "SourceName"
+    assert args.sourceSettings == 'json:{"looping": true}'
+    assert args.sourceType == "ffmpeg_source"
