light9 Changeset - c1bf296b0a74

Changeset - c1bf296b0a74

Parent rev.

Child rev.

[Not reviewed]

default

0 7 1

Drew Perttula - 8 years ago 2017-05-17 07:39:21
drewp@bigasterisk.com

collector uses cyclone and gets a web ui showing output attrs
Ignore-this: 6dda48ab8d89344e0c8271429c8175af

8 files changed with 263 insertions and 35 deletions:

bin/collector

light9/collector/collector.py

light9/collector/collector_test.py

light9/collector/device.py

light9/collector/output.py

light9/collector/web/index.html

157

light9/io/udmx.py

light9/web/style.css

0 comments (0 inline, 0 general)

bin/collector

➞

Show inline comments

 #!bin/python
 from __future__ import division
-from rdflib import Graph, URIRef, Literal
 from rdflib import URIRef, Literal
 from twisted.internet import reactor, utils
 from twisted.web.server import Site
 from txzmq import ZmqEndpoint, ZmqFactory, ZmqPullConnection
 import json
 import logging
 import klein
 import optparse
 import time
 import traceback
 import cyclone.web, cyclone.websocket
 from greplin import scales
 from greplin.scales.twistedweb import StatsResource
 from run_local import log
 from lib.cycloneerr import PrettyErrorHandler
 from light9.collector.output import EnttecDmx, Udmx
 from light9.collector.collector import Collector
 from light9.namespaces import L9
 from light9 import networking
 from light9.rdfdb.syncedgraph import SyncedGraph
-from light9.rdfdb import clientsession
+from light9.greplin_cyclone import StatsForCyclone
 def parseJsonMessage(msg):
     body = json.loads(msg)
     settings = []
     for device, attr, value in body['settings']:
         settings.append((URIRef(device), URIRef(attr), Literal(value)))
     return body['client'], body['clientSession'], settings, body['sendTime']
 class WebServer(object):
     stats = scales.collection('/webServer',
                               scales.PmfStat('setAttr'))
     app = klein.Klein()
     def __init__(self, collector):
         self.collector = collector
     @app.route('/attrs', methods=['PUT'])
     def putAttrs(self, request):
         with WebServer.stats.setAttr.time():
             client, clientSession, settings, sendTime = parseJsonMessage(request.content.read())
             self.collector.setAttrs(client, clientSession, settings, sendTime)
             request.setResponseCode(202)
     @app.route('/stats', methods=['GET'])
     def getStats(self, request):
         return StatsResource('collector')
 def startZmq(port, collector):
     stats = scales.collection('/zmqServer',
                               scales.PmfStat('setAttr'))
     zf = ZmqFactory()
     addr = 'tcp://*:%s' % port
     log.info('creating zmq endpoint at %r', addr)
     e = ZmqEndpoint('bind', addr)
     s = ZmqPullConnection(zf, e)
     def onPull(message):
         with stats.setAttr.time():
             # todo: new compressed protocol where you send all URIs up
             # front and then use small ints to refer to devices and
             # attributes in subsequent requests.
             client, clientSession, settings, sendTime = parseJsonMessage(message[0])
             collector.setAttrs(client, clientSession, settings, sendTime)
     s.onPull = onPull
 class WebListeners(object):
     def __init__(self):
         self.clients = []
     def addClient(self, client):
         self.clients.append([client, {}])
     def delClient(self, client):
         self.clients = [[c, t] for c, t in self.clients if c != client]
     def outputAttrsSet(self, dev, attrs, outputMap):
         now = time.time()
         msg = self.makeMsg(dev, attrs, outputMap)
         for client, seen in self.clients:
             for m, t in seen.items():
                 if t < now - 5:
                     del seen[m]
             if msg in seen:
                 continue
             seen[msg] = now
             client.sendMessage(msg)
     def makeMsg(self, dev, attrs, outputMap):
         attrRows = []
         for attr, val in attrs.items():
             output, index = outputMap[(dev, attr)]
             attrRows.append({'attr': attr.rsplit('/')[-1],
                              'val': val,
                              'chan': (output.shortId(), index)})
         attrRows.sort(key=lambda r: r['chan'])
         for row in attrRows:
             row['chan'] = '%s %s' % (row['chan'][0], row['chan'][1])
         msg = json.dumps({'outputAttrsSet': {'dev': dev, 'attrs': attrRows}}, sort_keys=True)
         return msg
 class Updates(cyclone.websocket.WebSocketHandler):
     def connectionMade(self, *args, **kwargs):
         self.settings.listeners.addClient(self)
     def connectionLost(self, reason):
         self.settings.listeners.delClient(self)
     def messageReceived(self, message):
         json.loads(message)
 stats = scales.collection('/webServer', scales.PmfStat('setAttr'))
 class Attrs(PrettyErrorHandler, cyclone.web.RequestHandler):
     def put(self):
         with stats.setAttr.time():
             client, clientSession, settings, sendTime = parseJsonMessage(self.request.body)
             self.settings.collector.setAttrs(client, clientSession, settings, sendTime)
             self.set_status(202)
 def launch(graph, doLoadTest=False):
     try:
         # todo: drive outputs with config files
         outputs = [
             #EnttecDmx(L9['output/dmx0/'], '/dev/dmx0', 80),
             Udmx(L9['output/udmx/'], 510),
+        ]
     except Exception as e:
         log.error("setting up outputs: %r", e)
         traceback.print_exc()
         raise
     c = Collector(graph, outputs)
     server = WebServer(c)
     listeners = WebListeners()
     c = Collector(graph, outputs, listeners)
     startZmq(networking.collectorZmq.port, c)
     reactor.listenTCP(networking.collector.port,
                       Site(server.app.resource()),
                       cyclone.web.Application(handlers=[
                           (r'/()', cyclone.web.StaticFileHandler,
                            {"path" : "light9/collector/web", "default_filename" : "index.html"}),
                           (r'/updates', Updates),
                           (r'/attrs', Attrs),
                           (r'/stats', StatsForCyclone),
                       ], collector=c, listeners=listeners),
                       interface='::')
     log.info('serving http on %s, zmq on %s', networking.collector.port,
              networking.collectorZmq.port)
     if doLoadTest:
         # in a subprocess since we don't want this client to be
         # cooperating with the main event loop and only sending
         # requests when there's free time
         def afterWarmup():
             log.info('running collector_loadtest')
             d = utils.getProcessValue('bin/python', ['bin/collector_loadtest.py'])
             def done(*a):
                 log.info('loadtest done')
