VirtualInputDevice

def __init__(self, name: str, vendor: int, product: int,
             evbits: Iterable[int], keybits: Iterable[int] = (),
             relbits: Iterable[int] = (), sock_dir: str = "/tmp") -> None:
    self.name = name
    self.vendor, self.product = vendor, product
    self.evbits, self.keybits, self.relbits = list(evbits), list(keybits), list(relbits)
    self.sock_dir = sock_dir
    self.fd: Optional[int] = None
    self.event_num: Optional[int] = None
    self.server: Optional[asyncio.AbstractServer] = None
    self.clients: dict = {}
    self.node: Optional[str] = None

def _create_kernel(self) -> bool:
    """Register the device with the kernel; False when uinput refuses it."""
    try:
        fd = os.open(UINPUT_PATH, os.O_WRONLY | os.O_NONBLOCK)
    except OSError:
        return False
    try:
        for ev in self.evbits + [EV_SYN]:
            fcntl.ioctl(fd, UI_SET_EVBIT, ev)
        for code in self.keybits:
            fcntl.ioctl(fd, UI_SET_KEYBIT, code)
        for code in self.relbits:
            fcntl.ioctl(fd, UI_SET_RELBIT, code)
        fcntl.ioctl(fd, UI_DEV_SETUP, struct.pack(
            UINPUT_SETUP_FMT, BUS_VIRTUAL, self.vendor, self.product, 1,
            self.name.encode("utf-8")[:UINPUT_MAX_NAME_SIZE - 1], 0))
        fcntl.ioctl(fd, UI_DEV_CREATE)
    except OSError as e:
        os.close(fd)
        logger_webrtc_input.warning(f"{self.name}: kernel uinput setup failed ({e}).")
        return False
    self.fd = fd
    self.node = self._kernel_node() or UINPUT_PATH
    return True

def _kernel_node(self) -> Optional[str]:
    """The /dev/input node the kernel registered for this device, which is
    what an application opens; None when sysfs does not name one."""
    buffer = bytearray(UINPUT_SYSNAME_LEN)
    try:
        fcntl.ioctl(self.fd, UI_GET_SYSNAME, buffer, True)
    except OSError:
        return None
    sysname = bytes(buffer).split(b"\0", 1)[0].decode("utf-8", "replace")
    if not sysname:
        return None
    try:
        entries = os.listdir(os.path.join(UINPUT_SYSFS_BASE, sysname))
    except OSError:
        return None
    events = sorted(e for e in entries if e.startswith("event"))
    return os.path.join("/dev/input", events[0]) if events else None

def _stale(self, path: str) -> bool:
    """Whether a socket file is left over from a device nobody serves."""
    probe = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    try:
        probe.connect(path)
        return False
    except OSError:
        return True
    finally:
        probe.close()

async def _create_interposed(self) -> bool:
    """Claim a dynamic node: descriptor first, so a reader that sees the
    socket always finds the identity behind it."""
    desc = pack_udyn_desc(self.name, self.vendor, self.product, 1,
                          self.evbits, self.keybits, self.relbits)
    os.makedirs(self.sock_dir, exist_ok=True)
    for num in range(UDYN_EVENT_BASE, UDYN_EVENT_BASE + UDYN_MAX):
        sock = os.path.join(self.sock_dir, f"selkies_event{num}.sock")
        if os.path.exists(sock):
            if not self._stale(sock):
                continue
            try:
                os.unlink(sock)
            except OSError:
                continue
        desc_path = os.path.join(self.sock_dir, f"selkies_event{num}.desc")
        tmp = f"{desc_path}.tmp.{os.getpid()}"
        try:
            with open(tmp, "wb") as fh:
                fh.write(desc)
            os.replace(tmp, desc_path)
            self.server = await asyncio.start_unix_server(self._serve, path=sock)
        except OSError as e:
            logger_webrtc_input.debug(f"{self.name}: node {num} unavailable ({e}).")
            continue
        self.event_num, self.node = num, f"/dev/input/event{num}"
        return True
    logger_webrtc_input.warning(f"{self.name}: no free interposer node.")
    return False

async def _serve(self, reader: asyncio.StreamReader,
                 writer: asyncio.StreamWriter) -> None:
    """Hold one reader's connection open; the dynamic path carries events
    alone, the descriptor having already answered for the identity."""
    self.clients[writer] = True
    try:
        while self.server is not None and not writer.is_closing():
            if not await reader.read(256):
                break
    except (ConnectionResetError, BrokenPipeError, OSError):
        pass
    finally:
        self.clients.pop(writer, None)
        if not writer.is_closing():
            writer.close()

async def open(self, kernel: bool = True) -> bool:
    """Bring the device up on whichever backend this host offers; `kernel`
    off keeps it on the interposer pool where the kernel's copy would be
    read back by the display server."""
    ok = self._create_kernel() if kernel and uinput_writable() else await self._create_interposed()
    if ok:
        logger_webrtc_input.info(
            f"{self.name} available at {self.node} "
            f"({'kernel' if self.fd is not None else 'interposer'}).")
    return ok

def emit(self, ev_type: int, ev_code: int, value: int) -> None:
    """Deliver one event and the SYN_REPORT that closes its report."""
    data = get_evdev_events_packed(ev_type, ev_code, value, LOCAL_ARCH_BITS)
    if self.fd is not None:
        try:
            os.write(self.fd, data)
        except OSError as e:
            logger_webrtc_input.debug(f"{self.name}: kernel write failed ({e}).")
        return
    for writer in list(self.clients):
        try:
            writer.write(data)
        except (OSError, RuntimeError):
            self.clients.pop(writer, None)

async def close(self) -> None:
    """Retire the device and remove everything it published."""
    if self.fd is not None:
        try:
            fcntl.ioctl(self.fd, UI_DEV_DESTROY)
        except OSError:
            pass
        os.close(self.fd)
        self.fd = None
    if self.server is not None:
        self.server.close()
        await self.server.wait_closed()
        self.server = None
    for writer in list(self.clients):
        if not writer.is_closing():
            writer.close()
    self.clients.clear()
    if self.event_num is not None:
        for suffix in (".sock", ".desc"):
            try:
                os.unlink(os.path.join(self.sock_dir, f"selkies_event{self.event_num}{suffix}"))
            except OSError:
                pass
        self.event_num = None