Assembly Line

tutorial/assembly_line.py is a complete manufacturing-line model. It is larger than the tiny queue examples because it combines dynamic parts, station inboxes, exclusive processing resources, per-station measurements, whole-system measurements, and graph output in one runnable script.

Run it from the repository root with plotting dependencies installed:

uv run --extra plot python tutorial/assembly_line.py

The simulated system

The model represents a three-station assembly line:

• parts arrive after exponentially distributed interarrival times,

• each part enters Station 1, Station 2, and Station 3 in order,

• each station has an inbox and one processing resource,

• a station takes a part from its inbox, records its waiting time, acquires the resource, holds for an exponentially distributed processing time, releases the resource, and forwards the part downstream,

• the finished-parts sink records total cycle time and removes the part from the active system count.

The station processing means are 5, 7, and 4 minutes. The interarrival mean is 3 minutes, so the model is intentionally busy: the middle station is the bottleneck, queues can form, and utilization is an important output.

Model structure

Part is a Struct that stores the part id, arrival time, and the time when the part entered its current station. Those fields travel with the spawned part process.

Station is a reusable Component. Each station owns:

• inbox, a Store of waiting part handles,

• resource, a Resource representing the processor,

• wait_time, a Dataset for station waiting times,

• downstream, a reference to the next station or finished-parts sink.

The three stations are declared as fields on AssemblyLine and chained together through their downstream references. That keeps the station logic generic: Station 1, Station 2, and Station 3 all run the same process body with different processing-time parameters.

Process flow

The model has three main process patterns.

arrivals waits for the next arrival, spawns a part lifecycle process, and records the part’s id and system arrival time.

part_lifecycle increments the system population, stamps the current station entry time, and puts the current process handle into Station 1’s inbox. After that, station processes move the part by passing the same handle through stores.

Station.server loops forever. It takes a part handle from the station inbox, computes how long the part waited since its previous station-entry stamp, processes the part while holding the station resource, updates the station-entry stamp, and puts the handle into the next inbox.

FinishedParts.finish is the downstream endpoint. It records the total cycle time, decrements the system population, and hands the part to reclaim so the spawned process can be despawned.

Measurements

The example records both local station metrics and whole-system metrics.

Each station collects:

• average wait time from its wait_time dataset,

• utilization from the time average of its processing resource.

The model-level collector records:

• total parts produced,

• average and maximum cycle time,

• throughput rate,

• average, maximum, and final number of parts in the system.

The collector also writes raw cycle-time, wait-time, and system-population series to temporary files. The plotting helper reads those files back into NumPy arrays to produce histograms, utilization bars, and a step plot of parts in the system. main currently writes the process graph by default; the plot_results call is present but commented out.

Process graph output

After the experiment runs, plot_process_dag writes Mermaid and Graphviz representations under tutorial/assembly_line_plots/. If Graphviz dot is available, it also writes PNG and SVG renderings. This is useful for checking that the arrival process, part lifecycle, station servers, and finished-parts sink are connected the way the model intends.

Full source

"""
Cimba version of the SimPy/salabim assembly-line comparison model.

Run from the repository root:

    uv run --extra plot python tutorial/assembly_line.py
"""

from pathlib import Path
import shutil
import subprocess
from tempfile import TemporaryDirectory

import numpy as np

try:
    import matplotlib.pyplot as plt
except ModuleNotFoundError as exc:
    raise SystemExit(
        "This script needs matplotlib. Run it with: "
        "uv run --extra plot python tutorial/assembly_line.py"
    ) from exc

import cimba as cp
import cimba.random as random
import cimba.sim as sim


RANDOM_SEED = 45
STATION_1_NAME = "Station 1"
STATION_2_NAME = "Station 2"
STATION_3_NAME = "Station 3"
STATION_NAMES = (STATION_1_NAME, STATION_2_NAME, STATION_3_NAME)
NUM_STATIONS = 3
STATION_1_MEAN = 5.0
STATION_2_MEAN = 7.0
STATION_3_MEAN = 4.0
INTERARRIVAL_TIME = 3.0
SIMULATION_TIME = 10_000.0
PLOT_DIR = Path(__file__).with_name("assembly_line_plots")


