Return result from Daemon thread watching evdev input events

518 views Asked by At

I am trying to return the results of a Daemon thread. I am using evdev to monitor touch input events and if there is more than 0 touch events then I want to return True. I am trying to pass a list as an arg to the thread's target function and in this version I am appending to the list. However, I keep getting the error RuntimeError: There is no current event loop in thread on the first touch.

import time
def check_for_input(result):
    from evdev import InputDevice, codes
    dev = InputDevice('/dev/input/event0')
    for event in dev.read_loop():
        if event.type == ecodes.EV_KEY:
            result.append([True])

from threading import Thread
results = [False]
input_thread = Thread(target=check_for_input, args=(results), daemon=True)
input_thread.start()

# do some other stuff
time.sleep(5)

input_thread.join()
print(results)

How can I fix this and will it work once I have fixed that error? And as an additional question is this the best/easiest way to do this?

Edit: I have managed to remove the runtime error by creating a loop within the thread. Also, using Bhargav's suggestion, I can get the results as a global variable. So my code is currently as such:

import time
import asyncio
def check_for_input(stop_event):
    print('started')
    global results
    from evdev import InputDevice, ecodes
    dev = InputDevice('/dev/input/event0')
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    for event in dev.read_loop():
        if event.type == ecodes.EV_KEY:
            results = True
            break
        if stop_event.is_set():
            break
    print('EXITING')

from threading import Thread, Event
stop_event = Event()
stop_event.clear()
results = False
input_thread = Thread(target=check_for_input, args=(stop_event,), daemon=True)
print('about to start')
input_thread.start()

# do some other stuff
time.sleep(5)

stop_event.set()
print(results)

However, as per my additional question, is this the right way to do this? I'm sure people will say that a global variable is to be avoided here. Also, how can I kill the thread right before I print(results)? the stop_event.set() above doesn't seem to work

1

There are 1 answers

1
Bhargav Desai On

I think this will work for you :

import time
def check_for_input():
    global results
    from evdev import InputDevice, codes
    dev = InputDevice('/dev/input/event0')
    for event in dev.read_loop():
        if event.type == ecodes.EV_KEY:
            results = True

from threading import Thread
results = False
input_thread = Thread(target=check_for_input, daemon=True)
input_thread.start()

# do some other stuff
time.sleep(5)

input_thread.join()
print(results)