Listen for controller change events

Author: huntfx

mousetracks2/components/tracking.py

@@ -23,7 +23,7 @@
 from ..utils.monitor import MonitorData
 from ..utils.input import get_cursor_pos
 from ..utils.interface import Interfaces
-from ..utils.system import MonitorEventsListener, hide_child_process
+from ..utils.system import MonitorEventListener, ControllerEventListener, hide_child_process
 
 
 if XInput is None:
@@ -130,9 +130,12 @@ def __post_init__(self) -> None:
         self._pynput_mouse_listener.start()
         self._pynput_keyboard_listener.start()
 
-        self._monitor_listener = MonitorEventsListener()
+        self._monitor_listener = MonitorEventListener()
         self._monitor_listener.start()
 
+        self._controller_listener = ControllerEventListener()
+        self._controller_listener.start()
+
     def _receive_data(self) -> None:
         for message in self.receive_data():
             match message:
@@ -179,6 +182,11 @@ def _receive_data(self) -> None:
                 case ipc.SetGlobalGamepadTracking():
                     print(f'[Tracking] Tracking gamepad data: {message.enable}')
                     self.track_gamepad = message.enable
+                    if message.enable:
+                        self._controller_listener = ControllerEventListener()
+                        self._controller_listener.start()
+                    else:
+                        self._controller_listener.stop()
 
                 case ipc.SetGlobalNetworkTracking():
                     print(f'[Tracking] Tracking network data: {message.enable}')
@@ -197,7 +205,7 @@ def _receive_data(self) -> None:
                     if message.disable:
                         self._monitor_listener.stop()
                     else:
-                        self._monitor_listener = MonitorEventsListener()
+                        self._monitor_listener = MonitorEventListener()
                         self._monitor_listener.start()
 
     def _run_with_state(self) -> Iterator[tuple[int, DataState]]:
@@ -441,9 +449,8 @@ def run(self) -> None:
 
             # Determine which gamepads are connected
             if self.track_gamepad and XInput is not None:
-                if not tick % int(UPDATES_PER_SECOND * GlobalConfig.gamepad_check_frequency) or data.gamepad_force_recheck:
+                if self._controller_listener.triggered:
                     data.gamepads_current = XInput.get_connected()
-                    data.gamepad_force_recheck = False
 
                     if data.gamepads_current != data.gamepads_previous:
                         print('[Tracking] Gamepad change detected')

mousetracks2/config.py

@@ -26,7 +26,6 @@ class GlobalConfig:
         save_frequency: How often to autosave.
         max_loaded_profiles: Maximum amount of loaded profiles.
             This will only affect profiles without unsaved changes.
-        gamepad_check_frequency: How often to check the connected gamepads.
         application_check_frequency: How often to check the current focused application.
         component_check_frequency: How often to check all components are running.
             This is used once per message received.
@@ -46,7 +45,6 @@ class GlobalConfig:
     inactivity_time: float = 300.0
     save_frequency: float = 600.0
     max_loaded_profiles: int = 8
-    gamepad_check_frequency: float = 1.0
     application_check_frequency: float = 1.0
     component_check_frequency: float = 1.0
     shutdown_timeout: float = 15.0

mousetracks2/utils/system/__init__.py

@@ -3,12 +3,13 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Type
 
-from .base import Window as _Window, MonitorEventsListener as _MonitorEventsListener
+from . import base
 from ...constants import APP_EXECUTABLE, IS_BUILT_EXE
 
 if TYPE_CHECKING:
-    Window: Type[_Window]
-    MonitorEventsListener: Type[_MonitorEventsListener]
+    Window: Type[base.Window]
+    MonitorEventListener: Type[base.EventListener]
+    ControllerEventListener: Type[base.EventListener]
 
 match sys.platform:
     case 'win32':
@@ -17,7 +18,7 @@
         from .windows import get_autostart, set_autostart, remove_autostart
         from .windows import is_elevated, relaunch_as_elevated
         from .windows import Window
-        from .windows import MonitorEventsListener
+        from .windows import MonitorEventListener, ControllerEventListener
         from .base import hide_child_process
         from .windows import prepare_application_icon
         from .windows import update_installer_version_number
@@ -28,7 +29,7 @@
         from .macos import get_autostart, set_autostart, remove_autostart
         from .base import is_elevated, relaunch_as_elevated
         from .macos import Window
-        from .base import MonitorEventsListener
+        from .base import MonitorEventListener, ControllerEventListener
         from .macos import hide_child_process, prepare_application_icon
         from .base import update_installer_version_number
 
@@ -38,7 +39,7 @@
         from .linux import get_autostart, set_autostart, remove_autostart
         from .base import is_elevated, relaunch_as_elevated
         from .linux import Window
