Posts by cs301

#!/usr/bin/python3

import os
import sys
import time
import glob
import re
import struct
import fcntl
import signal
import select

VENDOR_ID = "0406"
PRODUCT_ID = "2814"

DEVICE_NAME_HINT = ""

ALLOW_REPEAT = {
    103,  # KEY_UP
    108,  # KEY_DOWN
    105,  # KEY_LEFT
    106,  # KEY_RIGHT
    104,  # KEY_PAGEUP
    109,  # KEY_PAGEDOWN
    102,  # KEY_HOME
    107,  # KEY_END
    114,  # KEY_VOLUMEDOWN
    115,  # KEY_VOLUMEUP
}

VENDOR_ID = VENDOR_ID.lower().replace("0x", "").zfill(4)
PRODUCT_ID = PRODUCT_ID.lower().replace("0x", "").zfill(4)

LOCK_FILE = "/storage/.config/remote-filter.lock"

EV_SYN = 0x00
EV_KEY = 0x01
EV_MSC = 0x04
MSC_SCAN = 0x04
BUS_VIRTUAL = 0x06

UINPUT_MAX_NAME_SIZE = 80
KEY_MAX = 0x2ff

_IOC_NRBITS = 8
_IOC_TYPEBITS = 8
_IOC_SIZEBITS = 14

_IOC_NRSHIFT = 0
_IOC_TYPESHIFT = _IOC_NRSHIFT + _IOC_NRBITS
_IOC_SIZESHIFT = _IOC_TYPESHIFT + _IOC_TYPEBITS
_IOC_DIRSHIFT = _IOC_SIZESHIFT + _IOC_SIZEBITS

_IOC_NONE = 0
_IOC_WRITE = 1


def _IOC(direction, type_, nr, size):
    return (
        (direction << _IOC_DIRSHIFT)
        | (ord(type_) << _IOC_TYPESHIFT)
        | (nr << _IOC_NRSHIFT)
        | (size << _IOC_SIZESHIFT)
    )


def _IO(type_, nr):
    return _IOC(_IOC_NONE, type_, nr, 0)


def _IOW(type_, nr, size):
    return _IOC(_IOC_WRITE, type_, nr, size)


INT_SIZE = struct.calcsize("i")

EVIOCGRAB = _IOW("E", 0x90, INT_SIZE)

UI_DEV_CREATE = _IO("U", 1)
UI_DEV_DESTROY = _IO("U", 2)
UI_SET_EVBIT = _IOW("U", 100, INT_SIZE)
UI_SET_KEYBIT = _IOW("U", 101, INT_SIZE)
UI_SET_MSCBIT = _IOW("U", 104, INT_SIZE)

EVENT_STRUCT = struct.Struct("llHHi")
EVENT_SIZE = EVENT_STRUCT.size

running = True


def log(message):
    print("[remote-filter] " + message, flush=True)


def signal_handler(signum, frame):
    global running
    running = False


signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)


def read_text(filename):
    try:
        with open(filename, "r") as f:
            return f.read().strip()
    except Exception:
        return ""


def get_event_ids(path):
    real = os.path.realpath(path)
    event = os.path.basename(real)
    base = "/sys/class/input/%s/device" % event

    vendor = read_text(base + "/id/vendor").lower()
    product = read_text(base + "/id/product").lower()
    name = read_text(base + "/name")

    return vendor, product, name


def find_device_by_id():
    candidates = []

    for path in glob.glob("/dev/input/by-id/*-event-kbd"):
        vendor, product, name = get_event_ids(path)

        if vendor != VENDOR_ID or product != PRODUCT_ID:
            continue

        if DEVICE_NAME_HINT:
            if DEVICE_NAME_HINT.lower() not in name.lower():
                continue

        candidates.append((path, name))

    if not candidates:
        return None

    candidates.sort()
    path, name = candidates[0]

    log("Keyboard found: %s (%s)" % (path, name))
    return path


def find_device_from_proc():
    try:
        with open("/proc/bus/input/devices", "r") as f:
            contents = f.read()
    except Exception:
        return None

    candidates = []

    for block in contents.split("\n\n"):

        id_match = re.search(
            r"Vendor=([0-9a-fA-F]{4})\s+Product=([0-9a-fA-F]{4})",
            block,
        )

        if not id_match:
            continue

        vendor = id_match.group(1).lower()
        product = id_match.group(2).lower()

        if vendor != VENDOR_ID or product != PRODUCT_ID:
            continue

        name_match = re.search(r'N:\s+Name="(.*?)"', block)
        name = name_match.group(1) if name_match else ""

        handlers_match = re.search(r"H:\s+Handlers=(.*)", block)

        if not handlers_match:
            continue

        handlers = handlers_match.group(1)

        if "kbd" not in handlers:
            continue

        event_match = re.search(r"\bevent([0-9]+)\b", handlers)

        if not event_match:
            continue

        if DEVICE_NAME_HINT:
            if DEVICE_NAME_HINT.lower() not in name.lower():
                continue

        path = "/dev/input/event" + event_match.group(1)
        lower_name = name.lower()

        penalty = 0

        if "consumer control" in lower_name:
            penalty += 100
        if "system control" in lower_name:
            penalty += 100
        if "mouse" in lower_name:
            penalty += 100

        candidates.append((penalty, path, name))

    if not candidates:
        return None

    candidates.sort()
    penalty, path, name = candidates[0]

    log("Keyboard found: %s (%s)" % (path, name))
    return path


