light9 Changeset - a631e075a5bf

Changeset - a631e075a5bf

Parent rev.

Child rev.

[Not reviewed]

default

0 6 0

drewp@bigasterisk.com - 12 years ago 2012-07-19 04:23:06
drewp@bigasterisk.com

KC big rewrites, now multiple KC instances can sync with rdfdb
Ignore-this: 8c75ec3e2bd360c6eb87f7f4d4b3dcc4

6 files changed with 321 insertions and 120 deletions:

bin/keyboardcomposer

116

bin/rdfdb

light9/Submaster.py

light9/rdfdb/patch.py

light9/rdfdb/rdflibpatch.py

light9/rdfdb/syncedgraph.py

0 comments (0 inline, 0 general)

bin/keyboardcomposer

➞

Show inline comments

 #!bin/python
 from __future__ import division, nested_scopes
 import cgi, os, sys, time, subprocess, logging
+import cgi, os, sys, time, subprocess, logging, random
 from optparse import OptionParser
 import webcolors, colorsys
 from louie import dispatcher
 from twisted.internet import reactor, tksupport
 from twisted.web import xmlrpc, server, resource
 from rdflib import URIRef, Literal, RDF
 from Tix import *
 import Tix as tk
 import pickle
 import run_local
 from light9.Fadable import Fadable
 from light9.Submaster import Submasters, sub_maxes
 from light9.subclient import SubClient
 from light9 import dmxclient, showconfig, networking, prof
 from light9.uihelpers import toplevelat, bindkeys
 from light9.namespaces import L9
 from light9.tkdnd import initTkdnd, dragSourceRegister
 from light9.rdfdb.syncedgraph import SyncedGraph
 from light9.rdfdb.patch import Patch
 from bcf2000 import BCF2000
 nudge_keys = {
     'up'   : list('qwertyui'),
     'down' : list('asdfghjk')
+}
 class DummySliders:
     def valueOut(self, name, value):
         pass
     def close(self):
@@ @@ -46,76 +48,134 @@ class SubScale(Scale, Fadable): @@
                    'highlightthickness' : 1, 'bd' : 1,
                    'highlightcolor' : 'red', 'highlightbackground' : 'black',
                    'activebackground' : 'red'})
         Scale.__init__(self, master, *args, **kw)
         Fadable.__init__(self, var=self.scale_var, wheel_step=0.05)
         self.draw_indicator_colors()
     def draw_indicator_colors(self):
         if self.scale_var.get() == 0:
             self['troughcolor'] = 'black'
         else:
             self['troughcolor'] = 'blue'
 class SubmasterTk(Frame):
     def __init__(self, master, sub, current_level):
 class SubmasterBox(Frame):
     """
     this object owns the level of the submaster (the rdf graph is the
     real authority)
     """
     def __init__(self, master, sub):
         self.sub = sub
         bg = sub.graph.value(sub.uri, L9.color, default='#000000')
         rgb = webcolors.hex_to_rgb(bg)
         hsv = colorsys.rgb_to_hsv(*[x/255 for x in rgb])
         darkBg = webcolors.rgb_to_hex(tuple([x * 255 for x in colorsys.hsv_to_rgb(
             hsv[0], hsv[1], .3)]))
         Frame.__init__(self, master, bd=1, relief='raised', bg=bg)
         self.name = sub.name
         self.slider_var = DoubleVar()
