BLE Linux python-evdev joystick

joystick.py

from evdev import UInput, ecodes, AbsInfo
import asyncio
from bleak import BleakClient, BleakError

gearshift_uuid = "64bb31a7-432a-4d12-a646-efa0df31c789"
rotary_uuid = "4d9ef14c-e4f7-48c6-9cdc-8ded226b067d"
rumble_uuid = "798449fc-e2a1-4fa2-9e17-b77f87c98499"
accel_uuid = "f10d896b-a37c-43b7-b702-b324da1cfb99"
brake_uuid = "e62e34d0-3d83-4b9e-b3ff-0ab6f957a3be"
clutch_uuid = "c0a12cc1-302a-4847-819b-1e76a549611c"
speed_uuid = "831f3109-b19e-4d4f-abe7-84fb8b84ce83"
addy = "00:a0:50:c2:25:28"

re_amin = -740
re_amax = 740
re_abs_info = AbsInfo(value=0, min=re_amin, max=re_amax,
                      fuzz=0, flat=0, resolution=0)
ped_amin = 0
ped_amax = 255
ped_abs_info = AbsInfo(value=0, min=ped_amin, max=ped_amax,
                       fuzz=0, flat=0, resolution=0)
axis_info = (re_abs_info, ped_abs_info, ped_abs_info, ped_abs_info)
axies = (ecodes.ABS_X, ecodes.ABS_Y, ecodes.ABS_Z, ecodes.ABS_RX)
btns = (ecodes.BTN_A, ecodes.BTN_B, ecodes.BTN_X, ecodes.BTN_Y,
        ecodes.BTN_DPAD_DOWN, ecodes.BTN_DPAD_LEFT,
        ecodes.BTN_DPAD_RIGHT, ecodes.BTN_DPAD_UP)


async def rumble_ff(client, ui):
    async for e in ui.async_read_loop():
        if e.type != ecodes.EV_UINPUT:
            continue

        if e.code == ecodes.UI_FF_UPLOAD:
            upload = ui.begin_upload(e.value)
            upload.retval = 0
            await client.write_gatt_char(rumble_uuid, bytearray((1,)), False)
            await asyncio.sleep(.1)
            await client.write_gatt_char(rumble_uuid, bytearray((0,)), False)
            ui.end_upload(upload)

        elif e.code == ecodes.UI_FF_ERASE:
            erase = ui.begin_erase(e.value)
            erase.retval = 0
            ui.end_erase(erase)


async def async_main(ui):

    async def rotary_callback(sender, data):
        reval = int.from_bytes(data, "little", signed=True)
        if (reval > re_amax):
            reval = re_amax
        elif (reval < re_amin):
            reval = re_amin
        ui.write(ecodes.EV_ABS, ecodes.ABS_X, reval)
        ui.syn()

    async def gearshift_callback(sender, data):
        gsval = int.from_bytes(data, "little", signed=False)
        for i, btn in enumerate(btns):
            ui.write(ecodes.EV_KEY, btns[i], not (gsval & (1 << i)))
        ui.syn()

    async def accel_callback(sender, data):
        aval = int.from_bytes(data, "little", signed=False)
        ui.write(ecodes.EV_ABS, ecodes.ABS_Y, aval)
        ui.syn()

    async def brake_callback(sender, data):
        bval = int.from_bytes(data, "little", signed=False)
        ui.write(ecodes.EV_ABS, ecodes.ABS_Z, bval)
        ui.syn()

    async def clutch_callback(sender, data):
        cval = int.from_bytes(data, "little", signed=False)
        ui.write(ecodes.EV_ABS, ecodes.ABS_RX, cval)
        ui.syn()

    async with BleakClient(addy) as client:
        print("Connected")

        try:
            asyncio.create_task(client.start_notify(rotary_uuid,
                                                    rotary_callback))
            asyncio.create_task(client.start_notify(gearshift_uuid,
                                                    gearshift_callback))
            asyncio.create_task(client.start_notify(accel_uuid,
                                                    accel_callback))
            asyncio.create_task(client.start_notify(brake_uuid,
                                                    brake_callback))
            asyncio.create_task(client.start_notify(clutch_uuid,
                                                    clutch_callback))
            asyncio.create_task(rumble_ff(client, ui))
            await asyncio.Event().wait()
        except BleakError as e:
            print(f"Failed to read BLE device: {e}")


def main():
    cap = {ecodes.EV_FF: (ecodes.FF_CONSTANT,),
           ecodes.EV_KEY: btns,
           ecodes.EV_ABS: zip(axies, axis_info)}
    ui = UInput(cap, name='test-controller')
    asyncio.run(async_main(ui))
    return 0


if __name__ == "__main__":
    exit(main())