A Ciw Implementation of SimPy’s Car Example With a Process Interaction

This post is a continuation of A Ciw Implementation of SimPy’s Car Example, and may be considered a prerequisite for reading this post.

The SimPy documentation extends the car example to have the cycling between driving and parking be interrupted. They introduce this driver Python generator:

import simpy
from typing import Generator
from .car import Car


def driver(env: simpy.Environment, car: Car) -> Generator[simpy.events.Timeout, None, None]:
    """
    Simulates a driver interrupting the action of a car after a timeout.

    Parameters:
    - env (simpy.Environment): The simulation environment in which the driver operates.
    - car (Car): The car object representing the vehicle whose action will be interrupted.

    Yields:
    - simpy.events.Timeout: A simulation event representing the passage of time.
    """
    yield env.timeout(3)
    car.action.interrupt()

And they upgrade their car process to a class definition:

from typing import Generator
from simpy import Environment, Process, Timeout, Interrupt


class Car(object):
    """A simulation of an electric car that alternates between parking and charging,
    and driving states within a simulation environment.
    """

    def __init__(self, env: Environment) -> None:
        """
        Initializes a new instance of the Car class.

        Parameters:
            env (Environment): The simulation environment.
        """
        self.env: Environment = env
        self.action: Process = env.process(self.run())

    def run(self) -> Generator[Timeout, None, None]:
        """
        The main behavior loop of the car, alternating between parking and charging,
        and driving states.
        """
        while True:
            print("Start parking and charging at %d" % self.env.now)
            charge_duration: float = 5

            try:
                yield self.env.process(self.charge(charge_duration))
            except Interrupt:
                print("Was interrupted. Hope, the battery is full enough ...")

            print("Start driving at %d" % self.env.now)
            trip_duration: float = 2
            yield self.env.timeout(trip_duration)

    def charge(self, duration: float) -> Generator[Timeout, None, None]:
        """
        Simulates the charging process of the car for a specified duration.

        Parameters:
            duration (float): The duration for which the car charges its battery.
        """
        yield self.env.timeout(duration)

The overall behavior of Car is not fundamentally different from before in terms of parking (with the added fluff of “charging”). The main thing that changes any of the consequent log of events is self.action: Process = env.process(self.run()). That is, at \(t=3\) we need to print: Was interrupted. Hope, the battery is full enough ....

This isn’t the most interesting example, and there are more interesting examples to get to in the SimPy docs, so let’s make this quick. All we need to do is update the logic in PrintDistWrapper.sample from last time so that it prints the message. Here is the source with a sufficient edit:

import ciw


class StepDist(ciw.dists.Distribution):
    """A custom distribution representing a step function with two values."""

    def __init__(self, first_value: float, second_value: float, step_t: float) -> None:
        """
        Initializes a new instance of the StepDist class.
        """
        super().__init__()
        self.first_value: float = first_value
        self.second_value: float = second_value
        self.step_t: float = step_t

    def sample(self, t: float = None, ind: ciw.Individual = None) -> float:
        """
        Samples the step function at a given time.
        """
        if t is not None and t <= self.step_t:
            return self.first_value
        else:
            return self.second_value


class PrintDistWrapper(ciw.dists.Distribution):
    """
    A wrapper class for a probability distribution that adds a print statement
    before sampling from the underlying distribution.
    """

    def __init__(self, dist: ciw.dists.Distribution, message) -> None:
        super().__init__()
        self.dist = dist
        self.message = message

    def sample(self, t=None, ind=None):
        """
        Generate a sample from the underlying distribution, printing a custom message.
        """
        if ind is not None and getattr(ind, "node", None) == 2 and t == 3:
            if t == 3:
                print("Was interrupted. Hope, the battery is full enough ...")
        else:
            print(self.message, t)

        return self.dist.sample(t, ind)


def build_network() -> ciw.network.Network:
    """Build the two-node car network used in the post."""
    arrival_dists = [ciw.dists.Sequential([0, float("inf")]), None]
    service_dists = [
        PrintDistWrapper(StepDist(3, 5, 3), "Start parking at"),
        PrintDistWrapper(ciw.dists.Deterministic(2), "Start driving at"),
    ]
    num_servers = [1, 1]
    R = [[0.0, 1.0], [1.0, 0.0]]

    return ciw.create_network(
        arrival_distributions=arrival_dists,
        service_distributions=service_dists,
        number_of_servers=num_servers,
        routing=R,
    )


def run_simulation(max_time: float = 15) -> ciw.Simulation:
    """Run the car simulation up to ``max_time`` and return it."""
    network = build_network()
    simulation = ciw.Simulation(network)
    simulation.simulate_until_max_time(max_time)
    return simulation


def main() -> None:
    run_simulation()


if __name__ == "__main__":
    main()

And the output will be the desired

Start parking at 0
Was interrupted. Hope, the battery is full enough ...
Start parking at 5
Start driving at 10
Start parking at 12