From bf25ad25a87e62d4d0ea626640f02acb99c3a28a Mon Sep 17 00:00:00 2001 From: sezanzeb <28510156+sezanzeb@users.noreply.github.com> Date: Fri, 3 Jan 2025 14:27:50 +0100 Subject: [PATCH] fixed wheel macro frequency, more tests for wheel --- .../injection/macros/tasks/mouse_xy.py | 21 +---- inputremapper/injection/macros/tasks/util.py | 45 ++++++++++ inputremapper/injection/macros/tasks/wheel.py | 9 +- tests/unit/test_macros/test_mouse.py | 82 +++++++++++-------- 4 files changed, 100 insertions(+), 57 deletions(-) create mode 100644 inputremapper/injection/macros/tasks/util.py diff --git a/inputremapper/injection/macros/tasks/mouse_xy.py b/inputremapper/injection/macros/tasks/mouse_xy.py index c6c8a79c9..c11a49db4 100644 --- a/inputremapper/injection/macros/tasks/mouse_xy.py +++ b/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): diff --git a/inputremapper/injection/macros/tasks/util.py b/inputremapper/injection/macros/tasks/util.py new file mode 100644 index 000000000..3b67714fb --- /dev/null +++ b/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 +# +# 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 . + +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 diff --git a/inputremapper/injection/macros/tasks/wheel.py b/inputremapper/injection/macros/tasks/wheel.py index a63ba843b..e92d77b6e 100644 --- a/inputremapper/injection/macros/tasks/wheel.py +++ b/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) diff --git a/tests/unit/test_macros/test_mouse.py b/tests/unit/test_macros/test_mouse.py index b8e4fdab1..15b6c0172 100644 --- a/tests/unit/test_macros/test_mouse.py +++ b/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__":