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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 48 additions & 0 deletions source/_magnifier/magnifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,13 @@
MagnifierParameters,
MagnifierAction,
MagnifiedView,
MagnifierTrackingType,
Direction,
Filter,
Coordinates,
)
from .config import (
getFollowState,
getZoomLevel,
getPanStep,
getFilter,
Expand All @@ -35,6 +37,7 @@
_isDebug,
)
from .utils.focusManager import FocusManager
from .utils.mouseHook import MagnifierMouseHook


class Magnifier:
Expand All @@ -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
Expand Down Expand Up @@ -184,6 +190,8 @@ def _startMagnifier(self) -> None:

self._isActive = True
self.currentCoordinates = self._focusManager.getCurrentFocusCoordinates()
self._mouseHook = MagnifierMouseHook(self._onMouseMove)
self._mouseHook.start()
Comment thread
Boumtchack marked this conversation as resolved.

def _updateMagnifier(self) -> None:
"""
Expand Down Expand Up @@ -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
Expand Down
95 changes: 95 additions & 0 deletions source/_magnifier/utils/mouseHook.py
Original file line number Diff line number Diff line change
@@ -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)
5 changes: 5 additions & 0 deletions tests/unit/test_magnifier/test_magnifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down
94 changes: 94 additions & 0 deletions tests/unit/test_magnifier/test_mouseHook.py
Original file line number Diff line number Diff line change
@@ -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()
Loading