#63 added e, mouse and wheel macros

data/key-mapper.glade

@@ -890,22 +890,29 @@ Don't hold down any keys while the injection starts.</property>
                         <property name="tooltip-text" translatable="yes">"disable" disables the key outside of combinations.
 Useful for turning a key into a modifier without any side effects.
 
+Macro help:
+- `r` repeats the execution of the second parameter
+- `w` waits in milliseconds
+- `k` writes a single keystroke
+- `e` writes an event
+- `m` holds a modifier while executing the second parameter
+- `h` executes the parameter as long as the key is pressed down
+- `.` executes two actions behind each other
+- `mouse` and `wheel` take direction and speed as parameters
+
 Macro examples:
-- k(a)
-- r(3, k(a).w(500))
-- h(k(a)).k(b)
-- m(Control_L, k(a).k(x))
+- `k(1).k(2)` 1, 2
+- `r(3, k(a).w(500))` a, a, a with 500ms pause
+- `m(Control_L, k(a).k(x))` CTRL + a, CTRL + x
+- `k(1).h(k(2)).k(3)` writes 1 2 2 ... 2 2 3 while the key is pressed
+- `e(EV_REL, REL_X, 10)` moves the mouse cursor 10px to the right
+- `mouse(right, 4)` which keeps moving the mouse while pressed
+- `wheel(down, 1)` keeps scrolling down while held
 
-Help:
-- r: repeats the execution of the second parameter
-- w: waits in milliseconds
-- k: writes a single keystroke
-- m: holds a modifier while executing the second parameter
-- h: executes the parameter as long as the key is pressed down
+Combine keycodes with `+`, for example: `control_l + a`, to write combinations
 
-Between calls to k, key down and key up events, macros
-will sleep for 10ms by default. This can be configured in
-~/.config/key-mapper/config</property>
+Between calls to k, key down and key up events, macros will sleep for 10ms by
+default. This can be configured in ~/.config/key-mapper/config</property>
                         <property name="margin-top">5</property>
                         <property name="margin-bottom">5</property>
                         <property name="label" translatable="yes">Mapping</property>

keymapper/injection/injector.py

@@ -251,7 +251,13 @@ class Injector(multiprocessing.Process):
 
         # and all keycodes that are injected by macros
         for macro in self.context.macros.values():
-            capabilities[EV_KEY] += list(macro.get_capabilities())
+            macro_capabilities = macro.get_capabilities()
+            for ev_type in macro_capabilities:
+                if len(macro_capabilities[ev_type]) == 0:
+                    continue
+                if ev_type not in capabilities:
+                    capabilities[ev_type] = []
+                capabilities[ev_type] += list(macro_capabilities[ev_type])
 
         if gamepad and self.context.joystick_as_mouse():
             # REL_WHEEL was also required to recognize the gamepad

keymapper/injection/keycode_mapper.py

@@ -225,9 +225,9 @@ class KeycodeMapper:
                         f'but got {key}'
                     )
 
-    def macro_write(self, code, value):
+    def macro_write(self, ev_type, code, value):
         """Handler for macros."""
-        self.context.uinput.write(EV_KEY, code, value)
+        self.context.uinput.write(ev_type, code, value)
         self.context.uinput.syn()
 
     def write(self, key):

keymapper/injection/macros.py

@@ -37,19 +37,15 @@ w(1000).m(Shift_L, r(2, k(a))).w(10).k(b): <1s> A A <10ms> b
 
 import asyncio
 import re
+import copy
+
+from evdev.ecodes import ecodes, EV_KEY, EV_REL, REL_X, REL_Y, REL_WHEEL, \
+    REL_HWHEEL
 
 from keymapper.logger import logger
 from keymapper.state import system_mapping
 
 
-MODIFIER = 1
-CHILD_MACRO = 2
-SLEEP = 3
-REPEAT = 4
-KEYSTROKE = 5
-DEBUG = 6
-
-
 def is_this_a_macro(output):
     """Figure out if this is a macro."""
     if not isinstance(output, str):
@@ -73,7 +69,7 @@ class _Macro:
 
         Parameters
         ----------
-        code : string
+        code : string or None
             The original parsed code, for logging purposes.
         mapping : Mapping
             The preset object, needed for some config stuff
@@ -88,7 +84,10 @@ class _Macro:
         self.running = False
 
         # all required capabilities, without those of child macros
-        self.capabilities = set()
+        self.capabilities = {
+            EV_KEY: set(),
+            EV_REL: set(),
+        }
 
         self.child_macros = []
 
@@ -98,9 +97,16 @@ class _Macro:
 
     def get_capabilities(self):
         """Resolve all capabilities of the macro and those of its children."""
-        capabilities = self.capabilities.copy()
+        capabilities = copy.deepcopy(self.capabilities)
+
         for macro in self.child_macros:
-            capabilities.update(macro.get_capabilities())
+            macro_capabilities = macro.get_capabilities()
+            for ev_type in macro_capabilities:
+                if ev_type not in capabilities:
+                    capabilities[ev_type] = set()
+
+                capabilities[ev_type].update(macro_capabilities[ev_type])
+
         return capabilities
 
     async def run(self, handler):
