rpikvm/hidinput.py

from flask import Response
from flask import Flask
from flask import render_template
from flask import request
from flask_socketio import SocketIO
import argparse
import json
import struct

from hidkeycodes import hidkeycodes
from jskeycodes import jscodehidmap


# initialize a flask object
app = Flask(__name__)
socketio = SocketIO(app)

hiddev = None
hidmouseabs = None


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


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

    hiddev.write(data)
    hiddev.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("/kvm.html")
def kvm():
    return render_template("kvm.html")

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

@app.route("/keyboard.html")
def keyboardIndex():
    return render_template("keyboard.html")


@socketio.on('mouseEventRaw')
def handle_mouseEvent(data):
    print(data)

    btn = data[0]
    x = data[1]
    y = data[2]
    wheel = data[3]

    print("RAW Btn: {}, X: {}, Y: {}, W: {}".format(btn, x, y, wheel))

    hid_mouse_writeabs(btn, x, y, wheel)


@socketio.on('mouseEvent')
def handle_mouseEvent(data):
    mouseevent = json.loads(data)
    print(mouseevent)
    btn = mouseevent['btn']
    x = mouseevent['x']
    y = mouseevent['y']
    wheel = mouseevent['wheel']
    wheel = wheel if wheel >= 0 else 255-abs(wheel)

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

    hid_mouse_writeabs(btn, x, y, wheel)


@socketio.on('kbdEvent')
def handle_kbdEvent(data):
    keyevent = json.loads(data)
    print("Raw data: {}".format(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)


hid_init()

# 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())

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