Skip to content

Environment

Simulation environment.

Environment

Simulation environment.

Source code in src/asimpy/environment.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class Environment:
    """Simulation environment."""

    def __init__(self, logging: bool = False):
        """
        Construct a new simulation environment.

        Args:
            logging: print log messages while executing.
        """
        self.now = 0
        self._queue = []
        self._logging = logging

    def immediate(self, proc: Process):
        """
        Start a new process immediately.

        Args:
            proc: `Process`-derived object to schedule and run.
        """
        self.schedule(self.now, proc)

    def schedule(self, time: float | int, proc: Process):
        """
        Schedule a process to run at a specified time (*not* after a delay).

        Args:
            time: when the process should be scheduled to run.
            proc: `Process`-derived object to run.
        """
        heapq.heappush(self._queue, _Pending(time, proc))

    def sleep(self, delay: float | int) -> "_Sleep":
        """
        Suspend the caller for a specified length of time.

        Args:
            delay: how long to sleep.

        Returns: awaitable representing delay.
        """
        return _Sleep(self, delay)

    def run(self, until: float | int | None = None):
        """
        Run the whole simulation.

        Args:
            until: when to stop (run forever if not provided).
        """
        while self._queue:
            if self._logging:
                print(self)

            pending = heapq.heappop(self._queue)

            if until is not None and pending.time > until:
                break

            self.now = pending.time
            proc = pending.proc

            try:
                if proc._interrupt is None:
                    awaited = proc._coro.send(None)
                else:
                    exc, proc._interrupt = proc._interrupt, None
                    awaited = proc._coro.throw(exc)

                awaited.act(proc)

            except StopIteration:
                continue

    def __str__(self) -> str:
        """
        Format environment as printable string.

        Returns: string representation of environment time and queue.
        """
        return f"Env(t={self.now}, {' | '.join(str(p) for p in self._queue)})"

__init__(logging=False)

Construct a new simulation environment.

Parameters:

Name Type Description Default
logging bool

print log messages while executing.

 False
Source code in src/asimpy/environment.py 
12
13
14
15
16
17
18
19
20
21
def __init__(self, logging: bool = False):
    """
    Construct a new simulation environment.

    Args:
        logging: print log messages while executing.
    """
    self.now = 0
    self._queue = []
    self._logging = logging

__str__()

Format environment as printable string.

Returns: string representation of environment time and queue.

Source code in src/asimpy/environment.py 
84
85
86
87
88
89
90
def __str__(self) -> str:
    """
    Format environment as printable string.

    Returns: string representation of environment time and queue.
    """
    return f"Env(t={self.now}, {' | '.join(str(p) for p in self._queue)})"

immediate(proc)

Start a new process immediately.

Parameters:

Name Type Description Default
proc Process

Process-derived object to schedule and run.

 required
Source code in src/asimpy/environment.py 
23
24
25
26
27
28
29
30
def immediate(self, proc: Process):
    """
    Start a new process immediately.

    Args:
        proc: `Process`-derived object to schedule and run.
    """
    self.schedule(self.now, proc)

run(until=None)

Run the whole simulation.

Parameters:

Name Type Description Default
until float | int | None

when to stop (run forever if not provided).

 None
Source code in src/asimpy/environment.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def run(self, until: float | int | None = None):
    """
    Run the whole simulation.

    Args:
        until: when to stop (run forever if not provided).
    """
    while self._queue:
        if self._logging:
            print(self)

        pending = heapq.heappop(self._queue)

        if until is not None and pending.time > until:
            break

        self.now = pending.time
        proc = pending.proc

        try:
            if proc._interrupt is None:
                awaited = proc._coro.send(None)
            else:
                exc, proc._interrupt = proc._interrupt, None
                awaited = proc._coro.throw(exc)

            awaited.act(proc)

        except StopIteration:
            continue

schedule(time, proc)

Schedule a process to run at a specified time (not after a delay).

Parameters:

Name Type Description Default
time float | int

when the process should be scheduled to run.

 required
proc Process

Process-derived object to run.

 required
Source code in src/asimpy/environment.py 
32
33
34
35
36
37
38
39
40
def schedule(self, time: float | int, proc: Process):
    """
    Schedule a process to run at a specified time (*not* after a delay).

    Args:
        time: when the process should be scheduled to run.
        proc: `Process`-derived object to run.
    """
    heapq.heappush(self._queue, _Pending(time, proc))

sleep(delay)

Suspend the caller for a specified length of time.

Parameters:

Name Type Description Default
delay float | int

how long to sleep.

 required

Returns: awaitable representing delay.

Source code in src/asimpy/environment.py 
42
43
44
45
46
47
48
49
50
51
def sleep(self, delay: float | int) -> "_Sleep":
    """
    Suspend the caller for a specified length of time.

    Args:
        delay: how long to sleep.

    Returns: awaitable representing delay.
    """
    return _Sleep(self, delay)