fixed wheel macro frequency, more tests for wheel

inputremapper/injection/macros/tasks/mouse_xy.py

@@ -21,7 +21,6 @@
 from __future__ import annotations
 
 import asyncio
-import time
 from typing import Union
 
 from evdev._ecodes import REL_Y, REL_X
@@ -30,25 +29,7 @@
 from inputremapper.injection.macros.argument import ArgumentConfig
 from inputremapper.injection.macros.macro import InjectEventCallback
 from inputremapper.injection.macros.task import Task
-
-
-async def precise_iteration_frequency(rate: float):
-    """asyncio.sleep might end up sleeping too long, for whatever reason. Maybe other
-    async function calls that take longer than expected in the background. This
-    generator can be used to achieve the proper iteration frequency.
-    """
-    sleep = 1 / rate
-    corrected_sleep = sleep
-    error = 0
-
-    while True:
-        start = time.time()
-
-        yield
-
-        corrected_sleep -= error
-        await asyncio.sleep(corrected_sleep)
-        error = (time.time() - start) - sleep
+from inputremapper.injection.macros.tasks.util import precise_iteration_frequency
 
 
 class MouseXYTask(Task):

inputremapper/injection/macros/tasks/util.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+# input-remapper - GUI for device specific keyboard mappings
+# Copyright (C) 2024 sezanzeb <[email protected]>
+#
+# This file is part of input-remapper.
+#
+# input-remapper is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# input-remapper is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with input-remapper.  If not, see <https://www.gnu.org/licenses/>.
+
+from __future__ import annotations
+
+import asyncio
+import time
+from typing import AsyncIterator
+
+
+async def precise_iteration_frequency(frequency: float) -> AsyncIterator[None]:
+    """A generator to iterate over in a fixed frequency.
+
+    asyncio.sleep might end up sleeping too long, for whatever reason. Maybe there are
+    other async function calls that take longer than expected in the background.
+    """
+    sleep = 1 / frequency
+    corrected_sleep = sleep
+    error = 0
+
+    while True:
+        start = time.time()
+
+        yield
+
+        corrected_sleep -= error
+        await asyncio.sleep(corrected_sleep)
+        error = (time.time() - start) - sleep

inputremapper/injection/macros/tasks/wheel.py

@@ -20,7 +20,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import math
 
 from evdev.ecodes import (
@@ -33,6 +32,7 @@
 
 from inputremapper.injection.macros.argument import ArgumentConfig
 from inputremapper.injection.macros.task import Task
+from inputremapper.injection.macros.tasks.util import precise_iteration_frequency
 
 
 class WheelTask(Task):
@@ -63,10 +63,13 @@ async def run(self, callback) -> None:
 
         speed = self.get_argument("speed").get_value()
         remainder = [0.0, 0.0]
-        while self.is_holding():
+
+        async for _ in precise_iteration_frequency(self.mapping.rel_rate):
+            if not self.is_holding():
+                return
+
             for i in range(0, 2):
                 float_value = value[i] * speed + remainder[i]
                 remainder[i] = math.fmod(float_value, 1)
                 if abs(float_value) >= 1:
                     callback(EV_REL, code[i], int(float_value))
-            await asyncio.sleep(1 / self.mapping.rel_rate)

tests/unit/test_macros/test_mouse.py

@@ -21,7 +21,15 @@
 import asyncio
 import unittest
 
-from evdev._ecodes import REL_Y, EV_REL, REL_HWHEEL, REL_HWHEEL_HI_RES, REL_X
+from evdev._ecodes import (
+    REL_Y,
+    EV_REL,
+    REL_HWHEEL,
+    REL_HWHEEL_HI_RES,
+    REL_X,
+    REL_WHEEL,
+    REL_WHEEL_HI_RES,
+)
 
 from inputremapper.injection.macros.parse import Parser
 from tests.lib.test_setup import test_setup
@@ -133,47 +141,52 @@ async def test_mouse_xy_only_y(self):
             self._get_y_movement(),
         )
 
-    async def test_mouse_and_wheel(self):
+    async def test_wheel_left(self):
         wheel_speed = 60
