My own cython binding for lilv.

- replaces the ctypes based binding coming with lilv.
- easier to extend (because I understand cython better than ctypes).
- will need this anyway when converting the pipeline to nogil.
- all functions converted that are needed, but not much much.
- new:
  - implement map/unmap feature.
  - check for missing features and don't load plugins that we can't support.
looper
Ben Niemann 6 years ago
parent 87780526e9
commit 033dfb156d
  1. 2
      bin/setup_env.sh
  2. 17
      noisicaa/audioproc/nodes/lv2.py
  3. 49
      noisicaa/audioproc/nodes/lv2_test.py
  4. 1691
      noisicaa/bindings/lilv.pyx
  5. 14
      noisicaa/bindings/lilv.pyxbld
  6. 130
      noisicaa/bindings/lilv_test.py
  7. 16
      noisicaa/node_db/private/lv2_scanner.py

@ -8,7 +8,7 @@ fi
BASEDIR=$(readlink -f "$(dirname "$0")/..")
LIBSDIR="$BASEDIR/libs"
LILV_DEPS="libserd-dev libsord-dev libsratom-dev"
LILV_DEPS="libserd-dev libsord-dev libsratom-dev lv2-examples"
LADSPA_DEPS="ladspa-sdk"
CSOUND_DEPS="libsndfile1-dev libsamplerate0-dev libboost-dev flex bison cmake"

@ -3,9 +3,9 @@
import logging
import threading
import lilv
import numpy
from noisicaa.bindings import lilv
from noisicaa import node_db
from .. import ports
@ -42,10 +42,11 @@ class LV2(node.Node):
logger.info("Loading plugin...")
plugins = self.__world.get_all_plugins()
uri_node = self.__world.new_uri(uri)
self.__plugin = plugins[uri_node]
self.__plugin = plugins.get_by_uri(uri_node)
assert self.__plugin is not None
logger.info("Creating instance...")
self.__instance = lilv.Instance(self.__plugin, 44100)
self.__instance = self.__plugin.instantiate(44100)
self.__instance.activate()
self.__buffers = {}
@ -58,7 +59,7 @@ class LV2(node.Node):
buf = numpy.zeros(shape=(initial_length,), dtype=numpy.float32)
self.__buffers[port.name] = buf
lv2_port = self.__plugin.get_port_by_symbol(port.name)
lv2_port = self.__plugin.get_port_by_symbol(self.__world.new_string(port.name))
assert lv2_port is not None, port.name
self.__instance.connect_port(lv2_port.get_index(), buf)
@ -69,7 +70,7 @@ class LV2(node.Node):
buf = numpy.zeros(shape=(1,), dtype=numpy.float32)
self.__buffers[parameter.name] = buf
lv2_port = self.__plugin.get_port_by_symbol(parameter.name)
lv2_port = self.__plugin.get_port_by_symbol(self.__world.new_string(parameter.name))
assert lv2_port is not None, parameter.name
self.__instance.connect_port(lv2_port.get_index(), buf)
@ -97,14 +98,14 @@ class LV2(node.Node):
if len(buf) < required_length:
buf.resize(required_length, refcheck=False)
lv2_port = self.__plugin.get_port_by_symbol(port.name)
lv2_port = self.__plugin.get_port_by_symbol(self.__world.new_string(port.name))
assert lv2_port is not None, port.name
self.__instance.connect_port(lv2_port.get_index(), buf)
for port_name, port in self.inputs.items():
buf = self.__buffers[port_name]
if isinstance(port, ports.AudioInputPort):
numpy.copyto(buf, port.frame.samples[0])
buf[0:ctxt.duration] = port.frame.samples[0]
elif isinstance(port, ports.ControlInputPort):
buf[0] = port.frame[0]
else:
@ -120,7 +121,7 @@ class LV2(node.Node):
buf = self.__buffers[port_name]
if isinstance(port, ports.AudioOutputPort):
port.frame.resize(ctxt.duration)
numpy.copyto(port.frame.samples[0], buf)
port.frame.samples[0] = buf[0:ctxt.duration]
elif isinstance(port, ports.ControlOutputPort):
port.frame.fill(buf[0])
else:

@ -0,0 +1,49 @@
#!/usr/bin/python3
import asynctest
import os
import os.path
from noisicaa import constants
from noisicaa import node_db
from noisicaa.audioproc import data
from . import lv2
class LV2Test(asynctest.TestCase):
async def test_foo(self):
description = node_db.NodeDescription(
ports=[
node_db.AudioPortDescription(
name='in',
direction=node_db.PortDirection.Input,
channels='mono'),
node_db.AudioPortDescription(
name='out',
direction=node_db.PortDirection.Output,
channels='mono'),
],
parameters=[
node_db.InternalParameterDescription(
name='uri', value='http://lv2plug.in/plugins/eg-amp'),
node_db.FloatParameterDescription(
name='gain',
display_name='Gain',
default=0.0,
min=-90.0,
max=24.0),
])
ctxt = data.FrameContext()
ctxt.sample_pos = 0
ctxt.duration = 1024
node = lv2.LV2(self.loop, description)
await node.setup()
try:
node.collect_inputs(ctxt)
node.run(ctxt)
finally:
await node.cleanup()
if __name__ == '__main__':
unittest.main()

File diff suppressed because it is too large Load Diff

@ -0,0 +1,14 @@
# -*- mode: python -*-
import os
import os.path
def make_ext(modname, pyxfilename):
from distutils.extension import Extension
return Extension(
name = modname,
sources=[pyxfilename],
include_dirs=[os.path.join(os.getenv('VIRTUAL_ENV'), 'include', 'lilv-0')],
library_dirs=[os.path.join(os.getenv('VIRTUAL_ENV'), 'lib')],
libraries=['lilv-0'],
)

