Add mouse_xy macro (#1029)

inputremapper/configs/preset.py

@@ -20,7 +20,6 @@
 """Contains and manages mappings."""
 
 from __future__ import annotations
-from evdev import ecodes
 
 import json
 import os
@@ -36,6 +35,8 @@
     overload,
 )
 
+from evdev import ecodes
+
 try:
     from pydantic.v1 import ValidationError
 except ImportError:
@@ -291,7 +292,11 @@ def _get_mappings_from_disc(self) -> Dict[InputCombination, MappingModel]:
 
         for mapping_dict in preset_list:
             if not isinstance(mapping_dict, dict):
-                logger.error("Expected mapping to be a dict: %s", mapping_dict)
+                logger.error(
+                    "Expected mapping to be a dict: %s %s",
+                    type(mapping_dict),
+                    mapping_dict,
+                )
                 continue
 
             try:

inputremapper/injection/macros/parse.py

@@ -44,6 +44,7 @@
 from inputremapper.injection.macros.tasks.mod_tap import ModTapTask
 from inputremapper.injection.macros.tasks.modify import ModifyTask
 from inputremapper.injection.macros.tasks.mouse import MouseTask
+from inputremapper.injection.macros.tasks.mouse_xy import MouseXYTask
 from inputremapper.injection.macros.tasks.repeat import RepeatTask
 from inputremapper.injection.macros.tasks.set import SetTask
 from inputremapper.injection.macros.tasks.wait import WaitTask
@@ -67,6 +68,7 @@ class Parser:
         "hold": HoldTask,
         "hold_keys": HoldKeysTask,
         "mouse": MouseTask,
+        "mouse_xy": MouseXYTask,
         "wheel": WheelTask,
         "if_eq": IfEqTask,
         "if_numlock": IfNumlockTask,

inputremapper/injection/macros/tasks/mouse.py

@@ -20,16 +20,14 @@
 
 from __future__ import annotations
 
-import asyncio
-
 from evdev._ecodes import REL_Y, REL_X
-from evdev.ecodes import EV_REL
 
 from inputremapper.injection.macros.argument import ArgumentConfig
-from inputremapper.injection.macros.task import Task
+from inputremapper.injection.macros.macro import InjectEventCallback
+from inputremapper.injection.macros.tasks.mouse_xy import MouseXYTask
 
 
-class MouseTask(Task):
+class MouseTask(MouseXYTask):
     """Move the mouse cursor."""
 
     argument_configs = [
@@ -46,40 +44,26 @@ class MouseTask(Task):
         ArgumentConfig(
             name="acceleration",
             position=2,
-            types=[int, float, None],
-            default=None,
+            types=[int, float],
+            default=1,
         ),
     ]
 
-    async def run(self, callback) -> None:
+    async def run(self, callback: InjectEventCallback) -> None:
         direction = self.get_argument("direction").get_value()
         speed = self.get_argument("speed").get_value()
         acceleration = self.get_argument("acceleration").get_value()
 
-        code, value = {
+        code, direction = {
             "up": (REL_Y, -1),
             "down": (REL_Y, 1),
             "left": (REL_X, -1),
             "right": (REL_X, 1),
         }[direction.lower()]
 
-        current_speed = 0.0
-        displacement_accumulator = 0.0
-        displacement = 0
-        if not acceleration:
-            displacement = speed
-
-        while self.is_holding():
-            # Cursors can only move by integers. To get smooth acceleration for
-            # small acceleration values, the cursor needs to move by a pixel every
-            # few iterations. This can be achieved by remembering the decimal
-            # places that were cast away, and using them for the next iteration.
-            if acceleration and current_speed < speed:
-                current_speed += acceleration
-                current_speed = min(current_speed, speed)
-                displacement_accumulator += current_speed
-                displacement = int(displacement_accumulator)
-                displacement_accumulator -= displacement
-
-            callback(EV_REL, code, value * displacement)
-            await asyncio.sleep(1 / self.mapping.rel_rate)
+        await self.axis(
+            code,
+            direction * speed,
+            acceleration,
+            callback,
+        )

inputremapper/injection/macros/tasks/mouse_xy.py

@@ -0,0 +1,99 @@
+#!/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
+from typing import Union
+
+from evdev._ecodes import REL_Y, REL_X
+from evdev.ecodes import EV_REL
+
+from inputremapper.injection.macros.argument import ArgumentConfig
+from inputremapper.injection.macros.macro import InjectEventCallback
+from inputremapper.injection.macros.task import Task
+from inputremapper.injection.macros.tasks.util import precise_iteration_frequency
+
+
+class MouseXYTask(Task):
+    """Move the mouse cursor."""
+
+    argument_configs = [
+        ArgumentConfig(
+            name="x",
+            position=0,
+            types=[int, float],
+            default=0,
+        ),
+        ArgumentConfig(
+            name="y",
+            position=1,
+            types=[int, float],
+            default=0,
+        ),
+        ArgumentConfig(
+            name="acceleration",
+            position=2,
+            types=[int, float],
+            default=1,
+        ),
+    ]
+
+    async def run(self, callback: InjectEventCallback) -> None:
+        x = self.get_argument("x").get_value()
+        y = self.get_argument("y").get_value()
+        acceleration = self.get_argument("acceleration").get_value()
+        await asyncio.gather(
+            self.axis(REL_X, x, acceleration, callback),
+            self.axis(REL_Y, y, acceleration, callback),
+        )
+
+    async def axis(
+        self,
+        code: int,
+        speed: Union[int, float],
+        fractional_acceleration: Union[int, float],
+        callback: InjectEventCallback,
+    ) -> None:
+        acceleration = speed * fractional_acceleration
+        direction = -1 if speed < 0 else 1
+        current_speed = 0.0
+        displacement_accumulator = 0.0
+        displacement = 0
+        if acceleration <= 0:
+            displacement = int(speed)
+
+        async for _ in precise_iteration_frequency(self.mapping.rel_rate):
+            if not self.is_holding():
+                return
+
+            # Cursors can only move by integers. To get smooth acceleration for
+            # small acceleration values, the cursor needs to move by a pixel every
+            # few iterations. This can be achieved by remembering the decimal
+            # places that were cast away, and using them for the next iteration.
+            if acceleration:
+                current_speed += acceleration
+                current_speed = direction * min(abs(current_speed), abs(speed))
+                displacement_accumulator += current_speed
+                displacement = int(displacement_accumulator)
+                displacement_accumulator -= displacement
+
+            if displacement != 0:
+                callback(EV_REL, code, displacement)

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.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):
@@ -54,6 +54,7 @@ class WheelTask(Task):
     async def run(self, callback) -> None:
         direction = self.get_argument("direction").get_value()
 
+        # 120, see https://www.kernel.org/doc/html/latest/input/event-codes.html#ev-rel
         code, value = {
             "up": ([REL_WHEEL, REL_WHEEL_HI_RES], [1 / 120, 1]),
             "down": ([REL_WHEEL, REL_WHEEL_HI_RES], [-1 / 120, -1]),
@@ -63,10 +64,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)