Merge pull request #52 from iluvcapra/pt-2024.10

2024.10 Additions

README.md

@@ -37,6 +37,8 @@ client.
   options to be set from the terminal.
 - [pt_pasteboard.py](examples/pt_pasteboard.py) - Demonstrates
   triggering cut/copy/paste actions.
+- [toolshell.py](examples/toolshell.py) - Implements a command line 
+  interface for Pro Tools.
 
 
 ### Sending Commands To Pro Tools with the `Engine` class

examples/toolshell.py

@@ -0,0 +1,141 @@
+# toolshell.py
+
+import cmd
+import shlex
+import os.path
+from typing import Optional
+
+import ptsl
+from ptsl import PTSL_pb2 as pt
+
+
+class ToolShell(cmd.Cmd):
+    intro = """
+Toolshell is a demonstration command interpreter that
+can remotely operate Pro Tools. Type `help` or `?` to
+list commands.
+
+To begin, type `connect`.
+    """
+    prompt = "(not connected) "
+
+    client = None
+
+    def run_command_on_session(self, command_id: pt.CommandId,
+                               args: dict) -> Optional[dict]:
+        if self.client is None:
+            print("Command failed, not connected")
+            return None
+
+        try:
+            r = self.client.run_command(command_id, args)
+            return r
+        except ptsl.errors.CommandError as e:
+            if e.error_type == pt.PT_NoOpenedSession:
+                print("command failed, no session is currently open")
+                return None
+        except Exception:
+            print("Command failed, Pro Tools may not be running")
+            return None
+
+    def do_connect(self, _):
+        'Connect to Pro Tools'
+        self.client = ptsl.client.Client(company_name="py-ptsl",
+                                         application_name="Toolshell")
+        if self.client is not None:
+            self.prompt = "(pt) "
+
+    def do_sinfo(self, _):
+        'Print info about the open session: SINFO'
+        r = self.run_command_on_session(pt.GetSessionName, {})
+
+        assert r, "Failed to receive a response"
+        session_name = r['session_name']
+        r = self.run_command_on_session(pt.GetSessionIDs, {})
+        assert r
+        print(f"Connected to Pro Tools session \"{session_name}\"")
+        print(f"Session origin ID: {r['origin_id']}")
+        print(f"Session instance ID: {r['instance_id']}")
+
+    def do_newsession(self, args):
+        'Create a new session: NEWSESSION name save-path sample-rate'
+        name, path, sr = shlex.split(args)
+        print(f"Creating new session {name} at {path} and SR {sr}")
+        command_args = {'session_name': name,
+                        'session_location': os.path.expanduser(path),
+                        'file_type': 'FT_WAVE',
+                        'sample_rate': 'SR_' + str(sr),
+                        'bit_depth': 'Bit24',
+                        'input_output_settings': "IO_Last",
+                        "is_interleaved": True,
+                        "is_cloud_project": False,
+                        "create_from_template": False,
+                        "template_group": "",
+                        "template_name": ""
+                        }
+        assert self.client
+        self.client.run_command(pt.CreateSession, command_args)
+
+    def do_newtracks(self, args):
+        'Create new audio track: NEWTRACKS count format'
+        count, fmt = shlex.split(args)
+        command_args = {'number_of_tracks': count,
+                        'track_name': "New Track",
+                        'track_format': 'TF_' + fmt,
+                        'track_type': 'TT_Audio',
+                        'track_timebase': 'TTB_Samples',
+                        'insertion_point_position': 'TIPoint_Unknown',
+                        }
+        self.run_command_on_session(pt.CreateNewTracks, command_args)
+
+    def do_locate(self, args):
+        'Locate to a given time: LOCATE time'
+        time = args.strip()
+        command_args = {'play_start_marker_time': time,
+                        'in_time': time,
+                        'out_time': time,
+                        }
+        self.run_command_on_session(pt.SetTimelineSelection, command_args)
+
+    def do_newmemloc(self, args):
+        'Create a new marker memory location: NEWMEMLOC start-time'
+        command_args = {'name': 'New Marker',
+                        'start_time': args.strip(),
+                        'end_time': args.strip(),
+                        'time_properties': 'TP_Marker',
+                        'reference': 'MLR_FollowTrackTimebase',
+                        'general_properties': {
+                            'zoom_settings': False,
+                            'pre_post_roll_times': False,
+                            'track_visibility': False,
+                            'track_heights': False,
+                            'group_enables': False,
+                            'window_configuration': False,
+                        },
+                        'comments': "Created by toolshell",
+                        'color_index': 1,
+                        'location': 'MLC_MainRuler'
+                        }
+
+        self.run_command_on_session(pt.CreateMemoryLocation, command_args)
+
+    def do_play(self, _):
+        'Toggle the play state of the transport: PLAY'
+        assert self.client
+        try:
+            self.client.run_command(pt.CommandId.TogglePlayState, {})
+        except ptsl.errors.CommandError as e:
+            if e.error_type == pt.PT_NoOpenedSession:
+                print("play failed, no session is currently open")
+                return False
+
+    def do_bye(self, _):
+        'Quit Toolshell and return to your shell: BYE'
+        print("Tooshell quitting...")
+        if self.client:
+            self.client.close()
+        return True
+
+
+if __name__ == '__main__':
+    ToolShell().cmdloop()

