mirror of
https://git.roussel.pro/telecom-paris/pact.git
synced 2026-02-09 10:30:17 +01:00
39 lines
1.2 KiB
Python
39 lines
1.2 KiB
Python
import asyncio
|
|
import json
|
|
import math
|
|
import websockets
|
|
import random
|
|
import os
|
|
import hands
|
|
import time
|
|
|
|
values = []
|
|
class WebsocketServer:
|
|
def __init__(self,getEffects,port=os.getenv("PORT"),host=os.getenv("HOST")) -> None:
|
|
self.host = host
|
|
self.port = port
|
|
self.getEffects = getEffects
|
|
|
|
async def run(self):
|
|
async with websockets.serve(self.handler, self.host, self.port):
|
|
await asyncio.Future()
|
|
|
|
|
|
async def handler(self,websocket):
|
|
while True:
|
|
start = time.time()
|
|
messages = self.getEffects()
|
|
hands.frame()
|
|
await websocket.send(json.dumps(messages))
|
|
# await asyncio.sleep(1/30)
|
|
delay = time.time() - start
|
|
values.append(1/delay)
|
|
avg = sum(values) / len(values)
|
|
dev = [(v - avg) ** 2 for v in values]
|
|
print(avg, math.sqrt(sum(dev)/len(dev)))
|
|
#Remplacer ça par la fonction qui récupère les effets (dans le module de reconnaissance de gestes)
|
|
def getEffects():
|
|
return {"type": "effects", "effects": [{"type": "thumbs_up", "x":random.randint(0,100), "y": random.randint(0,100), "width": 50, "height": 50}]}
|
|
|
|
server = WebsocketServer(getEffects)
|
|
asyncio.run(server.run()) |