hid_platform_bridge.py

# Project: G-Client-Python
# This project is a Python client for Gamepad-Core-Host.
# Copyright (c) 2026 valoto.games. All rights reserved.

import ctypes
from collections import deque
from typing import Callable

from gamepad_types import (
    BLUETOOTH_GUID,
    ConnectionType,
    DESCRIPTOR_SIZE,
    DeviceDescriptor,
    DeviceType,
    VENDOR_ID,
)


INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value
DIGCF_PRESENT = 0x00000002
DIGCF_DEVICEINTERFACE = 0x00000010
GENERIC_READ = 0x80000000
GENERIC_WRITE = 0x40000000
FILE_SHARE_READ = 0x00000001
FILE_SHARE_WRITE = 0x00000002
OPEN_EXISTING = 3
ERROR_NO_MORE_ITEMS = 259


class GUID(ctypes.Structure):
    _fields_ = [
        ("Data1", ctypes.c_ulong),
        ("Data2", ctypes.c_ushort),
        ("Data3", ctypes.c_ushort),
        ("Data4", ctypes.c_ubyte * 8),
    ]


class SP_DEVICE_INTERFACE_DATA(ctypes.Structure):
    _fields_ = [
        ("cbSize", ctypes.c_ulong),
        ("InterfaceClassGuid", GUID),
        ("Flags", ctypes.c_ulong),
        ("Reserved", ctypes.c_void_p),
    ]


class HIDD_ATTRIBUTES(ctypes.Structure):
    _fields_ = [
        ("Size", ctypes.c_ulong),
        ("VendorID", ctypes.c_ushort),
        ("ProductID", ctypes.c_ushort),
        ("VersionNumber", ctypes.c_ushort),
    ]


