Remote Flight Stick for WebGL Game

install_hotas_driver.sh

#!/usr/bin/env bash
git clone https://github.com/walterschell/tflight4.git
cd tflight4
make
sudo cp hid-tflight4.ko /lib/modules/$(shell uname -r)/kernel/drivers/hid/
sudo insmod /lib/modules/$(shell uname -r)/kernel/drivers/hid/hid-tflight4.ko

py_deps.sh

#!/usr/bin/env bash
sudo apt install python-pip python-sdl2
pip2 install websocket-client

receive-sample.js

const WebSocket = require('ws');

const wss = new WebSocket.Server({ port: 9001 });

wss.on('connection', function connection(ws) {
    ws.binaryType = 'arraybuffer';
    ws.on('message', function incoming(data) {
        console.log('received: %s', new Int16Array(data));
    });
});

remote_stick.py

#!/usr/bin/env python2

import ctypes
import time
import threading
import struct

try:
    import thread
except ImportError:
    import _thread as thread

from sdl2 import *
import websocket


RIGHT_X = 0
RIGHT_Y = 1
NONE1 = 2
NONE2 = 3
YAW = 4
THROTTLE = 5
RIGHT_DIR = 6

BT_LEFT1 = 0
BT_LEFT2 = 1
BT_LEFT3 = 2
BT_LEFT4 = 3
BT_LEFT_BACK1 = 6
BT_LEFT_BACK2 = 7
BT_RIGHT_THB1 = 4
BT_RIGHT_THB2 = 10
BT_RIGHT_IDX1 = 5
BT_RIGHT_IDX2 = 11


def main_loop(stop_event, ws, *args):
    joystick = Joystick()
    event = SDL_Event()
    while True:
        if stop_event.is_set():
            break
        while SDL_PollEvent(ctypes.byref(event)) != 0:
            joystick.event(event)
        ws.send(joystick.to_bytes(), websocket.ABNF.OPCODE_BINARY)
        time.sleep(0.1)
    time.sleep(1)
    ws.close()


class WebSocketClient(object):
    def __init__(self):
        self.stop_event = threading.Event()

    def on_message(self, ws, message):
        print(message)
        if message == 'stop':
            self.stop_event.set()

    def on_error(self, ws, error):
        print(error)

    def on_close(self, ws):
        print("### closed ###")

    def on_open(self, ws):
        thread.start_new_thread(main_loop, (self.stop_event, ws))


class Joystick:
    def __init__(self):
        SDL_Init(SDL_INIT_JOYSTICK)
        self.axis = {
            RIGHT_X:        0,
            RIGHT_Y:        0,
            NONE1:          0,
            NONE2:          0,
            YAW:            0,
            THROTTLE:       0,
            RIGHT_DIR:      0,
        }
        self.axis_seq = [
            RIGHT_X,
            RIGHT_Y,
            RIGHT_DIR,
            YAW,
            THROTTLE,
        ]
        self.button = {
            BT_LEFT1:       False,
            BT_LEFT2:       False,
            BT_LEFT3:       False,
            BT_LEFT4:       False,
            BT_LEFT_BACK1:  False,
            BT_LEFT_BACK2:  False,
            BT_RIGHT_THB1:  False,
            BT_RIGHT_THB2:  False,
            BT_RIGHT_IDX1:  False,
            BT_RIGHT_IDX2:  False,
        }
        self.button_values = {
            BT_LEFT1:       1 << 0,
            BT_LEFT2:       1 << 1,
            BT_LEFT3:       1 << 2,
            BT_LEFT4:       1 << 3,
            BT_LEFT_BACK1:  1 << 4,
            BT_LEFT_BACK2:  1 << 5,
            BT_RIGHT_THB1:  1 << 6,
            BT_RIGHT_THB2:  1 << 7,
            BT_RIGHT_IDX1:  1 << 8,
            BT_RIGHT_IDX2:  1 << 9,
        }
        self.hat = 0

    def event(self, e):
        if e.type == SDL_JOYDEVICEADDED:
            self.device = SDL_JoystickOpen(e.jdevice.which)
        elif e.type == SDL_JOYAXISMOTION:
            self.axis[e.jaxis.axis] = e.jaxis.value
        elif e.type == SDL_JOYBUTTONDOWN:
            self.button[e.jbutton.button] = True
        elif e.type == SDL_JOYBUTTONUP:
            self.button[e.jbutton.button] = False
        elif e.type == SDL_JOYHATMOTION:
            self.hat = e.jhat.value

    def to_bytes(self):
        res = bytes()
        btn = 0
        for key in self.axis_seq:
            res += struct.pack('<h', self.axis[key])
        for key, down in self.button.items():
            if down:
                btn += self.button_values[key]
        res += struct.pack('<h', btn)
        res += struct.pack('<h', self.hat)
        return res


def main():
    websocket.enableTrace(False)
    client = WebSocketClient()

    ws = websocket.WebSocketApp(
        "ws://localhost:9001/",
        on_message=lambda _ws, _msg: client.on_message(_ws, _msg),
        on_error=lambda _ws, _err: client.on_error(_ws, _err),
        on_close=lambda _ws: client.on_close(_ws))

    ws.on_open = lambda _ws: client.on_open(_ws)
    ws.run_forever()


if __name__ == "__main__":
    main()


# EOF