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
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.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:
            if self._interrupt is None:
                yielded = self._coro.send(value)
            else:
                exc = self._interrupt
                self._interrupt = None
                yielded = self._coro.throw(exc)

            yielded._add_waiter(self)

        except StopIteration:
            self._done = True

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

    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
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.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 
32
33
34
35
36
37
38
39
40
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 
61
62
63
64
65
66
67
68
69
70
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 
42
43
44
45
@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 
52
53
54
55
56
57
58
59
def timeout(self, delay: int | float):
    """
    Delay this process for a specified time.

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