class HidPlatformBridge:
    """Windows HID implementation supplied to GamepadCoreHost callbacks."""

    _product_types = {
        0x05C4: DeviceType.DualShock4,
        0x09CC: DeviceType.DualShock4,
        0x0CE6: DeviceType.DualSense,
        0x0DF2: DeviceType.DualSenseEdge,
    }

    def __init__(self, log: Callable[[str], None]) -> None:
        if not hasattr(ctypes, "WinDLL"):
            raise OSError("The HID platform bridge is only supported on Windows.")

        self._log = log
        self._known_paths: set[str] = set()
        self._pending_paths: deque[str] = deque()
        self._device_paths: dict[int, str] = {}
        self._handles: dict[int, str] = {}

        self._hid = ctypes.WinDLL("hid.dll")
        self._setupapi = ctypes.WinDLL("setupapi.dll")
        self._kernel32 = ctypes.WinDLL("kernel32.dll", use_last_error=True)
        self._configure_native_apis()

    def _configure_native_apis(self) -> None:
        self._hid.HidD_GetHidGuid.argtypes = [ctypes.POINTER(GUID)]
        self._hid.HidD_GetHidGuid.restype = None
        self._hid.HidD_GetAttributes.argtypes = [ctypes.c_void_p, ctypes.POINTER(HIDD_ATTRIBUTES)]
        self._hid.HidD_GetAttributes.restype = ctypes.c_bool

        self._setupapi.SetupDiGetClassDevsW.argtypes = [
            ctypes.POINTER(GUID), ctypes.c_wchar_p, ctypes.c_void_p, ctypes.c_ulong
        ]
        self._setupapi.SetupDiGetClassDevsW.restype = ctypes.c_void_p
        self._setupapi.SetupDiEnumDeviceInterfaces.argtypes = [
            ctypes.c_void_p, ctypes.c_void_p, ctypes.POINTER(GUID), ctypes.c_ulong,
            ctypes.POINTER(SP_DEVICE_INTERFACE_DATA),
        ]
        self._setupapi.SetupDiEnumDeviceInterfaces.restype = ctypes.c_bool
        self._setupapi.SetupDiGetDeviceInterfaceDetailW.argtypes = [
            ctypes.c_void_p, ctypes.POINTER(SP_DEVICE_INTERFACE_DATA), ctypes.c_void_p,
            ctypes.c_ulong, ctypes.POINTER(ctypes.c_ulong), ctypes.c_void_p,
        ]
        self._setupapi.SetupDiGetDeviceInterfaceDetailW.restype = ctypes.c_bool
        self._setupapi.SetupDiDestroyDeviceInfoList.argtypes = [ctypes.c_void_p]
        self._setupapi.SetupDiDestroyDeviceInfoList.restype = ctypes.c_bool

        self._kernel32.CreateFileW.argtypes = [
            ctypes.c_wchar_p, ctypes.c_ulong, ctypes.c_ulong, ctypes.c_void_p,
            ctypes.c_ulong, ctypes.c_ulong, ctypes.c_void_p,
        ]
        self._kernel32.CreateFileW.restype = ctypes.c_void_p
        self._kernel32.CloseHandle.argtypes = [ctypes.c_void_p]
        self._kernel32.CloseHandle.restype = ctypes.c_bool
        self._kernel32.ReadFile.argtypes = [
            ctypes.c_void_p, ctypes.c_void_p, ctypes.c_ulong, ctypes.POINTER(ctypes.c_ulong),
            ctypes.c_void_p,
        ]
        self._kernel32.ReadFile.restype = ctypes.c_bool
        self._kernel32.WriteFile.argtypes = self._kernel32.ReadFile.argtypes
        self._kernel32.WriteFile.restype = ctypes.c_bool

    def detect_devices(self, descriptors: int, max_devices: int) -> int:
        hid_guid = GUID()
        self._hid.HidD_GetHidGuid(ctypes.byref(hid_guid))
        device_info_set = self._setupapi.SetupDiGetClassDevsW(
            ctypes.byref(hid_guid), None, None, DIGCF_PRESENT | DIGCF_DEVICEINTERFACE
        )
        if device_info_set == INVALID_HANDLE_VALUE:
            self._log(f"Unable to enumerate HID devices (error {ctypes.get_last_error()}).")
            return 0

        found = 0
        try:
            index = 0
            while found < max_devices:
                interface_data = SP_DEVICE_INTERFACE_DATA()
                interface_data.cbSize = ctypes.sizeof(SP_DEVICE_INTERFACE_DATA)
                if not self._setupapi.SetupDiEnumDeviceInterfaces(
                    device_info_set, None, ctypes.byref(hid_guid), index, ctypes.byref(interface_data)
                ):
                    if ctypes.get_last_error() != ERROR_NO_MORE_ITEMS:
                        self._log(f"HID enumeration failed (error {ctypes.get_last_error()}).")
                    break
                index += 1
                path = self._get_device_path(device_info_set, interface_data)
                if path is None:
                    continue
                device_type = self._get_device_type(path)
                if device_type is None:
                    continue

                descriptor = DeviceDescriptor()
                descriptor.DeviceType = device_type
                descriptor.ConnectionType = (
                    ConnectionType.Bluetooth
                    if BLUETOOTH_GUID.lower() in path.lower()
                    else ConnectionType.Usb
                )
                descriptor.IsConnected = 1
                encoded_path = path.encode("ascii", errors="ignore")[:511]
                descriptor.Path[:len(encoded_path)] = encoded_path
                ctypes.memmove(descriptors + found * DESCRIPTOR_SIZE, ctypes.byref(descriptor), DESCRIPTOR_SIZE)
                found += 1
        finally:
            self._setupapi.SetupDiDestroyDeviceInfoList(device_info_set)
        return found

    def _get_device_path(self, device_info_set: int, interface_data: SP_DEVICE_INTERFACE_DATA) -> str | None:
        required_size = ctypes.c_ulong()
        self._setupapi.SetupDiGetDeviceInterfaceDetailW(
            device_info_set, ctypes.byref(interface_data), None, 0, ctypes.byref(required_size), None
        )
        if required_size.value == 0:
            return None
        detail = ctypes.create_string_buffer(required_size.value)
        ctypes.cast(detail, ctypes.POINTER(ctypes.c_ulong)).contents.value = 8
        if not self._setupapi.SetupDiGetDeviceInterfaceDetailW(
            device_info_set, ctypes.byref(interface_data), detail, required_size.value,
            ctypes.byref(required_size), None,
        ):
            return None
        return ctypes.wstring_at(ctypes.addressof(detail) + ctypes.sizeof(ctypes.c_ulong))

    def _get_device_type(self, path: str) -> DeviceType | None:
        handle = self._open_handle(path)
        if handle == INVALID_HANDLE_VALUE:
            return None
        try:
            attributes = HIDD_ATTRIBUTES()
            attributes.Size = ctypes.sizeof(HIDD_ATTRIBUTES)
            if not self._hid.HidD_GetAttributes(handle, ctypes.byref(attributes)):
                return None
            if attributes.VendorID != VENDOR_ID:
                return None
            return self._product_types.get(attributes.ProductID)
        finally:
            self._kernel32.CloseHandle(handle)

    def create_handle(self, descriptor_pointer: int) -> bool:
        descriptor = ctypes.cast(descriptor_pointer, ctypes.POINTER(DeviceDescriptor)).contents
        path = descriptor.path_str
        if not path:
            return False
        handle = self._open_handle(path)
        if handle == INVALID_HANDLE_VALUE:
            return False
        if path in self._known_paths:
            self._kernel32.CloseHandle(handle)
            return False

        self._known_paths.add(path)
        self._pending_paths.append(path)
        self._handles[int(handle)] = path
        descriptor.Handle = int(handle)
        descriptor.IsConnected = 1
        return True

    def bind_next_path_to_device(self, device_id: int) -> None:
        if self._pending_paths:
            self._device_paths[device_id] = self._pending_paths.popleft()

    def confirm_device_connected(self, device_id: int) -> None:
        path = self._device_paths.get(device_id)
        if path:
            self._log(f"Device connected: {device_id} ({path})")

    def remove_device(self, device_id: int) -> None:
        path = self._device_paths.pop(device_id, None)
        if path:
            self._known_paths.discard(path)

    def read(self, handle: int, buffer: int, length: int, bytes_read: int) -> bool:
        count = ctypes.c_ulong()
        success = self._kernel32.ReadFile(handle, buffer, length, ctypes.byref(count), None)
        if bytes_read:
            ctypes.cast(bytes_read, ctypes.POINTER(ctypes.c_ulong)).contents.value = count.value

        if not success:
            error_code = ctypes.get_last_error()
            if error_code != 0:  # Ignore error code 0 (no error)
                self._log(f"ReadFile failed with error code: {error_code}")
            self._remove_device_for_handle(handle)
        return bool(success)

    def write(self, handle: int, buffer: int, length: int, bytes_written: int) -> bool:
        count = ctypes.c_ulong()
        success = self._kernel32.WriteFile(handle, buffer, length, ctypes.byref(count), None)
        if bytes_written:
            ctypes.cast(bytes_written, ctypes.POINTER(ctypes.c_ulong)).contents.value = count.value
        return bool(success)

    def invalidate_handle(self, handle: int) -> None:
        if handle == INVALID_HANDLE_VALUE:
            return
        path = self._handles.pop(handle, None)
        if path is None:
            return
        self._known_paths.discard(path)
        self._kernel32.CloseHandle(handle)

    def _remove_device_for_handle(self, handle: int) -> None:
        path = self._handles.get(handle)
        if path is None:
            return

        self.invalidate_handle(handle)
        for device_id, device_path in tuple(self._device_paths.items()):
            if device_path == path:
                self.remove_device(device_id)
                return

    def _open_handle(self, path: str) -> int:
        return int(self._kernel32.CreateFileW(
            path, GENERIC_READ | GENERIC_WRITE, FILE_SHARE_READ | FILE_SHARE_WRITE,
            None, OPEN_EXISTING, 0, None,
        ))