@ -0,0 +1,130 @@
#!/usr/bin/python3
import unittest
import numpy
from . import lilv
class NodeTest(unittest.TestCase):
def setUp(self):
self.world = lilv.World()
def test_string(self):
n = self.world.new_string('foo')
self.assertIsInstance(n, lilv.Node)
self.assertTrue(n.is_string)
self.assertEqual(n, 'foo')
self.assertEqual(str(n), 'foo')
def test_uri(self):
n = self.world.new_uri('http://foo/bar')
self.assertIsInstance(n, lilv.Node)
self.assertTrue(n.is_uri)
self.assertEqual(n, 'http://foo/bar')
self.assertEqual(str(n), 'http://foo/bar')
def test_float(self):
n = self.world.new_float(2.0)
self.assertIsInstance(n, lilv.Node)
self.assertTrue(n.is_float)
self.assertEqual(n, 2.0)
self.assertEqual(float(n), 2.0)
def test_int(self):
n = self.world.new_float(2)
self.assertIsInstance(n, lilv.Node)
self.assertTrue(n.is_int)
self.assertEqual(n, 2)
self.assertEqual(int(n), 2)
def test_int(self):
n = self.world.new_bool(True)
self.assertIsInstance(n, lilv.Node)
self.assertTrue(n.is_bool)
self.assertEqual(n, True)
self.assertEqual(bool(n), True)
def test_compare(self):
n1 = self.world.new_string('foo')
n2 = self.world.new_string('foo')
n3 = self.world.new_string('bar')
self.assertTrue(n1 == n2)
self.assertFalse(n1 != n2)
self.assertFalse(n1 == n3)
self.assertTrue(n1 != n3)
class WorldTest(unittest.TestCase):
def test_ns(self):
world = lilv.World()
self.assertIsInstance(world.ns.lilv.map, lilv.BaseNode)
self.assertEqual(str(world.ns.lilv.map), 'http://drobilla.net/ns/lilv#map')
def test_get_all_plugins(self):
world = lilv.World()
world.load_all()
for plugin in world.get_all_plugins():
self.assertIsInstance(plugin.get_uri(), lilv.BaseNode)
class PluginTest(unittest.TestCase):
def setUp(self):
self.world = lilv.World()
self.world.load_all()
self.plugins = self.world.get_all_plugins()
def test_features(self):
uri_node = self.world.new_uri('http://lv2plug.in/plugins/eg-fifths')
plugin = self.plugins.get_by_uri(uri_node)
self.assertIsNot(plugin, None)
self.assertEqual(
list(plugin.get_required_features()),
['http://lv2plug.in/ns/ext/urid#map'])
self.assertEqual(
list(plugin.get_optional_features()),
['http://lv2plug.in/ns/lv2core#hardRTCapable'])
self.assertEqual(plugin.get_missing_features(), [])
instance = plugin.instantiate(44100)
self.assertIsNot(instance, None)
def test_not_supported(self):
uri_node = self.world.new_uri('http://guitarix.sourceforge.net/plugins/gx_cabinet#CABINET')
plugin = self.plugins.get_by_uri(uri_node)
self.assertIsNot(plugin, None)
self.assertEqual(
plugin.get_missing_features(),
['http://lv2plug.in/ns/ext/buf-size#boundedBlockLength',
'http://lv2plug.in/ns/ext/worker#schedule'])
def test_instantiate(self):
uri_node = self.world.new_uri('http://lv2plug.in/plugins/eg-amp')
plugin = self.plugins.get_by_uri(uri_node)
self.assertIsNot(plugin, None)
self.assertEqual(plugin.get_uri(), uri_node)
in_port = plugin.get_port_by_symbol(self.world.new_string('in'))
out_port = plugin.get_port_by_symbol(self.world.new_string('out'))
gain_port = plugin.get_port_by_symbol(self.world.new_string('gain'))
instance = plugin.instantiate(44100)
self.assertIsNot(instance, None)
in_buf = numpy.zeros(shape=(1024,), dtype=numpy.float32)
instance.connect_port(in_port.get_index(), in_buf)
out_buf = numpy.zeros(shape=(1024,), dtype=numpy.float32)
instance.connect_port(out_port.get_index(), out_buf)
gain_buf = numpy.zeros(shape=(1,), dtype=numpy.float32)
instance.connect_port(gain_port.get_index(), gain_buf)
instance.activate()
instance.run(1024)
instance.deactivate()

@ -4,8 +4,7 @@ import logging
import os
import os.path
import lilv
from noisicaa.bindings import lilv
from noisicaa import constants
from noisicaa import node_db
@ -26,13 +25,14 @@ class LV2Scanner(scanner.Scanner):
plugins = world.get_all_plugins()
for plugin in plugins:
logger.info("Adding LV2 plugin %s", plugin.get_uri())
missing_features = plugin.get_missing_features()
if missing_features:
logger.warning(
"Not adding LV2 plugin %s, because it requires unsupported features %s",
plugin.get_uri(), ", ".join(missing_features))
continue
# TODO: Check required features
for feature in plugin.get_required_features():
logger.info('REQUIRED FEATURE: %s', feature)
for feature in plugin.get_optional_features():
logger.info('OPTIONAL FEATURE: %s', feature)
logger.info("Adding LV2 plugin %s", plugin.get_uri())
ports = []
parameters = []

Loading…
Cancel
Save