bigbull/lv2.py

   1 import re
   2 import os
   3 import sys
   4 import glob
   5 import calfpytools
   6
   7 lv2 = "http://lv2plug.in/ns/lv2core#"
   8 lv2evt = "http://lv2plug.in/ns/ext/event#"
   9 lv2str = "http://lv2plug.in/ns/dev/string-port#"
  10 rdf = "http://www.w3.org/1999/02/22-rdf-syntax-ns#"
  11 rdfs = "http://www.w3.org/2000/01/rdf-schema#"
  12 epi = "http://lv2plug.in/ns/dev/extportinfo#"
  13 rdf_type = rdf + "type"
  14 tinyname_uri = "http://lv2plug.in/ns/dev/tiny-name"
  15 foaf = "http://xmlns.com/foaf/0.1/"
  16 doap = "http://usefulinc.com/ns/doap#"
  17
  18 event_type_names = {
  19     "http://lv2plug.in/ns/ext/midi#MidiEvent" : "MIDI"
  20 }
  21
  22 class DumpRDFModel:
  23     def addTriple(self, s, p, o):
  24         print "%s [%s] %s" % (s, p, repr(o))
  25
  26 class SimpleRDFModel:
  27     def __init__(self):
  28         self.bySubject = {}
  29         self.byPredicate = {}
  30     def getByType(self, classname):
  31         classes = self.bySubject["$classes"]
  32         if classname in classes:
  33             return classes[classname]
  34         return []
  35     def getByPropType(self, propname):
  36         if propname in self.byPredicate:
  37             return self.byPredicate[propname]
  38         return []
  39     def getProperty(self, subject, props, optional = False, single = False):
  40         if type(props) is list:
  41             prop = props[0]
  42         else:
  43             prop = props
  44         if type(subject) is str:
  45             subject = self.bySubject[subject]
  46         elif type(subject) is dict:
  47             pass
  48         else:
  49             if single:
  50                 return None
  51             else:
  52                 return []
  53         anyprops = set()
  54         if prop in subject:
  55             for o in subject[prop]:
  56                 anyprops.add(o)
  57         if type(props) is list:
  58             if len(props) > 1:
  59                 result = set()
  60                 for v in anyprops:
  61                     if single:
  62                         value = self.getProperty(v, props[1:], optional = optional, single = True)
  63                         if value != None:
  64                             return value
  65                     else:
  66                         result |= set(self.getProperty(v, props[1:], optional = optional, single = False))
  67                 if single:
  68                     return None
  69                 else:
  70                     return list(result)
  71         if single:
  72             if len(anyprops) > 0:
  73                 if len(anyprops) > 1:
  74                     raise Exception, "More than one value of " + prop
  75                 return list(anyprops)[0]
  76             else:
  77                 return None
  78         return list(anyprops)
  79
  80
  81     def addTriple(self, s, p, o):
  82         if p == rdf_type:
  83             p = "a"
  84         if s not in self.bySubject:
  85             self.bySubject[s] = {}
  86         if p not in self.bySubject[s]:
  87             self.bySubject[s][p] = []
  88         self.bySubject[s][p].append(o)
  89         if p not in self.byPredicate:
  90             self.byPredicate[p] = {}
  91         if s not in self.byPredicate[p]:
  92             self.byPredicate[p][s] = []
  93         self.byPredicate[p][s].append(o)
  94         if p == "a":
  95             self.addTriple("$classes", o, s)
  96     def copyFrom(self, src):
  97         for s in src.bySubject:
  98             po = src.bySubject[s]
  99             for p in po:
 100                 for o in po[p]:
 101                     self.addTriple(s, p, o)
 102     def dump(self):
 103         for s in self.bySubject.keys():
 104             for p in self.bySubject[s].keys():
 105                 print "%s %s %s" % (s, p, self.bySubject[s][p])
 106
 107 def parseTTL(uri, content, model, debug):
 108     # Missing stuff: translated literals, blank nodes
 109     if debug:
 110         print "Parsing: %s" % uri
 111     prefixes = {}
 112     spo_stack = []
 113     spo = ["", "", ""]
 114     item = 0
 115     anoncnt = 1
 116     for x in calfpytools.scan_ttl_string(content):
 117         if x[0] == '':
 118             continue
 119         if x[0] == 'prefix':
 120             spo[0] = "@prefix"
 121             item = 1
 122             continue
 123         elif (x[0] == '.' and spo_stack == []) or x[0] == ';' or x[0] == ',':
 124             if item == 3:
 125                 if spo[0] == "@prefix":
 126                     prefixes[spo[1][:-1]] = spo[2]
 127                 else:
 128                     model.addTriple(spo[0], spo[1], spo[2])
 129                 if x[0] == '.': item = 0
 130                 elif x[0] == ';': item = 1
 131                 elif x[0] == ',': item = 2
 132             else:
 133                 if x[0] == '.':
 134                     item = 0
 135                 elif item != 0:
 136                     raise Exception, uri+": Unexpected " + x[0]
 137         elif x[0] == "prnot" and item < 3:
 138             prnot = x[1].split(":")
 139             if item != 0 and spo[0] == "@prefix":
 140                 spo[item] = x[1]
 141             elif prnot[0] == "_":
 142                 spo[item] = uri + "#" + prnot[1]
 143             else:
 144                 if prnot[0] not in prefixes:
 145                     raise Exception, "Prefix %s not defined" % prnot[0]
 146                 else:
 147                     spo[item] = prefixes[prnot[0]] + prnot[1]
 148             item += 1
 149         elif (x[0] == 'URI' or x[0] == "string" or x[0] == "number" or (x[0] == "symbol" and x[1] == "a" and item == 1)) and (item < 3):
 150             if x[0] == "URI" and x[1] == "":
 151                 x = ("URI", uri)
 152             elif x[0] == "URI" and x[1].find(":") == -1 and x[1] != "" and x[1][0] != "/":
 153                 # This is quite silly
 154                 x = ("URI", os.path.dirname(uri) + "/" + x[1])
 155             spo[item] = x[1]
 156             item += 1
 157         elif x[0] == '[':
 158             if item != 2:
 159                 raise Exception, "Incorrect use of ["
 160             uri2 = uri + "$anon$" + str(anoncnt)
 161             spo[2] = uri2
 162             spo_stack.append(spo)
 163             spo = [uri2, "", ""]
 164             item = 1
 165             anoncnt += 1
 166         elif x[0] == ']' or x[0] == ')':
 167             if item == 3:
 168                 model.addTriple(spo[0], spo[1], spo[2])
 169                 item = 0
 170             spo = spo_stack[-1]
 171             spo_stack = spo_stack[:-1]
 172             item = 3
 173         elif x[0] == '(':
 174             if item != 2:
 175                 raise Exception, "Incorrect use of ("
 176             uri2 = uri + "$anon$" + str(anoncnt)
 177             spo[2] = uri2
 178             spo_stack.append(spo)
 179             spo = [uri2, "", ""]
 180             item = 2
 181             anoncnt += 1
 182         else:
 183             print uri + ": Unexpected: " + repr(x)
 184
 185 class LV2Port(object):
 186     def __init__(self):
 187         pass
 188     def connectableTo(self, port):
 189         if not ((self.isInput and port.isOutput) or (self.isOutput and port.isInput)):
 190             return False
 191         if self.isAudio != port.isAudio or self.isControl != port.isControl or self.isEvent != port.isEvent:
 192             return False
 193         if not self.isAudio and not self.isControl and not self.isEvent:
 194             return False
 195         return True
 196
 197 class LV2Plugin(object):
 198     def __init__(self):
 199         pass
 200
 201 class LV2DB:
 202     def __init__(self, debug = False):
 203         self.debug = debug
 204         self.initManifests()
 205
 206     def initManifests(self):
 207         lv2path = ["/usr/lib/lv2", "/usr/local/lib/lv2"]
 208         self.manifests = SimpleRDFModel()
 209         self.paths = {}
 210         self.plugin_info = dict()
 211         # Scan manifests
 212         for dir in lv2path:
 213             for bundle in glob.iglob(dir + "/*.lv2"):
 214                 fn = bundle+"/manifest.ttl"
 215                 if os.path.exists(fn):
 216                     parseTTL(fn, file(fn).read(), self.manifests, self.debug)
 217         # Read all specifications from all manifests
 218         if (lv2 + "Specification" in self.manifests.bySubject["$classes"]):
 219             specs = self.manifests.getByType(lv2 + "Specification")
 220             filenames = set()
 221             for spec in specs:
 222                 subj = self.manifests.bySubject[spec]
 223                 if rdfs+"seeAlso" in subj:
 224                     for fn in subj[rdfs+"seeAlso"]:
 225                         filenames.add(fn)
 226             for fn in filenames:
 227                 parseTTL(fn, file(fn).read(), self.manifests, self.debug)
 228         #fn = "/usr/lib/lv2/lv2core.lv2/lv2.ttl"
 229         #parseTTL(fn, file(fn).read(), self.manifests)
 230         self.plugins = self.manifests.getByType(lv2 + "Plugin")
 231         self.categories = set()
 232         self.category_paths = []
 233         self.add_category_recursive([], lv2 + "Plugin")
 234
 235     def add_category_recursive(self, tree_pos, category):
 236         cat_name = self.manifests.getProperty(category, rdfs + "label", single = True, optional = True)
 237         self.category_paths.append(((tree_pos + [cat_name])[1:], category))
 238         self.categories.add(category)
 239         items = self.manifests.byPredicate[rdfs + "subClassOf"]
 240         for subj in items:
 241             if subj in self.categories:
 242                 continue
 243             for o in items[subj]:
 244                 if o == category and subj not in self.categories:
 245                     self.add_category_recursive(list(tree_pos) + [cat_name], subj)
 246
 247     def get_categories(self):
 248         return self.category_paths
 249
 250     def getPluginList(self):
 251         return self.plugins
 252
 253     def getPluginInfo(self, uri):
 254         if uri not in self.plugin_info:
 255             world = SimpleRDFModel()
 256             world.copyFrom(self.manifests)
 257             seeAlso = self.manifests.bySubject[uri]["http://www.w3.org/2000/01/rdf-schema#seeAlso"]
 258             try:
 259                 for doc in seeAlso:
 260                     # print "Loading " + doc + " for plugin " + uri
 261                     parseTTL(doc, file(doc).read(), world, self.debug)
 262                 self.plugin_info[uri] = world
 263             except Exception, e:
 264                 print "ERROR %s: %s" % (uri, str(e))
 265                 return None
 266         info = self.plugin_info[uri]
 267         dest = LV2Plugin()
 268         dest.uri = uri
 269         dest.name = info.bySubject[uri][doap + 'name'][0]
 270         dest.license = info.bySubject[uri][doap + 'license'][0]
 271         dest.classes = info.bySubject[uri]["a"]
 272         dest.requiredFeatures = info.getProperty(uri, lv2 + "requiredFeature", optional = True)
 273         dest.optionalFeatures = info.getProperty(uri, lv2 + "optionalFeature", optional = True)
 274         dest.microname = info.getProperty(uri, tinyname_uri, optional = True)
 275         if len(dest.microname):
 276             dest.microname = dest.microname[0]
 277         else:
 278             dest.microname = None
 279         dest.maintainers = []
 280         if info.bySubject[uri].has_key(doap + "maintainer"):
 281             for maintainer in info.bySubject[uri][doap + "maintainer"]:
 282                 maintainersubj = info.bySubject[maintainer]
 283                 maintainerdict = {}
 284                 maintainerdict['name'] = info.getProperty(maintainersubj, foaf + "name")[0]
 285                 homepages = info.getProperty(maintainersubj, foaf + "homepage")
 286                 if homepages:
 287                     maintainerdict['homepage'] = homepages[0]
 288                 mboxes = info.getProperty(maintainersubj, foaf + "mbox")
 289                 if mboxes:
 290                     maintainerdict['mbox'] = mboxes[0]
 291                 dest.maintainers.append(maintainerdict)
 292         ports = []
 293         portDict = {}
 294         porttypes = {
 295             "isAudio" : lv2 + "AudioPort",
 296             "isControl" : lv2 + "ControlPort",
 297             "isEvent" : lv2evt + "EventPort",
 298             "isString" : lv2str + "StringPort",
 299             "isInput" : lv2 + "InputPort",
 300             "isOutput" : lv2 + "OutputPort",
 301             "isLarslMidi" : "http://ll-plugins.nongnu.org/lv2/ext/MidiPort",
 302         }
 303
 304         for port in info.bySubject[uri][lv2 + "port"]:
 305             psubj = info.bySubject[port]
 306             pdata = LV2Port()
 307             pdata.uri = port
 308             pdata.index = int(info.getProperty(psubj, lv2 + "index")[0])
 309             pdata.symbol = info.getProperty(psubj, lv2 + "symbol")[0]
 310             pdata.name = info.getProperty(psubj, lv2 + "name")[0]
 311             classes = set(info.getProperty(psubj, "a"))
 312             pdata.classes = classes
 313             for pt in porttypes.keys():
 314                 pdata.__dict__[pt] = porttypes[pt] in classes
 315             sp = info.getProperty(psubj, lv2 + "scalePoint")
 316             if sp and len(sp):
 317                 splist = []
 318                 for pt in sp:
 319                     name = info.getProperty(pt, rdfs + "label", optional = True, single = True)
 320                     if name != None:
 321                         value = info.getProperty(pt, rdf + "value", optional = True, single = True)
 322                         if value != None:
 323                             splist.append((name, value))
 324                 pdata.scalePoints = splist
 325             else:
 326                 pdata.scalePoints = []
 327             if pdata.isControl:
 328                 pdata.defaultValue = info.getProperty(psubj, [lv2 + "default"], optional = True, single = True)
 329             elif pdata.isString:
 330                 pdata.defaultValue = info.getProperty(psubj, [lv2str + "default"], optional = True, single = True)
 331             else:
 332                 pdata.defaultValue = None
 333             pdata.minimum = info.getProperty(psubj, [lv2 + "minimum"], optional = True, single = True)
 334             pdata.maximum = info.getProperty(psubj, [lv2 + "maximum"], optional = True, single = True)
 335             pdata.microname = info.getProperty(psubj, [tinyname_uri], optional = True, single = True)
 336             pdata.properties = set(info.getProperty(psubj, [lv2 + "portProperty"], optional = True))
 337             pdata.events = set(info.getProperty(psubj, [lv2evt + "supportsEvent"], optional = True))
 338             ports.append(pdata)
 339             portDict[pdata.uri] = pdata
 340         ports.sort(lambda x, y: cmp(x.index, y.index))
 341         dest.ports = ports
 342         dest.portDict = portDict
 343         return dest
 344