-        macro_1 = Parser.parse("mouse(up, 4)", self.context, DummyMapping)
-        macro_2 = Parser.parse(
-            f"wheel(left, {wheel_speed})", self.context, DummyMapping
-        )
-        macro_1.press_trigger()
-        macro_2.press_trigger()
-        asyncio.ensure_future(macro_1.run(self.handler))
-        asyncio.ensure_future(macro_2.run(self.handler))
-
         sleep = 0.1
-        await asyncio.sleep(sleep)
-        self.assertTrue(macro_1.tasks[0].is_holding())
-        self.assertTrue(macro_2.tasks[0].is_holding())
-        macro_1.release_trigger()
-        macro_2.release_trigger()
-
-        self.assertIn((EV_REL, REL_Y, -4), self.result)
-        expected_wheel_hi_res_event_count = sleep * DummyMapping.rel_rate
-        expected_wheel_event_count = int(
-            expected_wheel_hi_res_event_count / 120 * wheel_speed
-        )
-        actual_wheel_event_count = self.result.count((EV_REL, REL_HWHEEL, 1))
-        actual_wheel_hi_res_event_count = self.result.count(
+        await self._run_mouse_macro(f"wheel(left, {wheel_speed})", sleep)
+
+        expected_num_hires_events = sleep * DummyMapping.rel_rate
+        expected_num_wheel_events = int(expected_num_hires_events / 120 * wheel_speed)
+        actual_num_wheel_events = self.result.count((EV_REL, REL_HWHEEL, 1))
+        actual_num_hires_events = self.result.count(
             (
                 EV_REL,
                 REL_HWHEEL_HI_RES,
                 wheel_speed,
             )
         )
-        # this seems to have a tendency of injecting less wheel events,
-        # especially if the sleep is short
-        self.assertGreater(actual_wheel_event_count, expected_wheel_event_count * 0.8)
-        self.assertLess(actual_wheel_event_count, expected_wheel_event_count * 1.1)
+
         self.assertGreater(
-            actual_wheel_hi_res_event_count, expected_wheel_hi_res_event_count * 0.8
+            actual_num_wheel_events,
+            expected_num_wheel_events * 0.9,
         )
         self.assertLess(
-            actual_wheel_hi_res_event_count, expected_wheel_hi_res_event_count * 1.1
+            actual_num_wheel_events,
+            expected_num_wheel_events * 1.1,
+        )
+        self.assertGreater(
+            actual_num_hires_events,
+            expected_num_hires_events * 0.9,
         )
+        self.assertLess(
+            actual_num_hires_events,
+            expected_num_hires_events * 1.1,
+        )
+
+    async def test_wheel_up(self):
+        wheel_speed = 60
+        sleep = 0.1
+        await self._run_mouse_macro(f"wheel(up, {wheel_speed})", sleep)
+        self.assertIn((EV_REL, REL_WHEEL, 1), self.result)
+        self.assertIn((EV_REL, REL_WHEEL_HI_RES, 60), self.result)
+
+    async def test_wheel_down(self):
+        wheel_speed = 60
+        sleep = 0.1
+        await self._run_mouse_macro(f"wheel(down, {wheel_speed})", sleep)
+        self.assertIn((EV_REL, REL_WHEEL, -1), self.result)
+        self.assertIn((EV_REL, REL_WHEEL_HI_RES, -60), self.result)
 
     def _get_x_movement(self):
         return [event for event in self.result if event[1] == REL_X]
@@ -189,16 +202,17 @@ async def _run_mouse_macro(
     ):
         dummy_mapping = DummyMapping()
         dummy_mapping.rel_rate = rel_rate
-        macro_1 = Parser.parse(
+        macro = Parser.parse(
             code,
             self.context,
             dummy_mapping,
         )
-        macro_1.press_trigger()
-        asyncio.ensure_future(macro_1.run(self.handler))
+        macro.press_trigger()
+        asyncio.ensure_future(macro.run(self.handler))
 
         await asyncio.sleep(time)
-        macro_1.release_trigger()
+        self.assertTrue(macro.tasks[0].is_holding())
+        macro.release_trigger()
 
 
 if __name__ == "__main__":