rpikvm/hidinput.py

from flask import Response
from flask import Flask
from flask import render_template
from flask import request
import threading
import argparse
import json

from hidkeycodes import hidkeycodes
from jskeycodes import jscodehidmap


lock = threading.Lock()

# initialize a flask object
app = Flask(__name__)

hiddev = None
hidmouse = None
hidmouseabs = None


def hid_init():
    global hiddev, hidmouse, hidmouseabs
    hiddev = open('/dev/hidg0', 'rb+')
    hidmouse = open('/dev/hidg1', 'rb+')
    hidmouseabs = open('/dev/hidg2', 'rb+')


def hid_write(data):
    if hiddev is None:
        return False

    hiddev.write(data)
    hiddev.flush()


def hid_mouse_write(btn, x, y, wheel):
    if hidmouse is None:
        return False

    data = bytearray(4)
    data[0] = btn
    data[1] = x
    data[2] = y
    data[3] = wheel

    hidmouse.write(data)
    hidmouse.flush()


def hid_mouse_writeabs(btn, x, y, wheel):
    if hidmouseabs is None:
        return False

    data = bytearray(6)
    data[0] = btn
    data[1:3] = x.to_bytes(2, byteorder='little')
    data[3:5] = y.to_bytes(2, byteorder='little')
    data[5] = wheel

    hidmouseabs.write(data)
    hidmouseabs.flush()


def send_key(hidkey, shift, alt, ctlr, mod):
    data = bytearray(8)
    if shift:
        data[0] += hidkeycodes['KEY_MOD_LSHIFT']
    if alt:
        data[0] += hidkeycodes['KEY_MOD_LALT']
    if ctlr:
        data[0] += hidkeycodes['KEY_MOD_LCTRL']

    if mod and (hidkey == hidkeycodes['KEY_MOD_LMETA'] or hidkey == hidkeycodes['KEY_MOD_RMETA']):
        data[0] += hidkey
        hidkey = 0

    data[2] = hidkey

    hid_write(data)
    hid_write(bytearray(8))


def get_hid_by_jscode(rawkeycode):
    hidkeycode = None
    hidkeyname = None

    if rawkeycode in jscodehidmap:
        hidkeyname = jscodehidmap[rawkeycode]
        if hidkeyname in hidkeycodes:
            hidkeycode = hidkeycodes[hidkeyname]
        else:
            print("Code for key {} not found in HID mapping".format(hidkeyname))
    else:
        print("Javascript code for key {} not found".format(rawkeycode))

    print ("JS key: {} HID key: {}({})".format(rawkeycode, hidkeyname, hidkeycode))

    return hidkeycode, (hidkeyname is not None and "MOD" in hidkeyname)

@app.route("/")
def index():
    return render_template("hid.html")

@app.route("/mouse.html")
def mouseindex():
    return render_template("mouse.html")

@app.route("/hid/mouse", methods=["POST"])
def mouse():
    mouseevent = json.loads(request.data)
    print(mouseevent)

    btn = mouseevent['btn']
    x = mouseevent['x']
    y = mouseevent['y']
    wheel = mouseevent['wheel']

    x = x if x >= 0 else 255-abs(x)
    y = y if y >= 0 else 255-abs(y)

    print("X: {}, Y: {}".format(x, y))

    hid_mouse_write(btn, x, y, wheel)

    return Response("", mimetype="text/plain")

@app.route("/hid/mouseabs", methods=["POST"])
def mouseabs():
    mouseevent = json.loads(request.data)
    print(mouseevent)

    btn = mouseevent['btn']
    x = mouseevent['x']
    y = mouseevent['y']
    wheel = mouseevent['wheel']

    print("X: {}, Y: {}".format(x, y))

    hid_mouse_writeabs(btn, x, y, wheel)

    return Response("", mimetype="text/plain")


@app.route("/hid/keyboard", methods=["POST"])
def keypress():
    keyevent = json.loads(request.data)
    print("Raw data: {}".format(request.data))

    rawkeycode = keyevent['code']
    hidkeycode, mod = get_hid_by_jscode(rawkeycode)

    if hidkeycode is not None:
        try:
            send_key(
                hidkeycode,
                keyevent['shiftKey'],
                keyevent['altKey'],
                keyevent['ctrlKey'],
                mod)
        except Exception as e:
            print("Error sending HID message", e)

    return Response("Press {}".format(hidkeycode), mimetype="text/plain")


# check to see if this is the main thread of execution
if __name__ == '__main__':
    # construct the argument parser and parse command line arguments
    ap = argparse.ArgumentParser()
    ap.add_argument("-i", "--ip", type=str, required=True,
        help="ip address of the device")
    ap.add_argument("-o", "--port", type=int, required=True,
        help="ephemeral port number of the server (1024 to 65535)")
    ap.add_argument("-f", "--frame-count", type=int, default=32,
        help="# of frames used to construct the background model")
    args = vars(ap.parse_args())

    hid_init()

    app.run(host=args["ip"], port=args["port"], debug=True,
            threaded=True, use_reloader=False)

    # Clean up
    if hiddev is not None:
        print("Closing hid")
        hiddev.close()