97 lines
2.0 KiB
Python
97 lines
2.0 KiB
Python
from evdev import InputDevice, ecodes
|
|
import asyncio
|
|
import websockets
|
|
import json
|
|
import threading
|
|
|
|
# ----------------------------
|
|
# CONFIG
|
|
# ----------------------------
|
|
DEVICE_PATH = "/dev/input/event8"
|
|
WS_HOST = "0.0.0.0"
|
|
WS_PORT = 8765
|
|
|
|
device = InputDevice(DEVICE_PATH)
|
|
|
|
clients = set()
|
|
|
|
# ----------------------------
|
|
# KEY MAPPING
|
|
# ----------------------------
|
|
import json
|
|
|
|
with open("keymap.json", "r") as f:
|
|
KEY_MAP = json.load(f)
|
|
# ----------------------------
|
|
# WEBSOCKET HANDLER
|
|
# ----------------------------
|
|
async def handler(websocket):
|
|
print("Client connected")
|
|
clients.add(websocket)
|
|
|
|
try:
|
|
await websocket.wait_closed()
|
|
finally:
|
|
clients.remove(websocket)
|
|
print("Client disconnected")
|
|
|
|
# ----------------------------
|
|
# BROADCAST FUNCTION
|
|
# ----------------------------
|
|
async def broadcast(msg):
|
|
if clients:
|
|
await asyncio.gather(
|
|
*[c.send(msg) for c in clients],
|
|
return_exceptions=True
|
|
)
|
|
|
|
# ----------------------------
|
|
# INPUT LOOP
|
|
# ----------------------------
|
|
def input_loop(loop):
|
|
print("Input loop started")
|
|
|
|
for event in device.read_loop():
|
|
|
|
if event.type != ecodes.EV_KEY:
|
|
continue
|
|
|
|
if event.value not in (0, 1):
|
|
continue
|
|
|
|
key_name = KEY_MAP.get(str(event.code), "").upper()
|
|
|
|
payload = json.dumps({
|
|
"type": "keydown" if event.value == 1 else "keyup",
|
|
"key": key_name
|
|
})
|
|
|
|
print("send:", payload)
|
|
|
|
asyncio.run_coroutine_threadsafe(
|
|
broadcast(payload),
|
|
loop
|
|
)
|
|
|
|
# ----------------------------
|
|
# MAIN
|
|
# ----------------------------
|
|
async def main():
|
|
print(f"Starting WS server on ws://{WS_HOST}:{WS_PORT}")
|
|
|
|
loop = asyncio.get_running_loop()
|
|
|
|
server = await websockets.serve(handler, WS_HOST, WS_PORT)
|
|
|
|
print("WS server ready")
|
|
|
|
threading.Thread(
|
|
target=input_loop,
|
|
args=(loop,),
|
|
daemon=True
|
|
).start()
|
|
|
|
await server.wait_closed()
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main()) |