diff bin/debug/collector_loadtest.py @ 2376:4556eebe5d73

topdir reorgs; let pdm have its src/ dir; separate vite area from light9/
author drewp@bigasterisk.com
date Sun, 12 May 2024 19:02:10 -0700
parents bin/collector_loadtest.py@2c48e92ad5d3
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/bin/debug/collector_loadtest.py	Sun May 12 19:02:10 2024 -0700
@@ -0,0 +1,52 @@
+#!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())