diff options
Diffstat (limited to 'scripts')
| -rw-r--r-- | scripts/ingen.py | 308 | ||||
| -rwxr-xr-x | scripts/ingenams | 283 | ||||
| -rwxr-xr-x | scripts/ingenish | 121 |
3 files changed, 0 insertions, 712 deletions
diff --git a/scripts/ingen.py b/scripts/ingen.py deleted file mode 100644 index 594b98cf..00000000 --- a/scripts/ingen.py +++ /dev/null @@ -1,308 +0,0 @@ -#!/usr/bin/env python -# Ingen Python Interface -# Copyright 2012-2015 David Robillard <http://drobilla.net> -# -# Permission to use, copy, modify, and/or distribute this software for any -# purpose with or without fee is hereby granted, provided that the above -# copyright notice and this permission notice appear in all copies. -# -# THIS SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES -# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF -# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR -# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES -# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN -# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF -# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. - -import os -import rdflib -import re -import socket -import sys - -try: - import StringIO.StringIO as StringIO -except ImportError: - from io import StringIO as StringIO - -class NS: - atom = rdflib.Namespace('http://lv2plug.in/ns/ext/atom#') - ingen = rdflib.Namespace('http://drobilla.net/ns/ingen#') - ingerr = rdflib.Namespace('http://drobilla.net/ns/ingen/errors#') - lv2 = rdflib.Namespace('http://lv2plug.in/ns/lv2core#') - patch = rdflib.Namespace('http://lv2plug.in/ns/ext/patch#') - rdf = rdflib.Namespace('http://www.w3.org/1999/02/22-rdf-syntax-ns#') - rsz = rdflib.Namespace('http://lv2plug.in/ns/ext/resize-port#') - xsd = rdflib.Namespace('http://www.w3.org/2001/XMLSchema#') - -class Interface: - 'The core Ingen interface' - def put(self, subject, body): - pass - - def patch(self, subject, remove, add): - pass - - def get(self, subject): - pass - - def set(self, subject, key, value): - pass - - def connect(self, tail, head): - pass - - def disconnect(self, tail, head): - pass - - def delete(self, subject): - pass - -class Error(Exception): - def __init__(self, msg, cause): - Exception.__init__(self, '%s; cause: %s' % (msg, cause)) - -def lv2_path(): - path = os.getenv('LV2_PATH') - if path: - return path - elif sys.platform == 'darwin': - return os.pathsep.join(['~/Library/Audio/Plug-Ins/LV2', - '~/.lv2', - '/usr/local/lib/lv2', - '/usr/lib/lv2', - '/Library/Audio/Plug-Ins/LV2']) - elif sys.platform == 'haiku': - return os.pathsep.join(['~/.lv2', - '/boot/common/add-ons/lv2']) - elif sys.platform == 'win32': - return os.pathsep.join([ - os.path.join(os.getenv('APPDATA'), 'LV2'), - os.path.join(os.getenv('COMMONPROGRAMFILES'), 'LV2')]) - else: - return os.pathsep.join(['~/.lv2', - '/usr/lib/lv2', - '/usr/local/lib/lv2']) - -def ingen_bundle_path(): - for d in lv2_path().split(os.pathsep): - bundle = os.path.abspath(os.path.join(d, 'ingen.lv2')) - if os.path.exists(bundle): - return bundle - return None - -class Remote(Interface): - def __init__(self, uri='unix:///tmp/ingen.sock'): - self.msg_id = 1 - self.server_base = uri + '/' - self.model = rdflib.Graph() - self.ns_manager = rdflib.namespace.NamespaceManager(self.model) - self.ns_manager.bind('server', self.server_base) - for (k, v) in NS.__dict__.items(): - self.ns_manager.bind(k, v) - if uri.startswith('unix://'): - self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - self.sock.connect(uri[len('unix://'):]) - elif uri.startswith('tcp://'): - self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - parsed = re.split('[:/]', uri[len('tcp://'):]) - addr = (parsed[0], int(parsed[1])) - self.sock.connect(addr) - else: - raise Exception('Unsupported server URI `%s' % uri) - - # Parse error description from Ingen bundle for pretty printing - bundle = ingen_bundle_path() - if bundle: - self.model.parse(os.path.join(bundle, 'errors.ttl'), format='n3') - - def __del__(self): - self.sock.close() - - def _get_prefixes_string(self): - s = '' - for k, v in self.ns_manager.namespaces(): - s += '@prefix %s: <%s> .\n' % (k, v) - return s - - def msgencode(self, msg): - if sys.version_info[0] == 3: - return bytes(msg, 'utf-8') - else: - return msg - - def update_model(self, update): - for i in update.triples([None, NS.rdf.type, NS.patch.Put]): - put = i[0] - subject = update.value(put, NS.patch.subject, None) - body = update.value(put, NS.patch.body, None) - desc = {} - for i in update.triples([body, None, None]): - self.model.add([subject, i[1], i[2]]) - return update - - def uri_to_path(self, uri): - path = uri - if uri.startswith(self.server_base): - return uri[len(self.server_base)-1:] - return uri - - def recv(self): - 'Read from socket until a NULL terminator is received' - msg = u'' - while True: - c = self.sock.recv(1, 0).decode('utf-8') - if not c or ord(c[0]) == 0: # End of transmission - break - else: - msg += c[0] - return msg - - def blank_closure(self, graph, node): - def blank_walk(node, g): - for i in g.triples([node, None, None]): - if type(i[2]) == rdflib.BNode and i[2] != node: - yield i[2] - blank_walk(i[2], g) - - closure = [node] - for b in graph.transitiveClosure(blank_walk, node): - closure += [b] - - return closure - - def raise_error(self, code, cause): - klass = self.model.value(None, NS.ingerr.errorCode, rdflib.Literal(code)) - if not klass: - raise Error('error %d' % code, cause) - - fmt = self.model.value(klass, NS.ingerr.formatString, None) - if not fmt: - raise Error('%s' % klass, cause) - - raise Error(fmt, cause) - - def send(self, msg): - # Send message to server - payload = msg - if sys.version_info[0] == 3: - payload = bytes(msg, 'utf-8') - self.sock.send(self.msgencode(msg)) - - # Receive response and parse into a model - response_str = self._get_prefixes_string() + self.recv() - response_model = rdflib.Graph(namespace_manager=self.ns_manager) - - # Because rdflib has embarrassingly broken base URI resolution that - # just drops path components from the base URI entirely (seriously), - # unfortunate the real server base URI can not be used here. Use - # <ingen:/> instead to at least avoid complete nonsense - response_model.parse(StringIO(response_str), 'ingen:/', format='n3') - - # Add new prefixes to prepend to future responses because rdflib sucks - for line in response_str.split('\n'): - if line.startswith('@prefix'): - match = re.search('@prefix ([^:]*): <(.*)> *\.', line) - if match: - name = match.group(1) - uri = match.group(2) - self.ns_manager.bind(match.group(1), match.group(2)) - - # Handle response (though there should be only one) - blanks = [] - response_desc = [] - for i in response_model.triples([None, NS.rdf.type, NS.patch.Response]): - response = i[0] - subject = response_model.value(response, NS.patch.subject, None) - body = response_model.value(response, NS.patch.body, None) - - response_desc += [i] - blanks += [response] - if body != 0: - self.raise_error(int(body), msg) # Raise exception on server error - - # Find the blank node closure of all responses - blank_closure = [] - for b in blanks: - blank_closure += self.blank_closure(response_model, b) - - # Remove response descriptions from model - for b in blank_closure: - for t in response_model.triples([b, None, None]): - response_model.remove(t) - - # Remove triples describing responses from response model - for i in response_desc: - response_model.remove(i) - - # Update model with remaining information, e.g. patch:Put updates - return self.update_model(response_model) - - def get(self, subject): - return self.send(''' -[] - a patch:Get ; - patch:subject <%s> . -''' % subject) - - def put(self, subject, body): - return self.send(''' -[] - a patch:Put ; - patch:subject <%s> ; - patch:body [ -%s - ] . -''' % (subject, body)) - - def patch(self, subject, remove, add): - return self.send(''' -[] - a patch:Patch ; - patch:subject <%s> ; - patch:remove [ -%s - ] ; - patch:add [ -%s - ] . -''' % (subject, remove, add)) - - def set(self, subject, key, value): - return self.send(''' -[] - a patch:Set ; - patch:subject <%s> ; - patch:property <%s> ; - patch:value %s . -''' % (subject, key, value)) - - def connect(self, tail, head): - return self.send(''' -[] - a patch:Put ; - patch:subject <%s> ; - patch:body [ - a ingen:Arc ; - ingen:tail <%s> ; - ingen:head <%s> ; - ] . -''' % (os.path.commonprefix([tail, head]), tail, head)) - - def disconnect(self, tail, head): - return self.send(''' -[] - a patch:Delete ; - patch:body [ - a ingen:Arc ; - ingen:tail <%s> ; - ingen:head <%s> ; - ] . -''' % (tail, head)) - - def delete(self, subject): - return self.send(''' -[] - a patch:Delete ; - patch:subject <%s> . -''' % subject) diff --git a/scripts/ingenams b/scripts/ingenams deleted file mode 100755 index a183586a..00000000 --- a/scripts/ingenams +++ /dev/null @@ -1,283 +0,0 @@ -#!/usr/bin/env python -# Load an AlsaModularSynth patch file into Ingen -# Copyright 2012-2015 David Robillard <http://drobilla.net> -# -# Permission to use, copy, modify, and/or distribute this software for any -# purpose with or without fee is hereby granted, provided that the above -# copyright notice and this permission notice appear in all copies. -# -# THIS SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES -# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF -# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR -# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES -# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN -# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF -# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. - -import ingen -import rdflib -import rdflib.namespace -import sys - -ams_prefix = 'http://github.com/blablack/ams-lv2/' -fomp_prefix = 'http://drobilla.net/plugins/fomp/' -note_uri = 'http://drobilla.net/ns/ingen-internals#Note' - -class World: - def __init__(self, server_uri): - self.server_uri = server_uri - self.server = ingen.Remote(server_uri) - self.pending_arcs = [] - self.server.get('/') - self.mod_prototypes = {} - - def mod_sym(self, mod_id): - return 'mod%d' % int(mod_id) - - def add_block(self, mod_id, plugin_uri, x, y): - self.mod_prototypes[self.mod_sym(mod_id)] = plugin_uri - self.server.put('/' + self.mod_sym(mod_id), - ('\t\ta ingen:Block ;\n' - + 'lv2:prototype <%s> ;\n' % plugin_uri - + 'ingen:canvasX %f ;\n' % x - + 'ingen:canvasY %f' % y).replace('\n', '\n\t\t')) - - def add_arc(self, - head_port_id, tail_port_id, - head_mod_id, tail_mod_id, - jack_color, cable_color): - self.pending_arcs += [(head_port_id, tail_port_id, - head_mod_id, tail_mod_id, - jack_color, cable_color)] - - def get_ports(self, mod_uri, port_type): - ports = [] - for i in self.server.model.triples([None, ingen.NS.rdf.type, port_type]): - if str(i[0]).startswith(mod_uri + '/'): - if not [i[0], ingen.NS.rdf.type, ingen.NS.lv2.ControlPort] in self.server.model: - # Unfortunately ingen.NS.lv2.index is a method - index = self.server.model.value(i[0], ingen.NS.lv2['index'], None) - ports += [[int(index), i[0]]] - return ports - - def input_by_id(self, mod_id, port_id): - mod_uri = rdflib.URIRef(self.server.server_base + self.mod_sym(mod_id)) - - # Get all input ports on this module sorted by index - inputs = sorted(self.get_ports(mod_uri, ingen.NS.lv2.InputPort)) - - # Return the port_id'th port in the list - index = 0 - for i in inputs: - if index == int(port_id): - return i[1] - index += 1 - - return None - - def output_by_id(self, mod_id, port_id): - mod_uri = rdflib.URIRef(self.server.server_base + self.mod_sym(mod_id)) - - # Get all output ports on this module sorted by index - outputs = sorted(self.get_ports(mod_uri, ingen.NS.lv2.OutputPort)) - - port_index = int(port_id) - if world.mod_prototypes[self.mod_sym(mod_id)] == note_uri: - # Adapt MCV/ADVMCV port index to Note port index - port_mapping = [ 3, 0, 2, 4, 6, 5, -1, -1, -1, -1 ] - port_index = port_mapping[port_index] - if port_index == -1: - sys.stderr.write('warning: unsupported MCV port %d\n' % int(port_id)) - return - - # Return the port_id'th port in the list - if port_index < len(outputs): - return outputs[port_index][1] - - return None - - def create_arcs(self): - for (head_port_id, tail_port_id, - head_mod_id, tail_mod_id, - jack_color, cable_color) in self.pending_arcs: - print('%s:%s => %s:%s' % (tail_mod_id, tail_port_id, head_mod_id, head_port_id)) - try: - tail = self.output_by_id(tail_mod_id, tail_port_id) - head = self.input_by_id(head_mod_id, head_port_id) - if tail and head: - self.server.connect(self.server.uri_to_path(tail), - self.server.uri_to_path(head)) - except: - pass - -# Static enumeration of special module type IDs -class Special: - CUSTOM = 0 - LADSPA = 6 - MCV = 10 - SCMCV = 30 - SCQUANTIZER = 31 - ADVMCV = 35 - -# Module types list, indexed by numeric ID in file -# Except where otherwise commented, these correspond to internal modules, -# and the string is the suffix of the corresponding AMS LV2 plugin URI -module_types = [ - "custom", # 0 = custom (unsupported) - "vco", - "vca", - "lfo", - "delay", - "ringmod", - "ladspa", # 6 = LADSPA plugin - "pcmout", - "mix", - "vcf", - "mcv", - "env", - "seq", - "inv", - "noise", - "slew", - "quantizer", - "pcmin", - "cvs", - "sh", - "vcorgan", - "dynamicwaves", - "advenv", - "wavout", - "scope", - "spectrum", - "vcswitch", - "jackin", - "jackout", - "midiout", - "scmcv", # Scala module (different line format) - "scquantizer", # Scala module (different line format) - "stereomix", - "conv", - "vcenv", - "advmcv", - "function", - "vcpanning", - "vcenv2", - "vcdoubledecay", - "vquant", - "amp", - "ad", - "mphlfo", - "noise2", - "vco2" -] - -class Module: - def __init__(self, num, plugin_uri, properties={}): - self.num = num - self.plugin_uri = plugin_uri - self.properties = properties - self.ports = [] - -class Patch: - def __init__(self): - self.modules = [] - -def ladspa_module(world, mod_id, x, y, poly, lib, label): - lv2_uri = '' - # Kludge LADSPA library and label to LV2 URIs where applicable - if lib == 'blvco': - lv2_uri = fomp_prefix + label.lower().replace('-', '_') - elif lib == 'mvclpf24' or lib == 'mvchpf24': - lv2_uri = fomp_prefix + label.lower().replace('-', '') - elif lib == 'cs_chorus' or lib == 'cs_phaser': - lv2_uri = fomp_prefix + 'cs_' + label.lower().replace('+', '_') - - if lv2_uri: - world.add_block(mod_id, lv2_uri, x, y) - else: - print('MOD %3d LADSPA %s %s %s' % (mod_id, poly, lib, label)) - -def scala_module(world, mod_id, scala_name): - sys.stderr.write('warning: scala module %3d (%s) unsupported\n' % (d, scala_name)) - -def standard_module(world, mod_id, x, y, name, arg): - if name == 'vca': - if int(arg) > 0: - name += 'exp' - else: - name += 'lin' - elif name == 'mix': - name += 'er_%dch' % int(arg) - - lv2_uri = ams_prefix + name - world.add_block(mod_id, lv2_uri, x, y) - -def float_control(world, mod_id, port_index, value, - logarithmic, minimum, maximum, midi_sign): - #print('FLOAT CONTROL %s:%s = %s' % (mod_id, port_index, value)) - pass - -def control(world, mod_id, port_index, value, midi_sign): - #print('CONTROL %s:%s = %s' % (mod_id, port_index, value)) - pass - -if len(sys.argv) != 2 and len(sys.argv) != 3: - sys.stderr.write('Usage: %s AMS_PATCH_FILE [SERVER_URI]\n' % sys.argv[0]) - sys.exit(1) - -in_path = sys.argv[1] -server_uri = 'unix:///tmp/ingen.sock' -if len(sys.argv) == 3: - server_uri = sys.argv[2] - -world = World(server_uri) -in_file = open(in_path, 'r') - -in_comment = False -for l in in_file: - try: - expr = l.split() - if not expr: - continue - elif expr[0] == '#PARA#': - in_comment = True - elif in_comment and expr[0] == '#ARAP#': - in_comment = False - elif expr[0] == 'Module': - mod_type = int(expr[1]) - mod_id = int(expr[2]) - mod_x = int(expr[3]) - mod_y = int(expr[4]) - if mod_type > len(module_types): - sys.stderr.write('warning: unknown module type %d\n', mod_type) - elif mod_type == Special.CUSTOM: - sys.stderr.write('warning: custom module %d unsupported\n' % mod_id) - if mod_type == Special.LADSPA: - ladspa_module(world, mod_id, mod_x, mod_y, int(expr[5]), expr[6], expr[7]) - elif mod_type == Special.SCMCV or mod_type == Special.SCQUANTIZER: - scala_name = expr[5] - scala_module(world, mod_id, scala_name) - elif mod_type == Special.MCV or mod_type == Special.ADVMCV: - world.add_block(mod_id, note_uri, mod_x, mod_y) - else: - standard_module(world, mod_id, mod_x, mod_y, module_types[mod_type], expr[5]) - elif expr[0] == 'ColorP': - world.add_arc(expr[1], expr[2], expr[3], expr[4], - (expr[5], expr[6], expr[7]), - (expr[8], expr[9], expr[10])) - elif expr[0] == 'FSlider': - float_control(world, mod_id, - expr[2], expr[3], expr[4], expr[5], expr[6], expr[7]) - elif expr[0] == 'ISlider' or expr[0] == 'LSlider': - control(world, mod_id, expr[2], expr[3], expr[4]) - #else: - # sys.stderr.write('warning: unsupported form %s\n' % expr[0]) - except ingen.Error: - e = sys.exc_info()[1] - sys.stderr.write('ingen error: %s\n' % e.message) - -world.create_arcs() - -#print(world.server.model.serialize(format='n3')) - -in_file.close() diff --git a/scripts/ingenish b/scripts/ingenish deleted file mode 100755 index 97640645..00000000 --- a/scripts/ingenish +++ /dev/null @@ -1,121 +0,0 @@ -#!/usr/bin/env python -# Ingen Interactive Shell -# Copyright 2011-2015 David Robillard <http://drobilla.net> -# -# Permission to use, copy, modify, and/or distribute this software for any -# purpose with or without fee is hereby granted, provided that the above -# copyright notice and this permission notice appear in all copies. -# -# THIS SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES -# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF -# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR -# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES -# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN -# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF -# OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. - -import ingen -import os.path -import re -import shlex -import sys -import time -try: - import readline -except: - pass - -# Python 2 compatibility -try: - input = raw_input -except NameError: - pass - -def print_usage(): - print('''Usage: ingenish [OPTION]... [COMMAND [ARGUMENT]...] - -A command line interface to an Ingen server. A command can be given directly -on the command line, or when run with no arguments an interactive shell is -launched. - -Options: - -s ADDRESS The address of the Ingen server. Default is the local server - at unix:///tmp/ingen.sock but remote servers can be used with - an address like tcp:///my-ingen-server-host:16180 - -Commands: - put SUBJECT BODY - set SUBJECT KEY VALUE - connect TAIL HEAD - disconnect TAIL HEAD - patch SUBJECT REMOVE ADD - delete SUBJECT - help - exit - -Subjects are specified by URI, relative to the engine. The top level audio -graph has the path /main, so for example, a port on a block might have the -(relative) URI /main/osc/freq. - -Bodies are specified in fragments of Turtle, just as written in Ingen graph files. - -Example: - put /main/left_in 'a lv2:InputPort ; a lv2:AudioPort' - put /main/left_out 'a lv2:OutputPort ; a lv2:AudioPort' - put /main/tone 'a ingen:Block ; lv2:prototype <http://drobilla.net/plugins/mda/Shepard>' - put /main/combo 'a ingen:Block ; lv2:prototype <http://drobilla.net/plugins/mda/Combo>' - connect /main/left_in /main/tone/left_in - connect /main/tone/left_out /main/combo/left_in - connect /main/combo/left_out /main/left_out - set /main/tone/output ingen:value 0.7 -''') - -def run(cmd): - if cmd[0] == 'help': - print_usage() - elif cmd[0] == 'exit': - sys.exit(0) - elif cmd[0] == 'get' and len(cmd) == 2: - print(ingen.get(cmd[1]).serialize(format='n3')) - elif cmd[0] == 'put' and len(cmd) == 3: - return ingen.put(cmd[1], cmd[2]) - elif cmd[0] == 'patch' and len(cmd) == 4: - return ingen.patch(cmd[1], cmd[2], cmd[3]) - elif cmd[0] == 'set' and len(cmd) == 4: - return ingen.set(cmd[1], cmd[2], cmd[3]) - elif cmd[0] == 'connect' and len(cmd) == 3: - return ingen.connect(cmd[1], cmd[2]) - elif cmd[0] == 'disconnect' and len(cmd) == 3: - return ingen.disconnect(cmd[1], cmd[2]) - elif cmd[0] == 'delete' and len(cmd) == 2: - return ingen.delete(cmd[1]) - return False - -a = 1 -server = 'unix:///tmp/ingen.sock' -if len(sys.argv) > 1: - if sys.argv[a] == '-s': - server = sys.argv[a + 1] - a = a + 2 - elif sys.argv[a][0] == '-': - print_usage() - sys.exit(1) - -ingen = ingen.Remote(server) - -if len(sys.argv) - a == 0: - print('Ingen server at %s' % server) - while True: - try: - run(shlex.split(input('> '))) - except (EOFError, KeyboardInterrupt, SystemExit): - break - except: - print('error: %s' % sys.exc_info()[1]) -else: - try: - update = run(sys.argv[a:]) - if update: - print(update.serialize(format='n3')) - except: - print('error: %s' % sys.exc_info()[1]) |