class Part(sim.Struct):
    part_id: int
    arrival_system: float
    station_entry: float


class Station(sim.Component):
    avg_wait_time: sim.Output
    utilization: sim.Output
    inbox: sim.Store
    downstream: sim.Ref[sim.Component]
    resource: sim.Resource
    wait_time: sim.Dataset

    def __init__(self, name: str, mean_processing_time: float, *,
                 downstream=None):
        self.name = name
        self.mean_processing_time = mean_processing_time
        if downstream is not None:
            self.downstream = downstream

    @sim.collect
    def station_stats(self, env):
        self.avg_wait_time = self.wait_time.mean()
        self.utilization = 100.0 * self.resource.mean_in_use()

    @sim.process
    def server(self, env):
        while True:
            # Take a part from the inbox.
            handle = self.inbox.take()
            item = Part(handle)

            # Update the part's wait time.
            wait_time = sim.now() - item.station_entry
            self.wait_time.add(wait_time)

            # Hold the resource for the processing time.
            self.resource.acquire()
            sim.hold(random.exponential(self.mean_processing_time))
            self.resource.release()

            # Update the part's station entry time to the current time.
            item.station_entry = sim.now()

            # Put the part in the downstream station's inbox.
            self.downstream.inbox.put(handle)


class FinishedParts(sim.Component):
    inbox: sim.Store
    departed: sim.Store

    @sim.process
    def finish(self, env):
        while True:
            handle = self.inbox.take()
            item = Part(handle)
            env.cycle_time.add(sim.now() - item.arrival_system)
            env.system.get(1)
            self.departed.put(handle)

    @sim.process
    def reclaim(self, env):
        while True:
            sim.despawn(self.departed.take())


class AssemblyLine(sim.Model):
    total_parts_produced: sim.Output
    avg_cycle_time: sim.Output
    max_cycle_time: sim.Output
    throughput_rate: sim.Output
    avg_number_in_system: sim.Output
    max_number_in_system: sim.Output
    final_number_in_system: sim.Output

    generated_parts: sim.State
    part_lifecycle: sim.Spawnable
    system: sim.Queue
    cycle_time: sim.Dataset
    finished_parts: FinishedParts = FinishedParts()
    station_3: Station = Station(STATION_3_NAME, STATION_3_MEAN,
                                 downstream=finished_parts)
    station_2: Station = Station(STATION_2_NAME, STATION_2_MEAN,
                                 downstream=station_3)
    station_1: Station = Station(STATION_1_NAME, STATION_1_MEAN,
                                 downstream=station_2)


def build_model(raw_dir: Path) -> AssemblyLine:
    cycle_file = sim.log_text(str(raw_dir / "cycle_times.txt"))
    wait_files = (
        sim.log_text(str(raw_dir / "station_1_wait_times.txt")),
        sim.log_text(str(raw_dir / "station_2_wait_times.txt")),
        sim.log_text(str(raw_dir / "station_3_wait_times.txt")),
    )
    system_file = sim.log_text(str(raw_dir / "number_in_system.txt"))

    model = AssemblyLine("assembly_line")

    @model.process
    def arrivals(env: AssemblyLine):
        while True:
            sim.hold(random.exponential(INTERARRIVAL_TIME))
            handle = sim.spawn(env.part_lifecycle, env)
            part = Part(handle)

            env.generated_parts += 1
            part.part_id = env.generated_parts
            part.arrival_system = sim.now()

    @model.process
    def part_lifecycle(env: AssemblyLine, item: Part):
        env.system.put(1)
        item.station_entry = sim.now()
        env.station_1.inbox.put(sim.current())

    @model.collect
    def collect_stats(env: AssemblyLine):
        completed = env.cycle_time.count()
        env.total_parts_produced = completed
        env.avg_cycle_time = env.cycle_time.mean()
        env.max_cycle_time = env.cycle_time.max()
        env.throughput_rate = completed / env.duration_s
        env.avg_number_in_system = env.system.mean_level()
        env.max_number_in_system = env.system.history().max()
        env.final_number_in_system = env.system.level()

        env.cycle_time.print_file(cycle_file, 0)
        env.station_1.wait_time.print_file(wait_files[0], 0)
        env.station_2.wait_time.print_file(wait_files[1], 0)
        env.station_3.wait_time.print_file(wait_files[2], 0)
        env.system.history().print_file(system_file, 0)

    return model


