6

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.

  • Since this seems to be more a Python asyncio issue than a Pi issue, you may have better luck getting a response if you post on StackOverflow. I assume you've already looked through the Python docs here. – thephez Oct 17 '16 at 16:24
  • Yes, I also have thought to post it into Stackoverflow. But the library is related to raspberry directly. Also, I do not want to create a duplicated question in two different sites. And yes, before this I have take a look into the documentation of python and GPIO library, but my knowledge is not enought to understand the last one. – King Midas Oct 18 '16 at 13:01
  • You may consider using gpio sysfs interface with asyncio, so you won't have to delve into threading stuff. See twisted example here: github.com/sardok/txgpio. Easily convertible to asyncio. – sardok Mar 23 '17 at 9:01
6
+50

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

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
  • Really good explanation that points out how far I am from understanding all python tricks. Thanks for all your effort. This weekend I will try to deeply understand your proposal and test it with my raspberry. But your explanation looks really good. I am agree that the question can be also in stackoverflow due to seems a python issue, not a GPIO issue. But seems that is not need to duplicate the question now. – King Midas Oct 18 '16 at 13:08
  • Ok, I have found a few minutes to try your proposal. My Dummy-4 loop error has disappear. Still I have an error that is TypeError: coroutines cannot be used with call_soon(). In this case, I have added @asyncio.coroutine to the message_manager_f functions, and this error has disappeared. But now, RuntimeError('Event loop is running.') error is shown. Of course if I comment loop.run_forever() no any other error appear, but the message is never shown (neither a prinf). We are close, but I need probably to expend more time with this code. – King Midas Oct 19 '16 at 19:39
  • I just searched through the source code of asyncio for 'Event loop is running' and found the only place such an error is raised is when of BaseEventLoop.run_forever() validates its inputs and current state; which would indicate there is another direct or indirect call to it somewhere in your script. – Shreyas Murali Oct 19 '16 at 20:37
  • I will take a look on it. I have several dependencies and also can be there. – King Midas Oct 19 '16 at 20:52
  • Ok, I have seen the code of asyncio you have pointed out, and you must be correct. Searching in my code, I am not using the function run_forever() but I am using loop.run_until_complete(). This is used for example, for establishing a connection with a hangouts server. – King Midas Oct 19 '16 at 21:02
4

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()
  • Thanks. Your code looks promising! seems that call_soon_threadsafe is what I am looking for. Let me test it in my raspberry in the next days. – King Midas Oct 18 '16 at 13:04
  • I do not understand why you are calling gpio_event_on_loop_thread and not directly asyncio.async(stop_loop()). Does it has a scope? – King Midas Oct 19 '16 at 21:04
  • It will not work to call asyncio.async on the GPIO thread because the event loop isn't associated with the GPIO thread. on_gpio_event runs on the GPIO thread. The purpose of gpio_event_on_loop_thread is to run on the main thread so that it can call asyncio.async. – mkimball Oct 19 '16 at 21:40
  • In versions of Python since 3.5.1, I think you could use asyncio.run_coroutine_threadsafe(stop_loop(), loop) instead of call_soon_threadsafe. I was testing with a 3.4 version of Python, so I didn't use that. – mkimball Oct 19 '16 at 21:43
  • Your code has been really helpful, but as your answer is very similar to @Shreyas Murali, and you have answered later, I am obligated to give the reward to him. Still in some points, your code is really clear as a solution. – King Midas Oct 20 '16 at 13:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.