lazy imports cause a KeyError

[catbox305]

Description This was hard to reproduce, so I'm not fully sure what causes the bug.

I'm making an application similar to TinyTask. I have a global hotkey listener, and previously two other listeners for the recording - now they loop over events and aren't listeners, which was an attempt to fix this bug.

I'm constantly getting KeyError: 'CGEventGetIntegerValueField' in the hotkey listener callback. Sometimes I used to get it in the other listeners instead, seemingly at random.

I also get KeyError: 'CFMachPortCreateRunLoopSource' from starting two listeners too fast - using listener.wait() seems to fix it, but I haven't verified that.

Please, fix this. I've spent over two weeks trying to debug this with no success. Either the hotkeys work and the other "listeners" don't. I cannot find a way to prevent either the hotkeys or the other "listeners" from stopping themselves.

Platform and pynput version MacOS Sonama 14.1, Pynput 1.7.6

To Reproduce Honestly, I tried reproducing this but couldn't really. The closest I got was:

from pynput import keyboard

def on_press(key):
    print(key,"pressed")

def on_release(key):
    print(key,"released")

def triggered_hotkey():
    print("Hotkey triggered")

listener1 = keyboard.GlobalHotKeys(hotkeys={"<alt>+å":triggered_hotkey})
listener2 = keyboard.Listener(on_press=on_press,on_release=on_release)

listener1.start()
listener2.start()

input() # Keep the program alive

[moses-palmer]

Thank you for your report.

What version of pyobjc do you have installed? That library appears to be prone to race conditions, but in my communication with them, they indicated that those issues would be resolved in version 8.0+.

[rafaelBauer]

I am getting constantly the same error KeyError: 'CGEventGetIntegerValueField' I can reproduce it very consistently. My setup is macOS Monterey 12.7.4. The pyobjc version is 10.2.

The error stack is:

