-
Notifications
You must be signed in to change notification settings - Fork 0
/
ThrowBallClient.py
57 lines (44 loc) · 1.43 KB
/
ThrowBallClient.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import asyncio
import json
import time
from pynput import keyboard
import websockets
import multiprocessing as mp
from EventInfo import EventInfo
from EventInfo import Header
Event = mp.Manager().Value('i', 0)
def on_enter_press(key):
if key == keyboard.Key.enter:
print("Enter pressed")
Event.value = 1
time.sleep(1)
Event.value = 0
def throw_ball(sent_to):
header = Header("Trigger", sent_to, "throwBall", "001")
data = {"value": "1"}
event = EventInfo(header, data)
eventJson = json.dumps(event, default=vars)
return eventJson
async def client():
uri = "ws://localhost:3000"
disconnected = False
while True:
async with websockets.connect(uri) as websocket:
print("Connection successful")
while not disconnected:
try:
await websocket.connected
if bool(Event.value):
await websocket.send(throw_ball("Player App"))
await websocket.send(throw_ball("Detector"))
time.sleep(1)
except websockets.ConnectionClosed:
print("error")
disconnected = True
break
print("connection lost... reconnecting")
disconnected = False
time.sleep(5)
listener = keyboard.Listener(on_press=on_enter_press)
listener.start()
asyncio.run(client())