diff --git a/lib/hidapi/hidapi_impl.py b/lib/hidapi/hidapi_impl.py index 3ba31e055f..6d70ba5610 100644 --- a/lib/hidapi/hidapi_impl.py +++ b/lib/hidapi/hidapi_impl.py @@ -23,10 +23,13 @@ which is MIT licensed. """ +from __future__ import annotations + import atexit import ctypes import logging import platform +import typing from threading import Thread from time import sleep @@ -35,8 +38,9 @@ from hidapi.common import DeviceInfo -gi.require_version("Gdk", "3.0") -from gi.repository import GLib # NOQA: E402 +if typing.TYPE_CHECKING: + gi.require_version("Gdk", "3.0") + from gi.repository import GLib # NOQA: E402 logger = logging.getLogger(__name__) @@ -311,13 +315,21 @@ def find_paired_node_wpid(receiver_path, index): return None -def monitor_glib(callback, filterfn): +def monitor_glib(glib: GLib, callback, filterfn): + """Monitor GLib. + + Parameters + ---------- + glib + GLib instance. + """ + def device_callback(action, device): # print(f"device_callback({action}): {device}") if action == "add": d_info = _match(action, device, filterfn) if d_info: - GLib.idle_add(callback, action, d_info) + glib.idle_add(callback, action, d_info) elif action == "remove": # Removed devices will be detected by Solaar directly pass diff --git a/lib/hidapi/udev_impl.py b/lib/hidapi/udev_impl.py index 082f565bd0..4d1fe2dbc3 100644 --- a/lib/hidapi/udev_impl.py +++ b/lib/hidapi/udev_impl.py @@ -23,9 +23,12 @@ necessary. """ +from __future__ import annotations + import errno import logging import os +import typing import warnings @@ -39,8 +42,9 @@ from hidapi.common import DeviceInfo -gi.require_version("Gdk", "3.0") -from gi.repository import GLib # NOQA: E402 +if typing.TYPE_CHECKING: + gi.require_version("Gdk", "3.0") + from gi.repository import GLib # NOQA: E402 logger = logging.getLogger(__name__) @@ -213,13 +217,20 @@ def find_paired_node_wpid(receiver_path, index): return None -def monitor_glib(callback, filterfn): +def monitor_glib(glib: GLib, callback, filterfn): + """Monitor GLib. + + Parameters + ---------- + glib + GLib instance. + """ c = pyudev.Context() m = pyudev.Monitor.from_netlink(c) m.filter_by(subsystem="hidraw") def _process_udev_event(monitor, condition, cb, filterfn): - if condition == GLib.IO_IN: + if condition == glib.IO_IN: event = monitor.receive_device() if event: action, device = event @@ -227,7 +238,7 @@ def _process_udev_event(monitor, condition, cb, filterfn): if action == "add": d_info = _match(action, device, filterfn) if d_info: - GLib.idle_add(cb, action, d_info) + glib.idle_add(cb, action, d_info) elif action == "remove": # the GLib notification does _not_ match! pass @@ -235,13 +246,13 @@ def _process_udev_event(monitor, condition, cb, filterfn): try: # io_add_watch_full may not be available... - GLib.io_add_watch_full(m, GLib.PRIORITY_LOW, GLib.IO_IN, _process_udev_event, callback, filterfn) + glib.io_add_watch_full(m, glib.PRIORITY_LOW, glib.IO_IN, _process_udev_event, callback, filterfn) except AttributeError: try: # and the priority parameter appeared later in the API - GLib.io_add_watch(m, GLib.PRIORITY_LOW, GLib.IO_IN, _process_udev_event, callback, filterfn) + glib.io_add_watch(m, glib.PRIORITY_LOW, glib.IO_IN, _process_udev_event, callback, filterfn) except Exception: - GLib.io_add_watch(m, GLib.IO_IN, _process_udev_event, callback, filterfn) + glib.io_add_watch(m, glib.IO_IN, _process_udev_event, callback, filterfn) if logger.isEnabledFor(logging.DEBUG): logger.debug("Starting dbus monitoring") diff --git a/lib/logitech_receiver/base.py b/lib/logitech_receiver/base.py index 577fc6c7f4..ae97b10a77 100644 --- a/lib/logitech_receiver/base.py +++ b/lib/logitech_receiver/base.py @@ -29,6 +29,7 @@ from time import time from typing import Any +import gi import hidapi from . import base_usb @@ -41,6 +42,9 @@ from .common import LOGITECH_VENDOR_ID from .common import BusID +gi.require_version("Gdk", "3.0") +from gi.repository import GLib # NOQA: E402 + logger = logging.getLogger(__name__) _hidpp20 = hidpp20.Hidpp20() @@ -176,7 +180,7 @@ def receivers_and_devices(): def notify_on_receivers_glib(callback): """Watch for matching devices and notifies the callback on the GLib thread.""" - return hidapi.monitor_glib(callback, filter) + return hidapi.monitor_glib(GLib, callback, filter) def open_path(path):