114 lines
3.6 KiB
Python
Executable file
114 lines
3.6 KiB
Python
Executable file
#!/usr/bin/python3
|
|
|
|
from pynput.mouse import Button, Controller
|
|
from sanic import Sanic
|
|
from sanic.response import file
|
|
|
|
class MessageParser():
|
|
def __init__(self):
|
|
self.handlers = {}
|
|
def add_handler(self, code: str, args={}):
|
|
self.handlers[code] = {
|
|
"args": args
|
|
}
|
|
def parse(self, message: str):
|
|
if len(message) < 1:
|
|
return None, "Message is empty"
|
|
message_code = message[0]
|
|
if message_code not in self.handlers:
|
|
return None, "Message code is not handled"
|
|
message_arguments = message[1:].split(";")
|
|
handler = self.handlers[message_code]
|
|
decoded_arguments = {}
|
|
if len(handler["args"]) != len(message_arguments):
|
|
return None, "Got message with invalid argument count"
|
|
for i, argument_name in enumerate(handler["args"]):
|
|
argument_type = handler["args"][argument_name]
|
|
if argument_type == "int":
|
|
try:
|
|
decoded_arguments[argument_name] = int(message_arguments[i])
|
|
except ValueError:
|
|
return None, "Error parsing int argument for message (is it really an int?)"
|
|
elif argument_type == "str":
|
|
decoded_arguments[argument_name] = message_arguments[i]
|
|
else:
|
|
raise ValueError("parse(): Message handler references an invalid argument type")
|
|
|
|
return message_code, decoded_arguments
|
|
|
|
|
|
class InputController():
|
|
def __init__(self):
|
|
self.mouse_controller = Controller()
|
|
self.button_code_lookup = [
|
|
Button.left,
|
|
Button.right
|
|
]
|
|
self.parser = MessageParser()
|
|
|
|
# Relative mouse movement
|
|
self.parser.add_handler("r", {
|
|
"x": "int",
|
|
"y": "int"
|
|
})
|
|
# Mouse relative scroll
|
|
self.parser.add_handler("s", {
|
|
"x": "int",
|
|
"y": "int"
|
|
})
|
|
# Mouse button down
|
|
self.parser.add_handler("d", {
|
|
"button": "int"
|
|
})
|
|
# Mouse button up
|
|
self.parser.add_handler("u", {
|
|
"button": "int"
|
|
})
|
|
def button_code_to_object(self, button_code: int):
|
|
# HACK
|
|
obj = None
|
|
try:
|
|
obj = self.button_code_lookup[button_code]
|
|
except IndexError:
|
|
return self.button_code_lookup[0]
|
|
return obj
|
|
def process_message(self, message: str) -> bool:
|
|
code, args = self.parser.parse(message)
|
|
|
|
if code == None:
|
|
print("error while parsing message:", args)
|
|
return False
|
|
elif code == "r":
|
|
self.mouse_controller.move(args["x"], args["y"])
|
|
print("r", args["x"], args["y"])
|
|
elif code == "d":
|
|
print("d", args["button"])
|
|
self.mouse_controller.press(self.button_code_to_object(args["button"]))
|
|
elif code == "u":
|
|
print("u", args["button"])
|
|
self.mouse_controller.release(self.button_code_to_object(args["button"]))
|
|
elif code == "s":
|
|
print("s", args["x"], args["y"])
|
|
self.mouse_controller.scroll(args["x"], args["y"])
|
|
else:
|
|
print("got invalid code from parser (is this a bug with the MessageParser?)")
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
app = Sanic("capybara")
|
|
app.ctx.input_controller = InputController()
|
|
|
|
app.static("/", "./public/index.html")
|
|
|
|
@app.websocket("/gateway")
|
|
async def gateway(req, ws):
|
|
while True:
|
|
app.ctx.input_controller.process_message(await ws.recv())
|
|
|
|
def main():
|
|
app.run(host='0.0.0.0', port=4003, access_log=False)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|