@@ @@ -97,17 +147,17 @@ def launch(graph, doLoadTest=False): @@
 def main():
     parser = optparse.OptionParser()
     parser.add_option("-v", "--verbose", action="store_true",
                       help="logging.DEBUG")
     parser.add_option("--loadtest", action="store_true",
                       help="call myself with some synthetic load then exit")
     (options, args) = parser.parse_args()
     log.setLevel(logging.DEBUG if options.verbose else logging.INFO)
     graph = SyncedGraph(networking.rdfdb.url, "collector")
     graph.initiallySynced.addCallback(lambda _: launch(graph, options.loadtest))
+    graph.initiallySynced.addCallback(lambda _: launch(graph, options.loadtest)).addErrback(log.error)
     reactor.run()
 if __name__ == '__main__':
     main()

light9/collector/collector.py

➞

Show inline comments

@@ @@ -35,32 +35,33 @@ def outputMap(graph, outputs): @@
             universe = graph.value(dev, L9['dmxUniverse'])
             try:
                 output = outputByUri[universe]
             except Exception:
                 log.warn('dev %r :dmxUniverse %r', dev, universe)
                 raise
             dmxBase = int(graph.value(dev, L9['dmxBase']).toPython())
             for row in graph.objects(dc, L9['attr']):
                 outputAttr = graph.value(row, L9['outputAttr'])
                 offset = int(graph.value(row, L9['dmxOffset']).toPython())
                 index = dmxBase + offset - 1
                 ret[(dev, outputAttr)] = (output, index)
-                log.info('    map %s to %s,%s', outputAttr, output, index)
+                log.debug('    map %s to %s,%s', outputAttr, output, index)
     return ret
 class Collector(Generic[ClientType, ClientSessionType]):
     def __init__(self, graph, outputs, clientTimeoutSec=10):
+    def __init__(self, graph, outputs, listeners=None, clientTimeoutSec=10):
         # type: (Graph, List[Output], float) -> None
         self.graph = graph
         self.outputs = outputs
         self.listeners = listeners
         self.clientTimeoutSec = clientTimeoutSec
         self.graph.addHandler(self.rebuildOutputMap)
         # client : (session, time, {(dev,devattr): latestValue})
         self.lastRequest = {} # type: Dict[ClientType, Tuple[ClientSessionType, float, Dict[Tuple[URIRef, URIRef], float]]]
         # (dev, devAttr): value to use instead of 0
         self.stickyAttrs = {} # type: Dict[Tuple[URIRef, URIRef], float]
     def rebuildOutputMap(self):
         self.outputMap = outputMap(self.graph, self.outputs) # (device, outputattr) : (output, index)