-        self.slider_var.set(current_level)
+        self.pauseTrace = False
         self.scale = SubScale(self, variable=self.slider_var, width=20)
         self.namelabel = Label(self, font="Arial 7", bg=darkBg,
             fg='white', pady=0)
         self.sub.graph.addHandler(self.updateName)
         self.namelabel.pack(side=TOP)
         levellabel = Label(self, textvariable=self.slider_var, font="Arial 7",
             bg='black', fg='white', pady=0)
         levellabel.pack(side=TOP)
         self.scale.pack(side=BOTTOM, expand=1, fill=BOTH)
         bindkeys(self, "<Control-Key-l>", self.launch_subcomposer)
         for w in [self, self.namelabel, levellabel]:
             dragSourceRegister(w, 'copy', 'text/uri-list', sub.uri)
         self._slider_var_trace = self.slider_var.trace('w', self.slider_changed)
         sub.graph.addHandler(self.updateLevelFromGraph)
         # initial position
 #        self.send_to_hw(sub.name, col + 1) # needs fix
     def cleanup(self):
         self.slider_var.trace_vdelete('w', self._slider_var_trace)
     def slider_changed(self, *args):
         self.scale.draw_indicator_colors()
         if self.pauseTrace:
             return
         self.updateGraphWithLevel(self.sub.uri, self.slider_var.get())
         # dispatcher.send("level changed") # in progress
         ###self.send_levels() # use dispatcher?
         # needs fixing: plan is to use dispatcher or a method call to tell a hardware-mapping object who changed, and then it can make io if that's a current hw slider
         #if rowcount == self.current_row:
         #    self.send_to_hw(sub.name, col + 1)
     def updateGraphWithLevel(self, uri, level):
         """in our per-session graph, we maintain SubSetting objects like this:
            [a :SubSetting; :sub ?s; :level ?l]
         """
         ctx = URIRef("http://example.com/kc/session1")
         with self.sub.graph.currentState(context=ctx) as graph:
             adds = set([])
             for setting in graph.subjects(RDF.type, L9['SubSetting']):
                 if graph.value(setting, L9['sub']) == uri:
                     break
             else:
                 setting = URIRef("http://example.com/subsetting/%s" %
                                  random.randrange(999999))
                 adds.update([
                     (setting, RDF.type, L9['SubSetting'], ctx),
                     (setting, L9['sub'], uri, ctx),
                     ])
             dels = set([])
             for prev in graph.objects(setting, L9['level']):
                 dels.add((setting, L9['level'], prev, ctx))
             adds.add((setting, L9['level'], Literal(level), ctx))
             if adds != dels:
                 self.sub.graph.patch(Patch(delQuads=dels, addQuads=adds))
     def updateLevelFromGraph(self):
         """read rdf level, write it to subbox.slider_var"""
         graph = self.sub.graph
         for setting in graph.subjects(RDF.type, L9['SubSetting']):
             if graph.value(setting, L9['sub']) == self.sub.uri:
                 self.pauseTrace = True # don't bounce this update back to server
                 try:
                     self.slider_var.set(graph.value(setting, L9['level']))
                 finally:
                     self.pauseTrace = False
     def updateName(self):
         self.namelabel.config(text=self.sub.graph.label(self.sub.uri))
     def launch_subcomposer(self, *args):
         subprocess.Popen(["bin/subcomposer", "--no-geometry", self.name])
 class KeyboardComposer(Frame, SubClient):