-        from .base import MonitorEventsListener
+        from .base import MonitorEventListener, ControllerEventListener
         from .base import hide_child_process, prepare_application_icon
         from .base import update_installer_version_number
 
@@ -48,7 +49,7 @@
     'get_autostart', 'set_autostart', 'remove_autostart', 'remap_autostart',
     'is_elevated', 'relaunch_as_elevated',
     'Window',
-    'MonitorEventsListener',
+    'MonitorEventListener', 'ControllerEventListener',
     'hide_child_process', 'prepare_application_icon',
     'update_installer_version_number',
 ]

mousetracks2/utils/system/base.py

@@ -88,17 +88,17 @@ def size(self) -> tuple[int, int]:
         return (0, 0)
 
 
-class MonitorEventsListener(threading.Thread):
-    """Listen for monitor change events.
+class EventListener(threading.Thread):
+    """Base class to listen for events.
 
     The most basic implementation is to check every second for changes.
-    If an operating system has hooks then this class can be subclassed.
+    If an operating system has hooks then this should be overridden.
 
     The initial event is triggered on startup.
     """
 
     def __init__(self) -> None:
-        super().__init__(name='MonitorEventsListener', daemon=True)
+        super().__init__(name='EventListener', daemon=True)
         self._queue = queue.Queue()  # type: queue.Queue[None]
         self._running = True
 
@@ -127,6 +127,14 @@ def triggered(self) -> bool:
             count += 1
 
 
+class MonitorEventListener(EventListener):
+    """Listen for monitor change events."""
+
+
+class ControllerEventListener(EventListener):
+    """Listen for controller change events."""
+
+
 def hide_child_process() -> None:
     """This is here to allow macOS to hide the child processes."""
 

mousetracks2/utils/system/windows.py

@@ -13,7 +13,7 @@
 import subprocess
 import winreg
 
-from .base import Window as _Window, MonitorEventsListener as _MonitorEventsListener
+from . import base
 from ...constants import APP_EXECUTABLE, PACKAGE_IDENTIFIER
 from ...types import Rect, RectList
 from ...version import VERSION
@@ -49,6 +49,8 @@
 
 PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
 
+DBT_DEVNODES_CHANGED = 0x0007
+
 BOOL = ctypes.wintypes.BOOL
 
 DWORD = ctypes.wintypes.DWORD
@@ -415,7 +417,7 @@ def relaunch_as_elevated() -> None:
     sys.exit()
 
 
-class Window(_Window):
+class Window(base.Window):
     def __init__(self, hwnd: int) -> None:
         self._hwnd = hwnd
         self._handle = WindowHandle(self._hwnd)
@@ -454,24 +456,31 @@ def size(self) -> tuple[int, int]:
         return self._pid.size
 
 
-class MonitorEventsListener(_MonitorEventsListener):
-    """Listen for monitor change events."""
+class EventListener(base.EventListener):
+    """Base Windows event listener.
+
+    Override the `check` method to implement this.
+    """
 
     def __init__(self) -> None:
         super().__init__()
         self._hwnd = None  # type: int | None
 
+    def check(self, hwnd: int, msg: int, wparam: int, lparam: int) -> bool:
+        """Determine if a specific event has been fired."""
+        return False
+
     def run(self) -> None:
         """Create and start the message listener."""
         def wndproc(hwnd: int, msg: int, wparam: int, lparam: int) -> int:
-            if msg in (WM_DISPLAYCHANGE, WM_DEVICECHANGE):
+            if self.check(hwnd, msg, wparam, lparam):
                 self.trigger()
             return user32.DefWindowProcW(hwnd, msg, wparam, lparam)
 
         hinst = kernel32.GetModuleHandleW(None)
         wndproc_c = WNDPROCTYPE(wndproc)
 
-        class_name = 'MouseTracksHiddenWindowClass'
+        class_name = type(self).__name__
         wc = WNDCLASS()
         wc.lpfnWndProc = wndproc_c
         wc.lpszClassName = class_name
@@ -501,6 +510,20 @@ def stop(self) -> None:
             user32.PostMessageW(self._hwnd, WM_QUIT, 0, 0)
 
 
+class MonitorEventListener(EventListener):
+    """Listen for monitor change events."""
+
+    def check(self, hwnd: int, msg: int, wparam: int, lparam: int) -> bool:
+        return msg == WM_DISPLAYCHANGE
+
+
+class ControllerEventListener(EventListener):
+    """Listen for controller change events."""
+
+    def check(self, hwnd: int, msg: int, wparam: int, lparam: int) -> bool:
+        return msg == WM_DEVICECHANGE and wparam == DBT_DEVNODES_CHANGED
+
+
 def prepare_application_icon(icon_path: Path | str) -> None:
     """Register app so that setting an icon is possible."""
     ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID(PACKAGE_IDENTIFIER)