mirror of
https://git.roussel.pro/telecom-paris/pact.git
synced 2026-02-09 02:20:17 +01:00
45 lines
1.3 KiB
Python
45 lines
1.3 KiB
Python
import asyncio
|
|
import json
|
|
import math
|
|
import websockets
|
|
import random
|
|
import os
|
|
import time
|
|
from hands import getThumbState
|
|
|
|
|
|
class WebsocketServer:
|
|
def __init__(self,getEffects,port=os.getenv("PORT"),host=os.getenv("HOST")) -> None:
|
|
self.thumbResult = None
|
|
self.state = 0
|
|
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:
|
|
if(self.state == 0):
|
|
messages, result = self.getEffects()
|
|
if(messages != False):
|
|
if(result == False):
|
|
await websocket.send(json.dumps(messages))
|
|
else:
|
|
self.thumbResult = result
|
|
self.state = 1
|
|
await websocket.send('{"type":"state","state":2}')
|
|
|
|
def getEffects():
|
|
res = getThumbState()
|
|
if(res != False):
|
|
state, coords, size, result = res
|
|
return {"type": "effects", "effects": [{"type": state, "x":coords[0], "y": coords[1], "width": size, "height": size}]}, result
|
|
else:
|
|
return False,False
|
|
|
|
server = WebsocketServer(getEffects)
|
|
asyncio.run(server.run()) |