simple mouse listener for x window system

x11-mouse-listener.py

import math
from enum import Enum
from threading import Thread, Lock
from time import sleep
from Xlib.display import Display
from Xlib import X
from Xlib.ext.xtest import fake_input
from Xlib.ext import record
from Xlib.protocol import rq


class MouseEvent(int, Enum):
    MOVE = (1 << 0)
    CLICK_LEFT = (1 << 1)
    MIDDLE_CLICK = (1 << 2)
    CLICK_RIGHT = (1 << 3)
    SCROLL_UP = (1 << 4)
    SCROLL_DOWN = (1 << 5)
    ALL = MOVE | CLICK_LEFT | MIDDLE_CLICK | CLICK_RIGHT | SCROLL_UP | SCROLL_DOWN
    UNKNOWN = (1 << 32)


class ButtonEvent(int, Enum):
    PRESS = 4
    RELEASE = 5
    MOTION = 6
    UNKNOWN = 32


class MouseListener(Display):

    def __init__(self, suppress=False):
        super(MouseListener, self).__init__()
        self.__record = Display()
        self.__rec_ctx = None
        self.__worker = None
        self.__mtx = Lock()
        self.__listeners = {}
        self.__suppress = suppress
        self.__is_left_dragging = False
        self.__is_right_dragging = False
        self.__listening = False
        self.__entered = False

    def __enter__(self):
        self.__entered = True
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.record_disable_context(self.__rec_ctx)
        self.ungrab_pointer(X.CurrentTime)
        self.flush()
        self.__record.record_free_context(self.__rec_ctx)
        self.__record.record_disable_context(self.__rec_ctx)
        self.__record.ungrab_pointer(X.CurrentTime)
        self.__record.flush()
        self.__worker.join()

    def add_listener(self, flags: MouseEvent, handler):
        if (self.__listening):
            raise RuntimeError("cannot add listener while running")
        if (not self.__entered):
            raise RuntimeError(
                "cannot use this function without 'with' statement")
        self.__listeners[handler] = flags

    def listen(self):
        if (self.__listening):
            raise RuntimeError("already listening")
        if (not self.__entered):
            raise RuntimeError(
                "cannot use this function without 'with' statement")
        self.__listening = True
        self.__rec_ctx = self.__record.record_create_context(
            0,
            [record.AllClients],
            [{
                'core_requests': (0, 0),
                'core_replies': (0, 0),
                'ext_requests': (0, 0, 0, 0),
                'ext_replies': (0, 0, 0, 0),
                'delivered_events': (0, 0),
                'device_events': (X.ButtonPressMask, X.ButtonReleaseMask),
                'errors': (0, 0),
                'client_started': False,
                'client_died': False,
            }])
        if (self.__suppress):
            self.screen().root.grab_pointer(
                True, 0, X.GrabModeAsync, X.GrabModeAsync,
                0, 0, X.CurrentTime)
        self.__worker = Thread(target=self.__start_record, args=[])
        self.__worker.start()
        self.__entered = True

    def wait(self):
        if (not self.__listening):
            raise RuntimeError("not listening")
        self.__worker.join()

    def __start_record(self):
        self.__record.record_enable_context(self.__rec_ctx, self.__handler)

    def __handler(self, reply):
        data = reply.data
        while len(data):
            event, data = rq.EventField(None).parse_binary_value(
                data, self.display, None, None)

            # get mouse event
            mouse_event = MouseEvent.UNKNOWN
            try:
                mouse_event = MouseEvent(pow(2, event.detail))
            except:
                pass

            # get button event
            button_event = ButtonEvent.UNKNOWN
            try:
                button_event = ButtonEvent(event.type)
            except:
                pass

            # get left dragging
            if (mouse_event == MouseEvent.CLICK_LEFT):
                if event.type == X.ButtonPress:
                    self.__is_left_dragging = True
                elif event.type == X.ButtonRelease:
                    self.__is_left_dragging = False

            # get right dragging
            if (mouse_event == MouseEvent.CLICK_RIGHT):
                if event.type == X.ButtonPress:
                    self.__is_right_dragging = True
                elif event.type == X.ButtonRelease:
                    self.__is_right_dragging = False

            # handle listeners
            for handler, flags in self.__listeners.items():
                if (not (mouse_event & flags)):
                    continue
                try:
                    handler(
                        mouse_event,
                        button_event,
                        self.__is_left_dragging,
                        self.__is_right_dragging,
                        event.root_x,
                        event.root_y,
                        event.state)
                except TypeError as e:
                    print(e)


#######################################################


def on_mouse(
        event: MouseEvent,
        action: ButtonEvent,
        is_left_dragging: bool,
        is_right_dragging: bool,
        x: int,
        y: int,
        state: int):
    print(
        f"{event.name} {action.name} {is_left_dragging} {is_right_dragging} {x} {y} {state}")


def main():
    with MouseListener(suppress=False) as input:
        input.add_listener(MouseEvent.ALL, on_mouse)
        input.listen()

        #input.wait()
        # OR
        while True:
            sleep(1.0)
            # do work here
    return 0


if __name__ == "__main__":
    main()