def read_values(path: Path) -> np.ndarray:
    text = path.read_text().strip()
    if not text:
        return np.array([], dtype=float)
    return np.atleast_1d(np.loadtxt(path, dtype=float))


def read_timeseries(path: Path) -> tuple[np.ndarray, np.ndarray]:
    text = path.read_text().strip()
    if not text:
        empty = np.array([], dtype=float)
        return empty, empty
    rows = np.atleast_2d(np.loadtxt(path, dtype=float))
    return rows[:, 0], rows[:, 1]


def station_values(exp: sim.Experiment, field: str) -> np.ndarray:
    return np.array(
        [
            exp[f"station_1__{field}"][0],
            exp[f"station_2__{field}"][0],
            exp[f"station_3__{field}"][0],
        ],
        dtype=float,
    )


def print_results(exp: sim.Experiment) -> None:
    wait_times = station_values(exp, "avg_wait_time")
    utilization = station_values(exp, "utilization")

    print("--- Simulation Finished ---")
    print("\n--- Simulation Results Analysis (Cimba) ---")
    print(f"Total parts produced: {int(exp['total_parts_produced'][0])}")
    print(
        "Average cycle time per part: "
        f"{exp['avg_cycle_time'][0]:.2f} minutes"
    )
    print(
        "Maximum cycle time per part: "
        f"{exp['max_cycle_time'][0]:.2f} minutes"
    )
    print(f"Throughput rate: {exp['throughput_rate'][0]:.2f} parts per minute")
    for i in range(NUM_STATIONS):
        print(
            f"{STATION_NAMES[i]} - Average Wait Time: "
            f"{wait_times[i]:.2f} minutes"
        )
        print(f"{STATION_NAMES[i]} - Utilization: {utilization[i]:.2f}%")
    print(f"Average number in system: {exp['avg_number_in_system'][0]:.2f}")
    print(f"Maximum number in system: {exp['max_number_in_system'][0]:.0f}")
    print(f"Parts still in system: {exp['final_number_in_system'][0]:.0f}")


def plot_process_dag(model: AssemblyLine) -> None:
    PLOT_DIR.mkdir(parents=True, exist_ok=True)
    graph = model.process_dag()
    mermaid_path = PLOT_DIR / "process_dag.mmd"
    dot_path = PLOT_DIR / "process_dag.dot"

    mermaid_path.write_text(graph.to_mermaid(direction="TD") + "\n")
    dot_path.write_text(graph.to_dot(rankdir="TB") + "\n")

    dot = shutil.which("dot")
    if dot is not None:
        subprocess.run(
            [dot, "-Tpng", str(dot_path), "-o",
             str(PLOT_DIR / "process_dag.png")],
            check=True,
        )
        subprocess.run(
            [dot, "-Tsvg", str(dot_path), "-o",
             str(PLOT_DIR / "process_dag.svg")],
            check=True,
        )

    print(f"\nSaved process DAG in {PLOT_DIR}")


