Optimization Suggestions

[imtaizi2022-cpu]

1.Remove Heavy Libraries: No longer depend on the evdev library.Direct Memory Reading: Switch to directly reading the underlying binary event stream, skipping overheads such as object instantiation.
2.Read and Merge All Pending Mouse Movements in One Go: Package them into a single updated coordinate packet for transmission. If there are click operations, they will be sent immediately with priority to ensure zero-latency response for clicks.

Below is my own forwarding code.It can achieve a high response rate and ensure the mouse movement is highly responsive.

---

#!/usr/bin/env python3
import sys
import struct
import os
import time

Constants from linux/input-event-codes.h

EV_SYN = 0x00
EV_KEY = 0x01
EV_REL = 0x02

REL_X = 0x00
REL_Y = 0x01
REL_WHEEL = 0x08

BTN_LEFT = 0x110
BTN_RIGHT = 0x111
BTN_MIDDLE = 0x112

HID_DEV = "/dev/hidg0"

def find_mouse():
"""Finds a mouse device in /dev/input/ using simple string matching."""
try:
# Simple scan without evdev dependency
base = "/dev/input"
for handler in os.listdir(base):
if handler.startswith("event"):
path = os.path.join(base, handler)
try:
with open(os.path.join("/sys/class/input", handler, "device/name"), "r") as f:
name = f.read().lower()
if "mouse" in name:
return path
except:
continue
except Exception:
pass
return None

def main():
if len(sys.argv) > 1:
device_path = sys.argv[1]
else:
print("Scanning for mouse...")
device_path = find_mouse()
if not device_path:
print("No mouse found automatically. Usage: sudo python3 mouse_bridge.py /dev/input/eventX")
sys.exit(1)

print(f"Opening input device: {device_path}")
try:
    # Open as binary read-only
    in_fd = os.open(device_path, os.O_RDONLY)
except OSError as e:
    print(f"Failed to open device: {e}")
    sys.exit(1)

print(f"Opening HID gadget: {HID_DEV}")
try:
    # Unbuffered write
    hid_fd = os.open(HID_DEV, os.O_WRONLY)
except OSError as e:
    print(f"Failed to open {HID_DEV}. Is the gadget configured? Error: {e}")
    sys.exit(1)

# Grab device (ioctl EVIOCGRAB)
# _IOW('E', 0x90, int) -> 0x40044590
EVIOCGRAB = 0x40044590
try:
    import fcntl
    fcntl.ioctl(in_fd, EVIOCGRAB, 1)
except Exception as e:
    print(f"Warning: Could not grab device: {e}")

print("Forwarding started (Raw Mode). Press Ctrl+C to stop.")

buttons = 0
dx = 0
dy = 0
wheel = 0

# Event format: struct input_event
# info: 64-bit: time(16), type(2), code(2), value(4) = 24 bytes
# info: 32-bit: time(8), type(2), code(2), value(4) = 16 bytes
# We attempt to detect.

# Read one chunk to determine size? No, just assume 24 for Orange Pi 5 (ARM64).
EVENT_SIZE = 24
EVENT_FORMAT = 'llHHi' # long long, long long, unsigned short, unsigned short, signed int

try:
    while True:
        # Read a larger chunk of events (e.g. 64 events * 24 bytes = 1536 bytes)
        # Increased buffer size to catch more high-freq events for coalescing
        data = os.read(in_fd, 4096)
        if not data:
            break
        
        # Helper to send report
        def send_report():
            nonlocal buttons, dx, dy, wheel
            # Clamp values to int8 range
            wx = max(-127, min(127, dx))
            wy = max(-127, min(127, dy))
            ww = max(-127, min(127, wheel))

            # Only send if there is something to send (avoids zero-spam on empty loops if logic changes)
            # But here we only call send_report if we have data or buttons changed.
            
            report = struct.pack('Bbbb', buttons, wx, wy, ww)
            os.write(hid_fd, report)

            # Keep remainder (Negative Acceleration Fix)
            dx -= wx
            dy -= wy
            wheel -= ww

        # Iterate over the data
        for i in range(0, len(data), EVENT_SIZE):
            event_data = data[i:i+EVENT_SIZE]
            if len(event_data) < EVENT_SIZE:
                continue
            
            try:
                _, _, ev_type, ev_code, ev_val = struct.unpack(EVENT_FORMAT, event_data)
            except struct.error:
                continue

            if ev_type == EV_REL:
                if ev_code == REL_X:
                    dx += ev_val
                elif ev_code == REL_Y:
                    dy += ev_val
                elif ev_code == REL_WHEEL:
                    wheel += ev_val
            
            elif ev_type == EV_KEY:
                # Button state changing! 
                # We MUST flush pending motion/state BEFORE updating the button, 
                # OR update the button and flush immediately. 
                # Correct USB HID behavior: The report represents the state at that moment.
                # If we have pending motion and then a click, we can send them together.
                # BUT if we have "Click Press" then "Click Release" in the same chunk, we MUST separate them.
                
                # Strategy: Always flush BEFORE processing a button change to ensure the previous state is committed?
                # No, simply: Update button, separate reports if button changes again?
                
                # Simplest robust approach:
                # 1. Flush any pending motion/state. (Send what we have so far)
                if dx != 0 or dy != 0 or wheel != 0:
                    send_report()
                
                # 2. Update the button
                if ev_code == BTN_LEFT:
                    if ev_val: buttons |= 0x01
                    else:      buttons &= ~0x01
                elif ev_code == BTN_RIGHT:
                    if ev_val: buttons |= 0x02
                    else:      buttons &= ~0x02
                elif ev_code == BTN_MIDDLE:
                    if ev_val: buttons |= 0x04
                    else:      buttons &= ~0x04

                # 3. Flush AGAIN to send the new button state immediately (crucial for click latency)
                send_report()

            # Ignore EV_SYN (we coalesce automatically by buffer)

        # End of Read Chunk
        # Flush any remaining accumulated motion
        if dx != 0 or dy != 0 or wheel != 0:
            send_report()

except KeyboardInterrupt:
    print("\nStopping...")
except OSError as e:
    print(f"\nError loop: {e}")
finally:
    try:
        import fcntl
        fcntl.ioctl(in_fd, EVIOCGRAB, 0)
    except:
        pass
    os.close(hid_fd)
    os.close(in_fd)

if name == "main":
main()

[mcyork]

Okay, I like that code a lot. Where I have an issue is I don't love talking to the hardware directly so much, but the coalescing idea of events is phenomenal. And I think I can adopt that into the current code to bring clicks latency way down. I'll put this up on the dev tree later today if I can.