Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 18 additions & 6 deletions core/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
BUTTON_TO_EVENTS, GESTURE_DIRECTION_BUTTONS, save_config,
)
from core.app_detector import AppDetector
from core.mouse_hook_types import HidRuntimeState
from core.linux_permissions import (
linux_permission_log_message,
linux_permission_report,
Expand Down Expand Up @@ -58,7 +59,7 @@ def __init__(self):
)
self._battery_poll_stop = threading.Event()
self._battery_poll_thread = None # track the poller thread
self._last_connection_state = bool(self.hook.device_connected)
self._last_connection_state = bool(self._hid_runtime_state().input_ready)
self._last_hid_features_ready = bool(self.hid_features_ready)
self._hid_replay_requested_this_launch = False
self._replay_inflight = False
Expand All @@ -78,6 +79,18 @@ def __init__(self):
except Exception as e:
print(f"[Engine] Failed to set DPI: {e}")

def _hid_runtime_state(self):
state = getattr(self.hook, "hid_runtime_state", None)
if state is not None:
return state
hg = getattr(self.hook, "_hid_gesture", None)
hid_device = getattr(hg, "connected_device", None) if hg else None
return HidRuntimeState(
input_ready=bool(getattr(self.hook, "device_connected", False)),
hid_ready=hid_device is not None,
connected_device=getattr(self.hook, "connected_device", None),
)

# ------------------------------------------------------------------
# Hook wiring
# ------------------------------------------------------------------
Expand Down Expand Up @@ -676,25 +689,24 @@ def set_connection_change_callback(self, cb):
self._connection_change_cb = cb
if cb:
try:
cb(bool(self.hook.device_connected))
cb(bool(self._hid_runtime_state().input_ready))
except Exception:
pass

@property
def device_connected(self):
return self.hook.device_connected
return self._hid_runtime_state().input_ready

@property
def connected_device(self):
return getattr(self.hook, "connected_device", None)
return self._hid_runtime_state().connected_device

def dump_device_info(self):
return getattr(self.hook, "dump_device_info", lambda: None)()

@property
def hid_features_ready(self):
hg = self.hook._hid_gesture
return hg is not None and getattr(hg, "connected_device", None) is not None
return self._hid_runtime_state().hid_ready

@property
def enabled(self):
Expand Down
12 changes: 11 additions & 1 deletion core/mouse_hook_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
except Exception:
HidGestureListener = None

from core.mouse_hook_types import MouseEvent, format_debug_details
from core.mouse_hook_types import HidRuntimeState, MouseEvent, format_debug_details


class BaseMouseHook:
Expand Down Expand Up @@ -84,6 +84,16 @@ def device_connected(self):
def connected_device(self):
return self._connected_device

@property
def hid_runtime_state(self):
hg = getattr(self, "_hid_gesture", None)
hid_device = getattr(hg, "connected_device", None) if hg else None
return HidRuntimeState(
input_ready=bool(self._device_connected),
hid_ready=hid_device is not None,
connected_device=self._connected_device,
)

def dump_device_info(self):
hg = getattr(self, "_hid_gesture", None)
if hg and hasattr(hg, "dump_device_info"):
Expand Down
4 changes: 4 additions & 0 deletions core/mouse_hook_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@

from typing import Any, Callable, Protocol, runtime_checkable

from core.mouse_hook_types import HidRuntimeState


@runtime_checkable
class MouseHookLike(Protocol):
Expand Down Expand Up @@ -33,6 +35,8 @@ def set_gesture_callback(self, callback: Callable[[Any], None]) -> None: ...
def device_connected(self) -> bool: ...
@property
def connected_device(self) -> Any: ...
@property
def hid_runtime_state(self) -> HidRuntimeState: ...
def dump_device_info(self) -> Any: ...
def start(self) -> bool: ...
def stop(self) -> None: ...
12 changes: 11 additions & 1 deletion core/mouse_hook_linux.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@
resolve_device as _resolve_logi_device,
)
from core.mouse_hook_base import BaseMouseHook, HidGestureListener
from core.mouse_hook_types import MouseEvent
from core.mouse_hook_types import HidRuntimeState, MouseEvent

_LOGI_VENDOR = 0x046D
_LOG_ONCE_KEYS = set()
Expand Down Expand Up @@ -112,6 +112,16 @@ def evdev_ready(self):
def hid_ready(self):
return self._hid_ready

@property
def hid_runtime_state(self):
hg = getattr(self, "_hid_gesture", None)
hid_device = getattr(hg, "connected_device", None) if hg else None
return HidRuntimeState(
input_ready=bool(self._device_connected),
hid_ready=bool(self._hid_ready and hid_device is not None),
connected_device=self._connected_device,
)

