Skip to content
Draft
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
15 changes: 9 additions & 6 deletions source/_magnifier/magnifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
"""
Expand All @@ -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)
Expand Down Expand Up @@ -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:
"""
Expand Down
28 changes: 27 additions & 1 deletion source/winBindings/winmm.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@

DWORD_PTR = c_size_t
MMRESULT = c_long

TIMERR_NOERROR: int = 0

dll = windll.winmm

Expand Down Expand Up @@ -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
)
10 changes: 1 addition & 9 deletions tests/unit/test_magnifier/test_fullscreenMagnifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down
40 changes: 25 additions & 15 deletions tests/unit/test_magnifier/test_magnifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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()

Expand Down Expand Up @@ -144,29 +150,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):
Expand All @@ -189,18 +188,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):
Expand Down Expand Up @@ -229,13 +226,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."""
Expand Down Expand Up @@ -269,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
Expand Down
Loading