def plot_results(exp: sim.Experiment, raw_dir: Path) -> None:
    PLOT_DIR.mkdir(parents=True, exist_ok=True)

    cycle_times = read_values(raw_dir / "cycle_times.txt")
    wait_times = [
        read_values(raw_dir / f"station_{i + 1}_wait_times.txt")
        for i in range(NUM_STATIONS)
    ]
    time_points, parts_in_system = read_timeseries(
        raw_dir / "number_in_system.txt"
    )

    plt.figure(figsize=(10, 6))
    plt.hist(cycle_times, bins=20, color="skyblue", edgecolor="black")
    avg_cycle_time = exp["avg_cycle_time"][0]
    plt.axvline(
        avg_cycle_time,
        color="red",
        linestyle="dashed",
        linewidth=2,
        label=f"Avg: {avg_cycle_time:.2f}",
    )
    plt.title("Distribution of Part Cycle Times")
    plt.xlabel("Cycle Time (minutes)")
    plt.ylabel("Number of Parts")
    plt.legend()
    plt.tight_layout()
    plt.savefig(PLOT_DIR / "cycle_times.png", dpi=150)

    fig, axes = plt.subplots(
        NUM_STATIONS, 1, figsize=(10, 4 * NUM_STATIONS), sharex=True
    )
    fig.suptitle("Distribution of Waiting Times at Each Station", fontsize=16)
    avg_wait_times = station_values(exp, "avg_wait_time")
    for i, ax in enumerate(axes):
        ax.hist(wait_times[i], bins=15, color="lightcoral", edgecolor="black")
        ax.axvline(
            avg_wait_times[i],
            color="blue",
            linestyle="dashed",
            linewidth=2,
            label=f"Avg: {avg_wait_times[i]:.2f}",
        )
        ax.set_title(f"{STATION_NAMES[i]} Waiting Times")
        ax.set_ylabel("Number of Parts")
        ax.legend()
    axes[-1].set_xlabel("Waiting Time (minutes)")
    plt.tight_layout(rect=(0, 0, 1, 0.96))
    plt.savefig(PLOT_DIR / "station_wait_times.png", dpi=150)

    station_utilization = station_values(exp, "utilization")
    plt.figure(figsize=(10, 6))
    plt.bar(STATION_NAMES, station_utilization, color="mediumseagreen")
    plt.title("Average Station Utilization")
    plt.xlabel("Station")
    plt.ylabel("Utilization (%)")
    plt.ylim(0, 100)
    for i, value in enumerate(station_utilization):
        plt.text(i, value + 1, f"{value:.2f}%", ha="center")
    plt.tight_layout()
    plt.savefig(PLOT_DIR / "station_utilization.png", dpi=150)

    plt.figure(figsize=(12, 6))
    plt.step(time_points, parts_in_system, where="post", color="dodgerblue")
    avg_number = exp["avg_number_in_system"][0]
    plt.axhline(
        avg_number,
        color="blue",
        linewidth=1,
        label=f"Mean: {avg_number:.2f}",
    )
    plt.title("Number of Parts in the System Over Time")
    plt.xlabel("Time (minutes)")
    plt.ylabel("Number of Parts")
    plt.legend()
    plt.grid(True)
    plt.tight_layout()
    plt.savefig(PLOT_DIR / "number_in_system.png", dpi=150)

    print(f"\nSaved plots in {PLOT_DIR}")
    if "agg" not in plt.get_backend().lower():
        plt.show()


def main() -> None:
    print("--- Assembly Line Simulation Starting (Cimba) ---")
    print(f"cimba {cp.version()}")

    with TemporaryDirectory() as temp_dir:
        raw_dir = Path(temp_dir)
        model = build_model(raw_dir)
        exp = model.experiment(
            replications=1000,
            duration=SIMULATION_TIME,
            warmup=0.0,
            seed=RANDOM_SEED,
        )
        failures = exp.run()
        if failures:
            raise RuntimeError(f"{failures} trial(s) failed")

        print_results(exp)
        plot_process_dag(model)
        # plot_results(exp, raw_dir)

    print("\n--- End of Cimba Script ---")


if __name__ == "__main__":
    main()