Traffic light example

[ctimmer]

I posted this link in a post that asked about controlling a traffic light. There is much more documentation now and interrupt code has been added to handle a crosswalk request.

Link:
see traffic light under examples
There is no de-bouncing but it isn't needed in this implementation.

I started to include code to use uasyncio with button de-bouncing but that proved to be a little beyond my understanding of this module. A little help would be much appreciated.

Feel free to comment on the poll-looper module.

Thanks,
Curt

[jimmo]

I started to include code to use uasyncio with button de-bouncing but that proved to be a little beyond my understanding of this module. A little help would be much appreciated.

Sure! Can you give more information about what you'd like to know more about?

[ctimmer]

This has been real learning experience. I added the capability for poll-looper to be used with uasyncio. The main challenge the usage of time.sleep() in the poll_wait method. If running with uasyncio, poll_wait does not sleep and returns the sleep milliseconds. This is the uasyncio code from the trafficlight.py example:

if CROSSWALK_PIN != None :
    crosswalk_ir = Pin (CROSSWALK_PIN, Pin.IN)
    crosswalk_ir.irq (trigger = Pin.IRQ_RISING ,
                      handler = crossing_request.crossing_request)
        
if USE_ASYNCIO :
    import uasyncio as asyncio
    async def poll_plugins () :
        while True :
            #print ("poll_plugins: entry:")
            sleep_ms = poller.poll_wait ()
            #---- poll_wait returns ms to wait for next poll cycle
            await asyncio.sleep_ms (sleep_ms)
            #---- polls plugins added to poll-looper
            poller.poll_plugins ()
    async def main() :
        asyncio.create_task (poll_plugins())
        #---- Create additional tasks here
        while True:
            await asyncio.sleep(1)
    asyncio.run (main ())           # start the traffic lights
else :
    poller.poll_start ()            # normal startup

I am handling the pin input with an interrupt handler, this doesn't appear to cause problems with uasyncio. According to other posts, uasyncio does not currently handle GPIO input.

Curt

[jimmo]

According to other posts, uasyncio does not currently handle GPIO input.

asyncio is independent of gpio, however we do have plans to make it easier to integrate asyncio with hardware APIs (i.e. ability to await for hardware events).

To do this right now for GPIO (e.g. awaiting a pin change) see #6886 (comment)

[peterhinch]

To do this right now for GPIO (e.g. awaiting a pin change) see #6886 (comment)

To clarify, this solution caters for the case where a pulse is brief relative to scheduling delays. It is not a great way to interface switches (IMO).

Using interrupts with uasyncio does not reduce latency because the ThreadSafeFlag does not short-circuit the round-robin scheduling of uasyncio. For slow signals like those from switches and pushbuttons interrupts offer no gain over polling. Interrupts are a limited resource and (in general) require careful coding to use correctly.

For interfacing mechanical contacts consider these classes which are based on polling. They offer either a callback-based API or an "await event" interface.

[ctimmer]

I am marking this answered, not because all my questions have been answered but because I have a lot more research to do,

The final (current) version of the trafficlight example includes the Pushbutton class available here. Thanks Peter.

Revised code from above:

poller = PollLooper (POLL_INTERVAL_MS,
                     use_asyncio = USE_ASYNCIO)

tl_controller = TL_Controller (poller)
crossing_request = TL_CrossingRequest (poller)

poller.poll_add (tl_controller)
poller.poll_add (crossing_request)
poller.poll_add (TL_View (poller))

if USE_ASYNCIO :
    import uasyncio as asyncio
    from primitives import Pushbutton
    async def poll_plugins () :
        poller.poll_init ()        # Reset poll start time
        while poller.running () :
            #print ("poll_plugins: entry:")
            gc.collect ()
            #---- polls plugins added to poll-looper
            poller.poll_plugins ()
            #---- get wait time
            sleep_ms = poller.poll_wait ()
            #---- poll_wait returns ms to wait for next poll cycle
            await asyncio.sleep_ms (sleep_ms)
        print ("poller: exit")
    async def main() :
        asyncio.create_task (poll_plugins ())
        if CROSSWALK_PIN != None :
            crosswalk_pin = Pin (CROSSWALK_PIN, Pin.IN, Pin.PULL_UP)
            crosswalk_button = Pushbutton (crosswalk_pin)
            crosswalk_button.press_func (tl_controller.crossing_request, ()) 
        #---- Create additional tasks here
        while True:
            await asyncio.sleep(1)
    asyncio.run (main ())           # start the traffic lights
else :
    if CROSSWALK_PIN != None :
        crosswalk_pin = Pin (CROSSWALK_PIN, Pin.IN)
        crosswalk_pin.irq (trigger = Pin.IRQ_RISING ,
                          handler = crossing_request.crossing_request)
    poller.poll_start ()            # normal startup

Full code here

Curt