def find_physical_device():
    path = find_device_by_id()

    if path:
        return path

    return find_device_from_proc()


def open_physical_keyboard():
    while running:

        device = find_physical_device()

        if not device:
            log(
                "No keyboard with USB ID %s:%s found"
                % (VENDOR_ID, PRODUCT_ID)
            )
            time.sleep(2)
            continue

        fd = None

        try:
            fd = os.open(device, os.O_RDONLY | os.O_NONBLOCK)
            fcntl.ioctl(fd, EVIOCGRAB, 1)

            log("Receiver grabbed: %s" % device)
            return fd

        except OSError as e:
            log("Device currently unavailable: %s" % e)

            if fd is not None:
                try:
                    os.close(fd)
                except Exception:
                    pass

            time.sleep(2)

    return None


def create_virtual_keyboard():
    try:
        ui = os.open(
            "/dev/uinput",
            os.O_WRONLY | os.O_NONBLOCK,
        )
    except OSError:
        ui = os.open(
            "/dev/input/uinput",
            os.O_WRONLY | os.O_NONBLOCK,
        )

    fcntl.ioctl(ui, UI_SET_EVBIT, EV_KEY)
    fcntl.ioctl(ui, UI_SET_EVBIT, EV_MSC)
    fcntl.ioctl(ui, UI_SET_MSCBIT, MSC_SCAN)

    for keycode in range(KEY_MAX + 1):
        try:
            fcntl.ioctl(ui, UI_SET_KEYBIT, keycode)
        except OSError:
            pass

    name = b"LibreELEC HID Repeat Filter"
    name += b"\0" * (UINPUT_MAX_NAME_SIZE - len(name))

    input_id = struct.pack(
        "HHHH",
        BUS_VIRTUAL,
        0x0000,
        0x0000,
        0x0001,
    )

    uidev = (
        name
        + input_id
        + struct.pack("I", 0)
        + struct.pack("64i", *([0] * 64))
        + struct.pack("64i", *([0] * 64))
        + struct.pack("64i", *([0] * 64))
        + struct.pack("64i", *([0] * 64))
    )

    os.write(ui, uidev)
    fcntl.ioctl(ui, UI_DEV_CREATE)

    time.sleep(0.3)

    log("Virtual keyboard created")
    return ui


def acquire_lock():
    lock_fd = os.open(
        LOCK_FILE,
        os.O_CREAT | os.O_RDWR,
        0o644,
    )

    try:
        fcntl.flock(
            lock_fd,
            fcntl.LOCK_EX | fcntl.LOCK_NB,
        )
    except OSError:
        log("Another filter instance is already running.")
        sys.exit(1)

    os.ftruncate(lock_fd, 0)
    os.write(lock_fd, str(os.getpid()).encode())

    return lock_fd


def forward_event(ui, data):
    try:
        os.write(ui, data)
    except OSError as e:
        log("Error forwarding event: %s" % e)


def run_filter():
    lock_fd = acquire_lock()

    ui = None
    physical = None
    blocked = 0

    try:
        while running:

            physical = open_physical_keyboard()

            if physical is None:
                break

            if ui is None:
                ui = create_virtual_keyboard()

            try:
                while running:

                    readable, _, _ = select.select(
                        [physical],
                        [],
                        [],
                        1.0,
                    )

                    if not readable:
                        continue

                    try:
                        data = os.read(
                            physical,
                            EVENT_SIZE * 64,
                        )
                    except BlockingIOError:
                        continue

                    if not data:
                        raise OSError("Input device disconnected")

                    offset = 0

                    while offset + EVENT_SIZE <= len(data):

                        event_data = data[
                            offset:
                            offset + EVENT_SIZE
                        ]

                        offset += EVENT_SIZE

                        (
                            sec,
                            usec,
                            ev_type,
                            code,
                            value,
                        ) = EVENT_STRUCT.unpack(event_data)

                        if (
                            ev_type == EV_KEY
                            and value == 2
                            and code not in ALLOW_REPEAT
                        ):
                            blocked += 1

                            log(
                                "Repeat discarded: "
                                "keycode=%d total=%d"
                                % (code, blocked)
                            )

                            continue

                        if ev_type in (EV_SYN, EV_KEY, EV_MSC):
                            forward_event(ui, event_data)

            except OSError as e:
                if running:
                    log("Receiver disconnected/lost: %s" % e)

            finally:
                if physical is not None:
                    try:
                        fcntl.ioctl(physical, EVIOCGRAB, 0)
                    except Exception:
                        pass

                    try:
                        os.close(physical)
                    except Exception:
                        pass

                    physical = None

            if running:
                time.sleep(1)

    finally:
        log("Filter stopping")

        if physical is not None:
            try:
                fcntl.ioctl(physical, EVIOCGRAB, 0)
            except Exception:
                pass

            try:
                os.close(physical)
            except Exception:
                pass

        if ui is not None:
            try:
                fcntl.ioctl(ui, UI_DEV_DESTROY)
            except Exception:
                pass

            try:
                os.close(ui)
            except Exception:
                pass

        try:
            os.close(lock_fd)
        except Exception:
            pass


if __name__ == "__main__":

    log(
        "Starting HID repeat filter for %s:%s"
        % (VENDOR_ID, PRODUCT_ID)
    )

    try:
        run_filter()
    except Exception as e:
        log("ERROR: %s" % e)
        sys.exit(1)