diff --git a/README.md b/README.md index cfae0e7..e62b4f3 100644 --- a/README.md +++ b/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 diff --git a/examples/toolshell.py b/examples/toolshell.py new file mode 100644 index 0000000..b609f71 --- /dev/null +++ b/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() diff --git a/ptsl/client.py b/ptsl/client.py index 127fc89..fd517ed 100644 --- a/ptsl/client.py +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index c36b18b..a6fd345 100644 --- a/pyproject.toml +++ b/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 "] license = "BSD-3-Clause"