I'm trying to use the pynput library to listen for hotkeys and start a thread when the F12 key is pressed. The thread prints a message in an infinite loop to the console. Pressing ESC should stop the application. My Example class extends Thread and overrides the run method.
src/index.py
import pynput
from lib.listener import HotkeyListener
if __name__ == "__main__":
on_press = HotkeyListener()
listener = pynput.keyboard.Listener(on_press=on_press.on_press, suppress=True)
listener.run()
listener.py
from .Example import Example
import pynput
class HotkeyListener:
def __init__(self):
self.example = Example()
def on_press(self, key):
if key == pynput.keyboard.Key.f12:
print("Starting Thread...")
self.example.start()
if key == pynput.keyboard.Key.esc:
self.example.join()
print("Exiting application...")
return False # Stop listener
example.py
from threading import Thread
from .delay import rand_sleep
class Example(Thread):
def run(self):
while True:
print("Hello, I am an Example :) ?") # This message appears in the console
rand_sleep(1, 2)
Problem: When the F12 key is pressed, the thread starts and enters an infinite loop, printing messages to the console. This causes the application to not respond to other hotkeys, like ESC, to stop the execution.
Questions:
Any help would be greatly appreciated!
@EDIT
After @jupiterbjy's response, I arrived at the following solution (now it works as expected!), but there's a runtime error when I try to create it again by pressing the hotkey. The thread no longer exists and can only be started once:
class Example(Thread):
def __init__(self, event: Event) -> None:
super().__init__()
self.event = event
def run(self):
counter: Number = 0
while not self.event.is_set():
print("Hello, I am an Example :)? ")
rand_sleep(1, 2)
counter += 1
if counter > 5:
self.event.set()
# RuntimeError: threads can only be started once
class HotkeyListener:
def __init__(self):
self.thread_event = threading.Event()
self.example = Example(self.thread_event)
def on_press(self, key):
if key == pynput.keyboard.Key.f12:
print("---------------------", self.example)
# --------------------- <Example(Thread-1, stopped 9676)>
self.example.start()
if key == pynput.keyboard.Key.esc:
self.thread_event.set()
self.example.join()
print("Exiting application...")
return False # Stop listener
To quote from document:
join(timeout=None)
Wait until the thread terminates. This blocks the calling thread until the thread whose join() method is called terminates – either normally or through an unhandled exception – or until the optional timeout occurs.
This is used to wait until thread is stopped, not to stop the thread. Hence your infinite loop in Example class never had chance to stop.
To quote from Trio Documentation about threading:
Cancellation is a tricky issue here, because neither Python nor the operating systems it runs on provide any general mechanism for cancelling an arbitrary synchronous function running in a thread. This function will always check for cancellation on entry, before starting the thread. But once the thread is running, there are two ways it can handle being cancelled:
- If cancellable=False, the function ignores the cancellation and keeps going, just like if we had called sync_fn synchronously. This is the default behavior.
- If cancellable=True, then this function immediately raises Cancelled. In this case the thread keeps running in background – we just abandon it to do whatever it’s going to do, and silently discard any return value or errors that it raises.
So, from these we can learn that there's no good way to terminate infinite-loop running in thread.
Instead, we use Threading.Event.
import threading
import random
import time
import pynput
def thread_print(*arg, **kwargs):
"""Print with thread id"""
print(f"[T:{threading.get_ident():6}]", *arg, **kwargs)
def func_for_thread(event: threading.Event):
"""Function to be ran on thread"""
thread_print("Thread started")
while not event.is_set():
time.sleep(1)
thread_print(random.randint(1, 10))
thread_print("Thread finished")
def main():
thread_event = threading.Event()
# Create 5 threads, but don't start yet
threads = [
threading.Thread(target=func_for_thread, args=(thread_event,)) for _ in range(5)
]
threads_started = False
# main thread will start listening for key press
with pynput.keyboard.Events() as events:
# for each incoming keyboard event:
for event in events:
# change match-case to if-elif-else for python < 3.10
match event.key:
case pynput.keyboard.Key.esc:
print("ESC pressed! Stopping all threads!")
thread_event.set()
# wait for thread to stop gracefully
for thread in threads:
thread.join()
break
case pynput.keyboard.Key.space:
if threads_started:
continue
threads_started = True
print(f"SPACE pressed! Starting all threads!")
# start threads
for thread in threads:
thread.start()
if __name__ == "__main__":
main()
SPACE pressed! Starting all threads!
[T: 25216] Thread started
[T: 25796] Thread started
[T: 26688] Thread started
[T: 4216] Thread started
[T: 26480] Thread started
[T: 25216] 10
[T: 25796] 1
[T: 26688] 9
[T: 4216] 3
[T: 26480] 9
[T: 25216] 6
[T: 25796] 10
[T: 26688] 3
[T: 4216] 10
[T: 26480] 2
ESC pressed! Stopping all threads!
[T: 25216] 4
[T: 25216] Thread finished
[T: 25796] 3
[T: 25796] Thread finished
[T: 26688] 7
[T: 26688] Thread finished
[T: 4216] 9
[T: 4216] Thread finished
[T: 26480] 6
[T: 26480] Thread finished
Since what you actually wanted to do is creating hotkey that stoppes immediately, this should be sufficient. Subclassing Thread is overkill for that!
To expand further for restartability, and independant control I added convenience wrapper class.
import threading
import random
import time
import pynput
from typing import Callable
class RestartableThread:
def __init__(self, target: Callable[[threading.Event, ...], ...], *args, **kwargs):
"""Create a tiny thread manager that can be restarted.
Args:
target (Callable): A function/method to run on thread.
Must accept `threading.Event` as first argument.
args (tuple, optional): Argument for target. Defaults to ().
kwargs (dict, optional): Argument for target. Defaults to {}.
"""
self.target: Callable[[threading.Event, ...], ...] = target
self.thread: threading.Thread | None = None
self.event = threading.Event()
self.args = (self.event, *args)
self.kwargs = kwargs
@property
def is_running(self):
"""Check if thread is running"""
return self.thread is not None and self.thread.is_alive()
@property
def is_stopping(self):
"""Check if thread is stopping"""
return self.event.is_set()
def run(self):
"""Start thread. Doesn't do anything if already running"""
# ignore if already running
if self.is_running:
return
# make sure thread is stopped before starting new one
self.join()
# create new thread & start
self.thread = threading.Thread(
target=self.target, args=self.args, kwargs=self.kwargs
)
self.thread.start()
def join(self):
"""Wait for thread to stop gracefully"""
if self.thread is None:
return
self.thread.join()
self.event.clear()
# ^ also clear event so we can restart thread
def stop(self):
"""Signal thread to stop"""
self.event.set()
def thread_print(*arg, **kwargs):
"""Print with thread id"""
print(f"[T:{threading.get_ident():6}]", *arg, **kwargs)
def func_a_for_thread(event: threading.Event, *args, **kwargs):
"""Function to be ran on thread"""
thread_print("Thread started")
while not event.is_set():
time.sleep(random.randint(1, 3))
thread_print("Func A running with args:", args, kwargs)
thread_print("Thread finished")
# Your codes in Thread subclass goes here!
class SomeClass:
# skipping init
def func_b_for_thread(self, event: threading.Event, *args, **kwargs):
"""Function to be ran on thread"""
thread_print("Thread started")
while not event.is_set():
time.sleep(random.randint(1, 3))
thread_print("Func B running with args:", args, kwargs)
thread_print("Thread finished")
def main():
# create threads
thread_a = RestartableThread(func_a_for_thread, args=("arg1", "arg2"))
some_instance = SomeClass()
thread_b = RestartableThread(some_instance.func_b_for_thread, args=("arg3", "arg4"))
# main thread will start listening for key press
with pynput.keyboard.Events() as events:
# for each incoming keyboard event:
for event in events:
if event.key == pynput.keyboard.Key.esc:
print("ESC pressed! Stopping all threads!")
thread_a.stop()
thread_b.stop()
# wait for thread to stop gracefully
thread_a.join()
thread_b.join()
break
# some keys don't have char attribute
if not hasattr(event.key, "char"):
continue
match event.key.char:
case "a":
if not thread_a.is_running:
print("Starting thread A")
thread_a.run()
case "b":
if not thread_b.is_running:
print("Starting thread B")
thread_b.run()
case "c":
if thread_a.is_running and not thread_a.is_stopping:
print("Stopping thread A!")
thread_a.stop()
case "d":
if thread_b.is_running and not thread_b.is_stopping:
print("Stopping thread B!")
thread_b.stop()
if __name__ == "__main__":
main()
Starting thread A
[T: 30628] Thread started
Starting thread B
[T: 28796] Thread started
[T: 28796] Func B running with args: ('arg3', 'arg4') {}
[T: 30628] Func A running with args: ('arg1', 'arg2') {}
[T: 28796] Func B running with args: ('arg3', 'arg4') {}
[T: 30628] Func A running with args: ('arg1', 'arg2') {}
[T: 28796] Func B running with args: ('arg3', 'arg4') {}
Stopping thread A!
[T: 28796] Func B running with args: ('arg3', 'arg4') {}
[T: 30628] Func A running with args: ('arg1', 'arg2') {}
[T: 30628] Thread finished
Stopping thread B!
Starting thread A
[T: 11376] Thread started
[T: 28796] Func B running with args: ('arg3', 'arg4') {}
[T: 28796] Thread finished
ESC pressed! Stopping all threads!
[T: 11376] Func A running with args: ('arg1', 'arg2') {}
[T: 11376] Thread finished
ESC pressed! Stopping all threads!