56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
"""
|
|
Asynchronous I/O scheduler for writing concurrent code.
|
|
|
|
MicroPython module: https://docs.micropython.org/en/v1.21.0/library/asyncio.html
|
|
|
|
CPython module:
|
|
`asyncio `<https://docs.python.org/3.8/library/asyncio.html>
|
|
|
|
Example::
|
|
|
|
import asyncio
|
|
|
|
async def blink(led, period_ms):
|
|
while True:
|
|
led.on()
|
|
await asyncio.sleep_ms(5)
|
|
led.off()
|
|
await asyncio.sleep_ms(period_ms)
|
|
|
|
async def main(led1, led2):
|
|
asyncio.create_task(blink(led1, 700))
|
|
asyncio.create_task(blink(led2, 400))
|
|
await asyncio.sleep_ms(10_000)
|
|
|
|
# Running on a pyboard
|
|
from pyb import LED
|
|
asyncio.run(main(LED(1), LED(2)))
|
|
|
|
# Running on a generic board
|
|
from machine import Pin
|
|
asyncio.run(main(Pin(1), Pin(2)))
|
|
|
|
Core functions
|
|
--------------
|
|
"""
|
|
from _typeshed import Incomplete, Incomplete as Incomplete
|
|
from typing import Any, Coroutine, List, Tuple
|
|
|
|
class TaskQueue:
|
|
def push(self, *args, **kwargs) -> Incomplete: ...
|
|
def peek(self, *args, **kwargs) -> Incomplete: ...
|
|
def remove(self, *args, **kwargs) -> Incomplete: ...
|
|
def pop(self, *args, **kwargs) -> Incomplete: ...
|
|
def __init__(self, *argv, **kwargs) -> None: ...
|
|
|
|
class Task:
|
|
"""
|
|
This object wraps a coroutine into a running task. Tasks can be waited on
|
|
using ``await task``, which will wait for the task to complete and return
|
|
the return value of the task.
|
|
|
|
Tasks should not be created directly, rather use `create_task` to create them.
|
|
"""
|
|
|
|
def __init__(self) -> None: ...
|