-    def __init__(self, root, graph, current_sub_levels=None,
     def __init__(self, root, graph,
                  hw_sliders=True):
         Frame.__init__(self, root, bg='black')
         SubClient.__init__(self)
         self.graph = graph
         self.submasters = Submasters(graph)
         self.name_to_subtk = {}
         self.current_sub_levels = {}
         self.current_row = 0
         if current_sub_levels is not None:
             self.current_sub_levels = current_sub_levels
         else:
             try:
                 self.current_sub_levels, self.current_row = \
                     pickle.load(file('.keyboardcomposer.savedlevels'))
             except IOError:
                 pass
         self.subbox = {} # sub uri : SubmasterBox
         self.slider_table = {} # coords : SubmasterBox
         self.rows = [] # this holds Tk Frames for each row
         self.current_row = 0 # should come from session graph
         self.use_hw_sliders = hw_sliders
         self.connect_to_hw(hw_sliders)
         self.make_key_hints()
         self.make_buttons()
         self.graph.addHandler(self.redraw_sliders)
         self.send_levels_loop()
     def make_buttons(self):
         self.buttonframe = Frame(self, bg='black')
@@ @@ -132,104 +192,82 @@ class KeyboardComposer(Frame, SubClient) @@
         self.alltozerobutton = Button(self.buttonframe, text="All to Zero",
             command=self.alltozero, bg='black', fg='white')
         self.alltozerobutton.pack(side='left')
         self.save_stage_button = Button(self.buttonframe, text="Save",
             command=lambda: self.save_current_stage(self.sub_name.get()),
             bg='black', fg='white')
         self.save_stage_button.pack(side=LEFT)
         self.sub_name = Entry(self.buttonframe, bg='black', fg='white')
         self.sub_name.pack(side=LEFT)
     def redraw_sliders(self):
         self.slider_vars = {} # this holds suburi : sub Tk vars
         self.slider_table = {} # this holds coords:sub Tk vars
         self.name_to_subtk.clear() # subname : SubmasterTk instance
         self.graph.addHandler(self.draw_sliders)
         if len(self.rows):
             self.change_row(self.current_row)
             self.rows[self.current_row].focus()
         self.stop_frequent_update_time = 0
     def onNewSub(self, sub):
         log.info("new %s", sub)
         self.graph.addHandler(self.draw_sliders)
     def onLostSub(self, subUri):
         log.info("lost %s", subUri)
         self.graph.addHandler(self.draw_sliders)
     def draw_sliders(self):
         if hasattr(self, 'rows'):
             for r in self.rows:
                 r.destroy()
         self.rows = [] # this holds Tk Frames for each row
         for r in self.rows:
             r.destroy()
         self.rows = []
         for b in self.subbox.values():
             b.cleanup()
         self.subbox.clear()
         self.slider_table.clear()
         self.tk_focusFollowsMouse()
         rowcount = -1
         col = 0
         last_group = None
         # there are unlikely to be any subs at startup because we
         # probably haven't been called back with the graph data yet
         #read get_all_subs then watch 'new submaster' 'lost submaster' signals
         withgroups = sorted((self.graph.value(sub.uri, L9['group']),
                              self.graph.value(sub.uri, L9['order']),
                              sub)
             for sub in self.submasters.get_all_subs())
         dispatcher.connect(self.onNewSub, "new submaster")
         dispatcher.connect(self.onLostSub, "lost submaster")
         log.info("withgroups %s", withgroups)
         for group, order, sub in withgroups:
             group = self.graph.value(sub.uri, L9['group'])
-            if col == 0 or group != last_group: # make new row
             if col == 0 or group != last_group:
                 row = self.make_row()
                 rowcount += 1
                 col = 0
             current_level = self.current_sub_levels.get(sub.name, 0)
             subtk = self.draw_sub_slider(row, col, sub, current_level)
             self.slider_table[(rowcount, col)] = subtk
             self.name_to_subtk[sub.name] = subtk
             def slider_changed(x, y, z, subtk=subtk,
                                col=col, sub=sub, rowcount=rowcount):
                 subtk.scale.draw_indicator_colors()
                 self.send_levels()
                 if rowcount == self.current_row:
                     self.send_to_hw(sub.name, col + 1)
             subbox = SubmasterBox(row, sub)
             subbox.place(relx=col / 8, rely=0, relwidth=1 / 8, relheight=1)
             self.subbox[sub.uri] = self.slider_table[(rowcount, col)] = subbox
-            subtk.slider_var.trace('w', slider_changed)
+            self.setup_key_nudgers(subbox.scale)
             # initial position
             self.send_to_hw(sub.name, col + 1)
             col = (col + 1) % 8
             last_group = group
     def draw_sub_slider(self, row, col, sub, current_level):
         subtk = SubmasterTk(row, sub, current_level)
         subtk.place(relx=col / 8, rely=0, relwidth=1 / 8, relheight=1)
         self.setup_key_nudgers(subtk.scale)
         self.slider_vars[sub.uri] = subtk.slider_var
         return subtk
     def toggle_slider_connectedness(self):
         self.use_hw_sliders = not self.use_hw_sliders
         if self.use_hw_sliders:
             self.sliders.reopen()
         else:
             self.sliders.close()
         self.change_row(self.current_row)
         self.rows[self.current_row].focus()
     def connect_to_hw(self, hw_sliders):
         if hw_sliders:
             try:
@@ @@ -292,58 +330,58 @@ class KeyboardComposer(Frame, SubClient) @@
     def change_row(self, row):
         old_row = self.current_row
         self.current_row = row
         self.current_row = max(0, self.current_row)
         self.current_row = min(len(self.rows) - 1, self.current_row)
         self.unhighlight_row(old_row)
         self.highlight_row(self.current_row)
         row = self.rows[self.current_row]
         self.keyhints.pack_configure(before=row)
         for col in range(1, 9):
             try:
-                subtk = self.slider_table[(self.current_row, col - 1)]
+                subbox = self.slider_table[(self.current_row, col - 1)]
                 self.sliders.valueOut("button-upper%d" % col, True)
             except KeyError:
                 # unfilled bottom row has holes (plus rows with incomplete
                 # groups
                 self.sliders.valueOut("button-upper%d" % col, False)
                 self.sliders.valueOut("slider%d" % col, 0)
                 continue
-            self.send_to_hw(subtk.name, col)
+            self.send_to_hw(subbox.name, col)
     def got_nudger(self, number, direction, full=0):
         try:
-            subtk = self.slider_table[(self.current_row, number)]
+            subbox = self.slider_table[(self.current_row, number)]
         except KeyError:
             return
         if direction == 'up':
             if full:
-                subtk.scale.fade(1)
+                subbox.scale.fade(1)
             else:
-                subtk.scale.increase()
+                subbox.scale.increase()
         else:
             if full:
-                subtk.scale.fade(0)
+                subbox.scale.fade(0)
             else:
-                subtk.scale.decrease()
+                subbox.scale.decrease()
     def hw_slider_moved(self, col, value):
         value = int(value * 100) / 100
         try:
-            subtk = self.slider_table[(self.current_row, col)]
+            subbox = self.slider_table[(self.current_row, col)]
         except KeyError:
             return # no slider assigned at that column
-        subtk.scale.set(value)
+        subbox.scale.set(value)
     def send_to_hw(self, subUri, hwNum):
         if isinstance(self.sliders, DummySliders):
             return
         v = round(127 * self.slider_vars[subUri].get())
         chan = "slider%s" % hwNum
         # workaround for some rounding issue, where we receive one
         # value and then decide to send back a value that's one step
         # lower.  -5 is a fallback for having no last value.  hopefully
         # we won't really see it
@@ @@ -358,82 +396,83 @@ class KeyboardComposer(Frame, SubClient) @@
         self.rows.append(row)
         return row
     def highlight_row(self, row):
         row = self.rows[row]
         row['bg'] = 'red'
     def unhighlight_row(self, row):
         row = self.rows[row]
         row['bg'] = 'black'
     def get_levels(self):
         return dict([(uri, slidervar.get())
             for uri, slidervar in self.slider_vars.items()])
         return dict([(uri, box.slider_var.get())
             for uri, box in self.subbox.items()])
     def get_levels_as_sub(self):
         scaledsubs = [self.submasters.get_sub_by_uri(sub) * level
             for sub, level in self.get_levels().items() if level > 0.0]
         maxes = sub_maxes(*scaledsubs)
         return maxes
     def save_current_stage(self, subname):
         log.info("saving current levels as %s", subname)
         sub = self.get_levels_as_sub()
         sub.name = subname
         sub.temporary = 0
         sub.save()
     def save(self):
         pickle.dump((self.get_levels(), self.current_row),
                     file('.keyboardcomposer.savedlevels', 'w'))
     def send_frequent_updates(self):
         """called when we get a fade -- send events as quickly as possible"""
         if time.time() <= self.stop_frequent_update_time:
             self.send_levels()
             self.after(10, self.send_frequent_updates)
     def alltozero(self):
         for name, subtk in self.name_to_subtk.items():
             if subtk.scale.scale_var.get() != 0:
                 subtk.scale.fade(value=0.0, length=0)
         for uri, subbox in self.subbox.items():
             if subbox.scale.scale_var.get() != 0:
                 subbox.scale.fade(value=0.0, length=0)
 # move to web lib
 def postArgGetter(request):
     """return a function that takes arg names and returns string
     values. Supports args encoded in the url or in postdata. No
     support for repeated args."""
     # this is something nevow normally does for me
     request.content.seek(0)
     fields = cgi.FieldStorage(request.content, request.received_headers,
                               environ={'REQUEST_METHOD': 'POST'})
     def getArg(n):
         try:
             return request.args[n][0]
         except KeyError:
             return fields[n].value
     return getArg
 class LevelServerHttp(resource.Resource):
     isLeaf = True
     def __init__(self,name_to_subtk):
         self.name_to_subtk = name_to_subtk
     def __init__(self,name_to_subbox):
         self.name_to_subbox = name_to_subbox
     def render_POST(self, request):
         arg = postArgGetter(request)
         if request.path == '/fadesub':
             # fadesub?subname=scoop&level=0&secs=.2
-            self.name_to_subtk[arg('subname')].scale.fade(
+            self.name_to_subbox[arg('subname')].scale.fade(
                 float(arg('level')),
                 float(arg('secs')))
             return "set %s to %s" % (arg('subname'), arg('level'))
         else:
             raise NotImplementedError(repr(request))
 class Sliders(BCF2000):
     def __init__(self, kc):
         devices = ['/dev/snd/midiC1D0', '/dev/snd/midiC2D0', '/dev/snd/midiC3D0']
         for dev in devices:
             try:
                 BCF2000.__init__(self, dev=dev)
@@ @@ -468,58 +507,52 @@ class Sliders(BCF2000): @@
             if button_num == 1:
                 diff = -1
             elif button_num == 3:
                 diff = 1
             else:
                 return
             kc.change_row(kc.current_row + diff)
             self.valueOut(name, 0)
 if __name__ == "__main__":
     parser = OptionParser()
     parser.add_option('--nonpersistent', action="store_true",
                       help="don't load or save levels")
     parser.add_option('--no-sliders', action='store_true',
                       help="don't attach to hardware sliders")
     parser.add_option('-v', action='store_true', help="log info level")
     opts, args = parser.parse_args()
     logging.basicConfig(level=logging.INFO if opts.v else logging.WARN)
     log = logging.getLogger('keyboardcomposer')
     graph = SyncedGraph("keyboardcomposer")
     root = Tk()
     initTkdnd(root.tk, 'tkdnd/trunk/')
     tl = toplevelat("Keyboard Composer", existingtoplevel=root)
     startLevels = None
     if opts.nonpersistent:
         startLevels = {}
     kc = KeyboardComposer(tl, graph, startLevels,
     kc = KeyboardComposer(tl, graph,
                           hw_sliders=not opts.no_sliders)
     kc.pack(fill=BOTH, expand=1)
     for helpline in ["Bindings: B3 mute; C-l edit levels in subcomposer"]:
         tk.Label(root,text=helpline, font="Helvetica -12 italic",
                  anchor='w').pack(side='top',fill='x')
     import twisted.internet
     try:
         reactor.listenTCP(networking.keyboardComposer.port,
                           server.Site(LevelServerHttp(kc.name_to_subtk)))
     except twisted.internet.error.CannotListenError, e:
         log.warn("Can't (and won't!) start level server:")
         log.warn(e)
     if 0: # needs fixing, or maybe it's obsolete because other progs can just patch the rdf graph
         import twisted.internet
         try:
             reactor.listenTCP(networking.keyboardComposer.port,
                               server.Site(LevelServerHttp(kc.name_to_subbox)))
         except twisted.internet.error.CannotListenError, e:
             log.warn("Can't (and won't!) start level server:")
             log.warn(e)
     root.protocol('WM_DELETE_WINDOW', reactor.stop)
     if not opts.nonpersistent:
         reactor.addSystemEventTrigger('after', 'shutdown', kc.save)
     tksupport.install(root,ms=10)
 #    prof.watchPoint("/usr/lib/python2.4/site-packages/rdflib-2.3.3-py2.4-linux-i686.egg/rdflib/Graph.py", 615)
     prof.run(reactor.run, profile=False)

bin/rdfdb

➞

Show inline comments

@@ @@ -86,25 +86,25 @@ Ways to display patches, using labels an @@
   <creator> set <subj>'s <p> to <o>
   <creator> changed <subj>'s <pred> from <o1> to <o2>
   <creator> added <o> to <s> <p>
 """
 from twisted.internet import reactor
 import twisted.internet.error
 import sys, optparse, logging, json, os
 import cyclone.web, cyclone.httpclient, cyclone.websocket
 sys.path.append(".")
 from light9 import networking, showconfig
+from light9 import networking, showconfig, prof
 from rdflib import ConjunctiveGraph, URIRef, Graph
 from light9.rdfdb.graphfile import GraphFile
 from light9.rdfdb.patch import Patch, ALLSTMTS
 from light9.rdfdb.rdflibpatch import patchQuads
 from light9.rdfdb import syncedgraph
 from twisted.internet.inotify import INotify
 logging.basicConfig(level=logging.DEBUG)
 log = logging.getLogger()
 try:
     import sys
@@ @@ -158,27 +158,31 @@ class Db(object): @@
                                URIRef("http://example.com/file/%s" %
                                       os.path.basename(inFile)),
                                self.patch,
                                self.getSubgraph)
     def patch(self, p):
         """
         apply this patch to the master graph then notify everyone about it
         """
         log.info("patching graph -%d +%d" % (len(p.delQuads), len(p.addQuads)))
         patchQuads(self.graph, p.delQuads, p.addQuads, perfect=True)
+        senderUpdateUri = getattr(p, 'senderUpdateUri', None)
         self.summarizeToLog()
         for c in self.clients:
             print "send to %s? %s %s" % (c, c.updateUri, senderUpdateUri)
             if c.updateUri == senderUpdateUri:
                 # this client has self-applied the patch already
                 continue
             d = c.sendPatch(p)
             d.addErrback(self.clientErrored, c)
         sendToLiveClients(asJson=p.jsonRepr)
     def clientErrored(self, err, c):
         err.trap(twisted.internet.error.ConnectError)
         log.info("connection error- dropping client %r" % c)
         self.clients.remove(c)
         self.sendClientsToAllLivePages()
     def summarizeToLog(self):
         log.info("contexts in graph (%s total stmts):" % len(self.graph))
@@ @@ -284,13 +288,13 @@ if __name__ == "__main__": @@
     reactor.listenTCP(port, cyclone.web.Application(handlers=[
         (r'/', Index),
         (r'/live', Live),
         (r'/graph', GraphResource),
         (r'/patches', Patches),
         (r'/graphClients', GraphClients),
         (r"/(jquery-1\.7\.2\.min\.js)", cyclone.web.StaticFileHandler,
          dict(path='lib')),
         ], db=db))
     log.info("serving on %s" % port)
-    reactor.run()
+    prof.run(reactor.run, profile=False)

light9/Submaster.py

➞

Show inline comments

@@ @@ -141,36 +141,36 @@ class PersistentSubmaster(Submaster): @@
     def setName(self):
         log.info("sub update name %s %s", self.uri, self.graph.label(self.uri))
         self.name = self.graph.label(self.uri)
     def setLevels(self):
         log.info("sub update levels")
         oldLevels = getattr(self, 'levels', {}).copy()
         self.setLevelsFromGraph()
         if oldLevels != self.levels:
             log.info("sub %s changed" % self.name)
     def setLevelsFromGraph(self):
         patchGraph = showconfig.getGraph() # going away
         if hasattr(self, 'levels'):
             self.levels.clear()
         else:
             self.levels = {}
         for lev in self.graph.objects(self.uri, L9['lightLevel']):
             chan = self.graph.value(lev, L9['channel'])
             val = self.graph.value(lev, L9['level'])
             name = patchGraph.label(chan)
             name = self.graph.label(chan)
             if not name:
                 log.error("sub %r has channel %r with no name- leaving out that channel" % (self.name, chan))
                 continue
             val = self.graph.value(lev, L9['level'])
             self.levels[name] = float(val)
     def save(self):
         if self.temporary:
             log.info("not saving temporary sub named %s",self.name)
             return
         graph = Graph()
         subUri = L9['sub/%s' % self.name]
         graph.add((subUri, RDFS.label, Literal(self.name)))
         for chan in self.levels.keys():
             try:

light9/rdfdb/patch.py

➞

Show inline comments

 import json
 from rdflib import ConjunctiveGraph
 import json, unittest
 from rdflib import ConjunctiveGraph, URIRef
 from light9.rdfdb.rdflibpatch import graphFromNQuad, graphFromQuads, serializeQuad
 ALLSTMTS = (None, None, None)
 def graphFromQuads(q):
     g = ConjunctiveGraph()
     #g.addN(q) # no effect on nquad output
     for s,p,o,c in q:
         g.get_context(c).add((s,p,o))
         #g.store.add((s,p,o), c) # no effect on nquad output
     return g
 class Patch(object):
     """
     immutable
     the json representation includes the {"patch":...} wrapper
     """
     def __init__(self, jsonRepr=None, addQuads=None, delQuads=None,
                  addGraph=None, delGraph=None):
         self._jsonRepr = jsonRepr
         self._addQuads, self._delQuads = addQuads, delQuads
         self._addGraph, self._delGraph = addGraph, delGraph
         if self._jsonRepr is not None:
             body = json.loads(self._jsonRepr)
             self._delGraph = ConjunctiveGraph()
             self._delGraph.parse(data=body['patch']['deletes'], format='nquads')
             self._addGraph = ConjunctiveGraph()
             self._addGraph.parse(data=body['patch']['adds'], format='nquads')
             self._delGraph = graphFromNQuad(body['patch']['deletes'])
             self._addGraph = graphFromNQuad(body['patch']['adds'])
             if 'senderUpdateUri' in body:
                 self.senderUpdateUri = body['senderUpdateUri']
     @property
     def addQuads(self):
         if self._addQuads is None:
             if self._addGraph is not None:
                 self._addQuads = list(self._addGraph.quads(ALLSTMTS))
             else:
                 raise
         return self._addQuads
     @property
     def delQuads(self):
@@ @@ -52,17 +47,48 @@ class Patch(object): @@
             self._addGraph = graphFromQuads(self._addQuads)
         return self._addGraph
     @property
     def delGraph(self):
         if self._delGraph is None:
             self._delGraph = graphFromQuads(self._delQuads)
         return self._delGraph
     @property
     def jsonRepr(self):
         if self._jsonRepr is None:
             self._jsonRepr = json.dumps({"patch": {
                 'adds':self.addGraph.serialize(format='nquads'),
                 'deletes':self.delGraph.serialize(format='nquads'),
                 }})
             self._jsonRepr = self.makeJsonRepr()
         return self._jsonRepr
     def makeJsonRepr(self, extraAttrs={}):
         d = {"patch" : {
             'adds' : serializeQuad(self.addGraph),
             'deletes' : serializeQuad(self.delGraph),
             }}
         if len(self.addGraph) > 0 and d['patch']['adds'].strip() == "":
             # this is the bug that graphFromNQuad works around
             raise ValueError("nquads serialization failure")
         if '[<' in d['patch']['adds']:
             raise ValueError("[< found in %s" % d['patch']['adds'])
         d.update(extraAttrs)
         return json.dumps(d)
     def concat(self, more):
         """
         new Patch with the result of applying this patch and the
         sequence of other Patches
         """
         # not working yet
         adds = set(self.addQuads)
         dels = set(self.delQuads)
         for p2 in more:
             for q in p2.delQuads:
                 if q in adds:
                     adds.remove(q)
                 else:
                     dels.add(q)
             for q in p2.addQuads:
                 if q in dels:
                     dels.remove(q)
                 else:
                     adds.add(q)
         return Patch(delQuads=dels, addQuads=adds)

light9/rdfdb/rdflibpatch.py

➞

Show inline comments

 """
 this is a proposal for a ConjunctiveGraph method in rdflib
 """
 import unittest
 from rdflib import ConjunctiveGraph, URIRef as U
 def patchQuads(graph, deleteQuads, addQuads, perfect=False):
     """
     Delete the sequence of given quads. Then add the given quads just
     like addN would. If perfect is True, we'll error and not touch the
     graph if any of the deletes isn't in the graph or if any of the
     adds was already in the graph.
     """
     toDelete = []
     for s, p, o, c in deleteQuads:
         stmt = (s, p, o)
         if perfect:
@@ @@ -20,26 +22,75 @@ def patchQuads(graph, deleteQuads, addQu @@
         else:
             graph.store.remove(stmt, context=c)
     for c, stmt in toDelete:
         graph.store.remove(stmt, context=c)
     if perfect:
         addQuads = list(addQuads)
         for spoc in addQuads:
             if spoc in graph:
                 raise ValueError("%r already in %r" % (spoc[:3], spoc[3]))
     graph.addN(addQuads)
 import unittest
 from rdflib import ConjunctiveGraph, URIRef as U
 def graphFromQuads(q):
     g = ConjunctiveGraph()
     #g.addN(q) # no effect on nquad output
     for s,p,o,c in q:
         #g.get_context(c).add((s,p,o)) # kind of works with broken rdflib nquad serializer code
         g.store.add((s,p,o), c) # no effect on nquad output
     return g
 def graphFromNQuad(text):
     """
     g.parse(data=self.nqOut, format='nquads')
     makes a graph that serializes to nothing
     """
     g1 = ConjunctiveGraph()
     g1.parse(data=text, format='nquads')
     g2 = ConjunctiveGraph()
     for s,p,o,c in g1.quads((None,None,None)):
         #g2.get_context(c).add((s,p,o))
         g2.store.add((s,p,o), c)
     #import pprint; pprint.pprint(g2.store.__dict__)
     return g2
 from rdflib.plugins.serializers.nt import _xmlcharref_encode
 def serializeQuad(g):
     """replacement for graph.serialize(format='nquads')"""
     out = ""
     for s,p,o,c in g.quads((None,None,None)):
         out += u"%s %s %s %s .\n" % (s.n3(),
                                 p.n3(),
                                 _xmlcharref_encode(o.n3()),
                                 c.n3())
     return out
 class TestGraphFromQuads(unittest.TestCase):
     nqOut = '<http://example.com/> <http://example.com/> <http://example.com/> <http://example.com/> .\n'
     def testSerializes(self):
         n = U("http://example.com/")
         g = graphFromQuads([(n,n,n,n)])
         out = serializeQuad(g)
         self.assertEqual(out.strip(), self.nqOut.strip())
     def testNquadParserSerializes(self):
         g = graphFromNQuad(self.nqOut)
         self.assertEqual(len(g), 1)
         out = serializeQuad(g)
         self.assertEqual(out.strip(), self.nqOut.strip())
 stmt1 = U('http://a'), U('http://b'), U('http://c'), U('http://ctx1')
 stmt2 = U('http://a'), U('http://b'), U('http://c'), U('http://ctx2')
 class TestPatchQuads(unittest.TestCase):
     def testAddsToNewContext(self):
         g = ConjunctiveGraph()
         patchQuads(g, [], [stmt1])
         self.assert_(len(g), 1)
         quads = list(g.quads((None,None,None)))
         self.assertEqual(quads, [stmt1])
     def testDeletes(self):
         g = ConjunctiveGraph()

light9/rdfdb/syncedgraph.py

➞

Show inline comments

 from rdflib import ConjunctiveGraph, RDFS, RDF
+from rdflib import ConjunctiveGraph, RDFS, RDF, Graph
 import logging, cyclone.httpclient, traceback, urllib
 from twisted.internet import reactor
 log = logging.getLogger('syncedgraph')
 from light9.rdfdb.patch import Patch, ALLSTMTS
 from light9.rdfdb.rdflibpatch import patchQuads
 def sendPatch(putUri, patch):
     # this will take args for sender, etc
     body = patch.jsonRepr
     log.debug("send body: %r" % body)
 def sendPatch(putUri, patch, **kw):
     """
     kwargs will become extra attributes in the toplevel json object
     """
     body = patch.makeJsonRepr(kw)
     log.debug("send body: %r", body)
     def ok(done):
         if not str(done.code).startswith('2'):
             raise ValueError("sendPatch request failed %s: %s" % (done.code, done.body))
         log.debug("sendPatch finished, response: %s" % done.body)
         return done
     return cyclone.httpclient.fetch(
         url=putUri,
         method='PUT',
         headers={'Content-Type': ['application/json']},
         postdata=body,
         ).addCallback(ok)
@@ @@ -55,25 +57,25 @@ class GraphWatchers(object): @@
         except Exception:
             log.error("with key %r and func %r" % (key, func))
             raise
     def addPredObjWatcher(self, func, p, o):
         self._handlersPo.setdefault((p, o), set()).add(func)
     def whoCares(self, patch):
         """what handler functions would care about the changes in this patch?
         this removes the handlers that it gives you
         """
         self.dependencies()
+        #self.dependencies()
         affectedSubjPreds = set([(s, p) for s, p, o, c in patch.addQuads]+
                                 [(s, p) for s, p, o, c in patch.delQuads])
         affectedPredObjs = set([(p, o) for s, p, o, c in patch.addQuads]+
                                 [(p, o) for s, p, o, c in patch.delQuads])
         ret = set()
         for (s, p), funcs in self._handlersSp.iteritems():
             if (s, p) in affectedSubjPreds:
                 ret.update(funcs)
                 funcs.clear()
         for (p, o), funcs in self._handlersPo.iteritems():
@@ @@ -85,78 +87,146 @@ class GraphWatchers(object): @@
     def dependencies(self):
         """
         for debugging, make a list of all the active handlers and what
         data they depend on. This is meant for showing on the web ui
         for browsing.
         """
         log.info("whocares:")
         from pprint import pprint
         pprint(self._handlersSp)
 class PatchSender(object):
     """
     SyncedGraph may generate patches faster than we can send
     them. This object buffers and may even collapse patches before
     they go the server
     """
     def __init__(self, target, myUpdateResource):
         self.target = target
         self.myUpdateResource = myUpdateResource
         self._patchesToSend = []
         self._currentSendPatchRequest = None
     def sendPatch(self, p):
         self._patchesToSend.append(p)
         self._continueSending()
     def _continueSending(self):
         if not self._patchesToSend or self._currentSendPatchRequest:
             return
         if len(self._patchesToSend) > 1:
             log.info("%s patches left to send", len(self._patchesToSend))
             # this is where we could concatenate little patches into a
             # bigger one. Often, many statements will cancel each
             # other out. not working yet:
             if 0:
                 p = self._patchesToSend[0].concat(self._patchesToSend[1:])
                 print "concat down to"
                 print 'dels'
                 for q in p.delQuads: print q
                 print 'adds'
                 for q in p.addQuads: print q
                 print "----"
             else:
                 p = self._patchesToSend.pop(0)
         else:
             p = self._patchesToSend.pop(0)
         self._currentSendPatchRequest = sendPatch(
             self.target, p, senderUpdateUri=self.myUpdateResource)
         self._currentSendPatchRequest.addCallbacks(self._sendPatchDone,
                                                    self._sendPatchErr)
     def _sendPatchDone(self, result):
         self._currentSendPatchRequest = None
         self._continueSending()
     def _sendPatchErr(self, e):
         self._currentSendPatchRequest = None
         log.error(e)
         self._continueSending()
 class SyncedGraph(object):
     """
     graph for clients to use. Changes are synced with the master graph
     in the rdfdb process.
     This api is like rdflib.Graph but it can also call you back when
     there are graph changes to the parts you previously read.
     """
     def __init__(self, label):
         """
         label is a string that the server will display in association
         with your connection
         """
         _graph = self._graph = ConjunctiveGraph()
         self._watchers = GraphWatchers()
         def onPatch(p):
             patchQuads(_graph, p.delQuads, p.addQuads)
             """
             central server has sent us a patch
             """
             patchQuads(_graph, p.delQuads, p.addQuads, perfect=True)
             log.info("graph now has %s statements" % len(_graph))
             try:
                 self.updateOnPatch(p)
             except Exception:
                 # don't reflect this back to the server; we did
                 # receive its patch correctly.
                 traceback.print_exc()
         listen = reactor.listenTCP(0, cyclone.web.Application(handlers=[
             (r'/update', makePatchEndpoint(onPatch)),
         ]))
         port = listen._realPortNumber  # what's the right call for this?
         self.updateResource = 'http://localhost:%s/update' % port
         log.info("listening on %s" % port)
         self.register(label)
         self.currentFuncs = [] # stack of addHandler callers
         self._sender = PatchSender('http://localhost:8051/patches',
                                    self.updateResource)
     def register(self, label):
         def done(x):
             log.debug("registered with rdfdb")
         cyclone.httpclient.fetch(
             url='http://localhost:8051/graphClients',
             method='POST',
             headers={'Content-Type': ['application/x-www-form-urlencoded']},
             postdata=urllib.urlencode([('clientUpdate', self.updateResource),
                                        ('label', label)]),
             ).addCallbacks(done, log.error)
         log.info("registering with rdfdb")
     def patch(self, p):
         """send this patch to the server and apply it to our local graph and run handlers"""
         # currently this has to round-trip. But I could apply the
         # patch here and have the server not bounce it back to me
         return sendPatch('http://localhost:8051/patches', p)
         """send this patch to the server and apply it to our local
         graph and run handlers"""
         patchQuads(self._graph, p.delQuads, p.addQuads, perfect=True)
         self.updateOnPatch(p)
         self._sender.sendPatch(p)
     def patchObject(self, context, subject, predicate, newObject):
         """send a patch which removes existing values for (s,p,*,c)
         and adds (s,p,newObject,c). Values in other graphs are not affected"""
         raise NotImplementedError
         existing = []
         self.patch(Patch(
             delQuads=[],
             addQuads=[(subject, predicate, newObject, context)]))
     def addHandler(self, func):
         """
         run this (idempotent) func, noting what graph values it
         uses. Run it again in the future if there are changes to those
         graph values. The func might use different values during that
         future call, and those will be what we watch for next.
         """
         # if we saw this func before, we need to forget the old
         # callbacks it wanted and replace with the new ones we see
         # now.
@@ @@ -175,24 +245,41 @@ class SyncedGraph(object): @@
         finally:
             self.currentFuncs.pop()
     def updateOnPatch(self, p):
         """
         patch p just happened to the graph; call everyone back who
         might care, and then notice what data they depend on now
         """
         for func in self._watchers.whoCares(p):
             # todo: forget the old handlers for this func
             self.addHandler(func)
     def currentState(self, context=None):
         """
         a graph you can read without being in an addHandler
         """
         class Mgr(object):
             def __enter__(self2):
                 # this should be a readonly view of the existing graph
                 g = Graph()
                 for s in self._graph.triples((None, None, None), context):
                     g.add(s)
                 return g
             def __exit__(self, type, val, tb):
                 return
         return Mgr()
     def _getCurrentFunc(self):
         if not self.currentFuncs:
             # this may become a warning later
             raise ValueError("asked for graph data outside of a handler")
         # we add the watcher to the deepest function, since that
         # should be the cheapest way to update when this part of the
         # data changes
         return self.currentFuncs[-1]
     # these just call through to triples() so it might be possible to
     # watch just that one.

0 comments (0 inline, 0 general)