diff --git a/source/_magnifier/magnifier.py b/source/_magnifier/magnifier.py index 66e96c359a6..e628551a4c3 100644 --- a/source/_magnifier/magnifier.py +++ b/source/_magnifier/magnifier.py @@ -21,11 +21,13 @@ MagnifierParameters, MagnifierAction, MagnifiedView, + MagnifierTrackingType, Direction, Filter, Coordinates, ) from .config import ( + getFollowState, getZoomLevel, getPanStep, getFilter, @@ -35,6 +37,7 @@ _isDebug, ) from .utils.focusManager import FocusManager +from .utils.mouseHook import MagnifierMouseHook class Magnifier: @@ -57,6 +60,9 @@ def __init__(self): self._isManualPanning: bool = False self._consecutiveErrors: int = 0 self._recoveryAttempts: int = 0 + self._mouseHook: MagnifierMouseHook | None = None + self._pendingMouseCoordinates = Coordinates(0, 0) + self._mouseUpdatePending: bool = False # Register for display changes _displayTracking.displayChanged.register(self._onDisplayChanged) self._screenCurtainIsActive: bool = False @@ -184,6 +190,8 @@ def _startMagnifier(self) -> None: self._isActive = True self.currentCoordinates = self._focusManager.getCurrentFocusCoordinates() + self._mouseHook = MagnifierMouseHook(self._onMouseMove) + self._mouseHook.start() def _updateMagnifier(self) -> None: """ @@ -246,12 +254,52 @@ def _attemptRecovery(self) -> None: self._consecutiveErrors = 0 self._startTimer(self._updateMagnifier) + def _onMouseMove(self, x: int, y: int) -> None: + """ + Called from the mouse hook thread on every WM_MOUSEMOVE. + + This runs synchronously inside a global WH_MOUSE_LL hook chain, so it must + return immediately: it only records the latest coordinates and schedules + the actual update on the main thread. Calling into the Magnification API + (via _doUpdate) from here would delay delivery of the real WM_MOUSEMOVE to + whatever window is under the cursor, for every mouse move on the system, + not just NVDA's own windows. + + Only acts when mouse tracking is enabled and the magnifier is active. + """ + if not self._isActive or self._isManualPanning: + return + if not getFollowState(MagnifierTrackingType.MOUSE): + return + self._pendingMouseCoordinates = Coordinates(x, y) + if not self._mouseUpdatePending: + self._mouseUpdatePending = True + wx.CallAfter(self._applyPendingMousePosition) + + def _applyPendingMousePosition(self) -> None: + """ + Apply the latest mouse position recorded by _onMouseMove. + Runs on the main thread via wx.CallAfter, so this is the only place + where a mouse-driven update touches the Magnification API. + """ + self._mouseUpdatePending = False + if not self._isActive or self._isManualPanning: + return + try: + self.currentCoordinates = self._pendingMouseCoordinates + self._doUpdate() + except OSError: + pass + def _stopMagnifier(self) -> None: """ Stop the magnifier """ if not self._isActive: return + if self._mouseHook: + self._mouseHook.stop() + self._mouseHook = None self._stopTimer() self._isActive = False # Unregister from display changes diff --git a/source/_magnifier/utils/mouseHook.py b/source/_magnifier/utils/mouseHook.py new file mode 100644 index 00000000000..2a73dc2ec63 --- /dev/null +++ b/source/_magnifier/utils/mouseHook.py @@ -0,0 +1,95 @@ +# A part of NonVisual Desktop Access (NVDA) +# Copyright (C) 2026 NV Access Limited, Antoine Haffreingue +# This file may be used under the terms of the GNU General Public License, version 2 or later, as modified by the NVDA license. +# For full terms and any additional permissions, see the NVDA license file: https://github.com/nvaccess/nvda/blob/master/copying.txt + +"""Dedicated low-level mouse hook for the magnifier. + +Runs in its own thread so it stays responsive even when the wx main thread is +busy (e.g. at high Windows display scale factors), which is the root cause of +the DPI-related lag compared to the built-in Windows Magnifier. +""" + +import ctypes +import threading +from collections.abc import Callable +from ctypes import byref +from ctypes.wintypes import MSG +from logHandler import log +from winBindings import user32 +from winInputHook import MSLLHOOKSTRUCT, HC_ACTION, WH_MOUSE_LL + +WM_MOUSEMOVE: int = 0x0200 +WM_QUIT: int = 0x0012 +PM_NOREMOVE: int = 0x0000 + + +class MagnifierMouseHook: + """Installs a WH_MOUSE_LL hook in a dedicated thread. Calls onMouseMove(x, y) on every mouse move. + + WH_MOUSE_LL is a global hook: it runs for every mouse move on the system, and + delivery of the real WM_MOUSEMOVE to the window under the cursor waits for the + whole hook chain to return. onMouseMove must therefore return immediately (no + Magnification API calls, no blocking work) — defer any real work to another + thread, e.g. via wx.CallAfter. + """ + + def __init__(self, onMouseMove: Callable[[int, int], None]): + self._onMouseMove = onMouseMove + self._thread: threading.Thread | None = None + self._cCallback = None # kept alive to prevent GC of the ctypes callback + self._hookReady = threading.Event() + self._hookInstalled: bool = False + + def start(self) -> None: + self._thread = threading.Thread(target=self._run, name="magnifierMouseHook", daemon=True) + self._thread.start() + self._hookReady.wait(timeout=1.0) + + def stop(self) -> None: + thread = self._thread + if not thread: + self._cCallback = None + return + if thread.ident is None or not user32.PostThreadMessage(thread.ident, WM_QUIT, 0, 0): + log.error( + f"Failed to post WM_QUIT to magnifier mouse hook thread (error {ctypes.GetLastError()})", + ) + thread.join() + self._thread = None + self._cCallback = None + + def _onRawMouseEvent(self, code: int, eventType: int, mouseDataPointer: int) -> int: + if code == HC_ACTION and eventType == WM_MOUSEMOVE: + mouseData = MSLLHOOKSTRUCT.from_address(mouseDataPointer) + try: + self._onMouseMove(mouseData.pt.x, mouseData.pt.y) + except Exception: + log.exception("Error in magnifier mouse hook callback") + return user32.CallNextHookEx(0, code, eventType, mouseDataPointer) + + def _run(self) -> None: + windowsMessage = MSG() + # Ensure the thread message queue exists so PostThreadMessage(WM_QUIT) succeeds. + user32.PeekMessage(byref(windowsMessage), None, 0, 0, PM_NOREMOVE) + + self._cCallback = user32.HOOKPROC(self._onRawMouseEvent) + hookHandle = user32.SetWindowsHookEx(WH_MOUSE_LL, self._cCallback, None, 0) + self._hookInstalled = bool(hookHandle) + self._hookReady.set() + if not hookHandle: + log.error(f"Failed to install magnifier mouse hook (error {ctypes.GetLastError()})") + return + + try: + while True: + result = user32.GetMessage(byref(windowsMessage), None, 0, 0) + if result == 0: + break + if result == -1: + log.error( + f"GetMessage failed in magnifier mouse hook thread (error {ctypes.GetLastError()})", + ) + break + finally: + user32.UnhookWindowsHookEx(hookHandle) diff --git a/tests/unit/test_magnifier/test_magnifier.py b/tests/unit/test_magnifier/test_magnifier.py index 7a26dd883c3..0a778ff206f 100644 --- a/tests/unit/test_magnifier/test_magnifier.py +++ b/tests/unit/test_magnifier/test_magnifier.py @@ -33,9 +33,14 @@ def setUp(self): mock.MagUninitialize.return_value = True mock.MagSetFullscreenTransform.return_value = True mock.MagSetFullscreenColorEffect.return_value = True + self.mouseHook_patcher = patch("_magnifier.magnifier.MagnifierMouseHook") + self.MockMouseHook = self.mouseHook_patcher.start() + self.mock_hook_instance = MagicMock() + self.MockMouseHook.return_value = self.mock_hook_instance def tearDown(self): """Cleanup after each test.""" + self.mouseHook_patcher.stop() self.mag_fs_patcher.stop() self.mag_patcher.stop() diff --git a/tests/unit/test_magnifier/test_mouseHook.py b/tests/unit/test_magnifier/test_mouseHook.py new file mode 100644 index 00000000000..56f2c42daae --- /dev/null +++ b/tests/unit/test_magnifier/test_mouseHook.py @@ -0,0 +1,94 @@ +# A part of NonVisual Desktop Access (NVDA) +# Copyright (C) 2025-2026 NV Access Limited, Antoine Haffreingue +# This file may be used under the terms of the GNU General Public License, version 2 or later, as modified by the NVDA license. +# For full terms and any additional permissions, see the NVDA license file: https://github.com/nvaccess/nvda/blob/master/copying.txt + +from unittest.mock import MagicMock, patch +from _magnifier.fullscreenMagnifier import FullScreenMagnifier +from _magnifier.utils.types import Coordinates +from tests.unit.test_magnifier.test_magnifier import _TestMagnifier + + +class TestMouseHookLifecycle(_TestMagnifier): + """Tests for the WH_MOUSE_LL hook lifecycle managed by the base Magnifier class.""" + + def testHookStartedWithMagnifier(self): + """Mouse hook is started when the magnifier starts.""" + magnifier = FullScreenMagnifier() + magnifier._startMagnifier() + + self.MockMouseHook.assert_called_once_with(magnifier._onMouseMove) + self.mock_hook_instance.start.assert_called_once() + + magnifier._stopMagnifier() + + def testHookStoppedWithMagnifier(self): + """Mouse hook is stopped and cleared when the magnifier stops.""" + magnifier = FullScreenMagnifier() + magnifier._startMagnifier() + magnifier._stopMagnifier() + + self.mock_hook_instance.stop.assert_called_once() + self.assertIsNone(magnifier._mouseHook) + + +class TestOnMouseMove(_TestMagnifier): + """Tests for FullScreenMagnifier._onMouseMove — the hook callback. + + _onMouseMove runs synchronously inside a global WH_MOUSE_LL hook chain (see + utils/mouseHook.py), so it must never call into the Magnification API directly: + doing so would delay delivery of the real WM_MOUSEMOVE to whatever window is + under the cursor, for every mouse move on the system. It should only record + the latest coordinates and defer the actual update to the main thread via + wx.CallAfter (_applyPendingMousePosition). + """ + + def testDoesNotUpdateMagnifierSynchronously(self): + """_onMouseMove must not touch the Magnification API from the hook thread.""" + magnifier = FullScreenMagnifier() + magnifier._startMagnifier() + magnifier._fullscreenMagnifier = MagicMock() + + with ( + patch("_magnifier.magnifier.getFollowState", return_value=True), + patch("_magnifier.magnifier.wx.CallAfter"), + ): + magnifier._onMouseMove(500, 400) + + magnifier._fullscreenMagnifier.assert_not_called() + self.assertEqual(magnifier._pendingMouseCoordinates, Coordinates(500, 400)) + magnifier._stopMagnifier() + + def testSchedulesMainThreadUpdate(self): + """_onMouseMove schedules exactly one wx.CallAfter, even for bursts of moves.""" + magnifier = FullScreenMagnifier() + magnifier._startMagnifier() + + with ( + patch("_magnifier.magnifier.getFollowState", return_value=True), + patch("_magnifier.magnifier.wx.CallAfter") as mockCallAfter, + ): + magnifier._onMouseMove(500, 400) + magnifier._onMouseMove(510, 410) + magnifier._onMouseMove(520, 420) + + mockCallAfter.assert_called_once_with(magnifier._applyPendingMousePosition) + self.assertEqual(magnifier._pendingMouseCoordinates, Coordinates(520, 420)) + magnifier._stopMagnifier() + + def testApplyPendingMousePositionUpdatesMagnifier(self): + """_applyPendingMousePosition (run on the main thread) performs the real update.""" + magnifier = FullScreenMagnifier() + magnifier._startMagnifier() + magnifier._fullscreenMagnifier = MagicMock() + + with ( + patch("_magnifier.magnifier.getFollowState", return_value=True), + patch("_magnifier.magnifier.wx.CallAfter", side_effect=lambda func, *a, **kw: func(*a, **kw)), + ): + magnifier._onMouseMove(500, 400) + + magnifier._fullscreenMagnifier.assert_called_once() + self.assertEqual(magnifier.currentCoordinates, Coordinates(500, 400)) + self.assertFalse(magnifier._mouseUpdatePending) + magnifier._stopMagnifier()