Skip to content

academy.event

or_event 

or_event(*events: Event) -> Event

Create a combined event that is set when any input events are set.

Note

The creator can wait on the combined event, but must still check each individual event to see which was set.

Warning

This works by dynamically replacing methods on the inputs events with custom methods that trigger callbacks.

Note

Based on this Stack Overflow answer.

Parameters:

  • events (Event, default: () ) –

    One or more events to combine.

Returns:

  • Event –

    A single event that is set when any of the input events is set.

Source code in academy/event.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def or_event(*events: threading.Event) -> threading.Event:
    """Create a combined event that is set when any input events are set.

    Note:
        The creator can wait on the combined event, but must still check
        each individual event to see which was set.

    Warning:
        This works by dynamically replacing methods on the inputs events
        with custom methods that trigger callbacks.

    Note:
        Based on this Stack Overflow
        [answer](https://stackoverflow.com/a/12320352).

    Args:
        events: One or more events to combine.

    Returns:
        A single event that is set when any of the input events is set.
    """
    combined = threading.Event()

    def changed() -> None:
        bools = [e.is_set() for e in events]
        if any(bools):
            combined.set()
        else:
            combined.clear()

    for e in events:
        _orify(e, changed)

    changed()
    return combined

wait_event_async async 

wait_event_async(
    *events: Event, timeout: float | None = None
) -> Event

Wait for the first async event to be set.

Parameters:

  • events (Event, default: () ) –

    One or more events to wait on.

  • timeout (float | None, default: None ) –

    Maximum number of seconds to wait for an event to finish.

Returns:

  • Event –

    The first event to finish.

Raises:

  • TimeoutError –

    If no event finished within timeout seconds.

Source code in academy/event.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
async def wait_event_async(
    *events: asyncio.Event,
    timeout: float | None = None,
) -> asyncio.Event:
    """Wait for the first async event to be set.

    Args:
        events: One or more events to wait on.
        timeout: Maximum number of seconds to wait for an event to finish.

    Returns:
        The first event to finish.

    Raises:
        TimeoutError: If no event finished within `timeout` seconds.
    """
    tasks = {
        asyncio.create_task(event.wait(), name=f'or-event-waiter-{i}'): event
        for i, event in enumerate(events)
    }
    done, pending = await asyncio.wait(
        tasks.keys(),
        timeout=timeout,
        return_when=asyncio.FIRST_COMPLETED,
    )
    for task in pending:
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task

    if len(done) == 0:
        raise TimeoutError(f'No events were set within {timeout} seconds.')

    return tasks[done.pop()]