Source code for simvx.core.helpers.coroutines

"""Coroutine helpers: parallel, wait, wait_until, wait_signal, next_frame."""

import time
from collections.abc import Callable
from typing import TYPE_CHECKING

from ..descriptors import Coroutine

if TYPE_CHECKING:
    from ..descriptors import Signal




def parallel(*coroutines: Coroutine) -> Coroutine:
    """Run multiple coroutines simultaneously, finish when all complete."""
    active = list(coroutines)
    while active:
        finished = []
        for co in active:
            try:
                next(co)
            except StopIteration:
                finished.append(co)
        for co in finished:
            active.remove(co)
        if active:
            yield





def wait(seconds: float) -> Coroutine:
    """Pause a coroutine for given seconds."""
    end = time.monotonic() + seconds
    while time.monotonic() < end:
        yield





def wait_until(condition: Callable[[], bool]) -> Coroutine:
    """Yield until condition() returns True."""
    while not condition():
        yield





def wait_signal(signal: Signal) -> Coroutine:
    """Yield until signal is emitted. Returns signal args."""
    received = [None]
    done = [False]

    def _on_signal(*args):
        received[0] = args
        done[0] = True
    signal.connect(_on_signal)
    try:
        while not done[0]:
            yield
    finally:
        signal.disconnect(_on_signal)
    return received[0]





def next_frame() -> Coroutine:
    """Yield for exactly one frame."""
    yield