Skip to content

Process

Base class for processes.

Process

Bases: ABC

Base class for active processes.

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
class Process(ABC):
    """Base class for active processes."""

    def __init__(self, env: "Environment", *args: Any):
        """
        Construct a new process by performing common initialization,
        calling the user-defined `init()` method (no underscores),
        and registering the coroutine created by the `run()` method
        with the environment.

        Args:
            env: simulation environment.
            *args: to be passed to `init()` for custom initialization.
        """
        self.env = env
        self._interrupt = None

        self.init(*args)

        self._coro = self.run()
        self.env.immediate(self)

    def init(self, *args: Any, **kwargs: Any) -> None:
        """
        Default (do-nothing) post-initialization method.

        To satisfy type-checking, derived classes must also declare `*args`
        rather than listing specific parameters by name.
        """
        pass

    def interrupt(self, cause: Any):
        """
        Interrupt this process by raising an `Interrupt` exception the
        next time the process is scheduled to run.

        Args:
            cause: reason for interrupt (attacked to `Interrupt` exception).
        """
        self._interrupt = Interrupt(cause)

    @abstractmethod
    def run(self):
        """Actions for this process."""
        pass

__init__(env, *args)

Construct a new process by performing common initialization, calling the user-defined init() method (no underscores), and registering the coroutine created by the run() method with the environment.

Parameters:

Name Type Description Default
env Environment

simulation environment.

 required
*args Any

to be passed to init() for custom initialization.

 ()
Source code in src/asimpy/process.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def __init__(self, env: "Environment", *args: Any):
    """
    Construct a new process by performing common initialization,
    calling the user-defined `init()` method (no underscores),
    and registering the coroutine created by the `run()` method
    with the environment.

    Args:
        env: simulation environment.
        *args: to be passed to `init()` for custom initialization.
    """
    self.env = env
    self._interrupt = None

    self.init(*args)

    self._coro = self.run()
    self.env.immediate(self)

init(*args, **kwargs)

Default (do-nothing) post-initialization method.

To satisfy type-checking, derived classes must also declare *args rather than listing specific parameters by name.

Source code in src/asimpy/process.py 
34
35
36
37
38
39
40
41
def init(self, *args: Any, **kwargs: Any) -> None:
    """
    Default (do-nothing) post-initialization method.

    To satisfy type-checking, derived classes must also declare `*args`
    rather than listing specific parameters by name.
    """
    pass

interrupt(cause)

Interrupt this process by raising an Interrupt exception the next time the process is scheduled to run.

Parameters:

Name Type Description Default
cause Any

reason for interrupt (attacked to Interrupt exception).

 required
Source code in src/asimpy/process.py 
43
44
45
46
47
48
49
50
51
def interrupt(self, cause: Any):
    """
    Interrupt this process by raising an `Interrupt` exception the
    next time the process is scheduled to run.

    Args:
        cause: reason for interrupt (attacked to `Interrupt` exception).
    """
    self._interrupt = Interrupt(cause)

run() abstractmethod

Actions for this process.

Source code in src/asimpy/process.py 
53
54
55
56
@abstractmethod
def run(self):
    """Actions for this process."""
    pass