Add Wayland support by providing evdev and libinput backends

Makefile

@@ -1,7 +1,17 @@
 .PHONY: build test package clean
 
+POETRY_EXTRAS =
+
+ifeq ($(USE_EVDEV),true)
+	POETRY_EXTRAS := $(POETRY_EXTRAS) use_evdev
+endif
+
+# ifeq ($(USE_LIBINPUT),true)
+# 	POETRY_EXTRAS := $(POETRY_EXTRAS) use_libinput
+# endif
+
 build:
-	poetry install
+	poetry install --extras "$(POETRY_EXTRAS)"
 
 test:
 	poetry run aw-watcher-afk --help  # Ensures that it at least starts

aw_watcher_afk/evdev_devices.py

@@ -0,0 +1,51 @@
+from evdev import InputDevice, ecodes, list_devices
+
+def is_keyboard(capabilities):
+    if ecodes.EV_KEY not in capabilities:
+        return False
+
+    key_codes = set(capabilities[ecodes.EV_KEY])
+
+    required_keys = {ecodes.KEY_A, ecodes.KEY_Z, ecodes.KEY_ENTER}
+    if not required_keys.issubset(key_codes):
+        return False
+
+    if ecodes.EV_REL in capabilities or ecodes.EV_ABS in capabilities:
+        return False
+
+    return True
+
+def is_mouse(capabilities):
+    if ecodes.EV_REL not in capabilities and ecodes.EV_ABS not in capabilities:
+        return False
+
+    if ecodes.EV_KEY in capabilities:
+        key_events = capabilities[ecodes.EV_KEY]
+        if ecodes.BTN_MOUSE in key_events or ecodes.BTN_LEFT in key_events:
+            return True
+
+    return False
+
+def find_keyboards():
+
+    for path in list_devices():
+        try:
+            dev = InputDevice(path)
+            capabilities = dev.capabilities()
+
+            if is_keyboard(capabilities):
+                yield dev.fn
+        except OSError:
+            continue
+
+def find_mice():
+
+    for path in list_devices():
+        try:
+            dev = InputDevice(path)
+            capabilities = dev.capabilities()
+
+            if is_mouse(capabilities):
+                yield dev.fn
+        except OSError:
+            continue

aw_watcher_afk/listeners.py

