gamepad_client.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
import os
from typing import Callable, Optional

from gamepad_types import (
    AllocEngineDeviceCallback,
    DeviceIdCallback,
    InputDescriptor,
    LogCallback,
    PlatformConfigureFeaturesCallback,
    PlatformCreateHandleCallback,
    PlatformDetectCallback,
    PlatformInvalidateHandleCallback,
    PlatformProcessAudioHapticsCallback,
    PlatformReadCallback,
    PlatformWriteCallback,
)
from hid_platform_bridge import HidPlatformBridge


class GamepadHostClient:
    """Complete ctypes binding for the GamepadCoreHost native API."""

    def __init__(self, dll_path: str):
        self.dll_path = os.path.abspath(dll_path)
        self.dll: ctypes.CDLL
        self.connected_devices: set[int] = set()
        self._next_device_id = 0
        self._callbacks: list[object] = []
        self.on_device_connected: Optional[Callable[[int], None]] = None
        self.on_device_disconnected: Optional[Callable[[int], None]] = None

        self._load_dll()
        self._init_bindings()
        self._bridge = HidPlatformBridge(self._log)
        self._register_callbacks()

    def _load_dll(self) -> None:
        if not os.path.isfile(self.dll_path):
            raise FileNotFoundError(f"DLL not found at path: '{self.dll_path}'.")
        try:
            self.dll = ctypes.CDLL(self.dll_path)
        except OSError as error:
            raise OSError(f"Failed to load DLL '{self.dll_path}': {error}") from error

    def _bind(self, name: str, argtypes: list[object], restype: object) -> None:
        try:
            function = getattr(self.dll, name)
        except AttributeError as error:
            raise RuntimeError(f"The DLL does not export the required function '{name}'.") from error
        function.argtypes = argtypes
        function.restype = restype

    def _init_bindings(self) -> None:
        self._bind("GCH_SetLogCallback", [LogCallback], None)
        self._bind(
            "GCH_InitializePlatformBridge",
            [
                PlatformReadCallback, PlatformWriteCallback, PlatformDetectCallback,
                PlatformCreateHandleCallback, PlatformInvalidateHandleCallback,
                PlatformConfigureFeaturesCallback, PlatformProcessAudioHapticsCallback,
            ],
            None,
        )
        self._bind(
            "GCH_InitializeDeviceRegistryPolicy",
            [ctypes.c_int, AllocEngineDeviceCallback, DeviceIdCallback, DeviceIdCallback],
            None,
        )
        self._bind("GCH_DiscoverDevices", [ctypes.c_float], None)
        self._bind("GCH_UpdateInput", [ctypes.c_int, ctypes.c_float], None)
        self._bind("GCH_GetInputState", [ctypes.c_int, ctypes.POINTER(InputDescriptor)], ctypes.c_bool)
        self._bind("GCH_SetVibration", [ctypes.c_int, ctypes.c_ubyte, ctypes.c_ubyte], None)
        self._bind(
            "GCH_Lightbar", [ctypes.c_int, ctypes.c_ubyte, ctypes.c_ubyte, ctypes.c_ubyte], None
        )
        self._bind("GCH_UpdateOutput", [ctypes.c_int], None)
        self._bind(
            "GCH_CustomTrigger",
            [ctypes.c_int, ctypes.POINTER(ctypes.c_ubyte), ctypes.c_int, ctypes.c_int],
            ctypes.c_bool,
        )
        self._bind("GCH_StopTrigger", [ctypes.c_int, ctypes.c_int], None)
        self._bind("GCH_Shutdown", [], None)

    def _register_callbacks(self) -> None:
        @LogCallback
        def log_callback(_level: int, message: bytes | None) -> None:
            if message:
                self._log(message.decode("utf-8", errors="replace"))

        @AllocEngineDeviceCallback
        def allocate_device_id() -> int:
            self._next_device_id += 1
            self._bridge.bind_next_path_to_device(self._next_device_id)
            return self._next_device_id

        @DeviceIdCallback
        def dispatch_device(device_id: int) -> None:
            self._bridge.confirm_device_connected(device_id)
            self.connected_devices.add(device_id)
            if self.on_device_connected:
                self.on_device_connected(device_id)

        @DeviceIdCallback
        def disconnect_device(device_id: int) -> None:
            self._bridge.remove_device(device_id)
            self.connected_devices.discard(device_id)
            if self.on_device_disconnected:
                self.on_device_disconnected(device_id)

        callbacks = [
            log_callback, allocate_device_id, dispatch_device, disconnect_device,
            PlatformDetectCallback(self._bridge.detect_devices),
            PlatformReadCallback(self._bridge.read),
            PlatformWriteCallback(self._bridge.write),
            PlatformCreateHandleCallback(self._bridge.create_handle),
            PlatformInvalidateHandleCallback(self._bridge.invalidate_handle),
            PlatformConfigureFeaturesCallback(lambda *_: None),
            PlatformProcessAudioHapticsCallback(lambda *_: None),
        ]
        self._callbacks = callbacks
        self.dll.GCH_SetLogCallback(callbacks[0])
        self.dll.GCH_InitializePlatformBridge(
            callbacks[5], callbacks[6], callbacks[4], callbacks[7],
            callbacks[8], callbacks[9], callbacks[10],
        )
        self.dll.GCH_InitializeDeviceRegistryPolicy(0, callbacks[1], callbacks[2], callbacks[3])

    def _log(self, message: str) -> None:
        pass

    def discover_devices(self, delta_time: float = 0.016) -> None:
        self.dll.GCH_DiscoverDevices(delta_time)

    def update_input(self, device_id: int, delta_time: float = 0.0166) -> InputDescriptor | None:
        self.dll.GCH_UpdateInput(device_id, delta_time)
        state = InputDescriptor()
        return state if self.dll.GCH_GetInputState(device_id, ctypes.byref(state)) else None

    def set_basic_output(self, device_id: int, left: int, right: int, red: int, green: int, blue: int) -> None:
        self.dll.GCH_SetVibration(device_id, left, right)
        self.dll.GCH_Lightbar(device_id, red, green, blue)
        self.dll.GCH_UpdateOutput(device_id)

    def set_trigger_effect(self, device_id: int, effect: bytes, hand: int) -> bool:
        payload = (ctypes.c_ubyte * len(effect)).from_buffer_copy(effect)
        success = bool(self.dll.GCH_CustomTrigger(device_id, payload, len(effect), hand))
        self.dll.GCH_UpdateOutput(device_id)
        return success

    def stop_triggers(self, device_id: int) -> None:
        self.dll.GCH_StopTrigger(device_id, 0)
        self.dll.GCH_StopTrigger(device_id, 1)
        self.dll.GCH_UpdateOutput(device_id)

    def shutdown(self) -> None:
        self.dll.GCH_Shutdown()