Implement a GPIO function with a callback calling a asyncio method

Question

I am testing a sensor attached to the GPIO in raspberry using python 3.5. The idea is something similar to this link.

My code is:

gpio.add_event_detect(4, gpio.RISING, callback=lambda x: self.motion_sensor(message_manager_f), bouncetime=500)

def motion_sensor(self, message_manager_f):
            asyncio.async(message_manager_f("Motion detected!"))

Where message_manager_f is a callback to a function that process the string message. This callback can have a complex operation (i.e. send to a server an alert) and therefore I am trying to perform an asyncio operation.

If I run this code, when the sensor detects a presence, I have this error:

RuntimeError: There is no current event loop in thread 'Dummy-4'.

If I remove the asyncio operator, and for example I only print the message to console, it works fine as the example of the previous link. Of course, I have tried to search for this error in the web (python is not my strongest language) but nothing that helps me to explain this error (only a few entries with this issue). I am wandering if maybe mixing gpio.add_event_detect with a function that uses asyncio is not possible, but I have no idea where to check it.

Is it is possible to achieve what I want? If it is, can somebody explain what really means the obtained error and how to solve it?

The complete code is available in this link if somebody thinks that the error can be elsewhere.

--EDIT--

Seems that same behavior is obtained with the next code:

self._message_manager = message_manager_f
gpio.add_event_detect(4, gpio.RISING, self.motion_sensor, bouncetime=500)

def motion_sensor(self):
    asyncio.async(self._message_manager("Motion detected!"))

Obviously, is related to the asyncio.async.

Answer 1

First off, I agree with @thephez that this question is probably better for stackoverflow. Since there is a overlap with the RPI.GPIO, I have taken the liberty to answer it the best I can.

Part 1 - RPI.GPIO

To really understand how the callback mechanism is implemented/works in RPI.GPIO, I dug into its
source code

Within it there is a file event_gpio.c and between lines 430 through 480 I found the definition of the function add_edge_detect Within this function a background thread is launched using pthread_create which essentially blocks until some interesting GPIO pin transition happens. When such an event occurs, the thread unblocks and executes the collection of callbacks registered.

int add_edge_detect(unsigned int gpio, unsigned int edge, int bouncetime)
// return values:
// 0 - Success
// 1 - Edge detection already added
// 2 - Other error
{
    // <<< snip >>>

    // start poll thread if it is not already running
    if (!thread_running) {
        if (pthread_create(&threads, NULL, poll_thread, (void *)t) != 0) {
           remove_edge_detect(gpio);
           return 2;
        }
    }
    return 0;
}

Part 2 - Python asyncio

Consider the simple async example here

import asyncio

def hello_world(loop):
    print('Hello World')
    loop.stop()

loop = asyncio.get_event_loop()

# Schedule a call to hello_world()
loop.call_soon(hello_world, loop)

# Blocking call interrupted by loop.stop()
loop.run_forever()
loop.close()

When a python process starts, there is by default 1 thread that executes the python code. In the above example one can see there is a blocking call to loop.run_forever(). Internally it would wait for some task to be submitted (execute hello_world in this case) into a priority queue and execute accordingly.

To verify this, we need to peek a bit under the hood. From the source code of loop.run_forever in base_events.py one can see it simply keeps calling _run_once() in a loop

def run_forever(self):
    """Run until stop() is called."""
    self._check_closed()
    if self.is_running():
        raise RuntimeError('Event loop is running.')
    self._thread_id = threading.get_ident()
    try:
        while True:
            try:
                self._run_once()
            except _StopError:
                break
    finally:
        self._thread_id = None

_run_once is implemented like so using a heap (priority queue)

def _run_once(self):
"""Run one full iteration of the event loop.

This calls all currently ready callbacks, polls for I/O,
schedules the resulting callbacks, and finally schedules
'call_later' callbacks.
"""

sched_count = len(self._scheduled)
if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
    self._timer_cancelled_count / sched_count >
        _MIN_CANCELLED_TIMER_HANDLES_FRACTION):
    # Remove delayed calls that were cancelled if their number
    # is too high
    new_scheduled = []
    for handle in self._scheduled:
        if handle._cancelled:
            handle._scheduled = False
        else:
            new_scheduled.append(handle)

    heapq.heapify(new_scheduled)
    self._scheduled = new_scheduled
    self._timer_cancelled_count = 0
else:
    # Remove delayed calls that were cancelled from head of queue.
    while self._scheduled and self._scheduled[0]._cancelled:
        self._timer_cancelled_count -= 1
        handle = heapq.heappop(self._scheduled)
        handle._scheduled = False

timeout = None
if self._ready:
    timeout = 0
elif self._scheduled:
    # Compute the desired timeout.
    when = self._scheduled[0]._when
    timeout = max(0, when - self.time())

if self._debug and timeout != 0:
    t0 = self.time()
    event_list = self._selector.select(timeout)
    dt = self.time() - t0
    if dt >= 1.0:
        level = logging.INFO
    else:
        level = logging.DEBUG
    nevent = len(event_list)
    if timeout is None:
        logger.log(level, 'poll took %.3f ms: %s events',
                   dt * 1e3, nevent)
    elif nevent:
        logger.log(level,
                   'poll %.3f ms took %.3f ms: %s events',
                   timeout * 1e3, dt * 1e3, nevent)
    elif dt >= 1.0:
        logger.log(level,
                   'poll %.3f ms took %.3f ms: timeout',
                   timeout * 1e3, dt * 1e3)
