Files
@ 3d28b1a54a29
Branch filter:
Location: light9/bin/collector
3d28b1a54a29
4.8 KiB
text/plain
reformat
Ignore-this: 28e00687983432a552e465291ef9b4b4
Ignore-this: 28e00687983432a552e465291ef9b4b4
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 | #!bin/python
"""
Collector receives device attrs from multiple senders, combines
them, and sends output attrs to hardware. The combining part has
custom code for some attributes.
Input can be over http or zmq.
"""
from run_local import log
from twisted.internet import reactor, utils
import json
import logging
import optparse
import traceback
import cyclone.web, cyclone.websocket
from greplin import scales
from cycloneerr import PrettyErrorHandler
from light9 import networking
from light9.collector.collector import Collector
from light9.collector.weblisteners import WebListeners
from greplin.scales.cyclonehandler import StatsHandler
from light9.namespaces import L9
from light9.zmqtransport import parseJsonMessage, startZmq
from rdfdb.syncedgraph import SyncedGraph
from standardservice.scalessetup import gatherProcessStats
from light9.collector.output import ArtnetDmx, DummyOutput # noqa
class Updates(cyclone.websocket.WebSocketHandler):
def connectionMade(self, *args, **kwargs):
log.info('socket connect %s', self)
self.settings.listeners.addClient(self)
def connectionLost(self, reason):
self.settings.listeners.delClient(self)
def messageReceived(self, message):
json.loads(message)
gatherProcessStats()
stats = scales.collection(
'/webServer',
scales.PmfStat('setAttr', recalcPeriod=1),
scales.RecentFpsStat('setAttrFps'),
)
class Attrs(PrettyErrorHandler, cyclone.web.RequestHandler):
def put(self):
stats.setAttrFps.mark()
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
rate = 30
outputs = [
ArtnetDmx(L9['output/dmxA/'],
host='127.0.0.1',
port=6445,
rate=rate),
#DummyOutput(L9['output/dmxA/']),
]
except Exception:
log.error("setting up outputs:")
traceback.print_exc()
raise
listeners = WebListeners()
c: Collector = Collector(graph, outputs, listeners)
startZmq(networking.collectorZmq.port, c)
reactor.listenTCP(networking.collector.port,
cyclone.web.Application(handlers=[
(r'/()', cyclone.web.StaticFileHandler, {
"path": "light9/collector/web",
"default_filename": "index.html"
}),
(r'/updates', Updates),
(r'/attrs', Attrs),
(r'/stats/(.*)', StatsHandler, {
'serverName': 'collector'
}),
],
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')
reactor.stop()
d.addCallback(done)
reactor.callLater(2, afterWarmup)
def main():
parser = optparse.OptionParser()
parser.add_option("-v",
"--verbose",
action="store_true",
help="logging.DEBUG")
parser.add_option("--logdmx", action="store_true", help="log all dmx sends")
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)
logging.getLogger('output').setLevel(logging.DEBUG)
logging.getLogger('output.allDmx').setLevel(
logging.DEBUG if options.logdmx else logging.INFO)
logging.getLogger('colormath').setLevel(logging.INFO)
graph = SyncedGraph(networking.rdfdb.url, "collector")
graph.initiallySynced.addCallback(lambda _: launch(graph, options.loadtest)
).addErrback(lambda e: reactor.crash())
reactor.run()
if __name__ == '__main__':
main()
|