Event

Events.

NO_TIME = object() module-attribute

Sentinel indicating that time should not advance.

Event

Manage an event.

Source code in src/asimpy/event.py

class Event:
    """Manage an event."""

    def __init__(self, env: "Environment"):
        """
        Construct a new event.

        Args:
            env: simulation environment.
        """
        self._env = env
        self._triggered = False
        self._cancelled = False
        self._value = None
        self._waiters = []
        self._on_cancel = None

    def succeed(self, value: Any = None):
        """
        Handle case of event succeeding.

        Args:
            value: value associated with event.
        """
        if self._triggered or self._cancelled:
            return
        self._triggered = True
        self._value = value
        for proc in self._waiters:
            proc.resume(value)
        self._waiters.clear()

    def cancel(self):
        """Cancel event."""
        if self._triggered or self._cancelled:
            return
        self._cancelled = True
        self._waiters.clear()
        if self._on_cancel:
            self._on_cancel()

    def _add_waiter(self, proc):
        if self._triggered:
            proc.resume(self._value)
        elif not self._cancelled:
            self._waiters.append(proc)

    def __await__(self):
        value = yield self
        return value

__init__(env)

Construct a new event.

Parameters:

Name	Type	Description	Default
env	Environment	simulation environment.	required

Source code in src/asimpy/event.py

def __init__(self, env: "Environment"):
    """
    Construct a new event.

    Args:
        env: simulation environment.
    """
    self._env = env
    self._triggered = False
    self._cancelled = False
    self._value = None
    self._waiters = []
    self._on_cancel = None

cancel()

Cancel event.

Source code in src/asimpy/event.py

def cancel(self):
    """Cancel event."""
    if self._triggered or self._cancelled:
        return
    self._cancelled = True
    self._waiters.clear()
    if self._on_cancel:
        self._on_cancel()

succeed(value=None)

Handle case of event succeeding.

Parameters:

Name	Type	Description	Default
value	Any	value associated with event.	None

Source code in src/asimpy/event.py

def succeed(self, value: Any = None):
    """
    Handle case of event succeeding.

    Args:
        value: value associated with event.
    """
    if self._triggered or self._cancelled:
        return
    self._triggered = True
    self._value = value
    for proc in self._waiters:
        proc.resume(value)
    self._waiters.clear()