Unhandled exception in listener callback
Traceback (most recent call last):
  File "/usr/local/anaconda3/envs/myenv/lib/python3.12/site-packages/pynput/_util/__init__.py", line 228, in inner
    return f(self, *args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/anaconda3/envs/myenv/lib/python3.12/site-packages/pynput/_util/darwin.py", line 265, in _handler
    self._handle(proxy, event_type, event, refcon)
  File "/usr/local/anaconda3/envs/myenv/lib/python3.12/site-packages/pynput/keyboard/_darwin.py", line 262, in _handle
    key = self._event_to_key(event)
          ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/anaconda3/envs/myenv/lib/python3.12/site-packages/pynput/keyboard/_darwin.py", line 320, in _event_to_key
    vk = Quartz.CGEventGetIntegerValueField(
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/anaconda3/envs/myenv/lib/python3.12/site-packages/objc/_lazyimport.py", line 178, in __getattr__
    value = getattr(p, name)
            ^^^^^^^^^^^^^^^^
  File "/usr/local/anaconda3/envs/myenv/lib/python3.12/site-packages/objc/_lazyimport.py", line 192, in __getattr__
    value = get_constant(name)
            ^^^^^^^^^^^^^^^^^^
  File "/usr/local/anaconda3/envs/myenv/lib/python3.12/site-packages/objc/_lazyimport.py", line 360, in get_constant
    funcmap.pop(name)
KeyError: 'CGEventGetIntegerValueField'

To reproduce, one can run the following and just press any key:

from __future__ import annotations
from enum import Enum
from collections.abc import Callable
from functools import partial
import time
import threading
from typing import Final

import numpy as np
from pynput import keyboard

class HumanFeedback(Enum):
    """
    Enum class for human feedback.

    Attributes:
        CORRECTED: A human had to correct the action. Value is -1.
        BAD: A human considered a bad action. Value is 0.
        GOOD: A human considered a good action Value is 1.
    """

    CORRECTED = -1
    BAD = 0
    GOOD = 1

class KeyboardObserver:
    """
    This class is used to observe keyboard inputs and map them to certain actions.
    """

    # Key mapping consists of (i, j), where:
    #   i = The index in the direction array
    #   j = The value of that index in the direction array
    __KEY_MAPPING: Final[dict[str, tuple[int, int]]] = {
        "s": (0, 1),  # backward
        "w": (0, -1),  # forward
        "a": (1, 1),  # left
        "d": (1, -1),  # right
        "q": (2, 1),  # down
        "e": (2, -1),  # up
        "j": (3, -1),  # look left
        "l": (3, 1),  # look right
        "i": (4, -1),  # look up
        "k": (4, 1),  # look down
        "u": (5, -1),  # rotate left
        "o": (5, 1),  # rotate right
    }

    def __init__(self):
        self.__label: HumanFeedback = HumanFeedback.GOOD
        self.__gripper: float | None = None
        self.__direction: np.array = np.zeros(6)

        self.__label_lock: threading.Lock = threading.Lock()
        self.__gripper_lock: threading.Lock = threading.Lock()
        self.__direction_lock: threading.Lock = threading.Lock()
        self.__hotkeys = keyboard.GlobalHotKeys(
            {
                "g": partial(self.__set_label, HumanFeedback.GOOD),
                "b": partial(self.__set_label, HumanFeedback.BAD),  # bad
                "c": partial(self.__set_gripper, -0.9),  # close
                "v": partial(self.__set_gripper, 0.9),  # open
                "f": partial(self.__set_gripper, None),  # gripper free
                "x": self.reset_episode,  # reset
            }
        )
        self.__listener = keyboard.Listener(
            on_press=self.__set_direction, on_release=self.__reset_direction
        )
        self.__direction_keys_pressed: set = set()  # store which keys are pressed

        self.__label_callbacks: [Callable[[int], None]] = []
        self.__gripper_callbacks: [Callable[[float], None]] = []
        self.__direction_callbacks: [Callable[[np.array], None]] = []
        self.__reset_callbacks: [Callable[[], None]] = []

    def __del__(self):
        self.stop()

    def start(self) -> None:
        self.__hotkeys.start()

        # Need to sleep because apparently there is a race condition problem with one dependent
        # library from pynput when running on macOS (pyobjc).
        # The solution is described in this comment:
        # https://github.com/moses-palmer/pynput/issues/55#issuecomment-1314410235
        time.sleep(1)
        self.__listener.start()

    def stop(self) -> None:
        self.__hotkeys.stop()
        self.__listener.stop()

    def __set_label(self, value: int) -> None:
        with self.__label_lock:
            self.__label = value
            self.__call_callbacks(self.__label_callbacks, self.__label)

    def subscribe_callback_to_label(self, callback_func: Callable[[int], None]) -> None:
        with self.__label_lock:
            self.__label_callbacks.append(callback_func)

    def unsubscribe_callback_to_label(
        self, callback_func: Callable[[int], None]
    ) -> None:
        with self.__label_lock:
            self.__label_callbacks.remove(callback_func)

    @property
    def label(self) -> HumanFeedback:
        with self.__label_lock:
            return self.__label

    def __set_gripper(self, value: float | None) -> None:
        with self.__gripper_lock:
            self.__gripper = value
            self.__call_callbacks(self.__gripper_callbacks, self.__gripper)

    def subscribe_callback_to_gripper(
        self, callback_func: Callable[[float], None]
    ) -> None:
        with self.__gripper_lock:
            self.__gripper_callbacks.append(callback_func)

    def unsubscribe_callback_to_gripper(
        self, callback_func: Callable[[float], None]
    ) -> None:
        with self.__gripper_lock:
            self.__gripper_callbacks.remove(callback_func)

    @property
    def gripper(self) -> float | None:
        with self.__gripper_lock:
            return self.__gripper

    @property
    def is_gripper_commanded(self) -> bool:
        """
        Method to verify weather there is any active gripper command
        :return: If there is any gripper direction command, otherwise False
        """
        with self.__gripper_lock:
            return self.__gripper is not None

    def __set_direction(self, key) -> None:
        with self.__direction_lock:
            try:
                idx, value = self.__KEY_MAPPING[key.char]
                if key not in self.__direction_keys_pressed:
                    self.__direction[idx] = value
                    self.__call_callbacks(self.__direction_callbacks, self.__direction)
                    self.__direction_keys_pressed.add(key)
            except (KeyError, AttributeError):
                pass

    def subscribe_callback_to_direction(
        self, callback_func: Callable[[np.array], None]
    ) -> None:
        with self.__direction_lock:
            self.__direction_callbacks.append(callback_func)

    def unsubscribe_callback_to_direction(
        self, callback_func: Callable[[np.array], None]
    ) -> None:
        with self.__direction_lock:
            self.__direction_callbacks.remove(callback_func)

    def __reset_direction(self, key) -> None:
        with self.__direction_lock:
            try:
                idx, _ = self.__KEY_MAPPING[key.char]
                old_direction_array = self.__direction.copy()
                self.__direction[idx] = 0
                if not np.array_equal(old_direction_array, self.__direction):
                    self.__call_callbacks(self.__direction_callbacks, self.__direction)
                self.__direction_keys_pressed.remove(key)
            except (KeyError, AttributeError):
                pass

    @property
    def direction(self) -> np.array:
        with self.__direction_lock:
            return self.__direction

    @property
    def is_direction_commanded(self) -> bool:
        """
        Method to verify weather there is any active direction command
        :return: If there is any active direction command, otherwise False
        """
        with self.__direction_lock:
            return np.count_nonzero(self.__direction) != 0

    def get_ee_action(self) -> np.array:
        with self.__direction_lock:
            return self.direction * 0.9

    def reset_episode(self) -> None:
        for callback in self.__reset_callbacks:
            callback()
        self.__set_label(1)
        self.__set_gripper(0)

    def subscribe_callback_to_reset(self, callback_func: Callable[[], None]) -> None:
        self.__reset_callbacks.append(callback_func)

    @classmethod
    def __call_callbacks(cls, callbacks, value) -> None:
        if len(callbacks) != 0:
            for callback in callbacks:
                callback(value)

# using the class

keyboard_obs = KeyboardObserver()

keyboard_obs.start()

print("Go!")
try:
    while True:
        # just need to sleep, since there is a thread in the agent doing the stepping and
        # everything else
        time.sleep(0.004)

except KeyboardInterrupt:
    print("Keyboard interrupt. Attempting graceful env shutdown ...")
    keyboard_obs.stop()

[rafaelBauer]

I changed from the official release to the master branch and got it working. I guess this is the same issue from #552

Is there a forecast for an official release?

[moses-palmer]

Thank you for your suggestion.

I have been waiting for confirmation that the feature branch for detection of injected events works, but today I realised that two years have passed since the last release, so a few hours ago I released pynput 1.7.7, which contains the fix för lazy imports.

I will close this issue now, but please reopen if the fix does not work.