circuitpython/tests/extmod/asyncio_threadsafeflag.py

# Test Event class

try:
    import asyncio
except ImportError:
    print("SKIP")
    raise SystemExit


import micropython

try:
    micropython.schedule
    # CIRCUITPY: no ThreadSafeFlag
    asyncio.ThreadSafeFlag
except AttributeError:
    print("SKIP")
    raise SystemExit


async def task(id, flag):
    print("task", id)
    await flag.wait()
    print("task", id, "done")


def set_from_schedule(flag):
    flag.set()


async def main():
    flag = asyncio.ThreadSafeFlag()

    # Set the flag from within the loop.
    t = asyncio.create_task(task(1, flag))
    print("yield")
    await asyncio.sleep(0)
    print("set event")
    flag.set()
    print("yield")
    await asyncio.sleep(0)
    print("wait task")
    await t

    # Set the flag from scheduler context.
    print("----")
    t = asyncio.create_task(task(2, flag))
    print("yield")
    await asyncio.sleep(0)
    print("set event")
    micropython.schedule(set_from_schedule, flag)
    print("yield")
    await asyncio.sleep(0)
    print("wait task")
    await t

    # Flag already set.
    print("----")
    print("set event")
    flag.set()
    t = asyncio.create_task(task(3, flag))
    print("yield")
    await asyncio.sleep(0)
    print("wait task")
    await t

    # Flag set, cleared, and set again.
    print("----")
    print("set event")
    flag.set()
    print("yield")
    await asyncio.sleep(0)
    print("clear event")
    flag.clear()
    print("yield")
    await asyncio.sleep(0)
    t = asyncio.create_task(task(4, flag))
    print("yield")
    await asyncio.sleep(0)
    print("set event")
    flag.set()
    print("yield")
    await asyncio.sleep(0)
    print("wait task")
    await t


asyncio.run(main())