Files
@ 1a96f8647126
Branch filter:
Location: light9/bin/collector_loadtest.py - annotation
1a96f8647126
1.6 KiB
text/x-python
big graph & autodep porting to make collector display labels from a syncedgraph
2951a690f1ba d5f1cc9615af 2951a690f1ba 2951a690f1ba 2951a690f1ba 2951a690f1ba d5f1cc9615af 2951a690f1ba 2951a690f1ba 2951a690f1ba 2951a690f1ba 4294ed82ee16 7772cc48e016 7772cc48e016 d5f1cc9615af f066d6e874db d5f1cc9615af d5f1cc9615af d5f1cc9615af 4294ed82ee16 4294ed82ee16 d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af d5f1cc9615af 7772cc48e016 d5f1cc9615af d5f1cc9615af d5f1cc9615af 4294ed82ee16 7772cc48e016 4294ed82ee16 d5f1cc9615af | #!bin/python
import asyncio
import logging
import time
from typing import cast
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=1
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"],
[DEV["houseSide"], L9["level"], .8],
[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())
|