improved macro tests

README.md

@@ -33,17 +33,20 @@ It is possible to write timed macros into the center column:
 - `k(1).w(10).k(2)` 12
 - `r(3, k(a).w(10))` aaa
 - `r(2, k(a).k(-)).k(b)` a-a-b
-- `w(1000).m(SHIFT_L, r(2, k(a))).w(10, 20).k(b)` AAb
+- `w(1000).m(SHIFT_L, r(2, k(a))).w(10).k(b)` AAb
 
 Documentation:
 - `r` repeats the execution of the second parameter
-- `w` waits in milliseconds (randomly with 2 parameters)
+- `w` waits in milliseconds
 - `k` writes a single keystroke
 - `m` holds a modifier while executing the second parameter
 - `.` executes two actions behind each other
 
 For a list of supported keystrokes and their names, check the output of `xmodmap -pke`
 
+Maybe you shouldn't use this feature in online PVP though. Might even get
+detected by the game.
+
 ## Git Installation
 
 ```bash

keymapper/config.py

@@ -33,9 +33,13 @@ from keymapper.logger import logger
 
 CONFIG_PATH = os.path.join(CONFIG, 'config')
 
-# an empty config with basic expected substructures
 INITIAL_CONFIG = {
-    'autoload': {}
+    'autoload': {},
+    'macros': {
+        # some time between keystrokes might be required for them to be
+        # detected properly in software.
+        'keystroke_sleep_ms': 10
+    }
 }
 
 
@@ -54,6 +58,11 @@ class _Config:
         elif self._config['autoload'].get(device) is not None:
             del self._config['autoload'][device]
 
+    def get_keystroke_sleep(self):
+        """Get the seconds of sleep between key down and up events."""
+        macros = self._config.get('macros', {})
+        return macros.get('keystroke_sleep_ms', 10)
+
     def iterate_autoload_presets(self):
         """Get tuples of (device, preset)."""
         return self._config.get('autoload', {}).items()

keymapper/dev/injector.py

@@ -22,7 +22,6 @@
 """Keeps injecting keycodes in the background based on the mapping."""
 
 
-import os
 import re
 import asyncio
 import time
@@ -220,11 +219,7 @@ class KeycodeInjector:
 
     def _write(self, device, keycode, value):
         """Actually inject."""
-        device.write(
-            evdev.ecodes.EV_KEY,
-            keycode - KEYCODE_OFFSET,
-            value
-        )
+        device.write(evdev.ecodes.EV_KEY, keycode - KEYCODE_OFFSET, value)
         device.syn()
 
     async def _injection_loop(self, device, keymapper_device):
@@ -237,6 +232,23 @@ class KeycodeInjector:
         keymapper_device : evdev.UInput
             where to write keycodes to
         """
+        # Parse all macros beforehand
+        logger.debug('Parsing macros')
+        macros = {}
+        for keycode, output in self.mapping:
+            if '(' in output and ')' in output and len(output) > 4:
+                # probably a macro
+                macros[keycode] = parse(
+                    output,
+                    lambda char, value: (
+                        self._write(
+                            keymapper_device,
+                            system_mapping.get_keycode(char),
+                            value
+                        )
+                    )
+                )
+
         logger.debug(
             'Started injecting into %s, fd %s',
             keymapper_device.device.path, keymapper_device.fd
@@ -261,24 +273,20 @@ class KeycodeInjector:
                 target_keycode = input_keycode
             elif '(' in character:
                 # must be a macro
+                if event.value == 0:
+                    continue
+
                 logger.spam(
                     'got code:%s value:%s, maps to macro %s',
                     event.code + KEYCODE_OFFSET,
                     event.value,
                     character
                 )
-                # TODO prepare this beforehand, not on each keystroke
+                macro = macros.get(input_keycode)
                 # TODO test if m(SHIFT_L, k(a)) prints A in injector tests
-                parse(
-                    character,
-                    handler=lambda keycode, value: (
-                        self._write(
-                            keymapper_device,
-                            keycode,
-                            value
-                        )
-                    )
-                ).run()
+                if macro is not None:
+                    asyncio.ensure_future(macro.run())
+                continue
             else:
                 target_keycode = system_mapping.get_keycode(character)
                 if target_keycode is None:

keymapper/dev/macros.py

@@ -37,9 +37,9 @@ w(1000).m(SHIFT_L, r(2, k(a))).w(10).k(b): <1s> A A <10ms> b
 
 import asyncio
 import re
-import random
 
 from keymapper.logger import logger
+from keymapper.config import config
 
 
 class _Macro:
@@ -77,8 +77,11 @@ class _Macro:
         macro : _Macro
         """
         self.tasks.append(lambda: self.handler(modifier, 1))
+        self.add_keycode_pause()
         self.tasks.append(macro.run)
+        self.add_keycode_pause()
         self.tasks.append(lambda: self.handler(modifier, 0))
+        self.add_keycode_pause()
         return self
 
     def repeat(self, repeats, macro):
@@ -93,21 +96,33 @@ class _Macro:
             self.tasks.append(macro.run)
         return self
 
+    def add_keycode_pause(self):
+        """To add a pause between keystrokes."""
+        sleeptime = config.get_keystroke_sleep() / 1000
+
+        async def sleep():
+            await asyncio.sleep(sleeptime)
+
+        self.tasks.append(sleep)
+
     def keycode(self, character):
         """Write the character."""
         self.tasks.append(lambda: self.handler(character, 1))
+        self.tasks.append(lambda: logger.spam(
+            'macro writes character %s',
+            character
+        ))
+        self.add_keycode_pause()
         self.tasks.append(lambda: self.handler(character, 0))
+        self.add_keycode_pause()
         return self
 
-    def wait(self, min_time, max_time=None):
-        """Wait a random time in milliseconds"""
-        async def sleep():
-            if max_time is None:
-                sleeptime = min_time
-            else:
-                sleeptime = random.random() * (max_time - min_time) + min_time
+    def wait(self, sleeptime):
+        """Wait time in milliseconds."""
+        sleeptime /= 1000
 
-            await asyncio.sleep(sleeptime / 1000)
+        async def sleep():
+            await asyncio.sleep(sleeptime)
 
         self.tasks.append(sleep)
         return self

tests/testcases/macros.py

@@ -24,6 +24,7 @@ import unittest
 import asyncio
 
 from keymapper.dev.macros import parse, _Macro
+from keymapper.config import config
 
 
 class TestMacros(unittest.TestCase):
@@ -50,19 +51,25 @@ class TestMacros(unittest.TestCase):
     
     def test_2(self):
         start = time.time()
-        macro = 'r(1, k(k))'
+        repeats = 20
+        macro = f'r({repeats}, k(k))'
         self.loop.run_until_complete(parse(macro, self.handler).run())
-        self.assertLess(time.time() - start, 0.1)
-        self.assertListEqual(self.result, [
-            ('k', 1), ('k', 0),
-        ])
+        sleep_time = 2 * repeats * config.get_keystroke_sleep() / 1000
+        self.assertGreater(time.time() - start, sleep_time * 0.9)
+        self.assertLess(time.time() - start, sleep_time * 1.1)
+        self.assertListEqual(self.result, [('k', 1), ('k', 0)] * repeats)
 
     def test_3(self):
         start = time.time()
-        macro = 'r(3, k(m).w(100, 200))'
+        macro = 'r(3, k(m).w(100))'
         self.loop.run_until_complete(parse(macro, self.handler).run())
-        self.assertGreater(time.time() - start, 0.1 * 3)
-        self.assertLess(time.time() - start, 0.21 * 3)
+
+        keystroke_time = 6 * config.get_keystroke_sleep()
+        total_time = keystroke_time + 300
+        total_time /= 1000
+
+        self.assertGreater(time.time() - start, total_time * 0.9)
+        self.assertLess(time.time() - start, total_time * 1.1)
         self.assertListEqual(self.result, [
             ('m', 1), ('m', 0),
             ('m', 1), ('m', 0),
@@ -84,8 +91,14 @@ class TestMacros(unittest.TestCase):
         start = time.time()
         macro = 'w(200).r(2,m(w,\rr(2,\tk(r))).w(10).k(k))'
         self.loop.run_until_complete(parse(macro, self.handler).run())
-        self.assertLess(time.time() - start, 0.23)
-        self.assertGreater(time.time() - start, 0.21)
+
+        num_pauses = 8 + 6 + 4
+        keystroke_time = num_pauses * config.get_keystroke_sleep()
+        wait_time = 220
+        total_time = (keystroke_time + wait_time) / 1000
+
+        self.assertLess(time.time() - start, total_time * 1.1)
+        self.assertGreater(time.time() - start, total_time * 0.9)
         expected = [('w', 1)]
         expected += [('r', 1), ('r', 0)] * 2
         expected += [('w', 0)]