Skip to content

Process

Base class for active process.

Process

Bases: ABC

Abstract base class for active process.

Source code in src/asimpy/process.py 
 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
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
class Process(ABC):
    """Abstract base class for active process."""

    def __init__(self, env: "Environment", *args: Any, **kwargs: Any):
        """
        Construct new process.

        Args:
            env: simulation environment.
            args: extra constructor arguments passed to `init()`.
        """
        self._env = env
        self._done = False
        self._interrupt = None
        self._current_event = None

        self.init(*args, **kwargs)

        self._coro = self.run()
        self._env.immediate(self._loop)

    def init(self, *args: Any, **kwargs: Any):
        """
        Extra construction after generic setup but before coroutine created.

        Args:
            args: extra constructor arguments passed to `init()`.
            kwargs: extra construct arguments passed to `init()`.
        """
        pass

    @abstractmethod
    def run(self):
        """Implementation of process behavior."""
        pass

    @property
    def now(self):
        """Shortcut to access simulation time."""
        return self._env.now

    def timeout(self, delay: int | float):
        """
        Delay this process for a specified time.

        Args:
            delay: how long to wait.
        """
        return self._env.timeout(delay)

    def interrupt(self, cause: Any):
        """
        Interrupt this process

        Args:
            cause: reason for interrupt.
        """
        if not self._done:
            self._interrupt = Interrupt(cause)
            self._env.immediate(self._loop)

    def _loop(self, value=None):
        if self._done:
            return

        try:
            self._env.active_process = self
            if self._interrupt is None:
                yielded = self._coro.send(value)
            else:
                if self._current_event is not None:
                    self._current_event.cancel()
                    self._current_event = None
                exc = self._interrupt
                self._interrupt = None
                yielded = self._coro.throw(exc)

            self._current_event = yielded
            yielded._add_waiter(self.resume)

        except StopIteration:
            self._done = True
            self._current_event = None

        except Exception as exc:
            self._done = True
            self._current_event = None
            raise exc

        finally:
            self._env.active_process = None

    def resume(self, value=None):
        if not self._done:
            self._env.immediate(lambda: self._loop(value))

now property

Shortcut to access simulation time.

__init__(env, *args, **kwargs)

Construct new process.

Parameters:

Name Type Description Default
env Environment

simulation environment.

 required
args Any

extra constructor arguments passed to init().

 ()
Source code in src/asimpy/process.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, env: "Environment", *args: Any, **kwargs: Any):
    """
    Construct new process.

    Args:
        env: simulation environment.
        args: extra constructor arguments passed to `init()`.
    """
    self._env = env
    self._done = False
    self._interrupt = None
    self._current_event = None

    self.init(*args, **kwargs)

    self._coro = self.run()
    self._env.immediate(self._loop)

init(*args, **kwargs)

Extra construction after generic setup but before coroutine created.

Parameters:

Name Type Description Default
args Any

extra constructor arguments passed to init().

 ()
kwargs Any

extra construct arguments passed to init().

 {}
Source code in src/asimpy/process.py 
33
34
35
36
37
38
39
40
41
def init(self, *args: Any, **kwargs: Any):
    """
    Extra construction after generic setup but before coroutine created.

    Args:
        args: extra constructor arguments passed to `init()`.
        kwargs: extra construct arguments passed to `init()`.
    """
    pass

interrupt(cause)

Interrupt this process

Parameters:

Name Type Description Default
cause Any

reason for interrupt.

 required
Source code in src/asimpy/process.py 
62
63
64
65
66
67
68
69
70
71
def interrupt(self, cause: Any):
    """
    Interrupt this process

    Args:
        cause: reason for interrupt.
    """
    if not self._done:
        self._interrupt = Interrupt(cause)
        self._env.immediate(self._loop)

run() abstractmethod

Implementation of process behavior.

Source code in src/asimpy/process.py 
43
44
45
46
@abstractmethod
def run(self):
    """Implementation of process behavior."""
    pass

timeout(delay)

Delay this process for a specified time.

Parameters:

Name Type Description Default
delay int | float

how long to wait.

 required
Source code in src/asimpy/process.py 
53
54
55
56
57
58
59
60
def timeout(self, delay: int | float):
    """
    Delay this process for a specified time.

    Args:
        delay: how long to wait.
    """
    return self._env.timeout(delay)