__author__ = "Ruben Svensson"
__copyright__ = """
Copyright 2017-2019 Modern Ancient Instruments Networked AB, dba Elk
elkpy is free software: you can redistribute it and/or modify it under the terms of the
GNU General Public License as published by the Free Software Foundation, either version 3
of the License, or (at your option) any later version.
elkpy is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License along with elkpy. If
not, see <http://www.gnu.org/licenses/>.
"""
__license__ = "GPL-3.0"
import grpc
from . import sushierrors
from . import grpc_gen
from . import sushi_info_types as info_types
from typing import List
####################################
# Sushi audio graph controller class #
####################################
[docs]class AudioGraphController(object):
"""
A class to control the audio graph in sushi via gRPC. It has functions to create, move, delete and get info
about audio graph object like tracks and processors.
Attributes:
_stub (AudioGraphControllerStub): Connection stubs to the gRPC audio graph interface implemented in sushi.
"""
[docs] def __init__(self,
address = 'localhost:51051',
sushi_proto_def = '/usr/share/sushi/sushi_rpc.proto'):
"""
The constructor for the AudioGraphController class setting up the gRPC connection with sushi.
Parameters:
address (str): 'ip-addres:port' The ip-addres and port at which to connect to sushi.
sushi_proto_def (str): path to .proto file with SUSHI's gRPC services definition
"""
try:
channel = grpc.insecure_channel(address)
except AttributeError as e:
raise TypeError("Parameter address = {}. Should be a string containing the ip-address and port of sushi ('ip-address:port')".format(address)) from e
self._sushi_proto, self._sushi_grpc = grpc_gen.modules_from_proto(sushi_proto_def)
self._stub = self._sushi_grpc.AudioGraphControllerStub(channel)
[docs] def get_all_processors(self) ->List[info_types.ProcessorInfo]:
"""
Gets a list of all available processors.
Returns:
List[info_types.ProcessorInfo]: A list with the info of all the available processors.
"""
try:
response = self._stub.GetAllProcessors(self._sushi_proto.GenericVoidValue())
processor_info_list = []
for processor_info in response.processors:
processor_info_list.append(info_types.ProcessorInfo(processor_info))
return processor_info_list
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e)
[docs] def get_all_tracks(self) -> List[info_types.TrackInfo]:
"""
Gets a list of all available tracks.
Returns:
List[info_types.TrackInfo]: A list with the info of all the available tracks.
"""
try:
response = self._stub.GetAllTracks(self._sushi_proto.GenericVoidValue())
track_info_list = []
for track_info in response.tracks:
track_info_list.append(info_types.TrackInfo(track_info))
return track_info_list
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e)
[docs] def get_track_id(self, track_name: str) -> int:
"""
Get the id of a track from its name.
Parameters:
track_name (str): The name of the track.
Returns:
int: The id of the track matching the name.
"""
try:
response = self._stub.GetTrackId(self._sushi_proto.GenericStringValue(
value = track_name
))
return response.id
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track name: {}".format(track_name))
[docs] def get_track_info(self, track_identifier: int) -> info_types.TrackInfo:
"""
Get the info of a track from its id.
Parameters:
track_identifier (int): The id of the track to get the info from.
Returns:
info_types.TrackInfo: The info of the track matching the id.
"""
try:
response = self._stub.GetTrackInfo(self._sushi_proto.TrackIdentifier(
id = track_identifier
))
return info_types.TrackInfo(response)
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track id: {}".format(track_identifier))
[docs] def get_track_processors(self, track_identifier: int) -> List[info_types.ProcessorInfo]:
"""
Get a list of processors assigned on the specified track.
Parameters:
track_identifier (int): The id of the track to get the processor list from.
Returns:
List[info_types.ProcessorInfo]: A list of the info of the processors assigned to the track matching the id.
"""
try:
response = self._stub.GetTrackProcessors(self._sushi_proto.TrackIdentifier(
id = track_identifier
))
processor_info_list = []
for processor_info in response.processors:
processor_info_list.append(info_types.ProcessorInfo(processor_info))
return processor_info_list
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track id: {}".format(track_identifier))
[docs] def get_processor_id(self, processor_name: str) -> int:
"""
Get the id of a processor from its name.
Parameters:
processor_name (str): The name of the processor to get the id from.
Returns:
int: The id of the processor matching the name.
"""
try:
response = self._stub.GetProcessorId(self._sushi_proto.GenericStringValue(
value = processor_name
))
return response.id
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor name: {}".format(processor_name))
[docs] def get_processor_info(self, processor_identifier: int) -> info_types.ProcessorInfo:
"""
Get the info of a processor from its id.
Parameters:
track_identifier (int): The id of the processor to get the info from.
Returns:
info_types.ProcessorInfo: The info of the processor matching the id.
"""
try:
response = self._stub.GetProcessorInfo(self._sushi_proto.ProcessorIdentifier(
id = processor_identifier
))
return info_types.ProcessorInfo(response)
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}".format(processor_identifier))
[docs] def get_processor_bypass_state(self, processor_identifier: int) -> bool:
"""
Get the bypass state of the specified processor.
Parameters:
processor_identifier (int): The id of processor to get the bypass state from.
Returns:
bool: The bypass state of the processor matching the id.
"""
try:
response = self._stub.GetProcessorBypassState(self._sushi_proto.ProcessorIdentifier(
id = processor_identifier
))
return response.value
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}".format(processor_identifier))
[docs] def get_processor_state(self, processor_identifier: int) -> info_types.ProcessorState:
"""
Get the full state of the specified processor.
Parameters:
processor_identifier (int): The id of processor to get the full state from.
Returns:
ProcessorState: An object describing the full the processor matching the id.
"""
try:
response = self._stub.GetProcessorState(self._sushi_proto.ProcessorIdentifier(
id = processor_identifier
))
return info_types.ProcessorState(response)
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}".format(processor_identifier))
[docs] def set_processor_bypass_state(self, processor_identifier: int, bypass_state: bool) -> None:
"""
Set the bypass state of the specified processor.
Parameters:
processor_identifier (int): The id of the processor to set the bypass state of.
bypass_sate (bool): The bypass state of the processor matching the id.
"""
try:
self._stub.SetProcessorBypassState(self._sushi_proto.ProcessorBypassStateSetRequest(
processor = self._sushi_proto.ProcessorIdentifier(id = processor_identifier),
value = bypass_state
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}, bypass state: {}".format(processor_identifier, bypass_state))
def set_processor_state(self, processor_identifier: int, program_id: int = None, bypassed: None = False,
property_values: (int, str) = [], parameter_values: (int, float) = []) -> None:
"""
Set the full or partial state of the specified processor.
Parameters:
program_id (int): The id of the program to set.
bypassed (bool): Whether the processor should be bypassed or not.
properties ((int, str)): A list of tuples (id, value) of string properties to set.
parameters ((int, float)): A list of tuples (id, value) of parameter values to set.
"""
try:
grpc_state = self._sushi_proto.ProcessorState()
if program_id:
grpc_state.program_id.value = program_id
grpc_state.program_id.has_value = True
if bypassed != None:
grpc_state.bypassed.value = bypassed
grpc_state.bypassed.has_value = True
for property in property_values:
grpc_property = grpc_state.properties.add()
grpc_property.property.property_id = property[0]
grpc_property.value = property[1]
for parameter in parameter_values:
grpc_parameter = grpc_state.parameters.add()
grpc_parameter.parameter.parameter_id = parameter[0]
grpc_parameter.value = parameter[1]
self._stub.SetProcessorState(self._sushi_proto.ProcessorStateSetRequest(
processor = self._sushi_proto.ProcessorIdentifier(id = processor_identifier),
state = grpc_state
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}".format(processor_identifier))
[docs] def set_processor_state(self, processor_identifier: int, state: info_types.ProcessorState) -> None:
"""
Set the full or partial state of the specified processor from an existing state object.
Parameters:
state (info_types.ProcessorState): a state object either populated manually or received from a call to get_processor_state.
"""
try:
grpc_state = self._sushi_proto.ProcessorState()
if state.program_id:
grpc_state.program_id.value = state.program_id
grpc_state.program_id.has_value = True
if state.bypassed != None:
grpc_state.bypassed.value = state.bypassed
grpc_state.bypassed.has_value = True
for property in state.properties:
grpc_property = grpc_state.properties.add()
grpc_property.property.property_id = property[0]
grpc_property.value = property[1]
for parameter in state.parameters:
grpc_parameter = grpc_state.parameters.add()
grpc_parameter.parameter.parameter_id = parameter[0]
grpc_parameter.value = parameter[1]
if len(state.binary_data) > 0:
grpc_state.binary_data = state.binary_data
self._stub.SetProcessorState(self._sushi_proto.ProcessorStateSetRequest(
processor = self._sushi_proto.ProcessorIdentifier(id = processor_identifier),
state = grpc_state
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}".format(processor_identifier))
[docs] def create_track(self, name: str, channels: int) -> None:
"""
Create a new track in sushi.
Parameters:
name (str): The name of the new track.
channels (int): The number of channels to assign the new track.
"""
try:
self._stub.CreateTrack(self._sushi_proto.CreateTrackRequest(
name = name,
channels = channels
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track name: {}, number of channels: {}".format(name, channels))
[docs] def create_multibus_track(self, name: str, buses: int) -> None:
"""
Create a new multibus track in sushi.
Parameters:
name (str): The name of the new track.
buses (int): The number of audio buses in the new track.
"""
try:
self._stub.CreateMultibusTrack(self._sushi_proto.CreateMultibusTrackRequest(
name = name,
buses = buses
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track name: {}, buses: {}".format(name, buses))
[docs] def create_pre_track(self, name: str) -> None:
"""
Create a new pre track in sushi.
Parameters:
name (str): The name of the new track.
"""
try:
self._stub.CreatePreTrack(self._sushi_proto.CreatePreTrackRequest(
name = name
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track name: {}".format(name))
[docs] def create_post_track(self, name: str) -> None:
"""
Create a new post track in sushi.
Parameters:
name (str): The name of the new track.
"""
try:
self._stub.CreatePostTrack(self._sushi_proto.CreatePostTrackRequest(
name = name
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track name: {}".format(name))
[docs] def create_processor_on_track(self, name: str, uid: str, path:
str, processor_type: info_types.PluginType, track_id: int,
before_processor: int, add_to_back: bool) -> None:
"""
Create a new processor on an existing track.
Parameters:
name (str): The name to assign the processor. Must be unique.
uid (str): The uid of an internal sushi processor or the URI of an LV2 processor. Not applicable for vst2 or vst3.
path (str): The path to the processor library. Only for vst2 or vst3 processors.
processor_type (info_type.PluginType): The type of processor to create.
track_id (int): The id of the track to add the processor to.
before_processor (int): Which existing processor to create the new processor in front of.
add_to_back (bool): Set to true to add the processor to the back of the processing chain on the track.
"""
try:
self._stub.CreateProcessorOnTrack(self._sushi_proto.CreateProcessorRequest(
name = name,
uid = uid,
path = path,
type = self._sushi_proto.PluginType(type = processor_type),
track = self._sushi_proto.TrackIdentifier(id = track_id),
position = self._sushi_proto.ProcessorPosition(add_to_back = add_to_back,
before_processor = self._sushi_proto.ProcessorIdentifier(id = before_processor))
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor name: {}, uid: {}, path: {}, type: {}, id: {}, position: {}, add_to_back: {}".format(name, uid, path, processor_type, track_id, before_processor, add_to_back))
[docs] def move_processor_on_track(self, processor: int, source_track: int, destination_track: int, before_processor: int, add_to_back: bool) -> None:
"""
Move an existing processor.
Parameters:
processor (int): The id of the processor to move.
source_track (int): The id of the track to move the processor from.
destination_track (int): The id of the track to move the processor to.
before_processor (int): The id of another processor to move this processor in front of.
add_to_back (bool): Set to true to add the processor to the back of the processing chain on the track.
"""
try:
self._stub.MoveProcessorOnTrack(self._sushi_proto.MoveProcessorRequest(
processor = self._sushi_proto.ProcessorIdentifier(id = processor),
source_track = self._sushi_proto.TrackIdentifier(id = source_track),
dest_track = self._sushi_proto.TrackIdentifier(id = destination_track),
position = self._sushi_proto.ProcessorPosition(add_to_back = add_to_back,
before_processor = self._sushi_proto.ProcessorIdentifier(id = before_processor))
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}, source_track: {}, dest_track: {}, before_processor: {}, add_to_back: {}".format(processor, source_track, destination_track, before_processor, add_to_back))
[docs] def delete_processor_from_track(self, processor: int, track: int) -> None:
"""
Delete an existing processor from a track.
Parameters:
processor (int): The id of the processor to delete.
track (int): The id of the track that contains the processor.
"""
try:
self._stub.DeleteProcessorFromTrack(self._sushi_proto.DeleteProcessorRequest(
processor = self._sushi_proto.ProcessorIdentifier(id = processor),
track = self._sushi_proto.TrackIdentifier(id = track)
))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With processor id: {}, track id: {}".format(processor, track))
[docs] def delete_track(self, track_id: int) -> None:
"""
Delet a track.
Parameters:
track_id (int): The id of the track to delete.
"""
try:
self._stub.DeleteTrack(self._sushi_proto.TrackIdentifier(id = track_id))
except grpc.RpcError as e:
sushierrors.grpc_error_handling(e, "With track id: {}".format(track_id))