ptsl/client.py

@@ -19,7 +19,7 @@
 from ptsl.ops import Operation
 
 
-PTSL_VERSION = 3
+PTSL_VERSION = 5
 
 
 @contextmanager
@@ -137,20 +137,54 @@ def __init__(self,
 
             raise grpc_error
 
-    def run(self, operation: Operation) -> None:
+    def run_command(self, command_id: pt.CommandId,
+                    request: dict) -> Optional[dict]:
+        """
+        Run a command on the client with a JSON request.
+
+        :param command_id: The command to run
+        :param request: The request parameters. This dict will be converted to
+            JSON.
+        :returns: The response if any. This is the JSON response returned by
+            the server and converted to a dict.
+        """
+        request_body_json = json.dumps(request)
+        response = self._send_sync_request(command_id, request_body_json)
+
+        if response.header.status == pt.Failed:
+            cleaned_response_error_json = response.response_error_json
+            # self._response_error_json_cleanup(
+            # response.response_error_json)
+            command_errors = json_format.Parse(cleaned_response_error_json,
+                                               pt.ResponseError())
+            raise CommandError(list(command_errors.errors))
+
+        elif response.header.status == pt.Completed:
+            if len(response.response_body_json) > 0:
+                return json.loads(response.response_body_json)
+            else:
+                return None
+        else:
+            # FIXME: dump out for now, will be on the lookout for when
+            # this happens
+            assert False, \
+                f"Unexpected response code {response.header.status} " + \
+                f"({pt.TaskStatus.Name(response.header.status)})"
+
+    def run(self, op: Operation) -> None:
         """
         Run an operation on the client.
 
         :raises: `CommandError` if the server returns an error
         """
 
-        self.auditor.run_called(operation.command_id())
+        self.auditor.run_called(op.command_id())
 
         # convert the request body into JSON
-        request_body_json = self._prepare_operation_request_json(operation)
-        response = self._send_sync_request(operation.command_id(),
+        request_body_json = self._prepare_operation_request_json(op)
+        response = self._send_sync_request(op.command_id(),
                                            request_body_json)
-        operation.status = response.header.status
+        op.status = response.header.status
 
         if response.header.status == pt.Failed:
             cleaned_response_error_json = response.response_error_json
@@ -161,7 +195,7 @@ def run(self, operation: Operation) -> None:
             raise CommandError(list(command_errors.errors))
 
         elif response.header.status == pt.Completed:
-            self._handle_completed_response(operation, response)
+            self._handle_completed_response(op, response)
         else:
             # FIXME: dump out for now, will be on the lookout for when
             # this happens

pyproject.toml

@@ -2,7 +2,7 @@
 
 [tool.poetry]
 name = "py-ptsl"
-version = "500.0.0"
+version = "500.1.0"
 description = "Native Python PTSL (Pro Tools Scripting Library) RPC interface"
 authors = ["Jamie Hardt <[email protected]>"]
 license = "BSD-3-Clause"