def _set_evdev_ready(self, ready):
if ready == self._evdev_ready:
return
Expand Down
11 changes: 11 additions & 0 deletions core/mouse_hook_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,18 @@
Shared mouse hook types and helpers.
"""

from dataclasses import dataclass
import time
from typing import Any


@dataclass(frozen=True)
class HidRuntimeState:
"""Read-only snapshot of hook input and HID++ readiness."""

input_ready: bool = False
hid_ready: bool = False
connected_device: Any = None


class MouseEvent:
Expand Down
21 changes: 21 additions & 0 deletions tests/test_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@

from core.config import DEFAULT_CONFIG
from core.mouse_hook import MouseEvent
from core.mouse_hook_types import HidRuntimeState


class _FakeMouseHook:
Expand Down Expand Up @@ -180,6 +181,26 @@ def test_hid_features_ready_requires_hid_identity(self):
)
self.assertTrue(engine.hid_features_ready)

def test_engine_projection_prefers_hid_runtime_state(self):
engine = self._make_engine()
device = SimpleNamespace(name="MX Master 3S")
engine.hook.device_connected = False
engine.hook.connected_device = SimpleNamespace(name="stale fallback")
engine.hook._hid_gesture = None
engine.hook.hid_runtime_state = HidRuntimeState(
input_ready=True,
hid_ready=True,
connected_device=device,
)

seen = []
engine.set_connection_change_callback(seen.append)

self.assertTrue(engine.device_connected)
self.assertIs(engine.connected_device, device)
self.assertTrue(engine.hid_features_ready)
self.assertEqual(seen, [True])

def test_duplicate_connected_refresh_does_not_restart_battery_poller(self):
engine = self._make_engine()
seen = []
Expand Down
51 changes: 51 additions & 0 deletions tests/test_mouse_hook.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@
from unittest.mock import Mock, MagicMock, call, patch

from core import mouse_hook
from core.mouse_hook_base import BaseMouseHook
from core.mouse_hook_types import HidRuntimeState


class _FakeEvdevDevice:
Expand Down Expand Up @@ -62,6 +64,25 @@ def from_device(*_args, **_kwargs):
return Mock()


class BaseMouseHookRuntimeStateTests(unittest.TestCase):
def test_default_runtime_state_is_disconnected(self):
hook = BaseMouseHook()

self.assertEqual(hook.hid_runtime_state, HidRuntimeState())

def test_runtime_state_projects_hid_identity(self):
hook = BaseMouseHook()
device = SimpleNamespace(name="MX Master 3S")
hook._hid_gesture = SimpleNamespace(connected_device=device)

hook._on_hid_connect()

state = hook.hid_runtime_state
self.assertTrue(state.input_ready)
self.assertTrue(state.hid_ready)
self.assertIs(state.connected_device, device)


class LinuxMouseHookReconnectTests(unittest.TestCase):
def _reload_for_linux(self):
fake_evdev = SimpleNamespace(
Expand Down Expand Up @@ -245,6 +266,14 @@ def test_hid_reconnect_requests_rescan_for_fallback_evdev_device(self):
self.assertFalse(hook.device_connected)
self.assertTrue(hook.hid_ready)
self.assertEqual(hook.connected_device, {"name": "MX Master 3S"})
self.assertEqual(
hook.hid_runtime_state,
HidRuntimeState(
input_ready=False,
hid_ready=True,
connected_device={"name": "MX Master 3S"},
),
)
self.assertTrue(hook._rescan_requested.is_set())
self.assertTrue(hook._evdev_wakeup.is_set())

Expand Down Expand Up @@ -274,6 +303,14 @@ def test_hid_reconnect_does_not_rescan_when_evdev_already_grabs_logitech(self):

self.assertTrue(hook.device_connected)
self.assertFalse(hook._rescan_requested.is_set())
self.assertEqual(
hook.hid_runtime_state,
HidRuntimeState(
input_ready=True,
hid_ready=True,
connected_device={"name": "MX Master 3S"},
),
)

def test_hid_connect_does_not_mark_device_connected_when_evdev_is_missing(self):
module = self._reload_for_linux()
Expand All @@ -299,6 +336,14 @@ def test_hid_disconnect_keeps_evdev_driven_connected_state(self):

self.assertTrue(hook.device_connected)
self.assertEqual(hook.connected_device, {"name": "MX Master 3S"})
self.assertEqual(
hook.hid_runtime_state,
HidRuntimeState(
input_ready=True,
hid_ready=False,
connected_device={"name": "MX Master 3S"},
),
)

def test_setup_evdev_marks_connected_and_populates_fallback_device_info(self):
module = self._reload_for_linux()
Expand All @@ -319,6 +364,12 @@ def test_setup_evdev_marks_connected_and_populates_fallback_device_info(self):
self.assertTrue(hook.device_connected)
self.assertEqual(getattr(hook.connected_device, "display_name", None), "MX Master 3S")
self.assertEqual(getattr(hook.connected_device, "source", None), "evdev")
self.assertEqual(hook.hid_runtime_state.input_ready, True)
self.assertEqual(hook.hid_runtime_state.hid_ready, False)
self.assertEqual(
getattr(hook.hid_runtime_state.connected_device, "source", None),
"evdev",
)

def test_listen_loop_exits_when_rescan_is_requested(self):
module = self._reload_for_linux()
Expand Down
12 changes: 11 additions & 1 deletion tests/test_mouse_hook_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@

from core import mouse_hook
from core.mouse_hook_contract import MouseHookLike
from core.mouse_hook_types import MouseEvent
from core.mouse_hook_types import HidRuntimeState, MouseEvent


class MouseHookContractTests(unittest.TestCase):
Expand All @@ -23,6 +23,16 @@ def test_selected_hook_exposes_engine_contract_surface(self):
hook = mouse_hook.MouseHook()
self.assertIsInstance(hook, MouseHookLike)

def test_selected_hook_exposes_hid_runtime_state(self):
hook = mouse_hook.MouseHook()

state = hook.hid_runtime_state

self.assertIsInstance(state, HidRuntimeState)
self.assertFalse(state.input_ready)
self.assertFalse(state.hid_ready)
self.assertIsNone(state.connected_device)

def test_dispatcher_monkeypatch_forwards_to_platform_module(self):
platform_module = sys.modules[mouse_hook.MouseHook.__module__]

Expand Down
Loading