From 27a313234a23c89f42a07517e0b4f6eebf6eee7c Mon Sep 17 00:00:00 2001 From: laziukdavid <241655708+laziukdavid@users.noreply.github.com> Date: Mon, 18 May 2026 19:08:51 -0700 Subject: [PATCH] Add MX Master 4 support: HID++ detection, gesture recognizer, UI & build fixes A multi-feature set developed and tested against a Logitech MX Master 4 (via a Logi Bolt receiver). Summary by area: HID++ detection speed - Receiver-aware device-index scan order, persistent device cache (device_cache.json), dead-slot early-out, 600ms discovery probe timeout applied across the whole discovery phase, and candidate dedup so a multi-interface device is probed once. Cuts cold-start connection from ~46s of wrong-device probing. Gesture swipe recognizer - New core/gesture_recognizer.py: a pure, thread-safe, stroke-segmenting GestureRecognizer replacing the old global-sum + fixed-cooldown logic (which dropped inputs, fired the opposite direction on return strokes, and could not do back-to-back repeats). Shared by the macOS, Windows and Linux hooks, removing the triplicated accumulate/detect code. - Config schema v9 -> v10: retires gesture_deadzone / gesture_timeout_ms / gesture_cooldown_ms; adds gesture_commit_window_ms, gesture_settle_ms, gesture_cross_ratio. Obsolete keys in existing configs are left ignored. - Click-jolt rejection (min sample count) and quick-flick capture. Sensitivity selector - Replaces the raw gesture_threshold px slider with a discrete 5-level Sensitivity dropdown. Stored key/units unchanged. Screenshot action - New screenshot action: Cmd+Shift+4 on macOS, Win+Shift+S on Windows. Application Log panel - core/log_setup.py ring-buffer log handler with listener fan-out; appLog property on the QML backend; a new Application Log section in the Debug Events panel showing captured console output. macOS Accessibility auto-recovery - Non-modal permission prompt with a deep link to the Accessibility pane, a 1s grant poll, and a clean re-exec once the grant lands -- no manual quit/reopen after granting Accessibility on first launch. macOS build - build_macos_app.sh redirects all build artifacts off network volumes to local disk (codesign rejects the resource-fork metadata SMB/AFP shares attach); auto-detects a non-local repo volume, overridable via MOUSER_BUILD_DIR. - New install_macos_app.sh: a documented build + install-to-/Applications wrapper around build_macos_app.sh. Tests: full suite green -- 555 passed, 1 skipped. Co-Authored-By: Claude Opus 4.7 (1M context) --- build_macos_app.sh | 46 ++- core/config.py | 85 +++++- core/engine.py | 6 +- core/gesture_recognizer.py | 467 +++++++++++++++++++++++++++++++ core/hid_gesture.py | 221 ++++++++++++++- core/key_simulator.py | 10 + core/log_setup.py | 76 +++++ core/mouse_hook_base.py | 156 ++++++----- core/mouse_hook_contract.py | 6 +- core/mouse_hook_linux.py | 181 +----------- core/mouse_hook_macos.py | 201 ++----------- core/mouse_hook_windows.py | 158 +---------- install_macos_app.sh | 94 +++++++ main_qml.py | 158 +++++++++-- tests/test_backend.py | 88 +++++- tests/test_config.py | 62 +++- tests/test_gesture_recognizer.py | 274 ++++++++++++++++++ tests/test_hid_gesture.py | 269 +++++++++++++++++- tests/test_key_simulator.py | 8 + tests/test_log_setup.py | 113 ++++++++ tests/test_macos_app_shell.py | 117 +++++++- tests/test_macos_build_script.py | 2 +- tests/test_smart_shift.py | 4 +- ui/backend.py | 54 +++- ui/locale_manager.py | 34 ++- ui/qml/MousePage.qml | 139 ++++++--- 26 files changed, 2309 insertions(+), 720 deletions(-) create mode 100644 core/gesture_recognizer.py create mode 100644 install_macos_app.sh create mode 100644 tests/test_gesture_recognizer.py diff --git a/build_macos_app.sh b/build_macos_app.sh index 28f1481..705d9df 100755 --- a/build_macos_app.sh +++ b/build_macos_app.sh @@ -11,7 +11,34 @@ SOURCE_ICON="$ROOT_DIR/images/logo_icon.png" ENTITLEMENTS="$ROOT_DIR/build_resources/Mouser.entitlements" TARGET_ARCH="${PYINSTALLER_TARGET_ARCH:-}" SIGN_IDENTITY="${MOUSER_SIGN_IDENTITY:-}" -export PYINSTALLER_CONFIG_DIR="$BUILD_DIR/pyinstaller" +# ── Build artifact location ───────────────────────────────────────── +# codesign refuses to sign Mach-O files that carry a resource fork or +# Finder-info metadata, and SMB/AFP network volumes attach exactly that to +# files written to them. When the repository itself lives on such a volume, +# redirect every build artifact (PyInstaller cache, work dir, dist bundle) to +# local disk so the in-build and post-build signing steps see clean files. +# Set MOUSER_BUILD_DIR to choose the location explicitly. +if [[ -n "${MOUSER_BUILD_DIR:-}" ]]; then + BUILD_OUTPUT_DIR="$MOUSER_BUILD_DIR" +else + ROOT_DEVICE="" + if command -v df >/dev/null 2>&1 && command -v awk >/dev/null 2>&1; then + ROOT_DEVICE="$(df -P "$ROOT_DIR" 2>/dev/null | awk 'NR==2 {print $1}' 2>/dev/null || true)" + fi + # Empty (detection unavailable) or a /dev/* device → treat as local disk. + if [[ -z "$ROOT_DEVICE" || "$ROOT_DEVICE" == /dev/* ]]; then + BUILD_OUTPUT_DIR="$ROOT_DIR" + else + BUILD_OUTPUT_DIR="$HOME/Library/Caches/Mouser/macos-build" + echo "Repository volume ($ROOT_DEVICE) is not local disk;" + echo "building in $BUILD_OUTPUT_DIR to avoid codesign metadata errors." + fi +fi +DIST_DIR="$BUILD_OUTPUT_DIR/dist" +WORK_DIR="$BUILD_OUTPUT_DIR/build" +DIST_APP="$DIST_DIR/Mouser.app" +mkdir -p "$DIST_DIR" "$WORK_DIR" +export PYINSTALLER_CONFIG_DIR="$BUILD_OUTPUT_DIR/pyinstaller" PYTHON="" PYTHON_SOURCE="" @@ -148,12 +175,17 @@ log_python_provenance() { run_pyinstaller() { # PYTHONHASHSEED=0 pins set iteration so PyInstaller's base_library.zip # layout is byte-identical across rebuilds for the same toolchain inputs. - PYTHONHASHSEED=0 "$PYTHON" -m PyInstaller "$ROOT_DIR/Mouser-mac.spec" --noconfirm + # Run inside ROOT_DIR so the spec's `ROOT = abspath(".")` resolves to the + # repo no matter where this script was invoked from, and pin work/dist + # paths to BUILD_OUTPUT_DIR (kept on local disk for network repos). + ( cd "$ROOT_DIR" && PYTHONHASHSEED=0 "$PYTHON" -m PyInstaller \ + "$ROOT_DIR/Mouser-mac.spec" --noconfirm \ + --workpath "$WORK_DIR" --distpath "$DIST_DIR" ) } sign_ad_hoc() { echo "Signing mode: ad-hoc" - codesign --force --deep --sign - "$ROOT_DIR/dist/Mouser.app" + codesign --force --deep --sign - "$DIST_APP" } entitlements_sha256() { @@ -161,7 +193,7 @@ entitlements_sha256() { } sign_nested_code() { - local frameworks_dir="$ROOT_DIR/dist/Mouser.app/Contents/Frameworks" + local frameworks_dir="$DIST_APP/Contents/Frameworks" [[ -d "$frameworks_dir" ]] || return 0 while IFS= read -r -d '' nested; do @@ -173,7 +205,7 @@ sign_nested_code() { } verify_bundle() { - codesign --verify --deep --strict --verbose=2 "$ROOT_DIR/dist/Mouser.app" + codesign --verify --deep --strict --verbose=2 "$DIST_APP" } sign_with_identity() { @@ -188,7 +220,7 @@ sign_with_identity() { codesign --force --options runtime --timestamp=none \ --entitlements "$ENTITLEMENTS" \ --sign "$SIGN_IDENTITY" \ - "$ROOT_DIR/dist/Mouser.app" + "$DIST_APP" verify_bundle } @@ -211,4 +243,4 @@ log_python_provenance run_pyinstaller sign_app -echo "Build complete: $ROOT_DIR/dist/Mouser.app" +echo "Build complete: $DIST_APP" diff --git a/core/config.py b/core/config.py index b8afa87..1cb697d 100644 --- a/core/config.py +++ b/core/config.py @@ -21,6 +21,9 @@ else: CONFIG_DIR = os.path.join(os.environ.get("APPDATA", os.path.expanduser("~")), "Mouser") CONFIG_FILE = os.path.join(CONFIG_DIR, "config.json") +# Last-good HID++ device identity, written by hid_gesture on every successful +# connect so the next launch can skip the slow multi-receiver device scan. +DEVICE_CACHE_FILE = os.path.join(CONFIG_DIR, "device_cache.json") # Which mouse events map to which friendly button names # Order matches the Logi Options+ diagram (top view then side view) @@ -42,6 +45,22 @@ "gesture_down", ) +# Swipe sensitivity presets — px of travel the recognizer needs to commit a +# swipe (stored as `gesture_threshold`). Ordered most → least sensitive; the +# UI exposes these as a discrete "Sensitivity" selector rather than a raw px +# slider. A lower value fires on shorter strokes. +GESTURE_SENSITIVITY_PX = (18, 25, 33, 44, 56) +GESTURE_DEFAULT_SENSITIVITY_INDEX = 1 # 25 px — leans toward the sensitive end + + +def gesture_sensitivity_index_for(threshold_px): + """Return the sensitivity preset index nearest to a stored px threshold.""" + return min( + range(len(GESTURE_SENSITIVITY_PX)), + key=lambda i: abs(GESTURE_SENSITIVITY_PX[i] - int(threshold_px)), + ) + + PROFILE_BUTTON_NAMES = { **BUTTON_NAMES, "gesture_left": "Gesture swipe left", @@ -67,7 +86,7 @@ } DEFAULT_CONFIG = { - "version": 9, + "version": 10, "active_profile": "default", "profiles": { "default": { @@ -98,10 +117,12 @@ "smart_shift_mode": "ratchet", "smart_shift_enabled": False, "smart_shift_threshold": 25, - "gesture_threshold": 50, - "gesture_deadzone": 40, - "gesture_timeout_ms": 3000, - "gesture_cooldown_ms": 500, + # px of travel along the swipe axis; UI sets this via the discrete + # GESTURE_SENSITIVITY_PX presets ("Sensitivity" selector). + "gesture_threshold": GESTURE_SENSITIVITY_PX[GESTURE_DEFAULT_SENSITIVITY_INDEX], + "gesture_commit_window_ms": 400, # that travel must finish this fast + "gesture_settle_ms": 90, # pause that ends a stroke + "gesture_cross_ratio": 0.5, # tolerated off-axis travel ratio "appearance_mode": "system", "debug_mode": False, "device_layout_overrides": {}, @@ -173,18 +194,20 @@ def load_config(): return json.loads(json.dumps(DEFAULT_CONFIG)) # deep copy -def save_config(cfg): - """Persist config to disk via atomic write with restrictive permissions.""" +def _atomic_write_json(path, obj): + """Write obj to path as JSON via a temp file plus atomic rename, with + restrictive (owner-only) permissions. Used for config.json and the HID + device cache so a crash mid-write can never leave a partial file.""" ensure_config_dir() fd, tmp_path = tempfile.mkstemp(suffix=".tmp", dir=CONFIG_DIR) try: with os.fdopen(fd, "w", encoding="utf-8") as f: - json.dump(cfg, f, indent=2) + json.dump(obj, f, indent=2) f.flush() os.fsync(f.fileno()) if sys.platform != "win32": os.chmod(tmp_path, stat.S_IRUSR | stat.S_IWUSR) - os.replace(tmp_path, CONFIG_FILE) + os.replace(tmp_path, path) except BaseException: try: os.unlink(tmp_path) @@ -193,6 +216,31 @@ def save_config(cfg): raise +def save_config(cfg): + """Persist config to disk via atomic write with restrictive permissions.""" + _atomic_write_json(CONFIG_FILE, cfg) + + +def load_device_cache(): + """Return the last-good HID++ device identity dict, or None. + + Lets hid_gesture._try_connect re-open the device that connected last time + before falling back to a full multi-receiver scan — the difference between + a ~1 s reconnect and a multi-second one. + """ + try: + with open(DEVICE_CACHE_FILE, "r", encoding="utf-8") as f: + data = json.load(f) + except (OSError, ValueError): + return None + return data if isinstance(data, dict) else None + + +def save_device_cache(identity): + """Persist the last-good HID++ device identity (atomic write).""" + _atomic_write_json(DEVICE_CACHE_FILE, identity) + + def get_active_mappings(cfg): """Return the mappings dict for the currently active profile.""" profile_name = cfg.get("active_profile", "default") @@ -272,10 +320,10 @@ def _migrate(cfg): if version < 3: settings = cfg.setdefault("settings", {}) - settings.setdefault("gesture_threshold", 50) - settings.setdefault("gesture_deadzone", 40) - settings.setdefault("gesture_timeout_ms", 3000) - settings.setdefault("gesture_cooldown_ms", 500) + settings.setdefault( + "gesture_threshold", + GESTURE_SENSITIVITY_PX[GESTURE_DEFAULT_SENSITIVITY_INDEX], + ) for pdata in cfg.get("profiles", {}).values(): mappings = pdata.setdefault("mappings", {}) mappings.setdefault("gesture", "none") @@ -329,6 +377,17 @@ def _migrate(cfg): settings.setdefault("ignore_trackpad", True) cfg["version"] = 9 + if version < 10: + # Gesture swipe detection was rewritten: the fixed-pixel-threshold + # accumulator (gesture_deadzone / gesture_timeout_ms / + # gesture_cooldown_ms) is replaced by a stroke-aware recognizer. + # The obsolete keys are left in place but ignored. + settings = cfg.setdefault("settings", {}) + settings.setdefault("gesture_commit_window_ms", 400) + settings.setdefault("gesture_settle_ms", 90) + settings.setdefault("gesture_cross_ratio", 0.5) + cfg["version"] = 10 + cfg.setdefault("settings", {}) cfg["settings"].setdefault("appearance_mode", "system") cfg["settings"].setdefault("debug_mode", False) diff --git a/core/engine.py b/core/engine.py index 6ab4156..30d538a 100644 --- a/core/engine.py +++ b/core/engine.py @@ -110,9 +110,9 @@ def _setup_hooks(self): enabled=any(mappings.get(key, "none") != "none" for key in GESTURE_DIRECTION_BUTTONS), threshold=settings.get("gesture_threshold", 50), - deadzone=settings.get("gesture_deadzone", 40), - timeout_ms=settings.get("gesture_timeout_ms", 3000), - cooldown_ms=settings.get("gesture_cooldown_ms", 500), + commit_window_ms=settings.get("gesture_commit_window_ms", 400), + settle_ms=settings.get("gesture_settle_ms", 90), + cross_ratio=settings.get("gesture_cross_ratio", 0.5), ) # Divert mode shift CID only when the device has the button and # at least one profile maps it to an action. When no device is diff --git a/core/gesture_recognizer.py b/core/gesture_recognizer.py new file mode 100644 index 0000000..9624801 --- /dev/null +++ b/core/gesture_recognizer.py @@ -0,0 +1,467 @@ +""" +core/gesture_recognizer.py — stroke-aware swipe recognizer for the gesture button. + +Replaces the previous fixed-pixel-threshold accumulator. That design summed +every movement delta into one running total and fired a swipe whenever the +total crossed a threshold, then blocked all input for a fixed "cooldown". +It produced three problems this module is built to eliminate: + + * Missed inputs — the cooldown swallowed the start of the next swipe, so + repeated swipes had to be performed slowly and deliberately. + * False positives — slow drift while merely holding the button summed up + past the threshold; and the motion that returns the mouse between two + swipes ("return stroke") triggered a swipe in the opposite direction. + * No real repeats — you could not hold the button and flick the same way + several times in a row to page through views. + +Approach +-------- +The recognizer never integrates position into one global sum. It segments +the motion into *strokes* and judges each stroke on its own: + + begin() — gesture button pressed; start a fresh hold. + sample(dx, dy) — one movement report captured while held. + end() -> bool — button released; True == it was a plain click. + +A stroke ends when the pointer pauses (`settle_ms` of no movement) or +sharply reverses. A stroke *commits* a swipe when it travels +`commit_distance` along one axis, quickly enough (within `commit_window_ms` +— this rejects slow drift), straight enough (off-axis travel within +`cross_ratio`), and across enough movement reports that the brief jolt of +clicking the button cannot be mistaken for a swipe. + +The first swipe of a hold locks the hold to that axis and direction. After +that the recognizer runs a peak detector on the locked axis: every fresh +flick in the locked direction fires again, while the return strokes between +flicks only re-arm it — they can never fire, not even the opposite swipe. +A pause (`settle_ms`) clears the lock so a different direction can be used. + +The class has no Qt / HID / platform imports so it is unit-tested directly +and shared unchanged by the macOS, Windows and Linux hooks. +""" + +import math +import threading +import time + +__all__ = ["GestureRecognizer", "LEFT", "RIGHT", "UP", "DOWN"] + +LEFT = "left" +RIGHT = "right" +UP = "up" +DOWN = "down" + +# Off-axis travel always tolerated on top of the cross_ratio allowance, so a +# short swipe is not rejected over a few stray pixels of hand wobble. +_CROSS_FLOOR = 14.0 + +# Minimum spacing between two emitted swipes — debounce only. Real repeats +# (return stroke + fresh flick) are always far slower than this. +_REFRACTORY_S = 0.09 + +# A turnaround only counts once the motion has reversed by this much; smaller +# jitter does not move the tracked extreme, so a noisy sensor cannot restart +# a flick mid-stroke. +_TURN_HYST = 4.0 + +# A stroke must span at least this many movement reports before it can commit +# a swipe. Pressing/releasing the gesture button jolts the mouse as a brief +# 1-2 report impulse; a real swipe is a sustained stream of reports. This +# (together with reversal detection) keeps a plain click from firing a swipe. +_MIN_SAMPLES = 4 + + +def _axis_of(direction): + return "x" if direction in (LEFT, RIGHT) else "y" + + +def _sign_of(direction): + # Screen coordinates: left / up are negative, right / down are positive. + return -1 if direction in (LEFT, UP) else 1 + + +class GestureRecognizer: + """Turns a stream of held-button movement deltas into swipe events. + + Feed ``begin()`` on button-down, ``sample(dx, dy)`` for every movement + report while the button is held, and ``end()`` on button-up. Recognized + swipes are delivered through the ``on_swipe(direction)`` callback, where + direction is one of ``LEFT`` / ``RIGHT`` / ``UP`` / ``DOWN``. + + All public methods are thread-safe; callbacks are invoked *outside* the + internal lock so they may safely call back into the owning hook. + """ + + # Detection phases. + _IDLE = "idle" # no swipe committed yet this hold — free detection. + _LOCKED = "locked" # a swipe fired — repeats locked to that axis + dir. + + def __init__(self, on_swipe=None, on_debug=None): + self._on_swipe = on_swipe + self._on_debug = on_debug + self._lock = threading.Lock() + + # Configuration — see configure(). + self._enabled = False + self._commit_distance = 50.0 + self._commit_window = 0.40 + self._settle = 0.09 + self._cross_ratio = 0.5 + self._dir_eps = 7.5 # derived: motion needed to start a leg + self._min_return = 22.5 # derived: return needed to arm a repeat + + self._reset_hold() + + # ── configuration ──────────────────────────────────────────────── + + def configure(self, *, enabled, threshold, commit_window_ms, + settle_ms, cross_ratio): + """Apply user-tunable gesture settings. + + threshold — px of travel along the dominant axis to fire. + commit_window_ms — that travel must complete this fast (drift gate). + settle_ms — no-movement gap that ends a stroke / clears lock. + cross_ratio — off-axis travel tolerated, as a fraction of the + dominant-axis travel. + """ + with self._lock: + self._enabled = bool(enabled) + self._commit_distance = max(8.0, float(threshold)) + self._commit_window = max(0.05, float(commit_window_ms) / 1000.0) + self._settle = max(0.02, float(settle_ms) / 1000.0) + self._cross_ratio = min(2.0, max(0.05, float(cross_ratio))) + # Derived thresholds scale with the commit distance. + self._dir_eps = max(5.0, self._commit_distance * 0.15) + self._min_return = max(14.0, self._commit_distance * 0.45) + + # ── per-hold lifecycle ─────────────────────────────────────────── + + def begin(self): + """Gesture button pressed — discard any prior state, start fresh.""" + with self._lock: + self._reset_hold() + self._active = True + + def end(self): + """Gesture button released. + + Returns True when the hold produced no swipe and should therefore be + treated as a plain gesture-button click. + """ + with self._lock: + was_click = self._active and not self._fired_any + self._active = False + self._phase = self._IDLE + return was_click + + def sample(self, dx, dy, source="hid_rawxy", now=None): + """Feed one movement delta captured while the button is held.""" + if now is None: + now = time.monotonic() + fires = [] + debugs = [] + with self._lock: + if not (self._active and self._enabled): + return + if dx == 0 and dy == 0: + return + if not self._accept_source(source): + return + self._step(float(dx), float(dy), now, fires, debugs) + # Callbacks run outside the lock. + for event in debugs: + self._emit_debug(event) + for direction in fires: + self._emit_swipe(direction) + + @property + def fired(self): + """True once a swipe has fired during the current / last hold.""" + with self._lock: + return self._fired_any + + def summary(self): + """Diagnostics for the current / most-recently-ended hold. + + Returned as a dict so the hook can print a one-line ``[Gesture]`` + trace on button-up — the quickest way to tune swipe-vs-click feel + against real hardware. + """ + with self._lock: + duration_ms = 0.0 + if self._hold_first_t is not None and self._last_t is not None: + duration_ms = (self._last_t - self._hold_first_t) * 1000.0 + return { + "samples": self._hold_samples, + "duration_ms": duration_ms, + "net_x": self._cx, + "net_y": self._cy, + "peak_speed": self._hold_peak_speed, + "source": self._source, + "fired": list(self._hold_fired), + } + + # ── internal: state management ─────────────────────────────────── + + def _reset_hold(self): + """Clear all per-hold state. Caller holds the lock.""" + self._phase = self._IDLE + self._active = False + self._fired_any = False + self._source = None + self._last_t = None + self._last_fire_t = -1.0 + # Cumulative pointer position since begin(). + self._cx = 0.0 + self._cy = 0.0 + # Per-hold diagnostics, surfaced by summary(). + self._hold_samples = 0 + self._hold_first_t = None + self._hold_peak_speed = 0.0 + self._hold_fired = [] + self._reset_leg(0.0, 0.0) + # LOCKED-phase peak-detector state. + self._lock_axis = None + self._lock_sign = 0 + self._latch_anchor = 0.0 # locked-axis position where it fired + self._latch_extreme = 0.0 # furthest point of the return stroke + self._latch_off_at_turn = 0.0 # off-axis position at that turnaround + self._latch_turn_t = 0.0 # time of that turnaround + self._latch_return_seen = False + + def _reset_leg(self, at_x, at_y): + """Re-base the free-detection leg at the given position.""" + self._pivot_x = at_x + self._pivot_y = at_y + self._pivot_t = None # set when motion actually starts + self._leg_peak = 0.0 + self._leg_samples = 0 # movement reports in the current leg + + def _accept_source(self, source): + """Source arbitration: lock to the first source; a real raw-XY + stream supersedes a coarse fallback (event-tap / evdev).""" + if self._source == source: + return True + if self._source is None: + self._source = source + return True + if source == "hid_rawxy": + # Promote: discard whatever the fallback source accumulated. + self._source = source + self._phase = self._IDLE + self._lock_axis = None + self._lock_sign = 0 + self._reset_leg(self._cx, self._cy) + return True + return False + + # ── internal: the recognizer ───────────────────────────────────── + + def _step(self, dx, dy, now, fires, debugs): + # Per-hold diagnostics (surfaced by summary()). + self._hold_samples += 1 + if self._hold_first_t is None: + self._hold_first_t = now + if self._last_t is not None: + dt = now - self._last_t + if dt > 0: + speed = math.hypot(dx, dy) / dt + if speed > self._hold_peak_speed: + self._hold_peak_speed = speed + + # A long-enough gap with no movement ends the current stroke and + # clears the hold's axis lock — the next stroke is judged afresh. + if self._last_t is not None and (now - self._last_t) > self._settle: + self._phase = self._IDLE + self._lock_axis = None + self._lock_sign = 0 + self._reset_leg(self._cx, self._cy) + self._last_t = now + + self._cx += dx + self._cy += dy + + if self._phase == self._LOCKED: + self._step_locked(now, fires, debugs) + else: + self._step_free(now, fires, debugs) + + def _step_free(self, now, fires, debugs): + """Pre-lock detection: watch one free stroke for the first swipe.""" + leg_x = self._cx - self._pivot_x + leg_y = self._cy - self._pivot_y + leg_len = math.hypot(leg_x, leg_y) + + # Wait for the leg to actually start moving before timing it. + if self._pivot_t is None: + if leg_len < self._dir_eps: + return + self._pivot_t = now + self._leg_peak = leg_len + self._leg_samples = 1 + debugs.append({"type": "tracking_started", "source": self._source}) + else: + self._leg_samples += 1 + + # A leg that shrinks back toward its pivot means the motion reversed + # or stalled — re-base so the new direction gets a clean stroke. + if leg_len < self._leg_peak - self._dir_eps: + self._reset_leg(self._cx, self._cy) + return + self._leg_peak = max(self._leg_peak, leg_len) + + debugs.append({"type": "segment", "source": self._source, + "dx": leg_x, "dy": leg_y}) + + # A leg that takes too long to reach commit distance is drift, not a + # swipe — re-base and keep watching. + if (now - self._pivot_t) > self._commit_window: + self._reset_leg(self._cx, self._cy) + return + + direction = self._evaluate_leg(leg_x, leg_y) + if direction is None: + return + if self._leg_samples < _MIN_SAMPLES: + # Far and fast enough to look like a swipe, but carried by too + # few movement reports — this is the jolt of clicking the + # button, not a sustained stroke. Hold off; if the motion is + # real the next reports will commit it. + return + if not self._fire(direction, now, leg_x, leg_y, fires, debugs): + return + + # First swipe of the hold — lock to this axis + direction. From now + # on only this direction repeats; return strokes cannot fire. + self._phase = self._LOCKED + self._lock_axis = _axis_of(direction) + self._lock_sign = _sign_of(direction) + pos = self._cx if self._lock_axis == "x" else self._cy + off = self._cy if self._lock_axis == "x" else self._cx + self._latch_anchor = pos + self._latch_extreme = pos + self._latch_off_at_turn = off + self._latch_turn_t = now + self._latch_return_seen = False + + def _step_locked(self, now, fires, debugs): + """Post-lock detection: a peak detector on the locked axis. + + Travel in the locked direction is a fresh flick; travel the other way + is a return stroke that only re-arms the detector. The opposite swipe + can never fire while the hold is locked. + """ + axis = self._lock_axis + sign = self._lock_sign + pos = self._cx if axis == "x" else self._cy + off = self._cy if axis == "x" else self._cx + + # Track the turnaround: the point furthest along the return stroke. + # The committed swipe's own decelerating tail moves further in the + # locked direction and is deliberately ignored here. + if sign * pos < sign * self._latch_extreme - _TURN_HYST: + self._latch_extreme = pos + self._latch_off_at_turn = off + self._latch_turn_t = now + + # The return must travel far enough to count as a genuine turnaround + # before the next flick is allowed to fire. + return_amount = sign * (self._latch_anchor - self._latch_extreme) + if return_amount >= self._min_return: + self._latch_return_seen = True + + # Flick = travel in the locked direction measured from the turnaround. + flick = sign * (pos - self._latch_extreme) + off_flick = off - self._latch_off_at_turn + + debugs.append({ + "type": "segment", "source": self._source, + "dx": (pos - self._latch_extreme) if axis == "x" else off_flick, + "dy": (pos - self._latch_extreme) if axis == "y" else off_flick, + }) + + if not self._latch_return_seen: + return + if flick < self._commit_distance: + return + if (now - self._latch_turn_t) > self._commit_window: + # The flick was too slow to be deliberate — re-arm from here. + self._latch_anchor = pos + self._latch_extreme = pos + self._latch_off_at_turn = off + self._latch_turn_t = now + self._latch_return_seen = False + return + if abs(off_flick) > self._cross_ratio * flick + _CROSS_FLOOR: + return + + direction = self._locked_direction() + seg_x = (pos - self._latch_extreme) if axis == "x" else off_flick + seg_y = (pos - self._latch_extreme) if axis == "y" else off_flick + if not self._fire(direction, now, seg_x, seg_y, fires, debugs): + return + # Re-arm for the next repeat from the current point. + self._latch_anchor = pos + self._latch_extreme = pos + self._latch_off_at_turn = off + self._latch_turn_t = now + self._latch_return_seen = False + + def _evaluate_leg(self, leg_x, leg_y): + """Return a swipe direction if (leg_x, leg_y) commits, else None. + + The caller has already enforced the speed (commit_window) gate. + """ + abs_x = abs(leg_x) + abs_y = abs(leg_y) + if abs_x >= abs_y: + dominant, cross = abs_x, abs_y + direction = RIGHT if leg_x > 0 else LEFT + else: + dominant, cross = abs_y, abs_x + direction = DOWN if leg_y > 0 else UP + if dominant < self._commit_distance: + return None # not far enough yet + if cross > self._cross_ratio * dominant + _CROSS_FLOOR: + return None # too diagonal to be sure + return direction + + def _locked_direction(self): + if self._lock_axis == "x": + return RIGHT if self._lock_sign > 0 else LEFT + return DOWN if self._lock_sign > 0 else UP + + def _fire(self, direction, now, seg_x, seg_y, fires, debugs): + """Record a recognized swipe. Returns False if debounced away.""" + if (now - self._last_fire_t) < _REFRACTORY_S: + return False + self._last_fire_t = now + self._fired_any = True + self._hold_fired.append(direction) + fires.append(direction) + debugs.append({ + "type": "detected", + "event_name": "gesture_swipe_" + direction, + "source": self._source, + "dx": seg_x, + "dy": seg_y, + }) + return True + + # ── internal: callbacks ────────────────────────────────────────── + + def _emit_swipe(self, direction): + if self._on_swipe is None: + return + try: + self._on_swipe(direction) + except Exception as exc: # pragma: no cover + print(f"[GestureRecognizer] swipe callback error: {exc}") + + def _emit_debug(self, event): + if self._on_debug is None: + return + try: + self._on_debug(event) + except Exception: # pragma: no cover + pass diff --git a/core/hid_gesture.py b/core/hid_gesture.py index 67498fa..7e96e55 100644 --- a/core/hid_gesture.py +++ b/core/hid_gesture.py @@ -24,6 +24,7 @@ clamp_dpi, resolve_device, ) +from core import config _HID_MODULE_NAME = None try: @@ -518,11 +519,18 @@ def read(self, _size, timeout_ms=0): remaining = deadline - time.monotonic() if remaining <= 0: return b"" - slice_seconds = min(remaining, 0.05) + # Run CoreFoundation in long slices. CFRunLoopRunInMode + # releases the GIL while it blocks, so one long slice keeps + # this HID thread waiting in C rather than thrashing back + # into Python every 50 ms. Device reads then stay + # responsive even while the Qt UI thread is busy holding + # the GIL — the old fine-grained slicing is the main + # reason detection stalled while the window was open. + slice_seconds = min(remaining, 1.0) else: slice_seconds = 0.05 - _cf.CFRunLoopRunInMode( + result = _cf.CFRunLoopRunInMode( _K_CF_RUN_LOOP_DEFAULT_MODE, slice_seconds, True, @@ -530,9 +538,13 @@ def read(self, _size, timeout_ms=0): try: return self._report_queue.get_nowait() except queue.Empty: - if deadline is not None: - continue + pass + if deadline is None: return b"" + # kCFRunLoopRunFinished (1): the mode had no sources left to + # service. Yield briefly so we don't busy-spin to the deadline. + if result == 1: + time.sleep(0.01) # ── Constants ───────────────────────────────────────────────────── LOGI_VID = 0x046D @@ -604,6 +616,20 @@ def _linux_logitech_hidraw_nodes(base="/sys/class/hidraw"): # Known Logi Bolt receiver PID. # Source: https://github.com/pwr-Solaar/Solaar/blob/master/lib/logitech_receiver/base_usb.py BOLT_RECEIVER_PID = 0xC548 +# Logitech wireless receivers (Bolt / Unifying / Nano / Lightspeed). A mouse +# paired through one of these answers HID++ on a numbered device index (1-6) — +# never on the direct-Bluetooth index 0xFF. Probing 0xFF first on a receiver +# just burns a full request timeout before the real slot is even tried. +RECEIVER_PIDS = frozenset({ + 0xC548, # Logi Bolt + 0xC52B, 0xC52F, 0xC532, 0xC534, # Unifying + 0xC517, 0xC518, 0xC51A, 0xC51B, # Nano / older receivers + 0xC53A, 0xC53D, 0xC53F, 0xC541, # Lightspeed / newer receivers +}) +# A present HID++ device answers an IRoot feature probe in well under 100 ms. +# Discovery probes use this short timeout so a wrong device-index slot is ruled +# out fast instead of stalling the whole connection attempt for 2 s per slot. +DISCOVERY_PROBE_TIMEOUT_MS = 600 FEAT_IROOT = 0x0000 FEAT_REPROG_V4 = 0x1B04 # Reprogrammable Controls V4 FEAT_ADJ_DPI = 0x2201 # Adjustable DPI @@ -728,6 +754,8 @@ def __init__(self, on_down=None, on_up=None, on_move=None, self._battery_idx = None self._battery_feature_id = None self._dev_idx = BT_DEV_IDX + self._last_good_dev_idx = None # device index that last answered REPROG_V4 + self._device_cache = None # last-good device identity; loaded in start() self._gesture_cid = DEFAULT_GESTURE_CID self._gesture_candidates = list(DEFAULT_GESTURE_CIDS) self._held = False @@ -770,6 +798,12 @@ def start(self): "[HidGesture] Linux hidraw module is unavailable; Bluetooth " "Logitech HID++ devices may not enumerate" ) + # Load the persisted last-good device so the first _try_connect can go + # straight to it. Done here rather than in __init__ so constructing a + # listener (e.g. in a unit test) never touches disk. + self._device_cache = config.load_device_cache() + if self._device_cache and self._last_good_dev_idx is None: + self._last_good_dev_idx = self._device_cache.get("dev_idx") self._running = True self._thread = threading.Thread( target=self._main_loop, daemon=True, name="HidGesture") @@ -977,8 +1011,32 @@ def add_info(info): and _MAC_NATIVE_OK and _BACKEND_PREFERENCE in ("auto", "iokit") ): + vendor_pids = set() + boot_only_receiver_pids = set() for info in _MacNativeHidDevice.enumerate_infos(): + up = int(info.get("usage_page", 0) or 0) + pid = int(info.get("product_id", 0) or 0) + # HID++ always lives on a vendor usage page (>=0xFF00). + # Standard pages (Generic Desktop, Consumer, Digitizer) are + # boot collections that cannot carry HID++ and only waste + # probe time. Keep usage_page 0 — some BLE devices omit it. + if 0 < up < 0xFF00: + if pid in RECEIVER_PIDS: + boot_only_receiver_pids.add(pid) + continue + vendor_pids.add(pid) add_info(info) + # A receiver that shows up with only boot collections has had its + # HID++ interface claimed by something else — almost always other + # Logitech software (G HUB / Options+) running in the background. + for pid in sorted(boot_only_receiver_pids - vendor_pids): + _log_once( + f"receiver-no-vendor-{pid:04X}", + f"[HidGesture] Receiver PID=0x{pid:04X} is present but " + "exposes no HID++ interface — it is most likely claimed " + "by other Logitech software (G HUB / Options+). Quit that " + "software so Mouser can use this receiver.", + ) return out @@ -1010,6 +1068,12 @@ def _rx(self, timeout_ms=2000): def _request(self, feat, func, params, timeout_ms=2000): """Send a long HID++ request, wait for matching response.""" req_params = list(params) + # Until a live session exists (self._connected), every request is part + # of device discovery — cap the wait short so a wrong or asleep slot is + # ruled out fast instead of stalling 2 s. Operational requests after + # connect keep the full timeout so a sleeping device still gets woken. + if not self._connected: + timeout_ms = min(timeout_ms, DISCOVERY_PROBE_TIMEOUT_MS) try: self._tx(LONG_ID, feat, func, req_params) except Exception as exc: @@ -1023,8 +1087,9 @@ def _request(self, feat, func, params, timeout_ms=2000): return None deadline = time.time() + timeout_ms / 1000 while time.time() < deadline: + remaining_ms = max(1, int((deadline - time.time()) * 1000)) try: - raw = self._rx(min(500, timeout_ms)) + raw = self._rx(min(500, remaining_ms)) except Exception as exc: print(f"[HidGesture] request rx failed feat=0x{feat:02X} func=0x{func:X} " f"params=[{_hex_bytes(req_params)}]: {exc}") @@ -1063,11 +1128,11 @@ def _request(self, feat, func, params, timeout_ms=2000): # ── feature helpers ─────────────────────────────────────────── - def _find_feature(self, feature_id): + def _find_feature(self, feature_id, timeout_ms=2000): """Use IRoot (feature 0x0000) to discover a feature index.""" hi = (feature_id >> 8) & 0xFF lo = feature_id & 0xFF - resp = self._request(0x00, 0, [hi, lo, 0x00]) + resp = self._request(0x00, 0, [hi, lo, 0x00], timeout_ms=timeout_ms) if resp: _, _, _, _, p = resp if p and p[0] != 0: @@ -1118,13 +1183,19 @@ def _set_cid_reporting(self, cid, flags): return self._request(self._feat_idx, 3, [hi, lo, flags, 0x00, 0x00]) def _discover_reprog_controls(self): + """Read the REPROG_V4 control table. + + Returns the list of controls, or None when the device did not answer + the control-count query — a slot that advertised REPROG_V4 on the root + probe but is not a reachable device (a stale or asleep pairing). + """ controls = [] if self._feat_idx is None: return controls resp = self._request(self._feat_idx, 0, []) if not resp: print("[HidGesture] Failed to read REPROG_V4 control count") - return controls + return None _, _, _, _, params = resp _MAX_REPROG_CONTROLS = 32 count = params[0] if params else 0 @@ -1670,9 +1741,64 @@ def _on_report(self, raw): # ── connect / main loop ─────────────────────────────────────── + def _is_receiver(self, product_id=None, product_name=None): + """True when this interface belongs to a Logitech wireless receiver.""" + name = (product_name or "").lower() + return ( + int(product_id or 0) in RECEIVER_PIDS + or "receiver" in name + or "bolt" in name + ) + + def _dev_index_scan_order(self, product_id=None, product_name=None): + """Order the HID++ device-index slots to probe for this candidate. + + A mouse paired through a wireless receiver answers on a numbered slot + (1-6); a direct-Bluetooth mouse answers on 0xFF. Probing the likely + slots first — and the last index that worked before anything else — + avoids spending a discovery timeout on indices that cannot match. + Every slot still stays in the list as a fallback in case the receiver + guess is wrong, so this only reorders, never narrows, the scan. + """ + receiver_slots = [1, 2, 3, 4, 5, 6] + if self._is_receiver(product_id, product_name): + order = receiver_slots + [BT_DEV_IDX] + else: + order = [BT_DEV_IDX] + receiver_slots + last_good = self._last_good_dev_idx + if last_good is not None and last_good in order: + order.remove(last_good) + order.insert(0, last_good) + return order + + def _save_device_cache(self, product_id, usage_page, usage, source): + """Remember the just-connected device so the next launch and every + reconnect can re-open it directly instead of scanning every Logitech + HID candidate. Written only when the identity actually changed.""" + identity = { + "product_id": int(product_id or 0), + "usage_page": int(usage_page or 0), + "usage": int(usage or 0), + "source": source or "", + "dev_idx": int(self._dev_idx), + } + if identity == self._device_cache: + return + self._device_cache = identity + try: + config.save_device_cache(identity) + print("[HidGesture] Cached device for fast reconnect: " + f"PID=0x{identity['product_id']:04X} " + f"devIdx=0x{identity['dev_idx']:02X}") + except Exception as exc: + print(f"[HidGesture] Could not write device cache: {exc}") + def _try_connect(self): """Open the vendor HID collection, discover features, divert.""" + enum_t0 = time.monotonic() infos = self._vendor_hid_infos() + print(f"[HidGesture] Enumerated {len(infos)} candidate interface(s) " + f"in {(time.monotonic() - enum_t0) * 1000:.0f} ms") if not infos: return False @@ -1684,6 +1810,25 @@ def _direct_device_first(info): infos.sort(key=_direct_device_first) + # Re-open the device that connected last time before scanning anything + # else. On a warm boot or a wake-from-sleep reconnect this collapses a + # multi-receiver scan into a single open; if the device is no longer + # present the loop just falls through to the full scan below. + cached = self._device_cache + if cached: + for i, info in enumerate(infos): + if ( + int(info.get("product_id", 0) or 0) == cached.get("product_id") + and int(info.get("usage_page", 0) or 0) == cached.get("usage_page") + and int(info.get("usage", 0) or 0) == cached.get("usage") + and (info.get("source") or "") == cached.get("source") + ): + infos.insert(0, infos.pop(i)) + print("[HidGesture] Probing cached device first: " + f"PID=0x{int(cached.get('product_id', 0)):04X} " + f"devIdx=0x{int(cached.get('dev_idx', 0)):02X}") + break + print(f"[HidGesture] Backend preference: {_BACKEND_PREFERENCE}") print(f"[HidGesture] Candidate HID interfaces: {len(infos)}") for info in infos: @@ -1698,8 +1843,15 @@ def _direct_device_first(info): f"usage=0x{usage:04X} transport={transport or '-'} " f"source={source} product={product} path={path or '-'}") + # Once a non-receiver interface is ruled out as a HID++ device, skip + # every other interface of that same physical device — e.g. an MX Brio + # webcam exposes six vendor collections that would otherwise each be + # opened and probed in turn. + dead_pids = set() for info in infos: pid = info.get("product_id", 0) + if pid in dead_pids: + continue up = info.get("usage_page", 0) usage = info.get("usage", 0) product = info.get("product_string") @@ -1741,6 +1893,7 @@ def _direct_device_first(info): open_attempts.append(("hidapi", info)) for transport, open_info in open_attempts: + open_t0 = time.monotonic() try: if transport.startswith("iokit"): d = _MacNativeHidDevice( @@ -1771,7 +1924,8 @@ def _direct_device_first(info): opened_up = int(open_info.get("usage_page", up) or 0) opened_usage = int(open_info.get("usage", usage) or 0) opened_path = _device_path_display(open_info.get("path")) - print(f"[HidGesture] Opened PID=0x{pid:04X} via {transport}") + print(f"[HidGesture] Opened PID=0x{pid:04X} via {transport} " + f"({(time.monotonic() - open_t0) * 1000:.0f} ms)") break except Exception as exc: print(f"[HidGesture] Can't open PID=0x{pid:04X} " @@ -1782,17 +1936,39 @@ def _direct_device_first(info): if self._dev is None: continue - # Try Bluetooth direct (0xFF) first, then Bolt receiver slots + # Probe HID++ device-index slots in the order most likely to hit + # for this candidate, with a short per-slot timeout so wrong slots + # are ruled out quickly instead of stalling for 2 s each. reprog_found = False hidpp_name = None - for idx in (0xFF, 1, 2, 3, 4, 5, 6): + is_receiver = self._is_receiver(pid, product) + probe_misses = 0 + for idx in self._dev_index_scan_order(pid, product): self._dev_idx = idx - fi = self._find_feature(FEAT_REPROG_V4) + probe_t0 = time.monotonic() + fi = self._find_feature( + FEAT_REPROG_V4, timeout_ms=DISCOVERY_PROBE_TIMEOUT_MS + ) + probe_ms = (time.monotonic() - probe_t0) * 1000 + if fi is None: + probe_misses += 1 + print(f"[HidGesture] devIdx 0x{idx:02X}: no REPROG_V4 " + f"({probe_ms:.0f} ms)") + # A non-receiver interface that answers neither direct + # Bluetooth (0xFF) nor the first slot is not a device we + # can drive — stop wasting ~0.6 s per remaining slot. + if not is_receiver and probe_misses >= 2: + print(f"[HidGesture] PID=0x{pid:04X} not a usable " + "HID++ device — skipping remaining slots") + dead_pids.add(pid) + break + continue if fi is not None: reprog_found = True self._feat_idx = fi + self._last_good_dev_idx = idx print(f"[HidGesture] Found REPROG_V4 @0x{fi:02X} " - f"PID=0x{pid:04X} devIdx=0x{idx:02X}") + f"PID=0x{pid:04X} devIdx=0x{idx:02X} ({probe_ms:.0f} ms)") # Query actual device name via HID++ (resolves # USB receivers that report a generic PID/name). hidpp_name = self._query_device_name() @@ -1806,6 +1982,15 @@ def _direct_device_first(info): or DEFAULT_GESTURE_CIDS ) controls = self._discover_reprog_controls() + if controls is None: + # REPROG_V4 answered the root probe but this slot will + # not answer real feature reads — a stale or asleep + # pairing on the receiver. Skip it now instead of + # burning a discovery timeout on every DPI / SmartShift + # / wheel / battery / divert probe that would follow. + print(f"[HidGesture] devIdx 0x{idx:02X}: REPROG_V4 " + "present but unresponsive — skipping slot") + continue self._last_controls = controls self._gesture_candidates = self._choose_gesture_candidates( controls, @@ -1882,6 +2067,7 @@ def _direct_device_first(info): "device_path": opened_path, }, ) + self._save_device_cache(pid, up, usage, source) return True continue # divert failed — try next receiver slot if not reprog_found: @@ -1905,16 +2091,21 @@ def _main_loop(self): """Outer loop: connect → listen → reconnect on error/disconnect.""" retry_logged = False while self._running: + attempt_t0 = time.monotonic() if not self._try_connect(): if not retry_logged: - print("[HidGesture] No compatible device; retrying in 5 s…") + print("[HidGesture] No compatible device; retrying every 2 s…") retry_logged = True - for _ in range(50): + print("[HidGesture] Connection attempt failed after " + f"{(time.monotonic() - attempt_t0) * 1000:.0f} ms") + for _ in range(20): if not self._running: return time.sleep(0.1) continue retry_logged = False + print("[HidGesture] Connected in " + f"{(time.monotonic() - attempt_t0) * 1000:.0f} ms") self._connected = True if self._on_connect: diff --git a/core/key_simulator.py b/core/key_simulator.py index 5511d8f..ce6f372 100644 --- a/core/key_simulator.py +++ b/core/key_simulator.py @@ -449,6 +449,11 @@ def _send(*events): "keys": [VK_LWIN, VK_TAB], "category": "Navigation", }, + "screenshot": { + "label": "Screenshot (Win+Shift+S)", + "keys": [VK_LWIN, VK_SHIFT, VK_S], + "category": "Navigation", + }, "space_left": { "label": "Previous Desktop", "keys": [VK_CONTROL, VK_LWIN, VK_LEFT], @@ -1059,6 +1064,11 @@ def _execute_mac_action(action_id): "keys": _MAC_ACTION_FALLBACKS["launchpad"], "category": "Navigation", }, + "screenshot": { + "label": "Screenshot (Cmd+Shift+4)", + "keys": [kVK_Command, kVK_Shift, kVK_ANSI_4], + "category": "Navigation", + }, "volume_up": { "label": "Volume Up", "keys": [], diff --git a/core/log_setup.py b/core/log_setup.py index 998d9e9..fbd772f 100644 --- a/core/log_setup.py +++ b/core/log_setup.py @@ -9,6 +9,7 @@ import os import sys import threading +from collections import deque def _get_log_dir() -> str: @@ -63,6 +64,75 @@ def isatty(self): return False +# ── In-memory log buffer ────────────────────────────────────────────── +# A bounded ring of recent formatted log lines, plus fan-out to live +# subscribers, so the UI can show the same console output that goes to +# mouser.log. Capture is always on (it is cheap); the UI only displays it +# while debug mode is enabled. +_LOG_BUFFER_CAPACITY = 1000 +_log_buffer = deque(maxlen=_LOG_BUFFER_CAPACITY) +_log_listeners = [] +_log_lock = threading.Lock() +_log_reentry = threading.local() + + +class _BufferLogHandler(logging.Handler): + """Keep the most recent formatted log lines in memory and notify + subscribers as each new line arrives. + + Subscriber callbacks run on whichever thread emitted the log record, so + they must be fast, non-blocking, and must not log or print themselves — + a re-entrancy guard drops nested notifications if one does anyway. + """ + + def emit(self, record): + try: + line = self.format(record) + except Exception: # pragma: no cover - logging must never crash + return + with _log_lock: + _log_buffer.append(line) + listeners = list(_log_listeners) + if getattr(_log_reentry, "active", False): + return + _log_reentry.active = True + try: + for listener in listeners: + try: + listener(line) + except Exception: + pass + finally: + _log_reentry.active = False + + +def get_recent_logs() -> list: + """Return a snapshot of the most recent captured log lines.""" + with _log_lock: + return list(_log_buffer) + + +def add_log_listener(callback) -> list: + """Register a callback to be invoked with each new formatted log line. + + Returns a snapshot of the lines already buffered, taken atomically with + the subscription so the caller sees every line exactly once — none + missed between the snapshot and the subscription, none delivered twice. + """ + with _log_lock: + _log_listeners.append(callback) + return list(_log_buffer) + + +def remove_log_listener(callback) -> None: + """Unregister a log listener added via add_log_listener (no-op if absent).""" + with _log_lock: + try: + _log_listeners.remove(callback) + except ValueError: + pass + + def setup_logging() -> str: """ Configure rotating file log and redirect stdout to it. @@ -107,6 +177,12 @@ def setup_logging() -> str: console_handler.setFormatter(fmt) root.addHandler(console_handler) + # In-memory ring buffer feeding the UI's "Application Log" view. Added + # unconditionally — it needs no file and works even if the log dir failed. + buffer_handler = _BufferLogHandler() + buffer_handler.setFormatter(fmt) + root.addHandler(buffer_handler) + root.setLevel(logging.DEBUG) # Redirect stdout — must come AFTER StreamHandler setup above. diff --git a/core/mouse_hook_base.py b/core/mouse_hook_base.py index 60e5623..44e93f9 100644 --- a/core/mouse_hook_base.py +++ b/core/mouse_hook_base.py @@ -3,7 +3,6 @@ """ import queue -import time try: from core.hid_gesture import HidGestureListener @@ -11,6 +10,16 @@ HidGestureListener = None from core.mouse_hook_types import HidRuntimeState, MouseEvent, format_debug_details +from core.gesture_recognizer import GestureRecognizer + + +# Recognizer swipe directions → dispatched MouseEvent types. +_SWIPE_EVENTS = { + "left": MouseEvent.GESTURE_SWIPE_LEFT, + "right": MouseEvent.GESTURE_SWIPE_RIGHT, + "up": MouseEvent.GESTURE_SWIPE_UP, + "down": MouseEvent.GESTURE_SWIPE_DOWN, +} class BaseMouseHook: @@ -30,18 +39,10 @@ def __init__(self): self.divert_mode_shift = False self.divert_dpi_switch = False self._gesture_direction_enabled = False - self._gesture_threshold = 50.0 - self._gesture_deadzone = 40.0 - self._gesture_timeout_ms = 3000 - self._gesture_cooldown_ms = 500 - self._gesture_tracking = False - self._gesture_triggered = False - self._gesture_started_at = 0.0 - self._gesture_last_move_at = 0.0 - self._gesture_delta_x = 0.0 - self._gesture_delta_y = 0.0 - self._gesture_cooldown_until = 0.0 - self._gesture_input_source = None + self._gesture_recognizer = GestureRecognizer( + on_swipe=self._on_recognized_swipe, + on_debug=self._emit_gesture_event, + ) self._connected_device = None self._dispatch_queue = None @@ -88,19 +89,18 @@ def configure_gestures( self, enabled=False, threshold=50, - deadzone=40, - timeout_ms=3000, - cooldown_ms=500, + commit_window_ms=400, + settle_ms=90, + cross_ratio=0.5, ): self._gesture_direction_enabled = bool(enabled) - self._gesture_threshold = float(max(5, threshold)) - self._gesture_deadzone = float(max(0, deadzone)) - self._gesture_timeout_ms = max(250, int(timeout_ms)) - self._gesture_cooldown_ms = max(0, int(cooldown_ms)) - if not self._gesture_direction_enabled: - self._gesture_tracking = False - self._gesture_triggered = False - self._gesture_input_source = None + self._gesture_recognizer.configure( + enabled=bool(enabled), + threshold=threshold, + commit_window_ms=commit_window_ms, + settle_ms=settle_ms, + cross_ratio=cross_ratio, + ) def set_connection_change_callback(self, cb): self._connection_change_cb = cb @@ -203,50 +203,6 @@ def _dispatch(self, event): def _hid_gesture_available(self): return self._hid_gesture is not None and self._device_connected - def _gesture_cooldown_active(self): - return time.monotonic() < self._gesture_cooldown_until - - def _start_gesture_tracking(self): - self._gesture_tracking = self._gesture_direction_enabled - self._gesture_started_at = time.monotonic() - self._gesture_last_move_at = self._gesture_started_at - self._gesture_delta_x = 0.0 - self._gesture_delta_y = 0.0 - self._gesture_input_source = None - - def _finish_gesture_tracking(self): - self._gesture_tracking = False - self._gesture_started_at = 0.0 - self._gesture_last_move_at = 0.0 - self._gesture_delta_x = 0.0 - self._gesture_delta_y = 0.0 - self._gesture_input_source = None - - def _detect_gesture_event(self): - delta_x = self._gesture_delta_x - delta_y = self._gesture_delta_y - - abs_x = abs(delta_x) - abs_y = abs(delta_y) - dominant = max(abs_x, abs_y) - if dominant < self._gesture_threshold: - return None - - cross_limit = max(self._gesture_deadzone, dominant * 0.35) - - if abs_x > abs_y: - if abs_y > cross_limit: - return None - if delta_x > 0: - return MouseEvent.GESTURE_SWIPE_RIGHT - return MouseEvent.GESTURE_SWIPE_LEFT - - if abs_x > cross_limit: - return None - if delta_y > 0: - return MouseEvent.GESTURE_SWIPE_DOWN - return MouseEvent.GESTURE_SWIPE_UP - def _build_extra_diverts(self): extra = {} if self.divert_mode_shift: @@ -295,13 +251,71 @@ def _on_hid_disconnect(self): self._set_device_connected(False) def _on_hid_gesture_down(self): - self._dispatch(MouseEvent(MouseEvent.GESTURE_DOWN)) + if getattr(self, "_ui_passthrough", False): + return + if self._gesture_active: + return + self._gesture_recognizer.begin() + self._gesture_active = True + self._emit_debug("HID gesture button down") + self._emit_gesture_event({"type": "button_down"}) def _on_hid_gesture_up(self): - self._dispatch(MouseEvent(MouseEvent.GESTURE_UP)) + if getattr(self, "_ui_passthrough", False): + return + if not self._gesture_active: + return + self._gesture_active = False + was_click = self._gesture_recognizer.end() + self._log_gesture_summary() + self._emit_debug( + f"HID gesture button up click_candidate={str(was_click).lower()}" + ) + self._emit_gesture_event( + {"type": "button_up", "click_candidate": was_click} + ) + if was_click: + self._dispatch(MouseEvent(MouseEvent.GESTURE_CLICK)) + + def _log_gesture_summary(self): + """Print a one-line trace of the gesture hold that just ended. + + Always on — one line per gesture-button press — so swipe-vs-click + behaviour is visible in ~/Library/Logs/Mouser/mouser.log. It is the + fastest way to tune the recognizer against real hardware. + """ + s = self._gesture_recognizer.summary() + outcome = "+".join(s["fired"]) if s["fired"] else "click" + print( + f"[Gesture] hold={s['duration_ms']:.0f}ms samples={s['samples']} " + f"net=({s['net_x']:+.0f},{s['net_y']:+.0f}) " + f"peak={s['peak_speed']:.0f}u/s src={s['source'] or '-'} " + f"-> {outcome}" + ) def _on_hid_gesture_move(self, dx, dy): - self._accumulate_gesture_delta(dx, dy, "hid_rawxy") + if getattr(self, "_ui_passthrough", False): + return + self._emit_gesture_event( + {"type": "move", "source": "hid_rawxy", "dx": dx, "dy": dy} + ) + self._gesture_recognizer.sample(dx, dy, "hid_rawxy") + + def _on_recognized_swipe(self, direction): + """Recognizer callback — dispatch a recognized swipe as a MouseEvent. + + Invoked on the HID listener thread (or, on macOS, the event-tap + thread, when raw XY is unavailable). + """ + event_type = _SWIPE_EVENTS.get(direction) + if event_type is not None: + self._emit_gesture_swipe(MouseEvent(event_type)) + + def _emit_gesture_swipe(self, mouse_event): + """Deliver a recognized swipe event. Platform hooks that own a + dedicated dispatch thread override this to hand the event off there + instead of dispatching inline.""" + self._dispatch(mouse_event) def _on_hid_mode_shift_down(self): self._dispatch(MouseEvent(MouseEvent.MODE_SHIFT_DOWN)) diff --git a/core/mouse_hook_contract.py b/core/mouse_hook_contract.py index c5e6397..541f734 100644 --- a/core/mouse_hook_contract.py +++ b/core/mouse_hook_contract.py @@ -24,9 +24,9 @@ def configure_gestures( self, enabled: bool = False, threshold: int = 50, - deadzone: int = 40, - timeout_ms: int = 3000, - cooldown_ms: int = 500, + commit_window_ms: int = 400, + settle_ms: int = 90, + cross_ratio: float = 0.5, ) -> None: ... def set_connection_change_callback(self, cb: Callable[[bool], None]) -> None: ... def set_debug_callback(self, callback: Callable[[str], None]) -> None: ... diff --git a/core/mouse_hook_linux.py b/core/mouse_hook_linux.py index ef7e9c9..47c7cd5 100644 --- a/core/mouse_hook_linux.py +++ b/core/mouse_hook_linux.py @@ -111,7 +111,6 @@ def __init__(self): self._evdev_ready = False self._hid_ready = False self._evdev_connected_device = None - self._gesture_lock = threading.Lock() self._evdev_device = None self._uinput = None self._evdev_thread = None @@ -305,160 +304,6 @@ def _refresh_device_state(self, force=False): def _hid_gesture_available(self): return self._hid_gesture is not None and self._evdev_ready - def _accumulate_gesture_delta(self, delta_x, delta_y, source): - dispatch_event = None - with self._gesture_lock: - if not (self._gesture_direction_enabled and self._gesture_active): - return - if self._gesture_cooldown_active(): - self._emit_debug( - f"Gesture cooldown active source={source} " - f"dx={delta_x} dy={delta_y}" - ) - self._emit_gesture_event( - { - "type": "cooldown_active", - "source": source, - "dx": delta_x, - "dy": delta_y, - } - ) - return - if not self._gesture_tracking: - self._emit_debug(f"Gesture tracking started source={source}") - self._emit_gesture_event( - { - "type": "tracking_started", - "source": source, - } - ) - self._start_gesture_tracking() - - now = time.monotonic() - idle_ms = (now - self._gesture_last_move_at) * 1000.0 - if idle_ms > self._gesture_timeout_ms: - self._emit_debug( - f"Gesture segment reset timeout source={source} " - f"accum_x={self._gesture_delta_x} accum_y={self._gesture_delta_y}" - ) - self._start_gesture_tracking() - - if source == "hid_rawxy" and self._gesture_input_source == "evdev": - self._emit_debug( - "Gesture source promoted from evdev to hid_rawxy " - f"prev_accum_x={self._gesture_delta_x} " - f"prev_accum_y={self._gesture_delta_y}" - ) - self._start_gesture_tracking() - - if self._gesture_input_source not in (None, source): - self._emit_debug( - f"Gesture source locked to {self._gesture_input_source}; " - f"ignoring {source} dx={delta_x} dy={delta_y}" - ) - return - self._gesture_input_source = source - - self._gesture_delta_x += delta_x - self._gesture_delta_y += delta_y - self._gesture_last_move_at = now - self._emit_debug( - f"Gesture segment source={source} " - f"accum_x={self._gesture_delta_x} accum_y={self._gesture_delta_y}" - ) - self._emit_gesture_event( - { - "type": "segment", - "source": source, - "dx": self._gesture_delta_x, - "dy": self._gesture_delta_y, - } - ) - - gesture_event = self._detect_gesture_event() - if not gesture_event: - return - - self._gesture_triggered = True - self._emit_debug( - "Gesture detected " - f"{gesture_event} source={source} " - f"delta_x={self._gesture_delta_x} delta_y={self._gesture_delta_y}" - ) - self._emit_gesture_event( - { - "type": "detected", - "event_name": gesture_event, - "source": source, - "dx": self._gesture_delta_x, - "dy": self._gesture_delta_y, - } - ) - dispatch_event = MouseEvent( - gesture_event, - { - "delta_x": self._gesture_delta_x, - "delta_y": self._gesture_delta_y, - "source": source, - }, - ) - self._gesture_cooldown_until = ( - time.monotonic() + self._gesture_cooldown_ms / 1000.0 - ) - self._emit_debug( - f"Gesture cooldown started source={source} " - f"for_ms={self._gesture_cooldown_ms}" - ) - self._emit_gesture_event( - { - "type": "cooldown_started", - "source": source, - "for_ms": self._gesture_cooldown_ms, - } - ) - self._finish_gesture_tracking() - - if dispatch_event: - self._dispatch(dispatch_event) - - def _on_hid_gesture_down(self): - if self._ui_passthrough: - return - with self._gesture_lock: - if not self._gesture_active: - self._gesture_active = True - self._gesture_triggered = False - self._emit_debug("HID gesture button down") - self._emit_gesture_event({"type": "button_down"}) - if self._gesture_direction_enabled and not self._gesture_cooldown_active(): - self._start_gesture_tracking() - else: - self._gesture_tracking = False - self._gesture_triggered = False - - def _on_hid_gesture_up(self): - if self._ui_passthrough: - return - dispatch_click = False - with self._gesture_lock: - if self._gesture_active: - should_click = not self._gesture_triggered - self._gesture_active = False - self._finish_gesture_tracking() - self._gesture_triggered = False - self._emit_debug( - f"HID gesture button up click_candidate={str(should_click).lower()}" - ) - self._emit_gesture_event( - { - "type": "button_up", - "click_candidate": should_click, - } - ) - dispatch_click = should_click - if dispatch_click: - self._dispatch(MouseEvent(MouseEvent.GESTURE_CLICK)) - def _on_hid_mode_shift_down(self): if self._ui_passthrough: return @@ -483,20 +328,6 @@ def _on_hid_dpi_switch_up(self): self._emit_debug("HID DPI switch button up") self._dispatch(MouseEvent(MouseEvent.DPI_SWITCH_UP)) - def _on_hid_gesture_move(self, delta_x, delta_y): - if self._ui_passthrough: - return - self._emit_debug(f"HID rawxy move dx={delta_x} dy={delta_y}") - self._emit_gesture_event( - { - "type": "move", - "source": "hid_rawxy", - "dx": delta_x, - "dy": delta_y, - } - ) - self._accumulate_gesture_delta(delta_x, delta_y, "hid_rawxy") - def _on_hid_connect(self): self._hid_ready = True self._refresh_device_state(force=True) @@ -519,8 +350,7 @@ def _on_hid_disconnect(self): self._hid_ready = False if self._gesture_active: self._gesture_active = False - self._finish_gesture_tracking() - self._gesture_triggered = False + self._gesture_recognizer.end() self._refresh_device_state(force=True) def _find_mouse_device(self): @@ -838,11 +668,10 @@ def _handle_rel(self, event): if code == _ecodes.REL_X or code == _ecodes.REL_Y: if self._gesture_direction_enabled and self._gesture_active: - if self._gesture_input_source != "hid_rawxy": - if code == _ecodes.REL_X: - self._accumulate_gesture_delta(value, 0, "evdev") - else: - self._accumulate_gesture_delta(0, value, "evdev") + if code == _ecodes.REL_X: + self._gesture_recognizer.sample(value, 0, "evdev") + else: + self._gesture_recognizer.sample(0, value, "evdev") return self._uinput.write_event(event) return diff --git a/core/mouse_hook_macos.py b/core/mouse_hook_macos.py index 2e971b7..67b911f 100644 --- a/core/mouse_hook_macos.py +++ b/core/mouse_hook_macos.py @@ -132,120 +132,6 @@ def _post_inverted_scroll_event(self, cg_event): Quartz.CGEventPost(Quartz.kCGHIDEventTap, inverted) return True - def _accumulate_gesture_delta(self, delta_x, delta_y, source): - if not (self._gesture_direction_enabled and self._gesture_active): - return - if self._gesture_cooldown_active(): - self._emit_debug( - f"Gesture cooldown active source={source} dx={delta_x} dy={delta_y}" - ) - self._emit_gesture_event( - { - "type": "cooldown_active", - "source": source, - "dx": delta_x, - "dy": delta_y, - } - ) - return - if not self._gesture_tracking: - self._emit_debug(f"Gesture tracking started source={source}") - self._emit_gesture_event( - { - "type": "tracking_started", - "source": source, - } - ) - self._start_gesture_tracking() - - now = time.monotonic() - idle_ms = (now - self._gesture_last_move_at) * 1000.0 - if idle_ms > self._gesture_timeout_ms: - self._emit_debug( - f"Gesture segment reset timeout source={source} " - f"accum_x={self._gesture_delta_x} accum_y={self._gesture_delta_y}" - ) - self._start_gesture_tracking() - - if source == "hid_rawxy" and self._gesture_input_source == "event_tap": - self._emit_debug( - "Gesture source promoted from event_tap to hid_rawxy " - f"prev_accum_x={self._gesture_delta_x} " - f"prev_accum_y={self._gesture_delta_y}" - ) - self._start_gesture_tracking() - - if self._gesture_input_source not in (None, source): - self._emit_debug( - f"Gesture source locked to {self._gesture_input_source}; " - f"ignoring {source} dx={delta_x} dy={delta_y}" - ) - return - self._gesture_input_source = source - - self._gesture_delta_x += delta_x - self._gesture_delta_y += delta_y - self._gesture_last_move_at = now - self._emit_debug( - f"Gesture segment source={source} " - f"accum_x={self._gesture_delta_x} accum_y={self._gesture_delta_y}" - ) - self._emit_gesture_event( - { - "type": "segment", - "source": source, - "dx": self._gesture_delta_x, - "dy": self._gesture_delta_y, - } - ) - - while True: - gesture_event = self._detect_gesture_event() - if not gesture_event: - return - - self._gesture_triggered = True - self._emit_debug( - "Gesture detected " - f"{gesture_event} source={source} " - f"delta_x={self._gesture_delta_x} delta_y={self._gesture_delta_y}" - ) - self._emit_gesture_event( - { - "type": "detected", - "event_name": gesture_event, - "source": source, - "dx": self._gesture_delta_x, - "dy": self._gesture_delta_y, - } - ) - self._enqueue_dispatch_event( - MouseEvent( - gesture_event, - { - "delta_x": self._gesture_delta_x, - "delta_y": self._gesture_delta_y, - "source": source, - }, - ) - ) - self._gesture_cooldown_until = ( - time.monotonic() + self._gesture_cooldown_ms / 1000.0 - ) - self._emit_debug( - f"Gesture cooldown started source={source} " - f"for_ms={self._gesture_cooldown_ms}" - ) - self._emit_gesture_event( - { - "type": "cooldown_started", - "source": source, - "for_ms": self._gesture_cooldown_ms, - } - ) - self._finish_gesture_tracking() - return - def _dispatch_worker(self): while self._running: try: @@ -295,35 +181,35 @@ def _event_tap_callback(self, proxy, event_type, cg_event, refcon): and self._gesture_direction_enabled and self._gesture_active ): + gesture_dx = Quartz.CGEventGetIntegerValueField( + cg_event, Quartz.kCGMouseEventDeltaX + ) + gesture_dy = Quartz.CGEventGetIntegerValueField( + cg_event, Quartz.kCGMouseEventDeltaY + ) self._emit_debug( - "Gesture move event " - f"type={int(event_type)} " - f"dx={Quartz.CGEventGetIntegerValueField(cg_event, Quartz.kCGMouseEventDeltaX)} " - f"dy={Quartz.CGEventGetIntegerValueField(cg_event, Quartz.kCGMouseEventDeltaY)}" + f"Gesture move event type={int(event_type)} " + f"dx={gesture_dx} dy={gesture_dy}" ) self._emit_gesture_event( { "type": "move", "source": "event_tap", - "dx": Quartz.CGEventGetIntegerValueField( - cg_event, Quartz.kCGMouseEventDeltaX - ), - "dy": Quartz.CGEventGetIntegerValueField( - cg_event, Quartz.kCGMouseEventDeltaY - ), + "dx": gesture_dx, + "dy": gesture_dy, } ) - if self._gesture_input_source == "hid_rawxy": - return None - self._accumulate_gesture_delta( - Quartz.CGEventGetIntegerValueField( - cg_event, Quartz.kCGMouseEventDeltaX - ), - Quartz.CGEventGetIntegerValueField( - cg_event, Quartz.kCGMouseEventDeltaY - ), - "event_tap", - ) + # Only use event-tap motion as the gesture source for devices + # that do NOT stream their own HID++ raw XY. When raw XY is + # active it is the source; feeding event-tap as well would + # double count the same motion, and the old source hand-off + # discarded the opening travel of a quick flick. The cursor + # is pinned (return None) during a gesture either way. + hg = self._hid_gesture + if not (hg is not None and getattr(hg, "_rawxy_enabled", False)): + self._gesture_recognizer.sample( + gesture_dx, gesture_dy, "event_tap" + ) return None if event_type == Quartz.kCGEventOtherMouseDown: @@ -419,36 +305,6 @@ def _event_tap_callback(self, proxy, event_type, cg_event, refcon): print(f"[MouseHook] event tap callback error: {exc}") return cg_event - def _on_hid_gesture_down(self): - if not self._gesture_active: - self._gesture_active = True - self._gesture_triggered = False - self._emit_debug("HID gesture button down") - self._emit_gesture_event({"type": "button_down"}) - if self._gesture_direction_enabled and not self._gesture_cooldown_active(): - self._start_gesture_tracking() - else: - self._gesture_tracking = False - self._gesture_triggered = False - - def _on_hid_gesture_up(self): - if self._gesture_active: - should_click = not self._gesture_triggered - self._gesture_active = False - self._finish_gesture_tracking() - self._gesture_triggered = False - self._emit_debug( - f"HID gesture button up click_candidate={str(should_click).lower()}" - ) - self._emit_gesture_event( - { - "type": "button_up", - "click_candidate": should_click, - } - ) - if should_click: - self._dispatch(MouseEvent(MouseEvent.GESTURE_CLICK)) - def _on_hid_mode_shift_down(self): self._emit_debug("HID mode shift button down") self._dispatch(MouseEvent(MouseEvent.MODE_SHIFT_DOWN)) @@ -465,17 +321,10 @@ def _on_hid_dpi_switch_up(self): self._emit_debug("HID DPI switch button up") self._dispatch(MouseEvent(MouseEvent.DPI_SWITCH_UP)) - def _on_hid_gesture_move(self, delta_x, delta_y): - self._emit_debug(f"HID rawxy move dx={delta_x} dy={delta_y}") - self._emit_gesture_event( - { - "type": "move", - "source": "hid_rawxy", - "dx": delta_x, - "dy": delta_y, - } - ) - self._accumulate_gesture_delta(delta_x, delta_y, "hid_rawxy") + def _emit_gesture_swipe(self, mouse_event): + # Hand the recognized swipe to the dispatch worker thread so that + # action execution never blocks the event-tap callback. + self._enqueue_dispatch_event(mouse_event) def _register_wake_observer(self): try: diff --git a/core/mouse_hook_windows.py b/core/mouse_hook_windows.py index efcee54..5c7e459 100644 --- a/core/mouse_hook_windows.py +++ b/core/mouse_hook_windows.py @@ -244,110 +244,6 @@ def __init__(self): self._init_dispatch_queue(maxsize=512) self._dispatch_worker_thread = None - def _accumulate_gesture_delta(self, delta_x, delta_y, source): - if not (self._gesture_direction_enabled and self._gesture_active): - return - if self._gesture_cooldown_active(): - self._emit_debug( - f"Gesture cooldown active source={source} dx={delta_x} dy={delta_y}" - ) - self._emit_gesture_event( - { - "type": "cooldown_active", - "source": source, - "dx": delta_x, - "dy": delta_y, - } - ) - return - if not self._gesture_tracking: - self._emit_debug(f"Gesture tracking started source={source}") - self._emit_gesture_event( - { - "type": "tracking_started", - "source": source, - } - ) - self._start_gesture_tracking() - - now = time.monotonic() - idle_ms = (now - self._gesture_last_move_at) * 1000.0 - if idle_ms > self._gesture_timeout_ms: - self._emit_debug( - f"Gesture segment reset timeout source={source} " - f"accum_x={self._gesture_delta_x} accum_y={self._gesture_delta_y}" - ) - self._start_gesture_tracking() - - if self._gesture_input_source not in (None, source): - self._emit_debug( - f"Gesture source locked to {self._gesture_input_source}; " - f"ignoring {source} dx={delta_x} dy={delta_y}" - ) - return - self._gesture_input_source = source - - self._gesture_delta_x += delta_x - self._gesture_delta_y += delta_y - self._gesture_last_move_at = now - self._emit_debug( - f"Gesture segment source={source} " - f"accum_x={self._gesture_delta_x} accum_y={self._gesture_delta_y}" - ) - self._emit_gesture_event( - { - "type": "segment", - "source": source, - "dx": self._gesture_delta_x, - "dy": self._gesture_delta_y, - } - ) - - gesture_event = self._detect_gesture_event() - if not gesture_event: - return - - self._gesture_triggered = True - self._emit_debug( - "Gesture detected " - f"{gesture_event} source={source} " - f"delta_x={self._gesture_delta_x} delta_y={self._gesture_delta_y}" - ) - self._emit_gesture_event( - { - "type": "detected", - "event_name": gesture_event, - "source": source, - "dx": self._gesture_delta_x, - "dy": self._gesture_delta_y, - } - ) - self._dispatch( - MouseEvent( - gesture_event, - { - "delta_x": self._gesture_delta_x, - "delta_y": self._gesture_delta_y, - "source": source, - }, - ) - ) - self._gesture_cooldown_until = ( - time.monotonic() + self._gesture_cooldown_ms / 1000.0 - ) - self._emit_debug( - f"Gesture cooldown started source={source} " - f"for_ms={self._gesture_cooldown_ms}" - ) - self._emit_gesture_event( - { - "type": "cooldown_started", - "source": source, - "for_ms": self._gesture_cooldown_ms, - } - ) - self._finish_gesture_tracking() - _WM_NAMES = { 0x0200: "WM_MOUSEMOVE", 0x0201: "WM_LBUTTONDOWN", @@ -558,15 +454,11 @@ def _check_raw_mouse_gesture(self, hDevice, buffer): if extra_now == extra_prev: return if extra_now and not extra_prev: - if not self._gesture_active: - self._gesture_active = True - self._gesture_triggered = False - print(f"[MouseHook] Gesture DOWN (rawBtns extra: 0x{extra_now:X})") + print(f"[MouseHook] Gesture DOWN (rawBtns extra: 0x{extra_now:X})") + self._on_hid_gesture_down() elif not extra_now and extra_prev: - if self._gesture_active: - self._gesture_active = False - print("[MouseHook] Gesture UP") - self._dispatch(MouseEvent(MouseEvent.GESTURE_CLICK)) + print("[MouseHook] Gesture UP") + self._on_hid_gesture_up() def _setup_raw_input(self): instance = GetModuleHandleW(None) @@ -704,36 +596,6 @@ def _reinstall_hook(self): else: print("[MouseHook] Failed to reinstall hook!") - def _on_hid_gesture_down(self): - if not self._gesture_active: - self._gesture_active = True - self._gesture_triggered = False - self._emit_debug("HID gesture button down") - self._emit_gesture_event({"type": "button_down"}) - if self._gesture_direction_enabled and not self._gesture_cooldown_active(): - self._start_gesture_tracking() - else: - self._gesture_tracking = False - self._gesture_triggered = False - - def _on_hid_gesture_up(self): - if self._gesture_active: - should_click = not self._gesture_triggered - self._gesture_active = False - self._finish_gesture_tracking() - self._gesture_triggered = False - self._emit_debug( - f"HID gesture button up click_candidate={str(should_click).lower()}" - ) - self._emit_gesture_event( - { - "type": "button_up", - "click_candidate": should_click, - } - ) - if should_click: - self._dispatch(MouseEvent(MouseEvent.GESTURE_CLICK)) - def _on_hid_mode_shift_down(self): self._emit_debug("HID mode shift button down") self._dispatch(MouseEvent(MouseEvent.MODE_SHIFT_DOWN)) @@ -750,18 +612,6 @@ def _on_hid_dpi_switch_up(self): self._emit_debug("HID DPI switch button up") self._dispatch(MouseEvent(MouseEvent.DPI_SWITCH_UP)) - def _on_hid_gesture_move(self, delta_x, delta_y): - self._emit_debug(f"HID rawxy move dx={delta_x} dy={delta_y}") - self._emit_gesture_event( - { - "type": "move", - "source": "hid_rawxy", - "dx": delta_x, - "dy": delta_y, - } - ) - self._accumulate_gesture_delta(delta_x, delta_y, "hid_rawxy") - def start(self): if self._hook_thread and self._hook_thread.is_alive(): return True diff --git a/install_macos_app.sh b/install_macos_app.sh new file mode 100644 index 0000000..b63309b --- /dev/null +++ b/install_macos_app.sh @@ -0,0 +1,94 @@ +#!/bin/zsh +# +# install_macos_app.sh — build Mouser.app and install it into /Applications. +# +# Thin wrapper around build_macos_app.sh: it runs the PyInstaller build +# (producing dist/Mouser.app), then replaces any existing install in +# /Applications with the fresh bundle and launches it. +# +# It honors the same environment variables as build_macos_app.sh +# (MOUSER_PYTHON, MOUSER_SIGN_IDENTITY, PYINSTALLER_TARGET_ARCH) since +# those are simply inherited by the child process. +# +# The build always runs on local disk: the repo may sit on a network volume, +# and codesign rejects the resource-fork metadata SMB/AFP shares attach to +# files (see build_macos_app.sh). The finished bundle is then copied into the +# Applications folder, which is on the Mac's internal drive. +# +# Usage: +# ./install_macos_app.sh # build + install + launch +# MOUSER_INSTALL_DIR="$HOME/Applications" ./install_macos_app.sh +# MOUSER_BUILD_DIR="$HOME/mouser-build" ./install_macos_app.sh +# MOUSER_NO_LAUNCH=1 ./install_macos_app.sh # install without launching +# +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "$0")" && pwd)" +APP_NAME="Mouser.app" +# Build artifacts go to local disk; tell build_macos_app.sh to use the same +# location so we know exactly where the finished bundle lands. +BUILD_OUTPUT_DIR="${MOUSER_BUILD_DIR:-$HOME/Library/Caches/Mouser/macos-build}" +export MOUSER_BUILD_DIR="$BUILD_OUTPUT_DIR" +BUILT_APP="$BUILD_OUTPUT_DIR/dist/$APP_NAME" +DEST_DIR="${MOUSER_INSTALL_DIR:-/Applications}" +DEST_APP="$DEST_DIR/$APP_NAME" + +if [[ "$(uname -s)" != "Darwin" ]]; then + echo "ERROR: this installer must be run on macOS." >&2 + exit 1 +fi + +# ── 1. Build ──────────────────────────────────────────────────────── +echo "==> Building $APP_NAME" +"$ROOT_DIR/build_macos_app.sh" + +if [[ ! -d "$BUILT_APP" ]]; then + echo "ERROR: build did not produce $BUILT_APP" >&2 + exit 1 +fi + +# ── 2. Quit any running copy so the bundle can be replaced cleanly ── +if pgrep -x Mouser >/dev/null 2>&1; then + echo "==> Quitting running Mouser" + osascript -e 'quit app "Mouser"' >/dev/null 2>&1 || true + for _ in 1 2 3 4 5 6; do + pgrep -x Mouser >/dev/null 2>&1 || break + sleep 0.5 + done + pkill -x Mouser >/dev/null 2>&1 || true +fi + +# ── 3. Install into the Applications folder ───────────────────────── +if [[ ! -d "$DEST_DIR" ]]; then + echo "ERROR: install directory does not exist: $DEST_DIR" >&2 + exit 1 +fi +if [[ ! -w "$DEST_DIR" ]]; then + echo "ERROR: $DEST_DIR is not writable by this user." >&2 + echo " Install to a user-owned folder instead, e.g.:" >&2 + echo " MOUSER_INSTALL_DIR=\"\$HOME/Applications\" $0" >&2 + exit 1 +fi + +echo "==> Installing to $DEST_APP" +rm -rf "$DEST_APP" +# ditto preserves bundle symlinks, extended attributes and the code signature. +ditto "$BUILT_APP" "$DEST_APP" + +echo "==> Installed: $DEST_APP" + +# ── 4. Launch (unless opted out) ──────────────────────────────────── +if [[ -z "${MOUSER_NO_LAUNCH:-}" ]]; then + echo "==> Launching Mouser" + open "$DEST_APP" +fi + +cat <<'EOF' + +Done. This is an ad-hoc-signed local build, so the first launch of a new +build may require re-granting Accessibility permission: + System Settings -> Privacy & Security -> Accessibility +If a stale "Mouser" entry is already listed there, remove it (the - button) +and re-add the new bundle, otherwise macOS may keep using the old permission +record and remapping will appear dead. +EOF diff --git a/main_qml.py b/main_qml.py index 40377f5..89b6cb1 100644 --- a/main_qml.py +++ b/main_qml.py @@ -47,7 +47,7 @@ def _resolve_root_dir(): _t1 = _time.perf_counter() from PySide6.QtWidgets import QApplication, QSystemTrayIcon, QMenu, QFileIconProvider, QMessageBox -from PySide6.QtGui import QAction, QColor, QIcon, QPainter, QPixmap, QWindow +from PySide6.QtGui import QAction, QColor, QDesktopServices, QIcon, QPainter, QPixmap, QWindow from PySide6.QtCore import QObject, Property, QCoreApplication, QRectF, Qt, QUrl, Signal, QFileInfo, QEvent, QTimer from PySide6.QtQml import QQmlApplicationEngine from PySide6.QtQuick import QQuickImageProvider @@ -346,6 +346,10 @@ def _rename_macos_bundle_for_dock(): _MACOS_NATIVE_STATUS_ITEM = None _MACOS_NATIVE_STATUS_TARGET = None _MACOS_QUIT_FILTER = None +# Held for the process lifetime so the non-modal Accessibility prompt and +# its grant-watcher QTimer are not garbage-collected before they fire. +_ACCESSIBILITY_PROMPT = None +_ACCESSIBILITY_WATCHER = None _MACOS_SYSTEM_QUIT_REASONS = { "quia", # kAEQuitAll "shut", # kAEShutDown @@ -879,29 +883,21 @@ def requestPixmap(self, icon_id, size, requested_size): return pixmap -def _check_accessibility(locale_mgr: "LocaleManager") -> bool: - """Verify the macOS Accessibility grant. Returns True only when - AXIsProcessTrustedWithOptions confirms the grant; any other path - (no grant, exception during the check) returns False so callers - fail closed. +def _check_accessibility() -> bool: + """Return True when the macOS Accessibility grant is in place. + + ``prompt=True`` asks macOS to show its own grant dialog and to list + Mouser under Accessibility, so granting it is a single toggle. Any + failure path returns False so callers fail closed -- the engine + stays off until the grant is actually confirmed. """ if sys.platform != "darwin": return True try: - trusted = is_process_trusted(prompt=True) + return bool(is_process_trusted(prompt=True)) except Exception as exc: print(f"[Mouser] Accessibility check failed: {exc}") return False - if not trusted: - print("[Mouser] Accessibility permission not granted") - msg = QMessageBox() - msg.setIcon(QMessageBox.Icon.Warning) - msg.setWindowTitle(locale_mgr.tr("accessibility.title")) - msg.setText(locale_mgr.tr("accessibility.text")) - msg.setInformativeText(locale_mgr.tr("accessibility.info")) - msg.setStandardButtons(QMessageBox.StandardButton.Ok) - msg.exec() - return bool(trusted) def _runtime_launch_path() -> str: @@ -921,6 +917,120 @@ def _schedule_engine_start(engine, *, accessibility_granted: bool) -> bool: return True +def _open_accessibility_settings() -> None: + """Deep-link into System Settings -> Privacy & Security -> Accessibility.""" + QDesktopServices.openUrl(QUrl( + "x-apple.systempreferences:com.apple.preference.security" + "?Privacy_Accessibility" + )) + + +def _show_accessibility_prompt(locale_mgr) -> QMessageBox: + """Show the non-blocking "grant Accessibility" notice. + + Deliberately non-modal: the grant watcher relaunches Mouser the + moment permission lands, so this dialog never needs an answer. It is + a signpost that explains the wait, plus a one-click shortcut into + the right System Settings pane. + """ + print("[Mouser] Accessibility permission not granted -- watching for grant") + msg = QMessageBox() + msg.setIcon(QMessageBox.Icon.Warning) + msg.setWindowTitle(locale_mgr.tr("accessibility.title")) + msg.setText(locale_mgr.tr("accessibility.text")) + msg.setInformativeText(locale_mgr.tr("accessibility.info")) + open_button = msg.addButton( + locale_mgr.tr("accessibility.open_settings"), + QMessageBox.ButtonRole.ActionRole, + ) + msg.addButton(QMessageBox.StandardButton.Close) + msg.setDefaultButton(open_button) + + def _on_button_clicked(button): + if button is open_button: + _open_accessibility_settings() + + msg.buttonClicked.connect(_on_button_clicked) + msg.setModal(False) + msg.show() + msg.raise_() + msg.activateWindow() + return msg + + +def _relaunch_app(single_server=None) -> None: + """Re-exec Mouser so every subsystem restarts with the current + macOS permission grants in place. + + macOS only hands a process its Accessibility (TCC) grant when the + process asks at startup; a CGEventTap that failed at the original + launch never revives in place. A clean re-exec is the reliable fix. + """ + # os.execv keeps the PID and inherits open file descriptors, so the + # single-instance socket would otherwise stay bound -- the re-exec'd + # process would then connect to itself and exit as a duplicate. + # Drop the lock first; the fresh process re-acquires it cleanly. + if single_server is not None: + try: + single_server.close() + except Exception as exc: + print(f"[Mouser] Could not release single-instance lock: {exc}") + target = sys.executable + if getattr(sys, "frozen", False): + # Frozen build: argv[0] is already the executable itself. + argv = [target, *sys.argv[1:]] + else: + # Source run: hand the interpreter back its script and args. + argv = [target, *sys.argv] + print(f"[Mouser] Relaunching to apply updated permissions: {target}") + try: + os.execv(target, argv) + except OSError as exc: + print(f"[Mouser] Relaunch failed: {exc}") + + +def _accessibility_grant_poll(timer, single_server) -> bool: + """Run one tick of the Accessibility-grant watcher. + + Returns True once the grant is observed (and the relaunch is + triggered), False while still waiting. Kept separate from the timer + wiring so it can be unit-tested without a running event loop. + """ + try: + granted = is_process_trusted(prompt=False) + except Exception as exc: + print(f"[Mouser] Accessibility re-check failed: {exc}") + return False + if not granted: + return False + # Stop before relaunching: if the re-exec somehow fails, the watcher + # must not keep firing it once per second. + timer.stop() + print("[Mouser] Accessibility granted -- relaunching Mouser") + _relaunch_app(single_server) + return True + + +_ACCESSIBILITY_POLL_INTERVAL_MS = 1000 + + +def _watch_accessibility_grant(app, single_server) -> QTimer: + """Poll for the Accessibility grant and relaunch Mouser once it lands. + + Without this, Mouser sits idle after the user grants permission -- + macOS does not retro-activate the grant for the already-running + process, so the user would otherwise have to quit and reopen it by + hand. + """ + timer = QTimer(app) + timer.setInterval(_ACCESSIBILITY_POLL_INTERVAL_MS) + timer.timeout.connect( + lambda: _accessibility_grant_poll(timer, single_server) + ) + timer.start() + return timer + + def _schedule_tray_minimized_notice(tray, locale_mgr) -> None: def _tray_minimized_notice(): tray.showMessage( @@ -1112,13 +1222,23 @@ def _on_second_instance_activate(): print(f"[Startup] TOTAL to window: {(_t8-_t0)*1000:7.1f} ms") # ── Accessibility check (macOS) ────────────────────────────── - accessibility_granted = _check_accessibility(locale_mgr) + accessibility_granted = _check_accessibility() if sys.platform == "linux": engine.set_ui_passthrough(not launch_hidden) # ── Start engine AFTER window is ready (deferred) ────────── - _schedule_engine_start(engine, accessibility_granted=accessibility_granted) + engine_started = _schedule_engine_start( + engine, accessibility_granted=accessibility_granted + ) + if not engine_started and sys.platform == "darwin": + # The Accessibility grant is missing, so the engine cannot run. + # Rather than leave Mouser idle until the user quits and reopens + # it by hand, prompt for the grant and relaunch automatically the + # moment it is given. + global _ACCESSIBILITY_PROMPT, _ACCESSIBILITY_WATCHER + _ACCESSIBILITY_PROMPT = _show_accessibility_prompt(locale_mgr) + _ACCESSIBILITY_WATCHER = _watch_accessibility_grant(app, single_server) # ── System Tray ──────────────────────────────────────────── tray = QSystemTrayIcon(_tray_icon(), app) diff --git a/tests/test_backend.py b/tests/test_backend.py index 1310276..41e5a12 100644 --- a/tests/test_backend.py +++ b/tests/test_backend.py @@ -7,8 +7,9 @@ from types import SimpleNamespace from unittest.mock import patch -from core.config import DEFAULT_CONFIG +from core.config import DEFAULT_CONFIG, GESTURE_SENSITIVITY_PX from core.updater import UpdateCheckState +from core import log_setup try: from PySide6.QtCore import QCoreApplication, Qt, QUrl @@ -1273,5 +1274,90 @@ def test_set_start_minimized_does_not_call_apply_login_startup(self): self.assertFalse(backend.startMinimized) +@unittest.skipIf(Backend is None, "PySide6 not installed in test environment") +class BackendAppLogTests(unittest.TestCase): + """The Application Log view — captured console output mirrored to QML.""" + + def tearDown(self): + # Backend.__init__ subscribes to log_setup; drop the test's listeners. + log_setup._log_listeners.clear() + + def _make_backend(self): + with ( + patch("ui.backend.load_config", + return_value=copy.deepcopy(DEFAULT_CONFIG)), + patch("ui.backend.save_config"), + patch("ui.backend.supports_login_startup", return_value=False), + ): + return Backend(engine=None) + + def test_init_subscribes_to_log_capture(self): + before = len(log_setup._log_listeners) + self._make_backend() + self.assertEqual(len(log_setup._log_listeners), before + 1) + + def test_handle_app_log_message_appends_to_app_log(self): + backend = self._make_backend() + backend.clearAppLog() + backend._handleAppLogMessage("[HidGesture] Connected in 1200 ms") + self.assertIn("[HidGesture] Connected in 1200 ms", backend.appLog) + + def test_clear_app_log_empties_the_view(self): + backend = self._make_backend() + backend._handleAppLogMessage("[Engine] Scroll mode changed") + self.assertNotEqual(backend.appLog, "") + backend.clearAppLog() + self.assertEqual(backend.appLog, "") + + def test_app_log_is_bounded(self): + backend = self._make_backend() + backend.clearAppLog() + for i in range(1100): + backend._handleAppLogMessage(f"[Test] line {i}") + self.assertLessEqual(len(backend.appLog.split("\n")), 1000) + + +@unittest.skipIf(Backend is None, "PySide6 not available") +class BackendGestureSensitivityTests(unittest.TestCase): + """The discrete swipe-sensitivity selector that replaced the px slider.""" + + def _make_backend(self, threshold=None): + cfg = copy.deepcopy(DEFAULT_CONFIG) + if threshold is not None: + cfg["settings"]["gesture_threshold"] = threshold + with ( + patch("ui.backend.load_config", return_value=cfg), + patch("ui.backend.save_config"), + patch("ui.backend.supports_login_startup", return_value=False), + ): + return Backend(engine=None) + + def test_default_config_lands_on_the_leaning_sensitive_preset(self): + backend = self._make_backend() + self.assertEqual(backend.gestureSensitivityIndex, 1) + + def test_index_snaps_to_nearest_preset_for_off_grid_threshold(self): + # 51 px is nearer the 56 px preset (index 4) than the 44 px one. + backend = self._make_backend(threshold=51) + self.assertEqual(backend.gestureSensitivityIndex, 4) + + def test_set_sensitivity_stores_the_preset_pixel_value(self): + backend = self._make_backend() + backend.setGestureSensitivity(0) + self.assertEqual( + backend._cfg["settings"]["gesture_threshold"], + GESTURE_SENSITIVITY_PX[0], + ) + self.assertEqual(backend.gestureSensitivityIndex, 0) + + def test_set_sensitivity_clamps_out_of_range_index(self): + backend = self._make_backend() + backend.setGestureSensitivity(99) + self.assertEqual( + backend._cfg["settings"]["gesture_threshold"], + GESTURE_SENSITIVITY_PX[-1], + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py index 54cb498..8bb858d 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -43,15 +43,15 @@ def test_migrate_v1_config_adds_profile_apps_and_gesture_defaults(self): migrated = config._migrate(legacy) - self.assertEqual(migrated["version"], 9) + self.assertEqual(migrated["version"], 10) self.assertEqual(migrated["profiles"]["default"]["apps"], []) self.assertFalse(migrated["settings"]["invert_hscroll"]) self.assertFalse(migrated["settings"]["invert_vscroll"]) self.assertEqual(migrated["settings"]["dpi"], 1000) - self.assertEqual(migrated["settings"]["gesture_threshold"], 50) - self.assertEqual(migrated["settings"]["gesture_deadzone"], 40) - self.assertEqual(migrated["settings"]["gesture_timeout_ms"], 3000) - self.assertEqual(migrated["settings"]["gesture_cooldown_ms"], 500) + self.assertEqual(migrated["settings"]["gesture_threshold"], 25) + self.assertEqual(migrated["settings"]["gesture_commit_window_ms"], 400) + self.assertEqual(migrated["settings"]["gesture_settle_ms"], 90) + self.assertEqual(migrated["settings"]["gesture_cross_ratio"], 0.5) self.assertEqual(migrated["settings"]["appearance_mode"], "system") self.assertFalse(migrated["settings"]["debug_mode"]) self.assertEqual(migrated["settings"]["device_layout_overrides"], {}) @@ -88,7 +88,7 @@ def test_migrate_updates_media_player_profile_apps(self): migrated = config._migrate(cfg) - self.assertEqual(migrated["version"], 9) + self.assertEqual(migrated["version"], 10) self.assertEqual( migrated["profiles"]["media"]["apps"], ["Microsoft.Media.Player.exe", "VLC.exe"], @@ -130,10 +130,10 @@ def test_load_config_merges_missing_defaults_from_disk(self): ): loaded = config.load_config() - self.assertEqual(loaded["version"], 9) + self.assertEqual(loaded["version"], 10) self.assertEqual(loaded["settings"]["dpi"], 800) self.assertFalse(loaded["settings"]["start_at_login"]) - self.assertEqual(loaded["settings"]["gesture_threshold"], 50) + self.assertEqual(loaded["settings"]["gesture_threshold"], 25) self.assertEqual(loaded["settings"]["appearance_mode"], "system") self.assertFalse(loaded["settings"]["debug_mode"]) self.assertEqual(loaded["settings"]["device_layout_overrides"], {}) @@ -157,7 +157,7 @@ def test_migrate_renames_start_with_windows_to_start_at_login(self): migrated = config._migrate(legacy) - self.assertEqual(migrated["version"], 9) + self.assertEqual(migrated["version"], 10) self.assertTrue(migrated["settings"]["start_at_login"]) self.assertEqual( migrated["profiles"]["default"]["mappings"]["mode_shift"], @@ -552,5 +552,49 @@ def test_resolve_app_spec_for_linux_runtime_path_prefers_catalog_entry(self): self.assertIn("/usr/lib64/firefox/firefox", resolved["aliases"]) +class DeviceCacheTests(unittest.TestCase): + """config.load_device_cache / save_device_cache — the last-good HID++ + device identity persisted so reconnects skip the slow device scan.""" + + @contextmanager + def _temp_cache(self): + """Patch the config dir + cache path onto a throwaway directory.""" + with tempfile.TemporaryDirectory() as temp_dir: + cache_file = os.path.join(temp_dir, "device_cache.json") + with ( + patch.object(config, "CONFIG_DIR", temp_dir), + patch.object(config, "DEVICE_CACHE_FILE", cache_file), + ): + yield cache_file + + def test_load_returns_none_when_no_cache_file_exists(self): + with self._temp_cache(): + self.assertIsNone(config.load_device_cache()) + + def test_save_then_load_round_trips_the_identity(self): + identity = { + "product_id": 0xC548, + "usage_page": 0xFF00, + "usage": 1, + "source": "hidapi-enumerate", + "dev_idx": 2, + } + with self._temp_cache(): + config.save_device_cache(identity) + self.assertEqual(config.load_device_cache(), identity) + + def test_load_returns_none_on_corrupt_cache_file(self): + with self._temp_cache() as cache_file: + with open(cache_file, "w", encoding="utf-8") as fh: + fh.write("{not valid json") + self.assertIsNone(config.load_device_cache()) + + def test_load_returns_none_when_cache_is_not_a_dict(self): + with self._temp_cache() as cache_file: + with open(cache_file, "w", encoding="utf-8") as fh: + json.dump([1, 2, 3], fh) + self.assertIsNone(config.load_device_cache()) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_gesture_recognizer.py b/tests/test_gesture_recognizer.py new file mode 100644 index 0000000..d08dc4d --- /dev/null +++ b/tests/test_gesture_recognizer.py @@ -0,0 +1,274 @@ +"""Tests for core.gesture_recognizer.GestureRecognizer. + +These exercise the behaviour the recognizer exists to guarantee: + + * a clean swipe fires exactly once; + * back-to-back flicks in one hold each register; + * the return stroke between flicks never fires the opposite swipe; + * slow drift while holding the button is ignored; + * there is no dead window swallowing the start of the next swipe. +""" + +from core.gesture_recognizer import GestureRecognizer, LEFT, RIGHT, UP, DOWN + + +# ── helpers ────────────────────────────────────────────────────────── + +def make(enabled=True, threshold=50, commit_window_ms=400, + settle_ms=90, cross_ratio=0.5): + """Build a recognizer that records every swipe it emits in ``.swipes``.""" + swipes = [] + rec = GestureRecognizer(on_swipe=swipes.append) + rec.configure( + enabled=enabled, + threshold=threshold, + commit_window_ms=commit_window_ms, + settle_ms=settle_ms, + cross_ratio=cross_ratio, + ) + rec.swipes = swipes + return rec + + +def feed(rec, deltas, t, step=0.012, gap=None, source="hid_rawxy"): + """Sample each (dx, dy) in turn. + + The first sample lands ``gap`` (default ``step``) seconds after ``t``; + every following sample is ``step`` seconds later. ``step`` (12 ms) is far + below the 90 ms settle window, so a single feed() is one continuous + stroke. Returns the final timestamp. + """ + for i, (dx, dy) in enumerate(deltas): + t += (gap if (gap is not None and i == 0) else step) + rec.sample(dx, dy, source, now=t) + return t + + +# A flick is a fast, continuous run of deltas covering well past the +# 50 px commit distance; a return is the same run reversed. +FLICK_LEFT = [(-10, 0)] * 7 +FLICK_RIGHT = [(10, 0)] * 7 +FLICK_UP = [(0, -10)] * 7 +FLICK_DOWN = [(0, 10)] * 7 +RETURN_RIGHT = [(10, 0)] * 7 +RETURN_LEFT = [(-10, 0)] * 7 + + +# ── basic recognition ──────────────────────────────────────────────── + +def test_single_left_swipe_fires_once(): + rec = make() + rec.begin() + feed(rec, FLICK_LEFT, t=0.0) + assert rec.end() is False # a swipe happened, not a click + assert rec.swipes == [LEFT] + + +def test_each_direction_is_recognised(): + for flick, expected in ( + (FLICK_LEFT, LEFT), + (FLICK_RIGHT, RIGHT), + (FLICK_UP, UP), + (FLICK_DOWN, DOWN), + ): + rec = make() + rec.begin() + feed(rec, flick, t=0.0) + rec.end() + assert rec.swipes == [expected] + + +def test_one_long_continuous_motion_fires_only_once(): + """Holding a single sweep across the desk is one swipe, not many.""" + rec = make() + rec.begin() + feed(rec, [(-10, 0)] * 40, t=0.0) # 400 px in one continuous push + rec.end() + assert rec.swipes == [LEFT] + + +# ── false-positive rejection ───────────────────────────────────────── + +def test_slow_drift_does_not_fire(): + """Creeping the mouse while merely holding the button is not a swipe.""" + rec = make() + rec.begin() + # 70 px over 840 ms (~83 px/s) — well under the commit-speed gate. + feed(rec, [(-1, 0)] * 70, t=0.0) + rec.end() + assert rec.swipes == [] + + +def test_small_movement_below_threshold_does_not_fire(): + rec = make() + rec.begin() + feed(rec, [(-6, 0)] * 5, t=0.0) # 30 px total, never reaches 50 + rec.end() + assert rec.swipes == [] + + +def test_steep_diagonal_is_rejected(): + rec = make() + rec.begin() + feed(rec, [(-9, -9)] * 8, t=0.0) # 45 degrees — too ambiguous + rec.end() + assert rec.swipes == [] + + +def test_mild_diagonal_still_resolves_to_dominant_axis(): + rec = make() + rec.begin() + feed(rec, [(-10, -3)] * 7, t=0.0) # mostly left, slight up + rec.end() + assert rec.swipes == [LEFT] + + +# ── back-to-back repeats (the headline scenario) ───────────────────── + +def test_repeated_left_flicks_in_one_hold_each_fire(): + """Hold the button and flick left, return, flick left, return, flick + left — all in one continuous motion. Every flick must register and no + return stroke may fire a right swipe.""" + rec = make() + rec.begin() + t = 0.0 + t = feed(rec, FLICK_LEFT, t) # flick 1 + t = feed(rec, RETURN_RIGHT, t) # return (must not fire RIGHT) + t = feed(rec, FLICK_LEFT, t) # flick 2 + t = feed(rec, RETURN_RIGHT, t) # return (must not fire RIGHT) + t = feed(rec, FLICK_LEFT, t) # flick 3 + rec.end() + assert rec.swipes == [LEFT, LEFT, LEFT] + + +def test_return_stroke_alone_never_fires_opposite(): + """A single flick followed only by a big, fast return to home position + yields exactly one swipe.""" + rec = make() + rec.begin() + t = feed(rec, FLICK_LEFT, t=0.0) + feed(rec, [(12, 0)] * 9, t) # 108 px fast return right + rec.end() + assert rec.swipes == [LEFT] + + +def test_repeated_flicks_across_separate_holds(): + """Press / flick / release, repeated quickly, fires every time — there + is no cooldown bleeding across button presses.""" + rec = make() + t = 0.0 + for _ in range(4): + rec.begin() + t = feed(rec, FLICK_LEFT, t) + assert rec.end() is False + t += 0.03 # tiny pause between presses + assert rec.swipes == [LEFT, LEFT, LEFT, LEFT] + + +def test_direction_change_after_a_pause(): + """Once the motion settles, the hold unlocks and a new direction may + be swiped.""" + rec = make() + rec.begin() + t = feed(rec, FLICK_LEFT, t=0.0) + # Pause longer than the 90 ms settle window, then flick the other way. + feed(rec, FLICK_RIGHT, t, gap=0.20) + rec.end() + assert rec.swipes == [LEFT, RIGHT] + + +def test_opposite_flick_within_a_locked_hold_is_absorbed(): + """Without a pause, the hold stays locked to the first direction, so a + reversed flick is treated as a return stroke rather than a swipe.""" + rec = make() + rec.begin() + t = feed(rec, FLICK_LEFT, t=0.0) + feed(rec, FLICK_RIGHT, t) # continuous — no settle pause + rec.end() + assert rec.swipes == [LEFT] + + +# ── click vs swipe ─────────────────────────────────────────────────── + +def test_press_release_without_motion_is_a_click(): + rec = make() + rec.begin() + assert rec.end() is True + + +def test_press_release_with_tiny_motion_is_a_click(): + rec = make() + rec.begin() + feed(rec, [(-3, 0)] * 3, t=0.0) + assert rec.end() is True + + +def test_hold_with_a_swipe_is_not_a_click(): + rec = make() + rec.begin() + feed(rec, FLICK_LEFT, t=0.0) + assert rec.end() is False + + +# ── enable / source handling ───────────────────────────────────────── + +def test_disabled_recognizer_emits_nothing(): + rec = make(enabled=False) + rec.begin() + feed(rec, FLICK_LEFT, t=0.0) + assert rec.end() is True # still a click candidate + assert rec.swipes == [] + + +def test_raw_xy_supersedes_event_tap_source(): + """An event-tap leg is discarded once a real raw-XY stream arrives, and + later event-tap samples are ignored.""" + rec = make() + rec.begin() + t = feed(rec, [(-10, 0)] * 4, t=0.0, source="event_tap") # partial + t = feed(rec, FLICK_LEFT, t, source="hid_rawxy") # the real one + feed(rec, [(-40, 0)], t, source="event_tap") # ignored + rec.end() + assert rec.swipes == [LEFT] + + +# ── click-jolt rejection (clicking the button jostles the mouse) ───── + +def test_two_report_jolt_does_not_fire(): + """The button click jolts the mouse as a brief 1-2 report impulse. + Even past the commit distance, that must stay a click, not a swipe.""" + rec = make() + rec.begin() + feed(rec, [(-34, 0), (-34, 0)], t=0.0) # 68 px, only 2 reports + assert rec.end() is True + assert rec.swipes == [] + + +def test_single_giant_report_does_not_fire(): + rec = make() + rec.begin() + feed(rec, [(-95, 5)], t=0.0) # one big spike, one report + assert rec.end() is True + assert rec.swipes == [] + + +def test_quick_flick_with_enough_reports_still_fires(): + """A genuine quick flick — short, but a sustained stream of reports — + must still register.""" + rec = make() + rec.begin() + feed(rec, [(-14, 0)] * 5, t=0.0) # 70 px over 5 reports + rec.end() + assert rec.swipes == [LEFT] + + +def test_summary_reports_hold_stats(): + rec = make() + rec.begin() + feed(rec, FLICK_LEFT, t=0.0) + rec.end() + s = rec.summary() + assert s["fired"] == [LEFT] + assert s["samples"] == len(FLICK_LEFT) + assert s["net_x"] < 0 + assert s["source"] == "hid_rawxy" diff --git a/tests/test_hid_gesture.py b/tests/test_hid_gesture.py index 91c08a5..7df8a01 100644 --- a/tests/test_hid_gesture.py +++ b/tests/test_hid_gesture.py @@ -2,6 +2,7 @@ import os import sys import tempfile +import time import unittest from types import SimpleNamespace from unittest.mock import Mock, patch @@ -9,6 +10,24 @@ from core import hid_gesture +_save_device_cache_patch = None + + +def setUpModule(): + """Stop _try_connect's success path from writing a real device_cache.json + into the developer's config dir while the suite runs.""" + global _save_device_cache_patch + _save_device_cache_patch = patch.object( + hid_gesture.config, "save_device_cache" + ) + _save_device_cache_patch.start() + + +def tearDownModule(): + if _save_device_cache_patch is not None: + _save_device_cache_patch.stop() + + class HidModuleImportTests(unittest.TestCase): def tearDown(self): importlib.reload(hid_gesture) @@ -126,6 +145,51 @@ def test_choose_gesture_candidates_falls_back_to_defaults(self): ) +class DevIndexScanOrderTests(unittest.TestCase): + def test_receiver_pid_probes_numbered_slots_before_bluetooth(self): + listener = hid_gesture.HidGestureListener() + + order = listener._dev_index_scan_order( + product_id=hid_gesture.BOLT_RECEIVER_PID + ) + + self.assertEqual(order, [1, 2, 3, 4, 5, 6, hid_gesture.BT_DEV_IDX]) + + def test_receiver_detected_by_product_name(self): + listener = hid_gesture.HidGestureListener() + + order = listener._dev_index_scan_order( + product_id=0x0000, product_name="USB Receiver" + ) + + self.assertEqual(order[0], 1) + self.assertEqual(order[-1], hid_gesture.BT_DEV_IDX) + + def test_direct_device_probes_bluetooth_first(self): + listener = hid_gesture.HidGestureListener() + + order = listener._dev_index_scan_order( + product_id=0xB034, product_name="MX Master 4" + ) + + self.assertEqual(order[0], hid_gesture.BT_DEV_IDX) + self.assertEqual(order[1:], [1, 2, 3, 4, 5, 6]) + + def test_last_good_index_is_probed_first(self): + listener = hid_gesture.HidGestureListener() + listener._last_good_dev_idx = 3 + + order = listener._dev_index_scan_order( + product_id=hid_gesture.BOLT_RECEIVER_PID + ) + + self.assertEqual(order[0], 3) + # Every slot is still present — the cache only reorders the scan. + self.assertEqual( + sorted(order), sorted([1, 2, 3, 4, 5, 6, hid_gesture.BT_DEV_IDX]) + ) + + class DeviceInfoDumpTests(unittest.TestCase): def test_dump_device_info_includes_runtime_capability_inventory(self): listener = hid_gesture.HidGestureListener() @@ -202,7 +266,7 @@ def test_try_connect_accepts_known_device_without_usage_metadata(self): } fake_dev = _FakeHidDevice() - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id == hid_gesture.FEAT_REPROG_V4: return 0x10 return None @@ -364,7 +428,7 @@ def test_try_connect_success_path_keeps_existing_reprog_discovery_diagnostics(se listener, info = self._make_listener() fake_dev = _FakeHidDevice() - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id == hid_gesture.FEAT_REPROG_V4: return 0x10 return None @@ -417,7 +481,7 @@ def test_try_connect_rearms_extra_diverts_on_reconnect(self): } fake_devs = [_FakeHidDevice(), _FakeHidDevice()] - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id == hid_gesture.FEAT_REPROG_V4: return 0x10 return None @@ -512,7 +576,7 @@ def test_divert_failure_continues_to_next_receiver_slot(self): fake_dev = _FakeHidDevice() divert_call_count = [0] - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id == hid_gesture.FEAT_REPROG_V4: return 0x09 return None @@ -578,7 +642,7 @@ def test_transport_label_bluetooth_for_direct_connection(self): } fake_dev = _FakeHidDevice() - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id == hid_gesture.FEAT_REPROG_V4: return 0x09 return None @@ -623,7 +687,7 @@ def test_try_connect_applies_runtime_supported_buttons(self): ] fake_dev = _FakeHidDevice() - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id == hid_gesture.FEAT_REPROG_V4: return 0x09 return None @@ -670,7 +734,7 @@ def test_try_connect_preserves_directional_gestures_after_rawxy_divert(self): ] fake_dev = _FakeHidDevice() - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id == hid_gesture.FEAT_REPROG_V4: return 0x09 return None @@ -717,7 +781,7 @@ def test_transport_label_logi_bolt_for_bolt_receiver(self): fake_dev = _FakeHidDevice() call_count = [0] - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id != hid_gesture.FEAT_REPROG_V4: return None call_count[0] += 1 @@ -759,7 +823,7 @@ def test_transport_label_usb_receiver_for_non_bolt(self): fake_dev = _FakeHidDevice() call_count = [0] - def fake_find_feature(feature_id): + def fake_find_feature(feature_id, **_kwargs): if feature_id != hid_gesture.FEAT_REPROG_V4: return None call_count[0] += 1 @@ -806,5 +870,192 @@ def test_force_release_stale_holds_clears_gesture_and_extra_buttons(self): extra_up.assert_called_once_with() +class HidDeviceCacheTests(unittest.TestCase): + """Persistent device cache and the discovery short-circuits in + _try_connect that keep boot and reconnects fast.""" + + @staticmethod + def _receiver_info(**overrides): + info = { + "product_id": 0xC548, + "usage_page": 0xFF00, + "usage": 0x0001, + "source": "hidapi-enumerate", + "product_string": "USB Receiver", + "path": b"/dev/hidraw-test", + } + info.update(overrides) + return info + + def test_unresponsive_reprog_slot_is_skipped(self): + """A slot that advertises REPROG_V4 but will not answer the + control-count query is abandoned, and the scan moves on.""" + listener = hid_gesture.HidGestureListener() + info = self._receiver_info() + fake_dev = _FakeHidDevice() + + def fake_find_feature(feature_id, **_kwargs): + return 0x09 if feature_id == hid_gesture.FEAT_REPROG_V4 else None + + # devIdx 1 answers the root probe but is dead; devIdx 2 is real. + controls_by_slot = [None, []] + + with ( + patch.object(listener, "_vendor_hid_infos", return_value=[info]), + patch.object(listener, "_find_feature", side_effect=fake_find_feature), + patch.object( + listener, + "_discover_reprog_controls", + side_effect=lambda: controls_by_slot.pop(0), + ), + patch.object(listener, "_divert", return_value=True), + patch.object(listener, "_divert_extras"), + patch.object(hid_gesture, "HIDAPI_OK", True), + patch.object(hid_gesture, "_BACKEND_PREFERENCE", "hidapi"), + patch.object(hid_gesture, "_HID_API_STYLE", "hidapi"), + patch.object( + hid_gesture, + "_hid", + SimpleNamespace(device=lambda: fake_dev), + create=True, + ), + patch("builtins.print"), + ): + self.assertTrue(listener._try_connect()) + + # Connected on devIdx 2 — the dead devIdx 1 was skipped. + self.assertEqual(listener._dev_idx, 2) + + def test_cached_device_is_probed_before_other_candidates(self): + """A cached identity is moved to the front of the candidate list.""" + listener = hid_gesture.HidGestureListener() + listener._device_cache = { + "product_id": 0xC548, + "usage_page": 0xFF00, + "usage": 0x0001, + "source": "hidapi-enumerate", + "dev_idx": 2, + } + infos = [ + self._receiver_info(product_id=0xC541), # other receiver + self._receiver_info(product_id=0xC548), # cached device + ] + fake_dev = _FakeHidDevice() + + with ( + patch.object(listener, "_vendor_hid_infos", return_value=infos), + patch.object(listener, "_find_feature", return_value=None), + patch.object(hid_gesture, "HIDAPI_OK", True), + patch.object(hid_gesture, "_BACKEND_PREFERENCE", "hidapi"), + patch.object(hid_gesture, "_HID_API_STYLE", "hidapi"), + patch.object( + hid_gesture, + "_hid", + SimpleNamespace(device=lambda: fake_dev), + create=True, + ), + patch("builtins.print"), + ): + listener._try_connect() + + # _try_connect reorders infos in place — the cached PID now leads. + self.assertEqual(infos[0]["product_id"], 0xC548) + + def test_dead_non_receiver_pid_is_probed_once_not_per_interface(self): + """A webcam (non-receiver) exposing several vendor collections is + ruled out once, then its remaining interfaces are skipped.""" + listener = hid_gesture.HidGestureListener() + infos = [ + { + "product_id": 0x0944, + "usage_page": 0xFF00, + "usage": usage, + "source": "hidapi-enumerate", + "product_string": "MX Brio", + "path": b"/dev/hidraw-webcam", + } + for usage in (0x0001, 0x0002, 0x0003) + ] + open_count = [0] + + def make_dev(): + open_count[0] += 1 + return _FakeHidDevice() + + with ( + patch.object(listener, "_vendor_hid_infos", return_value=infos), + patch.object(listener, "_find_feature", return_value=None), + patch.object(hid_gesture, "HIDAPI_OK", True), + patch.object(hid_gesture, "_BACKEND_PREFERENCE", "hidapi"), + patch.object(hid_gesture, "_HID_API_STYLE", "hidapi"), + patch.object( + hid_gesture, + "_hid", + SimpleNamespace(device=make_dev), + create=True, + ), + patch("builtins.print"), + ): + self.assertFalse(listener._try_connect()) + + # Opened once — the 2nd and 3rd MX Brio interfaces were skipped. + self.assertEqual(open_count[0], 1) + + def test_discovery_request_timeout_is_capped_short(self): + """Before a session is live, _request caps the wait at the short + discovery timeout no matter how long the caller asked for.""" + listener = hid_gesture.HidGestureListener() + self.assertFalse(listener._connected) + + with ( + patch.object(hid_gesture, "DISCOVERY_PROBE_TIMEOUT_MS", 50), + patch.object(listener, "_tx"), + patch.object(listener, "_rx", return_value=None), + ): + started = time.monotonic() + self.assertIsNone( + listener._request(0x0E, 0, [], timeout_ms=10_000) + ) + elapsed_ms = (time.monotonic() - started) * 1000 + + # Capped near 50 ms — nowhere near the 10 s the caller requested. + self.assertLess(elapsed_ms, 1000) + + def test_successful_connect_persists_device_identity(self): + """A successful connection writes the device identity to the cache.""" + listener = hid_gesture.HidGestureListener() + info = self._receiver_info() + fake_dev = _FakeHidDevice() + + def fake_find_feature(feature_id, **_kwargs): + return 0x09 if feature_id == hid_gesture.FEAT_REPROG_V4 else None + + with ( + patch.object(listener, "_vendor_hid_infos", return_value=[info]), + patch.object(listener, "_find_feature", side_effect=fake_find_feature), + patch.object(listener, "_discover_reprog_controls", return_value=[]), + patch.object(listener, "_divert", return_value=True), + patch.object(listener, "_divert_extras"), + patch.object(hid_gesture, "HIDAPI_OK", True), + patch.object(hid_gesture, "_BACKEND_PREFERENCE", "hidapi"), + patch.object(hid_gesture, "_HID_API_STYLE", "hidapi"), + patch.object( + hid_gesture, + "_hid", + SimpleNamespace(device=lambda: fake_dev), + create=True, + ), + patch.object(hid_gesture.config, "save_device_cache") as save_mock, + patch("builtins.print"), + ): + self.assertTrue(listener._try_connect()) + + save_mock.assert_called_once() + identity = save_mock.call_args[0][0] + self.assertEqual(identity["product_id"], 0xC548) + self.assertEqual(identity["source"], "hidapi-enumerate") + self.assertEqual(identity["dev_idx"], 1) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_key_simulator.py b/tests/test_key_simulator.py index 5c6d704..eaab462 100644 --- a/tests/test_key_simulator.py +++ b/tests/test_key_simulator.py @@ -24,6 +24,14 @@ def test_tab_switch_actions_exist(self): self.assertTrue(len(key_simulator.ACTIONS["next_tab"]["keys"]) > 0) self.assertTrue(len(key_simulator.ACTIONS["prev_tab"]["keys"]) > 0) + @unittest.skipUnless(sys.platform in ("darwin", "win32"), "screenshot action is platform-specific") + def test_screenshot_action_exists(self): + self.assertIn("screenshot", key_simulator.ACTIONS) + action = key_simulator.ACTIONS["screenshot"] + self.assertIn("Screenshot", action["label"]) + # A real, synthesizable key chord (modifiers + a main key). + self.assertGreaterEqual(len(action["keys"]), 2) + class CustomShortcutParsingTests(unittest.TestCase): def test_build_custom_key_name_map_adds_common_aliases(self): diff --git a/tests/test_log_setup.py b/tests/test_log_setup.py index f3b2d84..29df730 100644 --- a/tests/test_log_setup.py +++ b/tests/test_log_setup.py @@ -203,5 +203,118 @@ def test_encoding_is_utf8(self): self.assertEqual(stream.encoding, "utf-8") +class LogBufferTests(unittest.TestCase): + """The in-memory log buffer + listener fan-out feeding the UI app log.""" + + def setUp(self): + self._orig_stdout = sys.stdout + self._orig_handlers = logging.root.handlers[:] + self._orig_level = logging.root.level + logging.root.handlers.clear() + log_setup._log_buffer.clear() + log_setup._log_listeners.clear() + self._tmp_dir = tempfile.TemporaryDirectory() + self.tmp = self._tmp_dir.name + + def tearDown(self): + sys.stdout = self._orig_stdout + for h in logging.root.handlers[:]: + h.close() + logging.root.handlers.clear() + logging.root.handlers.extend(self._orig_handlers) + logging.root.setLevel(self._orig_level) + log_setup._log_listeners.clear() + log_setup._log_buffer.clear() + self._tmp_dir.cleanup() + + def _setup(self): + with patch.object(log_setup, "_get_log_dir", return_value=self.tmp): + log_setup.setup_logging() + + def test_setup_logging_installs_buffer_handler(self): + self._setup() + self.assertTrue(any( + isinstance(h, log_setup._BufferLogHandler) + for h in logging.root.handlers + )) + + def test_print_output_is_captured_in_recent_logs(self): + self._setup() + print("[Test] buffered line") + self.assertTrue(any( + "[Test] buffered line" in ln for ln in log_setup.get_recent_logs() + )) + + def test_recent_logs_keep_the_timestamped_format(self): + import re + self._setup() + print("[Test] stamped line") + line = next( + ln for ln in log_setup.get_recent_logs() if "stamped line" in ln + ) + self.assertRegex(line, r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2} ") + + def test_add_log_listener_returns_existing_buffer_snapshot(self): + self._setup() + print("[Test] logged before subscribe") + snapshot = log_setup.add_log_listener(lambda line: None) + self.assertTrue(any( + "logged before subscribe" in ln for ln in snapshot + )) + + def test_listener_receives_new_lines(self): + self._setup() + received = [] + log_setup.add_log_listener(received.append) + print("[Test] logged after subscribe") + self.assertTrue(any( + "logged after subscribe" in ln for ln in received + )) + + def test_removed_listener_stops_receiving(self): + self._setup() + received = [] + log_setup.add_log_listener(received.append) + log_setup.remove_log_listener(received.append) + print("[Test] logged after removal") + self.assertEqual(received, []) + + def test_buffer_is_bounded_by_capacity(self): + self._setup() + for i in range(log_setup._LOG_BUFFER_CAPACITY + 50): + print(f"[Test] line {i}") + self.assertLessEqual( + len(log_setup.get_recent_logs()), log_setup._LOG_BUFFER_CAPACITY + ) + + def test_listener_exception_does_not_break_logging(self): + self._setup() + + def bad_listener(line): + raise RuntimeError("listener boom") + + log_setup.add_log_listener(bad_listener) + print("[Test] survives a bad listener") # must not raise + self.assertTrue(any( + "survives a bad listener" in ln + for ln in log_setup.get_recent_logs() + )) + + def test_listener_that_logs_does_not_recurse(self): + self._setup() + calls = [] + + def logging_listener(line): + calls.append(line) + if len(calls) < 3: # bounded so a broken guard still ends + print("[Test] re-entrant line") + + log_setup.add_log_listener(logging_listener) + print("[Test] trigger line") + # The re-entrancy guard means the listener is not re-invoked for the + # line it logged itself — exactly one call, no recursion. + self.assertEqual(len(calls), 1) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_macos_app_shell.py b/tests/test_macos_app_shell.py index 1241de5..db59611 100644 --- a/tests/test_macos_app_shell.py +++ b/tests/test_macos_app_shell.py @@ -281,13 +281,11 @@ def test_engine_start_schedules_when_accessibility_is_granted(self): engine.start.assert_called_once() def test_accessibility_check_exception_fails_closed(self): - locale_mgr = SimpleNamespace(tr=lambda key: key) - with ( patch.object(main_qml.sys, "platform", "darwin"), patch.object(main_qml, "is_process_trusted", side_effect=RuntimeError("boom")), ): - self.assertFalse(main_qml._check_accessibility(locale_mgr)) + self.assertFalse(main_qml._check_accessibility()) def test_tray_minimized_notice_is_scheduled_with_module_qtimer(self): tray = MagicMock() @@ -308,5 +306,118 @@ def test_tray_minimized_notice_is_scheduled_with_module_qtimer(self): ) +@unittest.skipIf(main_qml is None, "main_qml / PySide6 not available") +class AccessibilityGrantRecoveryTests(unittest.TestCase): + """The watcher that revives Mouser once the user grants the + Accessibility permission -- instead of leaving it idle until a + manual quit-and-reopen.""" + + def test_check_reports_grant_state_on_macos(self): + with ( + patch.object(main_qml.sys, "platform", "darwin"), + patch.object(main_qml, "is_process_trusted", return_value=True), + ): + self.assertTrue(main_qml._check_accessibility()) + + with ( + patch.object(main_qml.sys, "platform", "darwin"), + patch.object(main_qml, "is_process_trusted", return_value=False), + ): + self.assertFalse(main_qml._check_accessibility()) + + def test_relaunch_releases_lock_and_execs_source_argv(self): + single_server = MagicMock() + + with ( + patch.object(main_qml.sys, "executable", "/usr/bin/python3"), + patch.object(main_qml.sys, "argv", ["main_qml.py", "--show-window"]), + patch.object(main_qml.sys, "frozen", False, create=True), + patch.object(main_qml.os, "execv") as execv, + ): + main_qml._relaunch_app(single_server) + + single_server.close.assert_called_once_with() + execv.assert_called_once_with( + "/usr/bin/python3", + ["/usr/bin/python3", "main_qml.py", "--show-window"], + ) + + def test_relaunch_drops_argv0_for_frozen_build(self): + binary = "/Applications/Mouser.app/Contents/MacOS/Mouser" + + with ( + patch.object(main_qml.sys, "executable", binary), + patch.object(main_qml.sys, "argv", [binary, "--start-hidden"]), + patch.object(main_qml.sys, "frozen", True, create=True), + patch.object(main_qml.os, "execv") as execv, + ): + main_qml._relaunch_app(None) + + execv.assert_called_once_with(binary, [binary, "--start-hidden"]) + + def test_relaunch_survives_failed_execv(self): + with ( + patch.object(main_qml.sys, "executable", "/usr/bin/python3"), + patch.object(main_qml.sys, "argv", ["main_qml.py"]), + patch.object(main_qml.sys, "frozen", False, create=True), + patch.object(main_qml.os, "execv", side_effect=OSError("denied")), + ): + main_qml._relaunch_app(None) # must not raise + + def test_grant_poll_relaunches_once_permission_lands(self): + timer = MagicMock() + single_server = MagicMock() + + with ( + patch.object(main_qml, "is_process_trusted", return_value=True), + patch.object(main_qml, "_relaunch_app") as relaunch, + ): + triggered = main_qml._accessibility_grant_poll(timer, single_server) + + self.assertTrue(triggered) + timer.stop.assert_called_once_with() + relaunch.assert_called_once_with(single_server) + + def test_grant_poll_keeps_waiting_without_permission(self): + timer = MagicMock() + + with ( + patch.object(main_qml, "is_process_trusted", return_value=False), + patch.object(main_qml, "_relaunch_app") as relaunch, + ): + triggered = main_qml._accessibility_grant_poll(timer, MagicMock()) + + self.assertFalse(triggered) + timer.stop.assert_not_called() + relaunch.assert_not_called() + + def test_grant_poll_survives_check_exception(self): + timer = MagicMock() + + with ( + patch.object( + main_qml, "is_process_trusted", side_effect=RuntimeError("boom") + ), + patch.object(main_qml, "_relaunch_app") as relaunch, + ): + triggered = main_qml._accessibility_grant_poll(timer, MagicMock()) + + self.assertFalse(triggered) + timer.stop.assert_not_called() + relaunch.assert_not_called() + + def test_watch_accessibility_grant_starts_polling_timer(self): + fake_timer = MagicMock() + + with patch.object(main_qml, "QTimer", return_value=fake_timer): + returned = main_qml._watch_accessibility_grant(MagicMock(), MagicMock()) + + self.assertIs(returned, fake_timer) + fake_timer.setInterval.assert_called_once_with( + main_qml._ACCESSIBILITY_POLL_INTERVAL_MS + ) + fake_timer.start.assert_called_once_with() + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_macos_build_script.py b/tests/test_macos_build_script.py index 3d028c3..5f9fbd0 100644 --- a/tests/test_macos_build_script.py +++ b/tests/test_macos_build_script.py @@ -43,7 +43,7 @@ def setUp(self): self._write_manager("asdf") self._write_python(self.bin_dir / "python3", "path-python3") self._write_python(self.bin_dir / "python", "path-python") - for tool in ("dirname", "pwd", "mkdir", "find", "awk", "touch"): + for tool in ("dirname", "pwd", "mkdir", "find", "awk", "touch", "df"): resolved = shutil.which(tool) if resolved: os.symlink(resolved, self.tool_dir / tool) diff --git a/tests/test_smart_shift.py b/tests/test_smart_shift.py index 2534974..41cd7ef 100644 --- a/tests/test_smart_shift.py +++ b/tests/test_smart_shift.py @@ -749,7 +749,7 @@ def test_multiple_profiles_all_migrated(self): def test_version_bumped_to_current(self): from core.config import _migrate migrated = _migrate(self._v6_config()) - self.assertEqual(migrated["version"], 9) + self.assertEqual(migrated["version"], 10) # ────────────────────────────────────────────────────────────────────────────── @@ -811,7 +811,7 @@ def test_multiple_profiles_all_migrated(self): def test_version_bumped_to_current(self): from core.config import _migrate migrated = _migrate(self._v7_config()) - self.assertEqual(migrated["version"], 9) + self.assertEqual(migrated["version"], 10) class HidForceReconnectTests(unittest.TestCase): diff --git a/ui/backend.py b/ui/backend.py index db48f74..36fe71b 100644 --- a/ui/backend.py +++ b/ui/backend.py @@ -19,9 +19,9 @@ from core.config import ( BUTTON_NAMES, load_config, save_config, get_active_mappings, PROFILE_BUTTON_NAMES, set_mapping, create_profile, delete_profile, - get_icon_for_exe, + get_icon_for_exe, GESTURE_SENSITIVITY_PX, gesture_sensitivity_index_for, ) -from core import app_catalog +from core import app_catalog, log_setup from core.device_layouts import get_device_layout, get_manual_layout_choices from core.key_registry import ( ShortcutParseError, @@ -208,6 +208,7 @@ class Backend(QObject): batteryLevelChanged = Signal() debugLogChanged = Signal() debugEventsEnabledChanged = Signal() + appLogChanged = Signal() gestureStateChanged = Signal() gestureRecordsChanged = Signal() deviceInfoChanged = Signal() @@ -222,6 +223,7 @@ class Backend(QObject): _connectionChangeRequest = Signal(bool) _batteryChangeRequest = Signal(int) _debugMessageRequest = Signal(str) + _appLogMessageRequest = Signal(str) _gestureEventRequest = Signal(object) _smartShiftReadRequest = Signal() _statusMessageRequest = Signal(str) @@ -247,6 +249,7 @@ def __init__(self, engine=None, parent=None, root_dir=None): self._battery_level = -1 self._hid_features_ready = False self._debug_lines = [] + self._app_log_lines = [] self._debug_events_enabled = bool( self._cfg.get("settings", {}).get("debug_mode", False) ) @@ -292,6 +295,8 @@ def __init__(self, engine=None, parent=None, root_dir=None): self._handleBatteryChange, Qt.QueuedConnection) self._debugMessageRequest.connect( self._handleDebugMessage, Qt.QueuedConnection) + self._appLogMessageRequest.connect( + self._handleAppLogMessage, Qt.QueuedConnection) self._gestureEventRequest.connect( self._handleGestureEvent, Qt.QueuedConnection) self._smartShiftReadRequest.connect( @@ -307,6 +312,11 @@ def __init__(self, engine=None, parent=None, root_dir=None): self._updateInstallProgressRequest.connect( self._handleUpdateInstallProgress, Qt.QueuedConnection) + # Mirror the console log (everything print()s) into the UI. Capture + # has run since setup_logging(); seed from its buffer and subscribe + # for new lines — atomically, so none are missed or double-counted. + self._app_log_lines = log_setup.add_log_listener(self._onAppLogMessage) + # Wire engine callbacks if engine: engine.set_profile_change_callback(self._onEngineProfileSwitch) @@ -516,8 +526,11 @@ def ignoreTrackpad(self): return self._cfg.get("settings", {}).get("ignore_trackpad", True) @Property(int, notify=settingsChanged) - def gestureThreshold(self): - return int(self._cfg.get("settings", {}).get("gesture_threshold", 50)) + def gestureSensitivityIndex(self): + """Index into GESTURE_SENSITIVITY_PX nearest the stored px threshold.""" + px = self._cfg.get("settings", {}).get( + "gesture_threshold", GESTURE_SENSITIVITY_PX[1]) + return gesture_sensitivity_index_for(px) @Property(str, notify=settingsChanged) def appearanceMode(self): @@ -677,6 +690,10 @@ def batteryLevel(self): def debugLog(self): return "\n".join(self._debug_lines) + @Property(str, notify=appLogChanged) + def appLog(self): + return "\n".join(self._app_log_lines) + @Property(bool, notify=gestureStateChanged) def recordMode(self): return self._record_mode @@ -1303,9 +1320,11 @@ def setIgnoreTrackpad(self, value): self.settingsChanged.emit() @Slot(int) - def setGestureThreshold(self, value): - snapped = max(20, min(400, int(round(value / 5.0) * 5))) - self._cfg.setdefault("settings", {})["gesture_threshold"] = snapped + def setGestureSensitivity(self, index): + """Store the px threshold for sensitivity preset `index`.""" + index = max(0, min(len(GESTURE_SENSITIVITY_PX) - 1, int(index))) + self._cfg.setdefault("settings", {})["gesture_threshold"] = \ + GESTURE_SENSITIVITY_PX[index] save_config(self._cfg) if self._engine: self._engine.reload_mappings() @@ -1353,6 +1372,11 @@ def clearDebugLog(self): self._debug_lines = [] self.debugLogChanged.emit() + @Slot() + def clearAppLog(self): + self._app_log_lines = [] + self.appLogChanged.emit() + @Slot(bool) def setRecordMode(self, value): self._record_mode = bool(value) @@ -1579,6 +1603,14 @@ def _onEngineDebugMessage(self, message): """Called from engine/hook thread — posts to Qt main thread.""" self._debugMessageRequest.emit(message) + def _onAppLogMessage(self, line): + """Called from any thread by log_setup — posts to Qt main thread. + + Stays non-blocking and never logs/prints: it runs inside the logging + handler that captures stdout. + """ + self._appLogMessageRequest.emit(line) + def _onEngineGestureEvent(self, event): """Called from engine/hook thread — posts to Qt main thread.""" self._gestureEventRequest.emit(event) @@ -1824,6 +1856,14 @@ def _handleDebugMessage(self, message): """Runs on Qt main thread.""" self._append_debug_line(message) + @Slot(str) + def _handleAppLogMessage(self, line): + """Runs on Qt main thread — appends one captured console log line.""" + self._app_log_lines.append(line) + # Keep the UI view bounded, matching log_setup's ring buffer. + self._app_log_lines = self._app_log_lines[-1000:] + self.appLogChanged.emit() + @Slot(str) def _handleStatusMessage(self, message): """Runs on Qt main thread.""" diff --git a/ui/locale_manager.py b/ui/locale_manager.py index d6d1f34..9ad1c67 100644 --- a/ui/locale_manager.py +++ b/ui/locale_manager.py @@ -64,7 +64,12 @@ "mouse.swipe_right": "Swipe right", "mouse.swipe_up": "Swipe up", "mouse.swipe_down": "Swipe down", - "mouse.threshold": "Threshold", + "mouse.sensitivity": "Sensitivity", + "mouse.sensitivity_highest": "Highest", + "mouse.sensitivity_high": "High", + "mouse.sensitivity_medium": "Medium", + "mouse.sensitivity_low": "Low", + "mouse.sensitivity_lowest": "Lowest", # Mouse page — debug panel "mouse.debug_events": "Debug Events", @@ -211,10 +216,11 @@ "accessibility.text": ( "Mouser needs Accessibility permission to intercept " "mouse button events.\n\n" - "macOS should have opened the System Settings prompt.\n" - "Please grant permission, then restart Mouser." + "Grant it in System Settings and Mouser starts automatically " + "once the permission is enabled -- no need to quit and reopen." ), "accessibility.info": "System Settings -> Privacy & Security -> Accessibility", + "accessibility.open_settings": "Open System Settings", # About dialog "about.title": "About Mouser", @@ -282,7 +288,12 @@ "mouse.swipe_right": "\u5411\u53f3\u6ed1\u52a8", "mouse.swipe_up": "\u5411\u4e0a\u6ed1\u52a8", "mouse.swipe_down": "\u5411\u4e0b\u6ed1\u52a8", - "mouse.threshold": "\u9608\u5024", + "mouse.sensitivity": "\u7075\u654f\u5ea6", + "mouse.sensitivity_highest": "\u6700\u9ad8", + "mouse.sensitivity_high": "\u9ad8", + "mouse.sensitivity_medium": "\u4e2d", + "mouse.sensitivity_low": "\u4f4e", + "mouse.sensitivity_lowest": "\u6700\u4f4e", "mouse.debug_events": "\u8c03\u8bd5\u4e8b\u4ef6", "mouse.debug_events_desc": "\u6536\u96c6\u68c0\u6d4b\u5230\u7684\u6309\u952e\u3001\u624b\u52bf\u548c\u6620\u5c04\u52a8\u4f5c", @@ -421,10 +432,10 @@ "accessibility.title": "\u9700\u8981\u8f85\u52a9\u529f\u80fd\u6743\u9650", "accessibility.text": ( "Mouser \u9700\u8981\u8f85\u52a9\u529f\u80fd\u6743\u9650\u4ee5\u62e6\u622a\u9f20\u6807\u6309\u952e\u4e8b\u4ef6\u3002\n\n" - "macOS \u5e94\u5df2\u6253\u5f00\u7cfb\u7edf\u8bbe\u7f6e\u63d0\u793a\u3002\n" - "\u8bf7\u6388\u4e88\u6743\u9650\uff0c\u7136\u540e\u91cd\u65b0\u542f\u52a8 Mouser\u3002" + "\u5728\u7cfb\u7edf\u8bbe\u7f6e\u4e2d\u6388\u4e88\u6743\u9650\u540e\uff0cMouser \u5c06\u81ea\u52a8\u542f\u52a8\uff0c\u65e0\u9700\u9000\u51fa\u91cd\u5f00\u3002" ), "accessibility.info": "\u7cfb\u7edf\u8bbe\u7f6e -> \u9690\u79c1\u4e0e\u5b89\u5168\u6027 -> \u8f85\u52a9\u529f\u80fd", + "accessibility.open_settings": "\u6253\u5f00\u7cfb\u7edf\u8bbe\u7f6e", "about.title": "\u5173\u4e8e Mouser", "about.subtitle": "\u7528\u4e8e\u652f\u6301\u548c\u8c03\u8bd5\u7684\u8fd0\u884c\u65f6\u4e0e\u6784\u5efa\u4fe1\u606f\u3002", @@ -490,7 +501,12 @@ "mouse.swipe_right": "\u5411\u53f3\u6ed1\u52d5", "mouse.swipe_up": "\u5411\u4e0a\u6ed1\u52d5", "mouse.swipe_down": "\u5411\u4e0b\u6ed1\u52d5", - "mouse.threshold": "\u95be\u5024", + "mouse.sensitivity": "\u9748\u654f\u5ea6", + "mouse.sensitivity_highest": "\u6700\u9ad8", + "mouse.sensitivity_high": "\u9ad8", + "mouse.sensitivity_medium": "\u4e2d", + "mouse.sensitivity_low": "\u4f4e", + "mouse.sensitivity_lowest": "\u6700\u4f4e", "mouse.debug_events": "\u9664\u932f\u4e8b\u4ef6", "mouse.debug_events_desc": "\u6536\u96c6\u5075\u6e2c\u5230\u7684\u6309\u9375\u3001\u624b\u52e2\u548c\u5c0d\u6620\u52d5\u4f5c", @@ -629,10 +645,10 @@ "accessibility.title": "\u9700\u8981\u8f14\u52a9\u4f7f\u7528\u6b0a\u9650", "accessibility.text": ( "Mouser \u9700\u8981\u8f14\u52a9\u4f7f\u7528\u6b0a\u9650\u4ee5\u6514\u622a\u6ed1\u9f20\u6309\u9375\u4e8b\u4ef6\u3002\n\n" - "macOS \u61c9\u5df2\u958b\u555f\u7cfb\u7d71\u8a2d\u5b9a\u63d0\u793a\u3002\n" - "\u8acb\u6388\u4e88\u6b0a\u9650\uff0c\u7136\u5f8c\u91cd\u65b0\u555f\u52d5 Mouser\u3002" + "\u5728\u7cfb\u7d71\u8a2d\u5b9a\u4e2d\u6388\u4e88\u6b0a\u9650\u5f8c\uff0cMouser \u5c07\u81ea\u52d5\u555f\u52d5\uff0c\u7121\u9700\u7d50\u675f\u91cd\u958b\u3002" ), "accessibility.info": "\u7cfb\u7d71\u8a2d\u5b9a -> \u96b1\u79c1\u6b0a\u8207\u5b89\u5168\u6027 -> \u8f14\u52a9\u4f7f\u7528", + "accessibility.open_settings": "\u958b\u555f\u7cfb\u7d71\u8a2d\u5b9a", "about.title": "\u95dc\u65bc Mouser", "about.subtitle": "\u63d0\u4f9b\u652f\u63f4\u8207\u9664\u932f\u7528\u7684\u57f7\u884c\u6642\u8207\u5efa\u7f6e\u8cc7\u8a0a\u3002", diff --git a/ui/qml/MousePage.qml b/ui/qml/MousePage.qml index bd6a342..cdca310 100644 --- a/ui/qml/MousePage.qml +++ b/ui/qml/MousePage.qml @@ -1381,55 +1381,32 @@ Item { color: theme.border } - Row { - width: parent.width - spacing: 12 - Text { - text: s["mouse.threshold"] - font { family: uiState.fontFamily; pixelSize: 12; bold: true } - color: theme.textPrimary - } - - Text { - text: ( - gestureThresholdSlider.pressed - ? Math.round(gestureThresholdSlider.value / 5.0) * 5 - : backend.gestureThreshold - ) + " px" - font { family: uiState.fontFamily; pixelSize: 12 } - color: theme.textSecondary - } + text: s["mouse.sensitivity"] + font { family: uiState.fontFamily; pixelSize: 11; + capitalization: Font.AllUppercase; letterSpacing: 1 } + color: theme.textDim } - WheelSafeSlider { - id: gestureThresholdSlider + ComboBox { width: parent.width - from: 20 - to: 400 - stepSize: 5 - value: backend.gestureThreshold - accentColor: theme.accent - accentDimColor: theme.accentDim - trackColor: theme.border - onMoved: gestureThresholdSave.restart() - onPressedChanged: { - if (!pressed) { - gestureThresholdSave.stop() - backend.setGestureThreshold( - Math.round(value / 5.0) * 5) - } + // Ordered most → least sensitive, matching + // core.config.GESTURE_SENSITIVITY_PX. + model: (lm.strings, [ + s["mouse.sensitivity_highest"], + s["mouse.sensitivity_high"], + s["mouse.sensitivity_medium"], + s["mouse.sensitivity_low"], + s["mouse.sensitivity_lowest"] + ]) + Material.accent: theme.accent + font { family: uiState.fontFamily; pixelSize: 11 } + currentIndex: backend.gestureSensitivityIndex + onActivated: function(index) { + backend.setGestureSensitivity(index) } } - Timer { - id: gestureThresholdSave - interval: 250 - repeat: false - onTriggered: backend.setGestureThreshold( - Math.round(gestureThresholdSlider.value / 5.0) * 5) - } - Text { text: s["mouse.swipe_actions"] font { family: uiState.fontFamily; pixelSize: 11; @@ -2134,6 +2111,84 @@ Item { } } } + + Column { + width: parent.width + spacing: 6 + + RowLayout { + width: parent.width + spacing: 12 + + Text { + Layout.fillWidth: true + text: "Application Log" + font { family: uiState.fontFamily; pixelSize: 11; bold: true } + color: theme.textPrimary + } + + Rectangle { + Layout.preferredWidth: clearAppLogText.implicitWidth + 20 + Layout.preferredHeight: 24 + radius: 8 + color: clearAppLogMa.containsMouse + ? Qt.rgba(1, 1, 1, 0.08) + : Qt.rgba(1, 1, 1, 0.04) + + Text { + id: clearAppLogText + anchors.centerIn: parent + text: s["mouse.clear"] + font { family: uiState.fontFamily; pixelSize: 11; bold: true } + color: theme.textPrimary + } + + MouseArea { + id: clearAppLogMa + anchors.fill: parent + hoverEnabled: true + cursorShape: Qt.PointingHandCursor + onClicked: backend.clearAppLog() + } + } + } + + Rectangle { + width: parent.width + height: 200 + radius: 10 + color: Qt.rgba(0, 0, 0, 0.18) + border.width: 1 + border.color: theme.border + + ScrollView { + anchors.fill: parent + anchors.margins: 1 + clip: true + + TextArea { + id: appLogArea + text: backend.appLog.length + ? backend.appLog + : "Application log output appears here while debug mode is on." + readOnly: true + wrapMode: TextEdit.NoWrap + selectByMouse: true + color: backend.appLog.length + ? theme.textPrimary + : theme.textSecondary + font.pixelSize: 11 + font.family: "Menlo" + background: null + padding: 10 + + onTextChanged: { + cursorPosition = length + } + } + } + } + } } }