asyncio used

Author: tomas-fryza

examples/30-asyncio/leds.py

@@ -1,35 +1,71 @@
 # This example makes two LEDs blink at different intervals,
-# showcasing how `asyncio` can handle multiple tasks concurrently
-# without blocking the execution of other task(s).
+# showcasing how `asyncio` can handle multiple tasks
+# concurrently without blocking the execution of other task(s).
 #
-# See:
-# https://docs.micropython.org/en/latest/library/asyncio.html
-# https://bbc.github.io/cloudfit-public-docs/asyncio/asyncio-part-2.html
-# https://gpiocc.github.io/learn/micropython/esp/2020/06/13/martin-ku-asynchronous-programming-with-uasyncio-in-micropython.html
-# https://randomnerdtutorials.com/micropython-esp32-esp8266-asynchronous-programming/
+# See also:
+#   https://docs.micropython.org/en/latest/library/asyncio.html
+#   https://bbc.github.io/cloudfit-public-docs/asyncio/asyncio-part-2.html
+#   https://gpiocc.github.io/learn/micropython/esp/2020/06/13/martin-ku-asynchronous-programming-with-uasyncio-in-micropython.html
+#   https://randomnerdtutorials.com/micropython-esp32-esp8266-asynchronous-programming/
 
-import asyncio
+import uasyncio as asyncio
+from machine import Pin
+import time
+
+# Get the starting time in milliseconds
+start = time.ticks_ms()
+
+# Setup LED pins
+red_led = Pin(7, Pin.OUT)
+green_led = Pin(5, Pin.OUT)
+
+# Elapsed time helper
+def get_elapsed():
+    return time.ticks_diff(time.ticks_ms(), start)
+
+# Define coroutine functions
+async def blink_red_led():
+    while True:
+        print(f"[{get_elapsed()}] Red toggling")
+        red_led.value(not red_led.value())
+
+        # Non-blocking async sleep
+        # Pauses only the current coroutine for 0.5 seconds and
+        # keeps other tasks running during this time
+        await asyncio.sleep(.5)
 
-async def blink(led, period_ms):
+async def blink_green_led():    
     while True:
-        print(f"[{led}] loop start")
-        led.on()
-        print(f"[{led}] on")
-        await asyncio.sleep_ms(20)  # Non-blocking delay
-        led.off()
-        print(f"[{led}] off")
-        await asyncio.sleep_ms(period_ms)
+        print(f"[{get_elapsed()}] Green toggling")
+        green_led.value(not green_led.value())
+        await asyncio.sleep(2)
 
 async def main():
-    t1 = asyncio.create_task(blink(led1, 700))
-    t2 = asyncio.create_task(blink(led2, 400))
-    print(t1)
-    print(t2)
-    await asyncio.sleep_ms(5_000)
-    
-from machine import Pin
+    # Log elapsed time
+    print(f"[{get_elapsed()}] Starting tasks...")
+
+    # Create tasks for blinking two LEDs concurrently
+    task1 = asyncio.create_task(blink_red_led())
+    task2 = asyncio.create_task(blink_green_led())
+
+    # Run async tasks concurrently, wait for all of them to complete,
+    # and keep main() alive while tasks run
+    await asyncio.gather(task1, task2)
 
-led1 = Pin(2, Pin.OUT)
-led2 = Pin(5, Pin.OUT)
+try:
+    # Gets the async "scheduler", the "brain" that controls
+    # and runs asynchronous code
+    loop = asyncio.get_event_loop()
+
+    # Runs that coroutine until it's completely done
+    loop.run_until_complete(main())
+
+except KeyboardInterrupt:
+    # This part runs when Ctrl+C is pressed
+    print("Program stopped. Exiting...")
+    
+    # Optional cleanup code
+    red_led.off()
+    green_led.off()
 
-asyncio.run(main())
+    loop.stop()  # Not available in all uasyncio versions