@@ @@ -142,39 +143,41 @@ class Collector(Generic[ClientType, Clie @@
             daDict = deviceAttrs.setdefault(d, {})
             if da not in daDict:
                 daDict[da] = v
         outputAttrs = {} # device: {outputAttr: value}
         for d in deviceAttrs:
             try:
                 devType = self.deviceType[d]
             except KeyError:
                 log.warn("request for output to unconfigured device %s" % d)
                 continue
             outputAttrs[d] = toOutputAttrs(devType, deviceAttrs[d])
             if self.listeners:
                 self.listeners.outputAttrsSet(d, outputAttrs[d], self.outputMap)
         pendingOut = {} # output : values
         for out in self.outputs:
             pendingOut[out] = [0] * out.numChannels
         for device, attrs in outputAttrs.iteritems():
             for outputAttr, value in attrs.iteritems():
                 self.setAttr(device, outputAttr, value, pendingOut)
         dt1 = 1000 * (time.time() - now)
         self.flush(pendingOut)
         dt2 = 1000 * (time.time() - now)
         if dt1 > 10:
-            print "slow setAttrs: %.1fms -> flush -> %.1fms. lr %s da %s oa %s" % (
+            log.warn("slow setAttrs: %.1fms -> flush -> %.1fms. lr %s da %s oa %s" % (
                 dt1, dt2, len(self.lastRequest), len(deviceAttrs), len(outputAttrs)
+            )
+            ))
     def setAttr(self, device, outputAttr, value, pendingOut):
         output, index = self.outputMap[(device, outputAttr)]
         outList = pendingOut[output]
         setListElem(outList, index, value, combine=max)
     def flush(self, pendingOut):
         """write any changed outputs"""
         for out, vals in pendingOut.iteritems():
             out.update(vals)
             out.flush()

light9/collector/collector_test.py

➞

Show inline comments

@@ @@ -75,25 +75,24 @@ class TestOutputMap(unittest.TestCase): @@
           dev:inst1 a :Device; :brightness dev:inst1Brightness .
         '''), [out0])
     def testMultipleOutputConnections(self):
         out0 = MockOutput(UDMX, [(0, DMX0['c1'])])
         self.assertRaises(ValueError, outputMap, MockSyncedGraph(PREFIX + '''
           dmx0:c1 :connectedTo dev:inst1Brightness .
           dmx0:c2 :connectedTo dev:inst1Brightness .
           dev:inst1 a :Device; :brightness dev:inst1Brightness .
         '''), [out0])
 class TestCollector(unittest.TestCase):
     def setUp(self):
         self.config = MockSyncedGraph(PREFIX + THEATER + '''
         dev:colorStrip a :Device, :ChauvetColorStrip;
           :dmxUniverse udmx:; :dmxBase 1;
           :red dev:colorStripRed;
           :green dev:colorStripGreen;
           :blue dev:colorStripBlue;
           :mode dev:colorStripMode .
         dev:inst1 a :Device, :SimpleDimmer;

light9/collector/device.py

➞

Show inline comments

@@ @@ -103,32 +103,31 @@ def toOutputAttrs(deviceType, deviceAttr @@
             L9['mode']: 215,
             L9['red']: r,
             L9['green']: g,
             L9['blue']: b
+            }
     elif deviceType == L9['SimpleDimmer']:
         return {L9['level']: _8bit(floatAttr(L9['brightness']))}
     elif deviceType == L9['Mini15']:
         inp = deviceAttrSettings
         rx8 = float(inp.get(L9['rx'], 0)) / 540 * 255
         ry8 = float(inp.get(L9['ry'], 0)) / 240 * 255
         r, g, b = hex_to_rgb(inp.get(L9['color'], '#000000'))
         return {
             L9['xRotation']: clamp255(int(math.floor(rx8))),
             # didn't find docs on this, but from tests it looks like 64 fine steps takes you to the next coarse step
             L9['xFine']: _8bit(1 - (rx8 % 1.0)),
             L9['yRotation']: clamp255(int(math.floor(ry8))),
             L9['yFine']: _8bit((ry8 % 1.0) / 4),
             L9['rotationSpeed']: 0,
+            L9['rotationSpeed']: 0, # seems to have no effect
             L9['dimmer']: 255,
             L9['red']: r,
             L9['green']: g,
             L9['blue']: b,
             L9['colorChange']: 0,
             L9['colorSpeed']: 0,
             L9['goboShake']: 0,
             L9['goboChoose']: 0,
+        }
     elif deviceType == L9['ChauvetHex12']:
         out = {}
         out[L9['red']], out[L9['green']], out[L9['blue']] = r, g, b = rgbAttr(L9['color'])

light9/collector/output.py

➞

Show inline comments

@@ @@ -40,24 +40,27 @@ class Output(object): @@
         """
         output takes a flattened list of values, maybe dmx channels, or
         pin numbers, etc
         """
         raise NotImplementedError
     def flush(self):
         """
         send latest data to output
         """
         raise NotImplementedError
     def shortId(self):
         """short string to distinguish outputs"""
         raise NotImplementedError
 class DmxOutput(Output):
     def __init__(self, uri, numChannels):
         self.uri = uri
         self.numChannels = numChannels
     def flush(self):
         pass
     def _loop(self):
         start = time.time()
         sendingBuffer = self.currentBuffer
@@ @@ -97,24 +100,27 @@ class EnttecDmx(DmxOutput): @@
         # I was outputting on 76 and it was turning on the light at
         # dmx75. So I added the 0 byte. No notes explaining the footer byte.
         self.currentBuffer = '\x00' + ''.join(map(chr, values)) + "\x00"
     @stats.write.time()
     def sendDmx(self, buf):
         self.dev.write(self.currentBuffer)
     def countError(self):
         pass
     def shortId(self):
         return 'enttec'
 class Udmx(DmxOutput):
     stats = scales.collection('/output/udmx',
                               scales.PmfStat('update'),
                               scales.PmfStat('write'),
                               scales.IntStat('usbErrors'))
     def __init__(self, uri, numChannels):
         DmxOutput.__init__(self, uri, numChannels)
         from light9.io.udmx import Udmx
         self.dev = Udmx()
         self.currentBuffer = ''
@@ @@ -143,14 +149,15 @@ class Udmx(DmxOutput): @@
         with Udmx.stats.write.time():
             try:
                 self.dev.SendDMX(buf)
                 return True
             except usb.core.USBError:
                 # not in main thread
                 return False
     def countError(self):
         # in main thread
         Udmx.stats.usbErrors += 1
     def shortId(self):
         return 'udmx' # and something unique from self.dev?

light9/collector/web/index.html

➞

Show inline comments

@@ new file 100644 @@
 <!doctype html>
 <html>
   <head>
     <title>collector</title>
     <meta charset="utf-8" />
     <script src="/lib/webcomponentsjs/webcomponents-lite.min.js"></script>
     <link rel="import" href="/lib/polymer/polymer.html">
     <link rel="import" href="/lib/iron-ajax/iron-ajax.html">
     <link rel="import" href="../rdfdb-synced-graph.html">
     <script src="/lib/N3.js-pull61/browser/n3-browser.js"></script>
     <script src="/lib/async/dist/async.js"></script>
     <script src="/lib/underscore/underscore-min.js"></script>
     <link rel="stylesheet"  href="/style.css">
   </head>
   <body>
     <dom-module id="light9-collector-device">
       <template>
         <style>
          :host {
              display: block;
              break-inside: avoid-column;
+         }
          td.nonzero {
              background: #310202;
              color: #e25757;
+         }
          td.full {
              background: #2b0000;
              color: red;
              font-weight: bold;
+         }
         </style>
         <h3>{{label}}</h3>
         <table class="borders">
           <tr>
             <th>output attr</th>
             <th>value</th>
             <th>output chan</th>
           </tr>
           <template is="dom-repeat" items="{{attrs}}">
             <tr>
               <td>{{item.attr}}</td>
               <td class$="{{item.valClass}}">{{item.val}} →</td>
               <td>{{item.chan}}</td>
             </tr>
           </template>
       </template>
       <script>
        HTMLImports.whenReady(function () {
            Polymer({
                is: "light9-collector-device",
                properties: {
                    graph: {type: Object, notify: true},
                    uri: {type: String, notify: true},
                    label: {type: String, notify: true},
                },
                observers: [
                    "setLabel(graph, uri)",
                    "initUpdates(updates)",
                ],
                initUpdates: function(updates) {
                    updates.addListener(function(msg) {
                        if (msg.outputAttrsSet && msg.outputAttrsSet.dev == this.uri) {
                            this.attrs = msg.outputAttrsSet.attrs;
                            this.attrs.forEach(function(row) {
                                row.valClass = row.val == 255 ? 'full' : (row.val ? 'nonzero' : '');
                            });
+                       }
                    }.bind(this));
                },
                setLabel: function(graph, uri) {
                    console.log('setlab', uri);
                    this.label = uri.replace(/.*\//, '');
                    graph.runHandler(function() {
                        this.label = graph.stringValue(uri, graph.Uri('rdfs:label'));
                    }.bind(this), 'setLabel');
+               }
            });
        });
       </script>
     </dom-module>
     <dom-module id="light9-collector-ui">
       <template>
         <rdfdb-synced-graph graph="{{graph}}"></rdfdb-synced-graph>
         <h1>Collector <a href="stats">[stats]</a></h1>
         <h2>Devices</h2>
         <div style="column-width: 18em">
         <template is="dom-repeat" items="{{devices}}">
           <light9-collector-device
               graph="{{graph}}" updates="{{updates}}"
               uri="{{item}}"></light9-collector-device>
         </template>
         </div>
       </template>
       <script>
        class Updates {
            constructor() {
                this.listeners = [];
+           }
            addListener(cb) {
                this.listeners.push(cb);
+           }
            onMessage(msg) {
                this.listeners.forEach(function(lis) {
                    lis(msg);
                });
+           }
+       }
        HTMLImports.whenReady(function () {
            Polymer({
                is: "light9-collector-ui",
                properties: {
                    graph: {type: Object, notify: true},
                    updates: {type: Object, notify: true},
                    devices: {type: Array},
                },
                observers: [
                    'onGraph(graph)',
                ],
                ready: function() {
                    this.updates = new Updates();
                    var sock = new WebSocket(
                        window.location.href.replace(/^http/, 'ws') + 'updates');
                    sock.onmessage = function(ev) {
                        this.updates.onMessage(JSON.parse(ev.data));
                    }.bind(this);
                },
                onGraph: function(graph) {
                    this.graph.runHandler(this.findDevices.bind(this), 'findDevices');
                },
                findDevices: function() {
                    var U = function(x) {
                        return this.graph.Uri(x);
                    };
                    this.set('devices', []);
                    _.uniq(_.sortBy(this.graph.subjects(U('rdf:type'), U(':DeviceClass'))), true).forEach(function(dc) {
                        _.sortBy(this.graph.subjects(U('rdf:type'), dc)).forEach(function(dev) {
                            this.push('devices', dev);
                        }.bind(this));
                    }.bind(this));
+               }
            });
        });
       </script>
     </dom-module>
     <light9-collector-ui></light9-collector-ui>
   </body>
 </html>

light9/io/udmx.py

➞

Show inline comments

 from __future__ import division
 import logging
 import usb.core
 from usb.util import CTRL_TYPE_VENDOR, CTRL_RECIPIENT_DEVICE, CTRL_OUT
 log = logging.getLogger('udmx')
 """
 Send dmx to one of these:
 http://www.amazon.com/Interface-Adapter-Controller-Lighting-Freestyler/dp/B00W52VIOS
 [4520784.059479] usb 1-2.3: new low-speed USB device number 6 using xhci_hcd
 [4520784.157410] usb 1-2.3: New USB device found, idVendor=16c0, idProduct=05dc
 [4520784.157416] usb 1-2.3: New USB device strings: Mfr=1, Product=2, SerialNumber=3
 [4520784.157419] usb 1-2.3: Product: uDMX
 [4520784.157422] usb 1-2.3: Manufacturer: www.anyma.ch
 [4520784.157424] usb 1-2.3: SerialNumber: ilLUTZminator001
 See https://www.illutzmination.de/udmxfirmware.html?&L=1
     sources/commandline/uDMX.c
 or https://github.com/markusb/uDMX-linux/blob/master/uDMX.c
 """
 cmd_SetChannelRange = 0x0002
 class Udmx(object):
     def __init__(self):
         self.dev = usb.core.find(idVendor=0x16c0, idProduct=0x05dc)
         log.info('found udmx at %r', self.dev)
     def SendDMX(self, buf):
         ret = self.dev.ctrl_transfer(
            bmRequestType=CTRL_TYPE_VENDOR | CTRL_RECIPIENT_DEVICE | CTRL_OUT,
            bRequest=cmd_SetChannelRange,
            wValue=len(buf),
            wIndex=0,
            data_or_wLength=buf)
         if ret < 0:
             raise ValueError("ctrl_transfer returned %r" % ret)

light9/web/style.css

➞

Show inline comments

@@ @@ -182,12 +182,21 @@ a button { @@
     font-size: 60%;
+}
 a.big {
     background-color: #384052;
     padding: 6px;
     text-shadow: rgba(0, 0, 0, 0.48) -1px -1px 0px;
     color: rgb(172, 172, 255);
     font-size: 160%;
     margin: 0px;
     display: inline-block;
     border-radius: 5px;
+}
 table {
     border-collapse: collapse;
+}
 table.borders td, table.borders th {
     border: 1px solid #4a4a4a;
     padding: 2px 8px;
+}
@@ \ No newline at end of file @@

0 comments (0 inline, 0 general)