Files
@ 5e4321405f54
Branch filter:
Location: light9/bin/collector_loadtest.py - annotation
5e4321405f54
1.7 KiB
text/x-python
redo services url structure and homepage config. (pages aren't updated this layout yet)
2951a690f1ba d5f1cc9615af 2951a690f1ba 2c48e92ad5d3 2c48e92ad5d3 2951a690f1ba 2951a690f1ba d5f1cc9615af 2951a690f1ba 2951a690f1ba 2951a690f1ba 2951a690f1ba 4294ed82ee16 7772cc48e016 7772cc48e016 d5f1cc9615af f066d6e874db d5f1cc9615af 2c48e92ad5d3 d5f1cc9615af 4294ed82ee16 4294ed82ee16 d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af 2c48e92ad5d3 2c48e92ad5d3 2c48e92ad5d3 2c48e92ad5d3 2c48e92ad5d3 2c48e92ad5d3 2c48e92ad5d3 2c48e92ad5d3 2c48e92ad5d3 d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af 7772cc48e016 d5f1cc9615af d5f1cc9615af d5f1cc9615af 4294ed82ee16 7772cc48e016 4294ed82ee16 d5f1cc9615af | #!bin/python
import asyncio
import logging
import random
from rdflib import URIRef
import time
from light9.collector.collector_client_asyncio import sendToCollector
from light9.effect.settings import DeviceSettings
from light9.namespaces import DEV, L9
from light9.run_local import log
log.setLevel(logging.DEBUG)
async def loadTest():
print("scheduling loadtest")
n = 200000
period=.02
times = []
session = "loadtest%s" % time.time()
for i in range(n):
if i % 100 == 0:
log.info('sendToCollector %s', i)
start = time.time()
await sendToCollector(
"http://localhost:8202/",
session,
DeviceSettings(
graph=None,
settingsList=[
# [DEV["backlight1"], L9["color"], "#ffffff"], #
# [DEV["backlight2"], L9["color"], "#ffffff"],
# [DEV["backlight3"], L9["color"], "#ffffff"],
# [DEV["backlight4"], L9["color"], "#ffffff"],
# [DEV["backlight5"], L9["color"], "#ffffff"],
# [DEV["down2"], L9["color"], "#ffffff"],
# [DEV["down3"], L9["color"], "#ffffff"],
# [DEV["down4"], L9["color"], "#ffffff"],
[URIRef('http://light9.bigasterisk.com/theater/skyline/device/down1'), L9["brightness"], random.random()],
[DEV["backlight5"], L9["uv"], 0.011]
]))
times.append(time.time() - start)
await asyncio.sleep(period)
print("loadtest done")
with open('/tmp/times', 'w') as f:
f.write(''.join('%s\n' % t for t in times))
if __name__ == '__main__':
asyncio.run(loadTest())
|