From cfe5d2d3d3349133bde886f7537ad1dbcfccce10 Mon Sep 17 00:00:00 2001 From: Antoine HAFFREINGUE Date: Wed, 24 Jun 2026 06:29:15 +0200 Subject: [PATCH 1/2] change timer logic --- source/_magnifier/magnifier.py | 15 ++++++---- source/winBindings/winmm.py | 28 ++++++++++++++++++- .../test_fullscreenMagnifier.py | 10 +------ tests/unit/test_magnifier/test_magnifier.py | 19 +++---------- 4 files changed, 41 insertions(+), 31 deletions(-) diff --git a/source/_magnifier/magnifier.py b/source/_magnifier/magnifier.py index 66e96c359a6..bf23cfd8057 100644 --- a/source/_magnifier/magnifier.py +++ b/source/_magnifier/magnifier.py @@ -15,6 +15,7 @@ import ui import speech import screenCurtain +from winBindings import winmm from winAPI import _displayTracking from winAPI._displayTracking import OrientationState, getPrimaryDisplayOrientation from .utils.types import ( @@ -38,7 +39,7 @@ class Magnifier: - _TIMER_INTERVAL_MS: int = 12 + _TIMER_INTERVAL_MS: int = 8 _MARGIN_BORDER: int = 50 _MAX_CONSECUTIVE_ERRORS: int = 3 _MAGNIFIED_VIEW: MagnifiedView @@ -182,14 +183,17 @@ def _startMagnifier(self) -> None: ui.message(message, speechPriority=speech.priorities.Spri.NOW) return + result = winmm.timeBeginPeriod(1) + if result != winmm.TIMERR_NOERROR: + log.warning(f"timeBeginPeriod(1) failed with code {result} — timer resolution stays at ~15ms") + self._isActive = True self.currentCoordinates = self._focusManager.getCurrentFocusCoordinates() def _updateMagnifier(self) -> None: """ Update the magnifier position and content. - This method is called repeatedly by the timer. - On transient errors (below threshold): reschedules itself to keep running. + This method is called repeatedly by the repeating timer. On repeated errors (at threshold): delegates rescheduling to _attemptRecovery. """ if not self._isActive: @@ -225,8 +229,6 @@ def _updateMagnifier(self) -> None: f"Transient error updating magnifier ({self._consecutiveErrors}/{self._MAX_CONSECUTIVE_ERRORS})", exc_info=True, ) - # Always reschedule the timer to keep the magnifier alive - self._startTimer(self._updateMagnifier) def _doUpdate(self) -> None: """ @@ -253,6 +255,7 @@ def _stopMagnifier(self) -> None: if not self._isActive: return self._stopTimer() + winmm.timeEndPeriod(1) self._isActive = False # Unregister from display changes _displayTracking.displayChanged.unregister(self._onDisplayChanged) @@ -367,7 +370,7 @@ def _startTimer(self, callback: Callable[[], None] = None) -> None: self._stopTimer() self._timer = wx.Timer() self._timer.Bind(wx.EVT_TIMER, lambda evt: callback()) - self._timer.Start(self._TIMER_INTERVAL_MS, oneShot=True) + self._timer.Start(self._TIMER_INTERVAL_MS, oneShot=False) def _stopTimer(self) -> None: """ diff --git a/source/winBindings/winmm.py b/source/winBindings/winmm.py index f0d814f80fe..67a67b459a1 100644 --- a/source/winBindings/winmm.py +++ b/source/winBindings/winmm.py @@ -19,7 +19,7 @@ DWORD_PTR = c_size_t MMRESULT = c_long - +TIMERR_NOERROR: int = 0 dll = windll.winmm @@ -48,3 +48,29 @@ DWORD_PTR, # dw1: Message parameter (DWORD_PTR) DWORD_PTR, # dw2: Message parameter (DWORD_PTR) ) + +timeBeginPeriod = WINFUNCTYPE(None)(("timeBeginPeriod", dll)) +""" +Sets the minimum timer resolution for the application. +Must be matched with a corresponding call to timeEndPeriod using the same uPeriod value. + +.. seealso:: + https://learn.microsoft.com/en-us/windows/win32/api/timeapi/nf-timeapi-timebeginperiod +""" +timeBeginPeriod.restype = MMRESULT +timeBeginPeriod.argtypes = ( + UINT, # uPeriod: Minimum timer resolution, in milliseconds +) + +timeEndPeriod = WINFUNCTYPE(None)(("timeEndPeriod", dll)) +""" +Clears a previously set minimum timer resolution. +uPeriod must match the value passed to the corresponding timeBeginPeriod call. + +.. seealso:: + https://learn.microsoft.com/en-us/windows/win32/api/timeapi/nf-timeapi-timeendperiod +""" +timeEndPeriod.restype = MMRESULT +timeEndPeriod.argtypes = ( + UINT, # uPeriod: Minimum timer resolution to clear, in milliseconds +) diff --git a/tests/unit/test_magnifier/test_fullscreenMagnifier.py b/tests/unit/test_magnifier/test_fullscreenMagnifier.py index 9f7f348aaa9..e4d28a4ab59 100644 --- a/tests/unit/test_magnifier/test_fullscreenMagnifier.py +++ b/tests/unit/test_magnifier/test_fullscreenMagnifier.py @@ -263,27 +263,19 @@ def testAttemptRecoveryFailureStopsMagnifier(self): self.assertEqual(magnifier._consecutiveErrors, 0) def testUpdateLoopSurvivesSingleDoUpdateError(self): - """A single _doUpdate error does not kill the update loop.""" + """A single _doUpdate error increments the counter; a subsequent success resets it.""" magnifier = FullScreenMagnifier() magnifier._startMagnifier() - magnifier._startTimer = MagicMock() magnifier._focusManager.getCurrentFocusCoordinates = MagicMock( return_value=(100, 200), ) - - # First call fails, second succeeds magnifier._doUpdate = MagicMock(side_effect=[OSError("Transient"), None]) - # First update — error magnifier._updateMagnifier() self.assertEqual(magnifier._consecutiveErrors, 1) - magnifier._startTimer.assert_called_with(magnifier._updateMagnifier) - # Second update — success - magnifier._startTimer.reset_mock() magnifier._updateMagnifier() self.assertEqual(magnifier._consecutiveErrors, 0) - magnifier._startTimer.assert_called_with(magnifier._updateMagnifier) magnifier._stopMagnifier() diff --git a/tests/unit/test_magnifier/test_magnifier.py b/tests/unit/test_magnifier/test_magnifier.py index 7a26dd883c3..729dd763565 100644 --- a/tests/unit/test_magnifier/test_magnifier.py +++ b/tests/unit/test_magnifier/test_magnifier.py @@ -144,29 +144,22 @@ def testUpdateMagnifier(self): 2, ) self.magnifier._doUpdate.assert_called_once() - self.magnifier._startTimer.assert_called_once_with( - self.magnifier._updateMagnifier, - ) + self.magnifier._startTimer.assert_not_called() self.assertEqual(self.magnifier.currentCoordinates, focusCoords) # Successful update should reset error counter self.assertEqual(self.magnifier._consecutiveErrors, 0) def testUpdateMagnifierResumesAfterSingleError(self): - """Timer must always be rescheduled even when _doUpdate raises an exception.""" + """Transient errors below the threshold increment the error counter without triggering recovery.""" self.magnifier._isActive = True focusCoords = Coordinates(self.screenWidth // 2, self.screenHeight // 2) self.magnifier._focusManager.getCurrentFocusCoordinates = MagicMock( return_value=focusCoords, ) self.magnifier._doUpdate = MagicMock(side_effect=OSError("COM failure")) - self.magnifier._startTimer = MagicMock() self.magnifier._updateMagnifier() - # Timer must still be rescheduled despite the error - self.magnifier._startTimer.assert_called_once_with( - self.magnifier._updateMagnifier, - ) self.assertEqual(self.magnifier._consecutiveErrors, 1) def testUpdateMagnifierTriggersRecoveryAfterMaxErrors(self): @@ -189,18 +182,16 @@ def testUpdateMagnifierTriggersRecoveryAfterMaxErrors(self): self.magnifier._startTimer.assert_not_called() def testUpdateMagnifierCatchesCOMError(self): - """COMError from UIA must be caught and the timer rescheduled.""" + """COMError is caught like OSError and increments the consecutive error counter.""" self.magnifier._isActive = True focusCoords = Coordinates(self.screenWidth // 2, self.screenHeight // 2) self.magnifier._focusManager.getCurrentFocusCoordinates = MagicMock( return_value=focusCoords, ) self.magnifier._doUpdate = MagicMock(side_effect=COMError(-2147417848, "RPC_E_DISCONNECTED", None)) - self.magnifier._startTimer = MagicMock() self.magnifier._updateMagnifier() - self.magnifier._startTimer.assert_called_once_with(self.magnifier._updateMagnifier) self.assertEqual(self.magnifier._consecutiveErrors, 1) def testUpdateMagnifierRecoveryFailureSafelyRestartsTimer(self): @@ -229,13 +220,11 @@ def testUpdateMagnifierResetsErrorCountOnSuccess(self): self.magnifier._focusManager.getCurrentFocusCoordinates = MagicMock( return_value=focusCoords, ) - self.magnifier._doUpdate = MagicMock() # Success - self.magnifier._startTimer = MagicMock() + self.magnifier._doUpdate = MagicMock() self.magnifier._updateMagnifier() self.assertEqual(self.magnifier._consecutiveErrors, 0) - self.magnifier._startTimer.assert_called_once() def testAttemptRecoveryBase(self): """Base _attemptRecovery resets errors and restarts timer.""" From 8a0f5542c814193de6929892cde3e296c1d1db04 Mon Sep 17 00:00:00 2001 From: Antoine HAFFREINGUE Date: Wed, 24 Jun 2026 14:20:08 +0200 Subject: [PATCH 2/2] tests --- tests/unit/test_magnifier/test_magnifier.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/unit/test_magnifier/test_magnifier.py b/tests/unit/test_magnifier/test_magnifier.py index 729dd763565..649fd748fdc 100644 --- a/tests/unit/test_magnifier/test_magnifier.py +++ b/tests/unit/test_magnifier/test_magnifier.py @@ -7,6 +7,7 @@ from _magnifier.magnifier import Magnifier from _magnifier.utils.types import Filter, Direction, Coordinates, MagnifierAction from comtypes import COMError +import time import unittest from winAPI._displayTracking import getPrimaryDisplayOrientation from unittest.mock import MagicMock, patch @@ -33,9 +34,14 @@ def setUp(self): mock.MagUninitialize.return_value = True mock.MagSetFullscreenTransform.return_value = True mock.MagSetFullscreenColorEffect.return_value = True + self.winmm_patcher = patch("_magnifier.magnifier.winmm") + self.mock_winmm = self.winmm_patcher.start() + self.mock_winmm.timeBeginPeriod.return_value = 0 + self.mock_winmm.TIMERR_NOERROR = 0 def tearDown(self): """Cleanup after each test.""" + self.winmm_patcher.stop() self.mag_fs_patcher.stop() self.mag_patcher.stop() @@ -258,6 +264,21 @@ def testStopMagnifier(self): self.magnifier._stopTimer.assert_called_once() self.assertFalse(self.magnifier._isActive) + def testTimeBeginPeriodReducesTimerResolution(self): + """timeBeginPeriod(1) must reduce actual sleep(1ms) duration below 5ms.""" + from winBindings import winmm as real_winmm + + real_winmm.timeBeginPeriod(1) + try: + durations_ms = [] + for _ in range(20): + t0 = time.perf_counter() + time.sleep(0.001) + durations_ms.append((time.perf_counter() - t0) * 1000) + self.assertLess(min(durations_ms), 5.0) + finally: + real_winmm.timeEndPeriod(1) + def testZoom(self): """zoom in and out with valid values and check boundaries.""" # Set initial zoom to 1.0 for predictable testing