else:
    event_list = self._selector.select(timeout)
self._process_events(event_list)

# Handle 'later' callbacks that are ready.
end_time = self.time() + self._clock_resolution
while self._scheduled:
    handle = self._scheduled[0]
    if handle._when >= end_time:
        break
    handle = heapq.heappop(self._scheduled)
    handle._scheduled = False
    self._ready.append(handle)

# This is the only place where callbacks are actually *called*.
# All other places just add them to ready.
# Note: We run all currently scheduled callbacks, but not any
# callbacks scheduled by callbacks run this time around --
# they will be run the next time (after another I/O poll).
# Use an idiom that is thread-safe without using locks.
ntodo = len(self._ready)
for i in range(ntodo):
    handle = self._ready.popleft()
    if handle._cancelled:
        continue
    if self._debug:
        try:
            self._current_handle = handle
            t0 = self.time()
            handle._run()
            dt = self.time() - t0
            if dt >= self.slow_callback_duration:
                logger.warning('Executing %s took %.3f seconds',
                               _format_handle(handle), dt)
        finally:
            self._current_handle = None
    else:
        handle._run()
handle = None  # Needed to break cycles when an exception occurs.

Part 3

Explaining the error you are getting

RuntimeError: There is no current event loop in thread 'Dummy-4'.

This would be because the thread 'Dummy-4' is the thread described in Part 1 which executes the GPIO callbacks. As there is no loop object created on this thread; there is no loop.run_forever() that is required to service your message_manager_f callback

Part 4

How to fix this problem ?

Simply introducing a event loop within your def motion_sensor is not the right way to do it

def motion_sensor(self, message_manager_f):
    loop = asyncio.get_event_loop()
    loop.call_soon(message_manager_f, loop)
    loop.run_forever()
    loop.close()

This wrong is because

• the call wouldnt really be async with respect to the GPIO callback thread
• this will block the RPI.GPIO's thread from servicing other callbacks

My solution would look something like this

import asyncio
import RPIO.GPIO as GPIO
import sys

loop = None

def message_manager_f():
    print ":P message_manager_f()"

def motion_sensor(self, message_manager_f):
    if loop is None:
        print(":(")
        return       # should not come to this
    # this enqueues a call to message_manager_f() 
    loop.call_soon_threadsafe(message_manager_f)

# this is the primary thread mentioned in Part 2
if __name__ == '__main__':
    try:
        # setup the GPIO
        GPIO.setwarnings(True)
        GPIO.setmode(GPIO.BCM)
        GPIO.setup(4, GPIO.IN) # adjust the PULL UP/PULL DOWN as applicable
        GPIO.add_event_detect(4, GPIO.RISING, callback=lambda x: self.motion_sensor(message_manager_f), bouncetime=500)

        # run the event loop
        loop = asyncio.get_event_loop()
        loop.run_forever()
        loop.close()
    except :
        print("Error:", sys.exc_info()[0])

    # cleanup
    GPIO.cleanup()

References

• https://stackoverflow.com/questions/32586794/submit-a-job-to-an-asyncio-event-loop
• https://docs.python.org/3/library/asyncio-dev.html#asyncio-multithreading

Notes

• As I don't have the required h/w handy, I haven't had a chance to verify the above code works correctly or not.
• Any errors or questions, please leave a comment and I will try to address it the best I can

Answer 2

RPi.GPIO callbacks are run on a callback thread that is not the main thread. asyncio event loops are associated with particular threads, and asyncio.async can only be used on a thread with an associated event loop. In order to get from the RPi.GPIO thread back to the asyncio event loop thread, you've got to use call_soon_threadsafe on the asyncio loop object. In order to do this, you must have a reference to the loop object obtained by calling get_event_loop on the asyncio thread. (Not on the RPi.GPIO thread.)

Here is an example of combining RPi.GPIO and asyncio. I have tested it on my Raspberry Pi by connecting a jumper cable between pins 16 and 22, which will allow it to send a signal from one pin to another, which will call the callback, and allow the queueing of the stop_loop coroutine in response to the GPIO event.

import asyncio
import RPi.GPIO as GPIO

LOOP_IN = 16
LOOP_OUT = 22

@asyncio.coroutine
def delayed_raise_signal():
    yield from asyncio.sleep(1)

    GPIO.output(LOOP_OUT, GPIO.HIGH)

@asyncio.coroutine
def stop_loop():
    yield from asyncio.sleep(1)

    print('Stopping Event Loop')
    asyncio.get_event_loop().stop()

def gpio_event_on_loop_thread():
    asyncio.async(stop_loop())

def setup():
    GPIO.setmode(GPIO.BOARD)
    GPIO.setup(LOOP_IN, GPIO.IN)
    GPIO.setup(LOOP_OUT, GPIO.OUT)
    GPIO.output(LOOP_OUT, GPIO.LOW)

    def on_gpio_event(channel):
        print('Rising event detected')
        loop.call_soon_threadsafe(gpio_event_on_loop_thread)

    loop = asyncio.get_event_loop()
    GPIO.add_event_detect(LOOP_IN, GPIO.RISING, callback=on_gpio_event)

    asyncio.async(delayed_raise_signal())

if __name__ == '__main__':
    setup()
    asyncio.get_event_loop().run_forever()
    GPIO.cleanup()