Environment

Simulation environment.

Environment

Simulation environment.

Source code in src/asimpy/environment.py

class Environment:
    """Simulation environment."""

    def __init__(self):
        self._now = 0
        self._pending = []
        self.active_process: "Process" | None = None
        """The process currently executing, or None between events."""

    @property
    def now(self):
        """Get the current simulated time."""
        return self._now

    def immediate(self, callback):
        """Schedule a callback for immediate execution."""
        self.schedule(self._now, callback)

    def run(self, until=None):
        """Run simulation."""
        while self._pending:
            pending = heapq.heappop(self._pending)
            if until is not None and pending.time > until:
                break
            result = pending.callback()
            if (result is not _NO_TIME) and (pending.time > self._now):
                self._now = pending.time

    def schedule(self, time, callback):
        """Schedule a callback to run at a specified future time."""
        heapq.heappush(self._pending, _Pending(time, callback))

    def timeout(self, delay):
        """Create delay."""
        return Timeout(self, delay)

    def __str__(self):
        return f"Env(t={self._now})"

active_process = None instance-attribute

The process currently executing, or None between events.

now property

Get the current simulated time.

immediate(callback)

Schedule a callback for immediate execution.

Source code in src/asimpy/environment.py

def immediate(self, callback):
    """Schedule a callback for immediate execution."""
    self.schedule(self._now, callback)

run(until=None)

Run simulation.

Source code in src/asimpy/environment.py

def run(self, until=None):
    """Run simulation."""
    while self._pending:
        pending = heapq.heappop(self._pending)
        if until is not None and pending.time > until:
            break
        result = pending.callback()
        if (result is not _NO_TIME) and (pending.time > self._now):
            self._now = pending.time

schedule(time, callback)

Schedule a callback to run at a specified future time.

Source code in src/asimpy/environment.py

def schedule(self, time, callback):
    """Schedule a callback to run at a specified future time."""
    heapq.heappush(self._pending, _Pending(time, callback))

timeout(delay)

Create delay.

Source code in src/asimpy/environment.py

def timeout(self, delay):
    """Create delay."""
    return Timeout(self, delay)