@@ -109,14 +115,17 @@ class _Macro:
         Parameters
         ----------
         handler : function
-            Will receive int code and value for an EV_KEY event to write
+            Will receive int type, code and value for an event to write
         """
         if self.running:
             logger.error('Tried to run already running macro "%s"', self.code)
             return
 
         self.running = True
-        for _, task in self.tasks:
+        for task in self.tasks:
+            # one could call tasks the compiled macros. it's lambda functions
+            # that receive the handler as an argument, so that they know
+            # where to send the event to.
             coroutine = task(handler)
             if asyncio.iscoroutine(coroutine):
                 await coroutine
@@ -161,7 +170,7 @@ class _Macro:
                     # released
                     logger.error('Failed h(): %s', error)
 
-            self.tasks.append((1234, task))
+            self.tasks.append(task)
         else:
             if not isinstance(macro, _Macro):
                 raise ValueError(
@@ -175,7 +184,7 @@ class _Macro:
                     # not-releasing any key
                     await macro.run(handler)
 
-            self.tasks.append((REPEAT, task))
+            self.tasks.append(task)
             self.child_macros.append(macro)
 
         return self
@@ -200,15 +209,15 @@ class _Macro:
         if code is None:
             raise KeyError(f'Unknown modifier "{modifier}"')
 
-        self.capabilities.add(code)
+        self.capabilities[EV_KEY].add(code)
 
         self.child_macros.append(macro)
 
-        self.tasks.append((MODIFIER, lambda handler: handler(code, 1)))
+        self.tasks.append(lambda handler: handler(EV_KEY, code, 1))
         self.add_keycode_pause()
-        self.tasks.append((CHILD_MACRO, macro.run))
+        self.tasks.append(macro.run)
         self.add_keycode_pause()
-        self.tasks.append((MODIFIER, lambda handler: handler(code, 0)))
+        self.tasks.append(lambda handler: handler(EV_KEY, code, 0))
         self.add_keycode_pause()
         return self
 
@@ -235,7 +244,7 @@ class _Macro:
             ) from error
 
         for _ in range(repeats):
-            self.tasks.append((CHILD_MACRO, macro.run))
+            self.tasks.append(macro.run)
 
         self.child_macros.append(macro)
 
@@ -248,7 +257,7 @@ class _Macro:
         async def sleep(_):
             await asyncio.sleep(sleeptime)
 
-        self.tasks.append((SLEEP, sleep))
+        self.tasks.append(sleep)
 
     def keycode(self, character):
         """Write the character."""
@@ -256,16 +265,77 @@ class _Macro:
         code = system_mapping.get(character)
 
         if code is None:
-            raise KeyError(f'aUnknown key "{character}"')
+            raise KeyError(f'Unknown key "{character}"')
 
-        self.capabilities.add(code)
+        if EV_KEY not in self.capabilities:
+            self.capabilities[EV_KEY] = set()
+        self.capabilities[EV_KEY].add(code)
 
-        self.tasks.append((KEYSTROKE, lambda handler: handler(code, 1)))
+        self.tasks.append(lambda handler: handler(EV_KEY, code, 1))
+        self.add_keycode_pause()
+        self.tasks.append(lambda handler: handler(EV_KEY, code, 0))
         self.add_keycode_pause()
-        self.tasks.append((KEYSTROKE, lambda handler: handler(code, 0)))
+        return self
+
+    def event(self, ev_type, code, value):
+        """Write any event.
+
+        Parameters
+        ----------
+        ev_type: str or int
+            examples: 2, 'EV_KEY'
+        code : int or int
+            examples: 52, 'KEY_A'
+        value : int
+        """
+        if isinstance(ev_type, str):
+            ev_type = ecodes[ev_type.upper()]
+        if isinstance(code, str):
+            code = ecodes[code.upper()]
+
+        if ev_type not in self.capabilities:
+            self.capabilities[ev_type] = set()
+
+        if ev_type == EV_REL:
+            # add all capabilities that are required for the display server
+            # to recognize the device as mouse
+            self.capabilities[EV_REL].add(REL_X)
+            self.capabilities[EV_REL].add(REL_Y)
+            self.capabilities[EV_REL].add(REL_WHEEL)
+
+        self.capabilities[ev_type].add(code)
+
+        self.tasks.append(lambda handler: handler(ev_type, code, value))
         self.add_keycode_pause()
+
         return self
 
+    def mouse(self, direction, speed):
+        """Shortcut for h(e(...))."""
+        code, value = {
+            'up': (REL_Y, -1),
+            'down': (REL_Y, 1),
+            'left': (REL_X, -1),
+            'right': (REL_X, 1),
+        }[direction.lower()]
+        value *= speed
+        child_macro = _Macro(None, self.mapping)
+        child_macro.event(EV_REL, code, value)
+        self.hold(child_macro)
+
+    def wheel(self, direction, speed):
+        """Shortcut for h(e(...))."""
+        code, value = {
+            'up': (REL_WHEEL, 1),
+            'down': (REL_WHEEL, -1),
+            'left': (REL_HWHEEL, 1),
+            'right': (REL_HWHEEL, -1),
+        }[direction.lower()]
+        child_macro = _Macro(None, self.mapping)
+        child_macro.event(EV_REL, code, value)
+        child_macro.wait(100 / speed)
+        self.hold(child_macro)
+
     def wait(self, sleeptime):
         """Wait time in milliseconds."""
         try:
@@ -281,7 +351,7 @@ class _Macro:
         async def sleep(_):
             await asyncio.sleep(sleeptime)
 
-        self.tasks.append((SLEEP, sleep))
+        self.tasks.append(sleep)
         return self
 
 
@@ -385,8 +455,11 @@ def _parse_recurse(macro, mapping, macro_instance=None, depth=0):
             'm': (macro_instance.modify, 2, 2),
             'r': (macro_instance.repeat, 2, 2),
             'k': (macro_instance.keycode, 1, 1),
+            'e': (macro_instance.event, 3, 3),
             'w': (macro_instance.wait, 1, 1),
-            'h': (macro_instance.hold, 0, 1)
+            'h': (macro_instance.hold, 0, 1),
+            'mouse': (macro_instance.mouse, 2, 2),
+            'wheel': (macro_instance.wheel, 2, 2)
         }
 
         function = functions.get(call)
@@ -396,7 +469,7 @@ def _parse_recurse(macro, mapping, macro_instance=None, depth=0):
         # get all the stuff inbetween
         position = _count_brackets(macro)
 
-        inner = macro[2:position - 1]
+        inner = macro[macro.index('(') + 1:position - 1]
 
         # split "3, k(a).w(10)" into parameters
         string_params = _extract_params(inner)
@@ -510,5 +583,5 @@ def parse(macro, mapping, return_errors=False):
         macro_object = _parse_recurse(macro, mapping)
         return macro_object if not return_errors else None
     except Exception as error:
-        logger.error('Failed to parse macro "%s": %s', macro, error)
+        logger.error('Failed to parse macro "%s": %s', macro, error.__repr__())
         return str(error) if return_errors else None

readme/development.md

@@ -35,7 +35,7 @@ key-mapper.
 - [x] inject in an additional device instead to avoid clashing capabilities
 - [ ] don't run any GTK code as root for wayland compatibility
 - [ ] injecting keys that aren't available in the systems keyboard layout
-- [ ] ship with a list of all keys known to xkb and validate input in the gui
+- [ ] add comprehensive tabbed help popup
 
 ## Tests
 

readme/usage.md

@@ -73,19 +73,24 @@ names can be chained using ` + `.
 ## Macros
 
 It is possible to write timed macros into the center column:
-- `k(1).k(2)` 1, 2
-- `r(3, k(a).w(500))` a, a, a with 500ms pause
-- `m(Control_L, k(a).k(x))` CTRL + a, CTRL + x
-- `k(1).h(k(2)).k(3)` writes 1 2 2 ... 2 2 3 while the key is pressed
-
-Documentation:
 - `r` repeats the execution of the second parameter
 - `w` waits in milliseconds
 - `k` writes a single keystroke
+- `e` writes an event
 - `m` holds a modifier while executing the second parameter
 - `h` executes the parameter as long as the key is pressed down
 - `.` executes two actions behind each other
 
+Examples:
+- `k(1).k(2)` 1, 2
+- `r(3, k(a).w(500))` a, a, a with 500ms pause
+- `m(Control_L, k(a).k(x))` CTRL + a, CTRL + x
+- `k(1).h(k(2)).k(3)` writes 1 2 2 ... 2 2 3 while the key is pressed
+- `e(EV_REL, REL_X, 10)` moves the mouse cursor 10px to the right
+- `mouse(right, 4)` which keeps moving the mouse while pressed.
+  Made out of `h(e(...))` internally
+- `wheel(down, 1)` keeps scrolling down while held
+
 Syntax errors are shown in the UI on save. Each `k` function adds a short
 delay of 10ms between key-down, key-up and at the end. See
 [Configuration Files](#configuration-files) for more info.

tests/testcases/test_keycode_mapper.py

@@ -547,10 +547,10 @@ class TestKeycodeMapper(unittest.TestCase):
 
         # 6 keycodes written, with down and up events
         self.assertEqual(len(history), 12)
-        self.assertIn((code_a, 1), history)
-        self.assertIn((code_a, 0), history)
-        self.assertIn((code_b, 1), history)
-        self.assertIn((code_b, 0), history)
+        self.assertIn((EV_KEY, code_a, 1), history)
+        self.assertIn((EV_KEY, code_a, 0), history)
+        self.assertIn((EV_KEY, code_b, 1), history)
+        self.assertIn((EV_KEY, code_b, 0), history)
 
         # releasing stuff
         self.assertIn((EV_KEY, 1), unreleased)
@@ -611,14 +611,14 @@ class TestKeycodeMapper(unittest.TestCase):
         self.assertGreater(len(history), events * 0.9)
         self.assertLess(len(history), events * 1.1)
 
-        self.assertIn((code_a, 1), history)
-        self.assertIn((code_a, 0), history)
-        self.assertIn((code_b, 1), history)
-        self.assertIn((code_b, 0), history)
-        self.assertIn((code_c, 1), history)
-        self.assertIn((code_c, 0), history)
-        self.assertGreater(history.count((code_b, 1)), 1)
-        self.assertGreater(history.count((code_b, 0)), 1)
+        self.assertIn((EV_KEY, code_a, 1), history)
+        self.assertIn((EV_KEY, code_a, 0), history)
+        self.assertIn((EV_KEY, code_b, 1), history)
+        self.assertIn((EV_KEY, code_b, 0), history)
+        self.assertIn((EV_KEY, code_c, 1), history)
+        self.assertIn((EV_KEY, code_c, 0), history)
+        self.assertGreater(history.count((EV_KEY, code_b, 1)), 1)
+        self.assertGreater(history.count((EV_KEY, code_b, 0)), 1)
 
         # it's stopped and won't write stuff anymore
         count_before = len(history)
@@ -667,8 +667,8 @@ class TestKeycodeMapper(unittest.TestCase):
 
         loop.run_until_complete(asyncio.sleep(0.1))
         # starting code_c written
-        self.assertEqual(history.count((code_c, 1)), 1)
-        self.assertEqual(history.count((code_c, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 0)), 1)
 
         # spam garbage events
         for _ in range(5):
@@ -684,8 +684,8 @@ class TestKeycodeMapper(unittest.TestCase):
 
         # there should only be one code_c in the events, because no key
         # up event was ever done so the hold just continued
-        self.assertEqual(history.count((code_c, 1)), 1)
-        self.assertEqual(history.count((code_c, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 0)), 1)
         # without an key up event on 2, it won't write code_d
         self.assertNotIn((code_d, 1), history)
         self.assertNotIn((code_d, 0), history)
@@ -695,17 +695,17 @@ class TestKeycodeMapper(unittest.TestCase):
         loop.run_until_complete(asyncio.sleep(0.1))
 
         # it stopped and didn't restart, so the count stays at 1
-        self.assertEqual(history.count((code_c, 1)), 1)
-        self.assertEqual(history.count((code_c, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 0)), 1)
         # and the trailing d was written
-        self.assertEqual(history.count((code_d, 1)), 1)
-        self.assertEqual(history.count((code_d, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_d, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_d, 0)), 1)
 
         # it's stopped and won't write stuff anymore
-        count_before = history.count((code_a, 1))
+        count_before = history.count((EV_KEY, code_a, 1))
         self.assertGreater(count_before, 1)
         loop.run_until_complete(asyncio.sleep(0.1))
-        count_after = history.count((code_a, 1))
+        count_after = history.count((EV_KEY, code_a, 1))
         self.assertEqual(count_before, count_after)
 
         """restart macro 2"""
@@ -714,8 +714,8 @@ class TestKeycodeMapper(unittest.TestCase):
 
         keycode_mapper.handle_keycode(new_event(EV_KEY, 2, 1))
         loop.run_until_complete(asyncio.sleep(0.1))
-        self.assertEqual(history.count((code_c, 1)), 1)
-        self.assertEqual(history.count((code_c, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 0)), 1)
 
         # spam garbage events again, this time key-up events on all other
         # macros
@@ -734,11 +734,11 @@ class TestKeycodeMapper(unittest.TestCase):
         keycode_mapper.handle_keycode(new_event(EV_KEY, 2, 0))
         loop.run_until_complete(asyncio.sleep(0.1))
         # was started only once
-        self.assertEqual(history.count((code_c, 1)), 1)
-        self.assertEqual(history.count((code_c, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 0)), 1)
         # and the trailing d was also written only once
-        self.assertEqual(history.count((code_d, 1)), 1)
-        self.assertEqual(history.count((code_d, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_d, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_d, 0)), 1)
 
         # stop all macros
         keycode_mapper.handle_keycode(new_event(EV_KEY, 1, 0))
@@ -794,18 +794,18 @@ class TestKeycodeMapper(unittest.TestCase):
             loop.run_until_complete(asyncio.sleep(0.05))
 
         # duplicate key down events don't do anything
-        self.assertEqual(history.count((code_a, 1)), 1)
-        self.assertEqual(history.count((code_a, 0)), 1)
-        self.assertEqual(history.count((code_c, 1)), 0)
-        self.assertEqual(history.count((code_c, 0)), 0)
+        self.assertEqual(history.count((EV_KEY, code_a, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_a, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 1)), 0)
+        self.assertEqual(history.count((EV_KEY, code_c, 0)), 0)
 
         # stop
         keycode_mapper.handle_keycode(new_event(EV_KEY, 1, 0))
         loop.run_until_complete(asyncio.sleep(0.1))
-        self.assertEqual(history.count((code_a, 1)), 1)
-        self.assertEqual(history.count((code_a, 0)), 1)
-        self.assertEqual(history.count((code_c, 1)), 1)
-        self.assertEqual(history.count((code_c, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_a, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_a, 0)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 1)), 1)
+        self.assertEqual(history.count((EV_KEY, code_c, 0)), 1)
         self.assertFalse(active_macros[(EV_KEY, 1)].is_holding())
         self.assertFalse(active_macros[(EV_KEY, 1)].running)
 
@@ -925,22 +925,22 @@ class TestKeycodeMapper(unittest.TestCase):
         self.assertGreater(len(history), events * 0.9)
         self.assertLess(len(history), events * 1.1)
 
-        self.assertIn((code_a, 1), history)
-        self.assertIn((code_a, 0), history)
-        self.assertIn((code_b, 1), history)
-        self.assertIn((code_b, 0), history)
-        self.assertIn((code_c, 1), history)
-        self.assertIn((code_c, 0), history)
-        self.assertIn((code_1, 1), history)
-        self.assertIn((code_1, 0), history)
-        self.assertIn((code_2, 1), history)
-        self.assertIn((code_2, 0), history)
-        self.assertIn((code_3, 1), history)
-        self.assertIn((code_3, 0), history)
-        self.assertGreater(history.count((code_b, 1)), 1)
-        self.assertGreater(history.count((code_b, 0)), 1)
-        self.assertGreater(history.count((code_2, 1)), 1)
-        self.assertGreater(history.count((code_2, 0)), 1)
+        self.assertIn((EV_KEY, code_a, 1), history)
+        self.assertIn((EV_KEY, code_a, 0), history)
+        self.assertIn((EV_KEY, code_b, 1), history)
+        self.assertIn((EV_KEY, code_b, 0), history)
+        self.assertIn((EV_KEY, code_c, 1), history)
+        self.assertIn((EV_KEY, code_c, 0), history)
+        self.assertIn((EV_KEY, code_1, 1), history)
+        self.assertIn((EV_KEY, code_1, 0), history)
+        self.assertIn((EV_KEY, code_2, 1), history)
+        self.assertIn((EV_KEY, code_2, 0), history)
+        self.assertIn((EV_KEY, code_3, 1), history)
+        self.assertIn((EV_KEY, code_3, 0), history)
+        self.assertGreater(history.count((EV_KEY, code_b, 1)), 1)
+        self.assertGreater(history.count((EV_KEY, code_b, 0)), 1)
+        self.assertGreater(history.count((EV_KEY, code_2, 1)), 1)
+        self.assertGreater(history.count((EV_KEY, code_2, 0)), 1)
 
         # it's stopped and won't write stuff anymore
         count_before = len(history)
@@ -994,10 +994,10 @@ class TestKeycodeMapper(unittest.TestCase):
         sleeptime = config.get('macros.keystroke_sleep_ms') / 1000
         loop.run_until_complete(asyncio.sleep(1.1 * repeats * 2 * sleeptime))
 
-        self.assertEqual(history.count((code_1, 1)), 10)
-        self.assertEqual(history.count((code_1, 0)), 10)
-        self.assertEqual(history.count((code_2, 1)), 10)
-        self.assertEqual(history.count((code_2, 0)), 10)
+        self.assertEqual(history.count((EV_KEY, code_1, 1)), 10)
+        self.assertEqual(history.count((EV_KEY, code_1, 0)), 10)
+        self.assertEqual(history.count((EV_KEY, code_2, 1)), 10)
+        self.assertEqual(history.count((EV_KEY, code_2, 0)), 10)
         self.assertEqual(len(history), repeats * 4)
 
     def test_filter_trigger_spam(self):
@@ -1223,7 +1223,7 @@ class TestKeycodeMapper(unittest.TestCase):
         self.assertEqual(len(uinput_write_history), 0)
         self.assertGreater(len(macro_history), 1)
         self.assertIn(down_1[:2], unreleased)
-        self.assertIn((92, 1), macro_history)
+        self.assertIn((EV_KEY, 92, 1), macro_history)
 
         # combination triggered
         keycode_mapper.handle_keycode(new_event(*down_2))

tests/testcases/test_macros.py

@@ -23,6 +23,8 @@ import time
 import unittest
 import asyncio
 
+from evdev.ecodes import EV_REL, EV_KEY, REL_Y, REL_X, REL_WHEEL, REL_HWHEEL
+
 from keymapper.injection.macros import parse, _Macro, _extract_params, \
     is_this_a_macro, _parse_recurse, handle_plus_syntax
 from keymapper.config import config
@@ -43,9 +45,10 @@ class TestMacros(unittest.TestCase):
         self.mapping.clear_config()
         quick_cleanup()
 
-    def handler(self, code, value):
+    def handler(self, ev_type, code, value):
         """Where macros should write codes to."""
-        self.result.append((code, value))
+        print(f'\033[90mmacro wrote{(ev_type, code, value)}\033[0m')
+        self.result.append((ev_type, code, value))
 
     def test_is_this_a_macro(self):
         self.assertTrue(is_this_a_macro('k(1)'))
@@ -79,7 +82,7 @@ class TestMacros(unittest.TestCase):
 
     def test_run_plus_syntax(self):
         macro = parse('a + b + c + d', self.mapping)
-        self.assertSetEqual(macro.get_capabilities(), {
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {
             system_mapping.get('a'),
             system_mapping.get('b'),
             system_mapping.get('c'),
@@ -92,19 +95,19 @@ class TestMacros(unittest.TestCase):
         self.assertTrue(macro.is_holding())
 
         # starting from the left, presses each one down
-        self.assertEqual(self.result[0], (system_mapping.get('a'), 1))
-        self.assertEqual(self.result[1], (system_mapping.get('b'), 1))
-        self.assertEqual(self.result[2], (system_mapping.get('c'), 1))
-        self.assertEqual(self.result[3], (system_mapping.get('d'), 1))
+        self.assertEqual(self.result[0], (EV_KEY, system_mapping.get('a'), 1))
+        self.assertEqual(self.result[1], (EV_KEY, system_mapping.get('b'), 1))
+        self.assertEqual(self.result[2], (EV_KEY, system_mapping.get('c'), 1))
+        self.assertEqual(self.result[3], (EV_KEY, system_mapping.get('d'), 1))
 
         # and then releases starting with the previously pressed key
         macro.release_key()
         self.loop.run_until_complete(asyncio.sleep(0.2))
         self.assertFalse(macro.is_holding())
-        self.assertEqual(self.result[4], (system_mapping.get('d'), 0))
-        self.assertEqual(self.result[5], (system_mapping.get('c'), 0))
-        self.assertEqual(self.result[6], (system_mapping.get('b'), 0))
-        self.assertEqual(self.result[7], (system_mapping.get('a'), 0))
+        self.assertEqual(self.result[4], (EV_KEY, system_mapping.get('d'), 0))
+        self.assertEqual(self.result[5], (EV_KEY, system_mapping.get('c'), 0))
+        self.assertEqual(self.result[6], (EV_KEY, system_mapping.get('b'), 0))
+        self.assertEqual(self.result[7], (EV_KEY, system_mapping.get('a'), 0))
 
     def test_extract_params(self):
         def expect(raw, expectation):
@@ -142,15 +145,16 @@ class TestMacros(unittest.TestCase):
     def test_0(self):
         macro = parse('k(1)', self.mapping)
         one_code = system_mapping.get('1')
-        self.assertSetEqual(macro.get_capabilities(), {one_code})
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {one_code})
+        self.assertSetEqual(macro.get_capabilities()[EV_REL], set())
 
         self.loop.run_until_complete(macro.run(self.handler))
-        self.assertListEqual(self.result, [(one_code, 1), (one_code, 0)])
+        self.assertListEqual(self.result, [(EV_KEY, one_code, 1), (EV_KEY, one_code, 0)])
         self.assertEqual(len(macro.child_macros), 0)
 
     def test_1(self):
         macro = parse('k(1).k(a).k(3)', self.mapping)
-        self.assertSetEqual(macro.get_capabilities(), {
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {
             system_mapping.get('1'),
             system_mapping.get('a'),
             system_mapping.get('3')
@@ -158,9 +162,9 @@ class TestMacros(unittest.TestCase):
 
         self.loop.run_until_complete(macro.run(self.handler))
         self.assertListEqual(self.result, [
-            (system_mapping.get('1'), 1), (system_mapping.get('1'), 0),
-            (system_mapping.get('a'), 1), (system_mapping.get('a'), 0),
-            (system_mapping.get('3'), 1), (system_mapping.get('3'), 0),
+            (EV_KEY, system_mapping.get('1'), 1), (EV_KEY, system_mapping.get('1'), 0),
+            (EV_KEY, system_mapping.get('a'), 1), (EV_KEY, system_mapping.get('a'), 0),
+            (EV_KEY, system_mapping.get('3'), 1), (EV_KEY, system_mapping.get('3'), 0),
         ])
         self.assertEqual(len(macro.child_macros), 0)
 
@@ -196,7 +200,7 @@ class TestMacros(unittest.TestCase):
 
     def test_hold(self):
         macro = parse('k(1).h(k(a)).k(3)', self.mapping)
-        self.assertSetEqual(macro.get_capabilities(), {
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {
             system_mapping.get('1'),
             system_mapping.get('a'),
             system_mapping.get('3')
@@ -212,17 +216,17 @@ class TestMacros(unittest.TestCase):
         self.loop.run_until_complete(asyncio.sleep(0.05))
         self.assertFalse(macro.is_holding())
 
-        self.assertEqual(self.result[0], (system_mapping.get('1'), 1))
-        self.assertEqual(self.result[-1], (system_mapping.get('3'), 0))
+        self.assertEqual(self.result[0], (EV_KEY, system_mapping.get('1'), 1))
+        self.assertEqual(self.result[-1], (EV_KEY, system_mapping.get('3'), 0))
 
         code_a = system_mapping.get('a')
-        self.assertGreater(self.result.count((code_a, 1)), 2)
+        self.assertGreater(self.result.count((EV_KEY, code_a, 1)), 2)
 
         self.assertEqual(len(macro.child_macros), 1)
 
     def test_dont_hold(self):
         macro = parse('k(1).h(k(a)).k(3)', self.mapping)
-        self.assertSetEqual(macro.get_capabilities(), {
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {
             system_mapping.get('1'),
             system_mapping.get('a'),
             system_mapping.get('3')
@@ -235,14 +239,14 @@ class TestMacros(unittest.TestCase):
         # and the child macro of hold is never called.
         self.assertEqual(len(self.result), 4)
 
-        self.assertEqual(self.result[0], (system_mapping.get('1'), 1))
-        self.assertEqual(self.result[-1], (system_mapping.get('3'), 0))
+        self.assertEqual(self.result[0], (EV_KEY, system_mapping.get('1'), 1))
+        self.assertEqual(self.result[-1], (EV_KEY, system_mapping.get('3'), 0))
 
         self.assertEqual(len(macro.child_macros), 1)
 
     def test_just_hold(self):
         macro = parse('k(1).h().k(3)', self.mapping)
-        self.assertSetEqual(macro.get_capabilities(), {
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {
             system_mapping.get('1'),
             system_mapping.get('3')
         })
@@ -261,14 +265,14 @@ class TestMacros(unittest.TestCase):
         self.assertFalse(macro.is_holding())
         self.assertEqual(len(self.result), 4)
 
-        self.assertEqual(self.result[0], (system_mapping.get('1'), 1))
-        self.assertEqual(self.result[-1], (system_mapping.get('3'), 0))
+        self.assertEqual(self.result[0], (EV_KEY, system_mapping.get('1'), 1))
+        self.assertEqual(self.result[-1], (EV_KEY, system_mapping.get('3'), 0))
 
         self.assertEqual(len(macro.child_macros), 0)
 
     def test_dont_just_hold(self):
         macro = parse('k(1).h().k(3)', self.mapping)
-        self.assertSetEqual(macro.get_capabilities(), {
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {
             system_mapping.get('1'),
             system_mapping.get('3')
         })
@@ -280,8 +284,8 @@ class TestMacros(unittest.TestCase):
         # completely
         self.assertEqual(len(self.result), 4)
 
-        self.assertEqual(self.result[0], (system_mapping.get('1'), 1))
-        self.assertEqual(self.result[-1], (system_mapping.get('3'), 0))
+        self.assertEqual(self.result[0], (EV_KEY, system_mapping.get('1'), 1))
+        self.assertEqual(self.result[-1], (EV_KEY, system_mapping.get('3'), 0))
 
         self.assertEqual(len(macro.child_macros), 0)
 
@@ -291,7 +295,7 @@ class TestMacros(unittest.TestCase):
 
         macro = parse(f'r({repeats}, k(k)).r(1, k(k))', self.mapping)
         k_code = system_mapping.get('k')
-        self.assertSetEqual(macro.get_capabilities(), {k_code})
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {k_code})
 
         self.loop.run_until_complete(macro.run(self.handler))
         keystroke_sleep = self.mapping.get('macros.keystroke_sleep_ms')
@@ -301,7 +305,7 @@ class TestMacros(unittest.TestCase):
 
         self.assertListEqual(
             self.result,
-            [(k_code, 1), (k_code, 0)] * (repeats + 1)
+            [(EV_KEY, k_code, 1), (EV_KEY, k_code, 0)] * (repeats + 1)
         )
 
         self.assertEqual(len(macro.child_macros), 2)
@@ -311,7 +315,7 @@ class TestMacros(unittest.TestCase):
         start = time.time()
         macro = parse('r(3, k(m).w(100))', self.mapping)
         m_code = system_mapping.get('m')
-        self.assertSetEqual(macro.get_capabilities(), {m_code})
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {m_code})
         self.loop.run_until_complete(macro.run(self.handler))
 
         keystroke_time = 6 * self.mapping.get('macros.keystroke_sleep_ms')
@@ -321,9 +325,9 @@ class TestMacros(unittest.TestCase):
         self.assertGreater(time.time() - start, total_time * 0.9)
         self.assertLess(time.time() - start, total_time * 1.1)
         self.assertListEqual(self.result, [
-            (m_code, 1), (m_code, 0),
-            (m_code, 1), (m_code, 0),
-            (m_code, 1), (m_code, 0),
+            (EV_KEY, m_code, 1), (EV_KEY, m_code, 0),
+            (EV_KEY, m_code, 1), (EV_KEY, m_code, 0),
+            (EV_KEY, m_code, 1), (EV_KEY, m_code, 0),
         ])
         self.assertEqual(len(macro.child_macros), 1)
         self.assertEqual(len(macro.child_macros[0].child_macros), 0)
@@ -335,15 +339,15 @@ class TestMacros(unittest.TestCase):
         minus = system_mapping.get('minus')
         m = system_mapping.get('m')
 
-        self.assertSetEqual(macro.get_capabilities(), {r, minus, m})
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {r, minus, m})
 
         self.loop.run_until_complete(macro.run(self.handler))
         self.assertListEqual(self.result, [
-            (r, 1), (r, 0),
-            (minus, 1), (minus, 0),
-            (r, 1), (r, 0),
-            (minus, 1), (minus, 0),
-            (m, 1), (m, 0),
+            (EV_KEY, r, 1), (EV_KEY, r, 0),
+            (EV_KEY, minus, 1), (EV_KEY, minus, 0),
+            (EV_KEY, r, 1), (EV_KEY, r, 0),
+            (EV_KEY, minus, 1), (EV_KEY, minus, 0),
+            (EV_KEY, m, 1), (EV_KEY, m, 0),
         ])
         self.assertEqual(len(macro.child_macros), 1)
         self.assertEqual(len(macro.child_macros[0].child_macros), 0)
@@ -359,7 +363,7 @@ class TestMacros(unittest.TestCase):
         left = system_mapping.get('bTn_lEfT')
         k = system_mapping.get('k')
 
-        self.assertSetEqual(macro.get_capabilities(), {w, left, k})
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {w, left, k})
 
         self.loop.run_until_complete(macro.run(self.handler))
 
@@ -370,10 +374,10 @@ class TestMacros(unittest.TestCase):
 
         self.assertLess(time.time() - start, total_time * 1.1)
         self.assertGreater(time.time() - start, total_time * 0.9)
-        expected = [(w, 1)]
-        expected += [(left, 1), (left, 0)] * 2
-        expected += [(w, 0)]
-        expected += [(k, 1), (k, 0)]
+        expected = [(EV_KEY, w, 1)]
+        expected += [(EV_KEY, left, 1), (EV_KEY, left, 0)] * 2
+        expected += [(EV_KEY, w, 0)]
+        expected += [(EV_KEY, k, 1), (EV_KEY, k, 0)]
         expected *= 2
         self.assertListEqual(self.result, expected)
 
@@ -431,9 +435,9 @@ class TestMacros(unittest.TestCase):
         self.assertFalse(macro.is_holding())
 
         expected = [
-            (a, 1), (a, 0),
-            (b, 1), (b, 0),
-            (c, 1), (c, 0),
+            (EV_KEY, a, 1), (EV_KEY, a, 0),
+            (EV_KEY, b, 1), (EV_KEY, b, 0),
+            (EV_KEY, c, 1), (EV_KEY, c, 0),
         ]
         self.assertListEqual(self.result, expected)
 
@@ -448,12 +452,57 @@ class TestMacros(unittest.TestCase):
         self.assertFalse(macro.is_holding())
 
         expected = [
-            (a, 1), (a, 0),
-            (b, 1), (b, 0),
-            (c, 1), (c, 0),
+            (EV_KEY, a, 1), (EV_KEY, a, 0),
+            (EV_KEY, b, 1), (EV_KEY, b, 0),
+            (EV_KEY, c, 1), (EV_KEY, c, 0),
         ] * 2
         self.assertListEqual(self.result, expected)
 
+    def test_mouse(self):
+        macro_1 = parse('mouse(up, 4)', self.mapping)
+        macro_2 = parse('wheel(left, 3)', self.mapping)
+        macro_1.press_key()
+        macro_2.press_key()
+        asyncio.ensure_future(macro_1.run(self.handler))
+        asyncio.ensure_future(macro_2.run(self.handler))
+        self.loop.run_until_complete(asyncio.sleep(0.1))
+        self.assertTrue(macro_1.is_holding())
+        self.assertTrue(macro_2.is_holding())
+        macro_1.release_key()
+        macro_2.release_key()
+
+        self.assertIn((EV_REL, REL_Y, -4), self.result)
+        self.assertIn((EV_REL, REL_HWHEEL, 1), self.result)
+
+        self.assertIn(REL_WHEEL, macro_1.get_capabilities()[EV_REL])
+        self.assertIn(REL_Y, macro_1.get_capabilities()[EV_REL])
+        self.assertIn(REL_X, macro_1.get_capabilities()[EV_REL])
+
+        self.assertIn(REL_WHEEL, macro_2.get_capabilities()[EV_REL])
+        self.assertIn(REL_Y, macro_2.get_capabilities()[EV_REL])
+        self.assertIn(REL_X, macro_2.get_capabilities()[EV_REL])
+
+    def test_event_1(self):
+        macro = parse('e(EV_KEY, KEY_A, 1)', self.mapping)
+        a_code = system_mapping.get('a')
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], {a_code})
+        self.assertSetEqual(macro.get_capabilities()[EV_REL], set())
+
+        self.loop.run_until_complete(macro.run(self.handler))
+        self.assertListEqual(self.result, [(EV_KEY, a_code, 1)])
+        self.assertEqual(len(macro.child_macros), 0)
+
+    def test_event_2(self):
+        macro = parse('e(5421, 324, 154)', self.mapping)
+        code = 324
+        self.assertSetEqual(macro.get_capabilities()[5421], {324})
+        self.assertSetEqual(macro.get_capabilities()[EV_REL], set())
+        self.assertSetEqual(macro.get_capabilities()[EV_KEY], set())
+
+        self.loop.run_until_complete(macro.run(self.handler))
+        self.assertListEqual(self.result, [(5421, code, 154)])
+        self.assertEqual(len(macro.child_macros), 0)
+
 
 if __name__ == '__main__':
     unittest.main()