Async + Event Loop

[timkpaine]

Putting this up for some discussion, we've been talking/integrating csp with lots of other frameworks, and many of them are based on asyncio. This PR adds some bridging utilities. For now they are mostly one direction, but we can also add the other direction which should be easy (e.g. running csp in a background thread and making edges awaitable / async generators).

import csp
from csp import ts
from datetime import timedelta, datetime
from typing import AsyncIterator
import asyncio


async def async_in() -> int:
    await asyncio.sleep(0.1)
    return 42

async def async_out(n: int) -> None:
    await asyncio.sleep(0.1)
    print(f"Output: {n}")

async def async_node(n: int) -> int:
    await asyncio.sleep(0.1)
    return n * 2

async def async_counter_node(n: int) -> AsyncIterator[int]:
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

@csp.node
def csp_counter_node() -> ts[int]:
    with csp.alarms():
        tick_alarm = csp.alarm(bool)

    with csp.state():
        s_counter = 0

    with csp.start():
        csp.schedule_alarm(tick_alarm, timedelta(), True)

    if csp.ticked(tick_alarm):
        s_counter += 1
        csp.schedule_alarm(tick_alarm, timedelta(seconds=0.1), True)
        return s_counter


@csp.node
def csp_async_node() -> ts[int]:
    """
    Example node that uses async operations with the alarm pattern.
    Uses AsyncAlarm for a clean alarm-like interface.
    """
    with csp.alarms():
        poll_alarm = csp.alarm(bool)
        async_alarm = csp.async_alarm(int)

    with csp.state():
        s_counter = 0
        s_pending = False

    with csp.start():
        # Schedule first poll
        csp.schedule_alarm(poll_alarm, timedelta(milliseconds=10), True)

    if csp.ticked(poll_alarm):
        # Only schedule a new async operation if one isn't already pending
        if not s_pending:
            s_counter += 1
            # Schedule async operation to double the counter
            csp.schedule_async_alarm(async_alarm, async_node(s_counter))
            s_pending = True

        # Keep polling
        csp.schedule_alarm(poll_alarm, timedelta(milliseconds=10), True)

    if csp.ticked(async_alarm):
        # Async operation completed - we can schedule another one now
        s_pending = False
        return async_alarm


@csp.graph
def graph():
    csp_counter = csp_counter_node()
    async_counter = csp.async_for(async_counter_node(15))

    csp.print("counter", csp_counter)
    csp.print("async_counter", async_counter)

    # async_in: coroutine that ticks once when ready
    async_in_result = csp.async_in(async_in())
    csp.print("async_in", async_in_result)

    # async_out: invoke async function when input ticks
    csp.async_out(csp_counter, async_out)

    # async_node: takes input, runs async, outputs result
    async_node_result = csp.async_node(csp_counter, async_node)
    csp.print("async_node", async_node_result)

    # async for within CSP node
    csp.print("csp_async_node", csp_async_node())

if __name__ == "__main__":
    csp.run(graph, realtime=True, endtime=timedelta(seconds=2))