@@ -6,104 +6,80 @@
 NOTE: Logging usage should be commented out before committed, for performance reasons.
 """
 
+
+"""
+Listeners for aggregated keyboard and mouse events.
+
+This is used for AFK detection on Linux, as well as used in aw-watcher-input to track input activity in general.
+
+NOTE: Logging usage should be commented out before committed, for performance reasons.
+"""
+
 import logging
-import threading
-from abc import ABCMeta, abstractmethod
-from collections import defaultdict
-from typing import Dict, Any
+import os
+import platform
+
+from .listeners_base import main_test
 
 logger = logging.getLogger(__name__)
 # logger.setLevel(logging.DEBUG)
 
+system = platform.system()
+
+def use_evdev():
+    """Use evdev backend"""
+    return system == "Linux" and os.getenv("USE_EVDEV") == "true"
+
+def use_libinput():
+    """Use libinput backend"""
+    return system == "Linux" and os.getenv("USE_LIBINPUT") == "true"
+
+def KeyboardListener():
+
+    """Returns keyboard listener using backends: evdev, libinput and pynput"""
+
+    if use_evdev():
+        # noreorder
+        from .listeners_evdev import KeyboardListener
+    # elif use_libinput():
+    #     # noreorder
+    #     from .listeners_libinput import KeyboardListener
+    else:
+        from .listeners_pynput import KeyboardListener
+
+    return KeyboardListener()
+
+def MouseListener():
+
+    """Returns mouse listener using backends: evdev, libinput, pynput"""
+
+    if use_evdev():
+        # noreorder
+        from .listeners_evdev import MouseListener  # fmt: skip
+    # elif use_libinput():
+    #     # noreorder
+    #     from .listeners_libinput import MouseListener
+    else:
+        # noreorder
+        from .listeners_pynput import MouseListener  # fmt: skip
+
+    return MouseListener()
+
+def MergedListener():
+
+    """Returns mouse and keyboard listener using backends: evdev, libinput, pynput"""
+
+    if use_evdev():
+        # noreorder
+        from .listeners_evdev import MergedListener
+    # elif use_libinput():
+    #     # noreorder
+    #     from .listeners_libinput import MergedListener
+    else:
+        # noreorder
+        from .listeners_pynput import MergedListener  # fmt: skip
+
+    return MergedListener()
 
-class EventFactory(metaclass=ABCMeta):
-    def __init__(self) -> None:
-        self.new_event = threading.Event()
-        self._reset_data()
-
-    @abstractmethod
-    def _reset_data(self) -> None:
-        self.event_data: Dict[str, Any] = {}
-
-    def next_event(self) -> dict:
-        """Returns an event and prepares the internal state so that it can start to build a new event"""
-        self.new_event.clear()
-        data = self.event_data
-        # self.logger.debug(f"Event: {data}")
-        self._reset_data()
-        return data
-
-    def has_new_event(self) -> bool:
-        return self.new_event.is_set()
-
-
-class KeyboardListener(EventFactory):
-    def __init__(self):
-        EventFactory.__init__(self)
-        self.logger = logger.getChild("keyboard")
-
-    def start(self):
-        from pynput import keyboard
-
-        listener = keyboard.Listener(on_press=self.on_press, on_release=self.on_release)
-        listener.start()
-
-    def _reset_data(self):
-        self.event_data = {"presses": 0}
-
-    def on_press(self, key):
-        # self.logger.debug(f"Press: {key}")
-        self.event_data["presses"] += 1
-        self.new_event.set()
-
-    def on_release(self, key):
-        # Don't count releases, only clicks
-        # self.logger.debug(f"Release: {key}")
-        pass
-
-
-class MouseListener(EventFactory):
-    def __init__(self):
-        EventFactory.__init__(self)
-        self.logger = logger.getChild("mouse")
-        self.pos = None
-
-    def _reset_data(self):
-        self.event_data = defaultdict(int)
-        self.event_data.update(
-            {"clicks": 0, "deltaX": 0, "deltaY": 0, "scrollX": 0, "scrollY": 0}
-        )
-
-    def start(self):
-        from pynput import mouse
-
-        listener = mouse.Listener(
-            on_move=self.on_move, on_click=self.on_click, on_scroll=self.on_scroll
-        )
-        listener.start()
-
-    def on_move(self, x, y):
-        newpos = (x, y)
-        # self.logger.debug("Moved mouse to: {},{}".format(x, y))
-        if not self.pos:
-            self.pos = newpos
-
-        delta = tuple(self.pos[i] - newpos[i] for i in range(2))
-        self.event_data["deltaX"] += abs(delta[0])
-        self.event_data["deltaY"] += abs(delta[1])
-
-        self.pos = newpos
-        self.new_event.set()
-
-    def on_click(self, x, y, button, down):
-        # self.logger.debug(f"Click: {button} at {(x, y)}")
-        # Only count presses, not releases
-        if down:
-            self.event_data["clicks"] += 1
-            self.new_event.set()
-
-    def on_scroll(self, x, y, scrollx, scrolly):
-        # self.logger.debug(f"Scroll: {scrollx}, {scrolly} at {(x, y)}")
-        self.event_data["scrollX"] += abs(scrollx)
-        self.event_data["scrollY"] += abs(scrolly)
-        self.new_event.set()
+if __name__ == "__main__":
+    main_test(MergedListener())

aw_watcher_afk/listeners_base.py

@@ -0,0 +1,48 @@
+"""
+Listeners for aggregated keyboard and mouse events.
+
+This is used for AFK detection on Linux, as well as used in aw-watcher-input to track input activity in general.
+
+NOTE: Logging usage should be commented out before committed, for performance reasons.
+"""
+
+
+"""
+Listeners for aggregated keyboard and mouse events.
+
+This is used for AFK detection on Linux, as well as used in aw-watcher-input to track input activity in general.
+
+NOTE: Logging usage should be commented out before committed, for performance reasons.
+"""
+
+import logging
+from abc import ABCMeta, abstractmethod
+
+logger = logging.getLogger(__name__)
+# logger.setLevel(logging.DEBUG)
+
+class BaseEventFactory(metaclass=ABCMeta):
+
+    @abstractmethod
+    def next_event(self):
+        """Returns new event data"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def start(self):
+        """Starts monitoring events in the background"""
+        raise NotImplementedError
+
+    @abstractmethod
+    def has_new_event(self) -> bool:
+        """Has new event data"""
+        raise NotImplementedError
+
+def main_test(listener):
+
+    listener.start()
+
+    while True:
+
+        if listener.has_new_event():
+            print(listener.next_event())