Production process optimisation¶

This production line simulation models the unit cost of a small manufacturing operation with the following components:

Components:¶

1. Input: Provides a fixed number of input items (10) per simulation step
2. InputStockpile:
   • Tracks inventory levels
   • Decrements based on machine operations (Machine 1: 5 items/step, Machine 2: 8 items/step)
   • Calculates storage costs ($10 per item above 50)
3. Controller (2 instances):
   • Controller 1: Activates Machine 1 when stockpile >= 30 items
   • Controller 2: Activates Machine 2 when stockpile >= 50 items
4. MachineCost (2 instances):
   • Machine 1: $100 per step when running
   • Machine 2: $200 per step when running
5. OutputStock: Tracks total items processed by both machines
6. TotalCost: Maintains running total of all costs (storage + machine operations)
7. CostPerUnit: Calculates the cost per unit produced (total cost ÷ total output)

Key Metrics:¶

• Production Efficiency: Percentage of input items successfully processed
• Cost Per Unit: Total operational cost divided by units produced
• Storage Utilization: How well the system manages inventory levels

The simulation runs for 1000 steps to provide stable long-term cost metrics. We can then use the model to find optimal values for the threshold levels on the two controllers.

In [ ]:

import typing as _t

from plugboard.component import Component, IOController as IO
from plugboard.schemas import ComponentArgsDict
from plugboard.connector import AsyncioConnector
from plugboard.process import LocalProcess
from plugboard.schemas import ConnectorSpec

import typing as _t from plugboard.component import Component, IOController as IO from plugboard.schemas import ComponentArgsDict from plugboard.connector import AsyncioConnector from plugboard.process import LocalProcess from plugboard.schemas import ConnectorSpec

Define the components for the model.

In [ ]:

class Input(Component):
    """Provides a fixed number of input items per step."""

    io = IO(outputs=["items"])

    def __init__(
        self,
        items_per_step: int = 10,
        total_steps: int = 1000,
        **kwargs: _t.Unpack[ComponentArgsDict],
    ) -> None:
        super().__init__(**kwargs)
        self._items_per_step = items_per_step
        self._total_steps = total_steps

    async def step(self) -> None:
        self.items = self._items_per_step
        if self._total_steps > 0:
            self._total_steps -= 1
        else:
            self.items = 0
            await self.io.close()


class InputStockpile(Component):
    """Tracks input stockpile, decrements based on machine operations, and calculates storage costs."""

    io = IO(
        inputs=["incoming_items", "machine1_running", "machine2_running"],
        outputs=["size", "storage_cost"],
    )

    def __init__(self, **kwargs: _t.Unpack[ComponentArgsDict]) -> None:
        super().__init__(**kwargs)
        self._size = 0

    async def step(self) -> None:
        # Add incoming items
        self._size += self.incoming_items

        # Remove items processed by machines
        if self.machine1_running:
            self._size = max(0, self._size - 5)  # Machine 1 processes 5 items
        if self.machine2_running:
            self._size = max(0, self._size - 8)  # Machine 2 processes 8 items

        # Calculate storage cost: $10 per item above 50
        storage_cost = max(0, self._size - 50) * 10

        self.size = self._size
        self.storage_cost = storage_cost


class Controller(Component):
    """Controls machine operation based on stockpile size."""

    io = IO(inputs=["stockpile_size"], outputs=["should_run"])

    def __init__(self, threshold: int = 30, **kwargs: _t.Unpack[ComponentArgsDict]) -> None:
        super().__init__(**kwargs)
        self._threshold = threshold

    async def step(self) -> None:
        self.should_run = self.stockpile_size >= self._threshold


class MachineCost(Component):
    """Calculates machine running costs."""

    io = IO(inputs=["is_running"], outputs=["cost"])

    def __init__(
        self, cost_per_step: float = 100.0, **kwargs: _t.Unpack[ComponentArgsDict]
    ) -> None:
        super().__init__(**kwargs)
        self._cost_per_step = cost_per_step

    async def step(self) -> None:
        self.cost = self._cost_per_step if self.is_running else 0.0


class OutputStock(Component):
    """Tracks total items processed by both machines."""

    io = IO(inputs=["machine1_running", "machine2_running"], outputs=["total_output"])

    def __init__(self, **kwargs: _t.Unpack[ComponentArgsDict]) -> None:
        super().__init__(**kwargs)
        self._total = 0

    async def step(self) -> None:
        if self.machine1_running:
            self._total += 5
        if self.machine2_running:
            self._total += 8
        self.total_output = self._total


class TotalCost(Component):
    """Keeps running total of all costs."""

    io = IO(inputs=["storage_cost", "machine1_cost", "machine2_cost"], outputs=["total_cost"])

    def __init__(self, **kwargs: _t.Unpack[ComponentArgsDict]) -> None:
        super().__init__(**kwargs)
        self._total = 0.0

    async def step(self) -> None:
        step_cost = self.storage_cost + self.machine1_cost + self.machine2_cost
        self._total += step_cost
        self.total_cost = self._total


class CostPerUnit(Component):
    """Calculates cost per unit produced."""

    io = IO(inputs=["total_cost", "total_output"], outputs=["cost_per_unit"])

    async def step(self) -> None:
        if self.total_output > 0:
            self.cost_per_unit = self.total_cost / self.total_output
        else:
            self.cost_per_unit = 0.0

class Input(Component): """Provides a fixed number of input items per step.""" io = IO(outputs=["items"]) def __init__( self, items_per_step: int = 10, total_steps: int = 1000, **kwargs: _t.Unpack[ComponentArgsDict], ) -> None: super().__init__(**kwargs) self._items_per_step = items_per_step self._total_steps = total_steps async def step(self) -> None: self.items = self._items_per_step if self._total_steps > 0: self._total_steps -= 1 else: self.items = 0 await self.io.close() class InputStockpile(Component): """Tracks input stockpile, decrements based on machine operations, and calculates storage costs.""" io = IO( inputs=["incoming_items", "machine1_running", "machine2_running"], outputs=["size", "storage_cost"], ) def __init__(self, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: super().__init__(**kwargs) self._size = 0 async def step(self) -> None: # Add incoming items self._size += self.incoming_items # Remove items processed by machines if self.machine1_running: self._size = max(0, self._size - 5) # Machine 1 processes 5 items if self.machine2_running: self._size = max(0, self._size - 8) # Machine 2 processes 8 items # Calculate storage cost: $10 per item above 50 storage_cost = max(0, self._size - 50) * 10 self.size = self._size self.storage_cost = storage_cost class Controller(Component): """Controls machine operation based on stockpile size.""" io = IO(inputs=["stockpile_size"], outputs=["should_run"]) def __init__(self, threshold: int = 30, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: super().__init__(**kwargs) self._threshold = threshold async def step(self) -> None: self.should_run = self.stockpile_size >= self._threshold class MachineCost(Component): """Calculates machine running costs.""" io = IO(inputs=["is_running"], outputs=["cost"]) def __init__( self, cost_per_step: float = 100.0, **kwargs: _t.Unpack[ComponentArgsDict] ) -> None: super().__init__(**kwargs) self._cost_per_step = cost_per_step async def step(self) -> None: self.cost = self._cost_per_step if self.is_running else 0.0 class OutputStock(Component): """Tracks total items processed by both machines.""" io = IO(inputs=["machine1_running", "machine2_running"], outputs=["total_output"]) def __init__(self, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: super().__init__(**kwargs) self._total = 0 async def step(self) -> None: if self.machine1_running: self._total += 5 if self.machine2_running: self._total += 8 self.total_output = self._total class TotalCost(Component): """Keeps running total of all costs.""" io = IO(inputs=["storage_cost", "machine1_cost", "machine2_cost"], outputs=["total_cost"]) def __init__(self, **kwargs: _t.Unpack[ComponentArgsDict]) -> None: super().__init__(**kwargs) self._total = 0.0 async def step(self) -> None: step_cost = self.storage_cost + self.machine1_cost + self.machine2_cost self._total += step_cost self.total_cost = self._total class CostPerUnit(Component): """Calculates cost per unit produced.""" io = IO(inputs=["total_cost", "total_output"], outputs=["cost_per_unit"]) async def step(self) -> None: if self.total_output > 0: self.cost_per_unit = self.total_cost / self.total_output else: self.cost_per_unit = 0.0

Now assemble into a Process and make connections between the components.

In [ ]:

connect = lambda in_, out_: AsyncioConnector(spec=ConnectorSpec(source=in_, target=out_))

process = LocalProcess(
    components=[
        # Input component provides items
        Input(name="input", items_per_step=10),
        # Stockpile manages inventory and storage costs
        # We need to include initial values here to resolve a circular graph
        InputStockpile(
            name="stockpile",
            initial_values={"machine1_running": [False], "machine2_running": [False]},
        ),
        # Controllers decide when machines should run
        Controller(name="controller1", threshold=30),  # Machine 1 threshold
        Controller(name="controller2", threshold=50),  # Machine 2 threshold
        # Machine cost components
        MachineCost(name="machine1_cost", cost_per_step=100.0),
        MachineCost(name="machine2_cost", cost_per_step=200.0),
        # Output tracking
        OutputStock(name="output_stock"),
        # Cost tracking and calculation
        TotalCost(name="total_cost"),
        CostPerUnit(name="cost_per_unit"),
    ],
    connectors=[
        # Input flow
        connect("input.items", "stockpile.incoming_items"),
        # Stockpile to controllers
        connect("stockpile.size", "controller1.stockpile_size"),
        connect("stockpile.size", "controller2.stockpile_size"),
        # Controllers to machine costs
        connect("controller1.should_run", "machine1_cost.is_running"),
        connect("controller2.should_run", "machine2_cost.is_running"),
        # Controllers to stockpile (for processing)
        connect("controller1.should_run", "stockpile.machine1_running"),
        connect("controller2.should_run", "stockpile.machine2_running"),
        # Controllers to output stock
        connect("controller1.should_run", "output_stock.machine1_running"),
        connect("controller2.should_run", "output_stock.machine2_running"),
        # All costs to total cost
        connect("stockpile.storage_cost", "total_cost.storage_cost"),
        connect("machine1_cost.cost", "total_cost.machine1_cost"),
        connect("machine2_cost.cost", "total_cost.machine2_cost"),
        # Total cost and output to cost per unit
        connect("total_cost.total_cost", "cost_per_unit.total_cost"),
        connect("output_stock.total_output", "cost_per_unit.total_output"),
    ],
)

print("Production line process created successfully!")
print(f"Process has {len(process.components)} components and {len(process.connectors)} connectors")

connect = lambda in_, out_: AsyncioConnector(spec=ConnectorSpec(source=in_, target=out_)) process = LocalProcess( components=[ # Input component provides items Input(name="input", items_per_step=10), # Stockpile manages inventory and storage costs # We need to include initial values here to resolve a circular graph InputStockpile( name="stockpile", initial_values={"machine1_running": [False], "machine2_running": [False]}, ), # Controllers decide when machines should run Controller(name="controller1", threshold=30), # Machine 1 threshold Controller(name="controller2", threshold=50), # Machine 2 threshold # Machine cost components MachineCost(name="machine1_cost", cost_per_step=100.0), MachineCost(name="machine2_cost", cost_per_step=200.0), # Output tracking OutputStock(name="output_stock"), # Cost tracking and calculation TotalCost(name="total_cost"), CostPerUnit(name="cost_per_unit"), ], connectors=[ # Input flow connect("input.items", "stockpile.incoming_items"), # Stockpile to controllers connect("stockpile.size", "controller1.stockpile_size"), connect("stockpile.size", "controller2.stockpile_size"), # Controllers to machine costs connect("controller1.should_run", "machine1_cost.is_running"), connect("controller2.should_run", "machine2_cost.is_running"), # Controllers to stockpile (for processing) connect("controller1.should_run", "stockpile.machine1_running"), connect("controller2.should_run", "stockpile.machine2_running"), # Controllers to output stock connect("controller1.should_run", "output_stock.machine1_running"), connect("controller2.should_run", "output_stock.machine2_running"), # All costs to total cost connect("stockpile.storage_cost", "total_cost.storage_cost"), connect("machine1_cost.cost", "total_cost.machine1_cost"), connect("machine2_cost.cost", "total_cost.machine2_cost"), # Total cost and output to cost per unit connect("total_cost.total_cost", "cost_per_unit.total_cost"), connect("output_stock.total_output", "cost_per_unit.total_output"), ], ) print("Production line process created successfully!") print(f"Process has {len(process.components)} components and {len(process.connectors)} connectors")

We can create a diagram of the process to make a visual check.

In [ ]:

# Visualize the process flow
from plugboard.diagram import MermaidDiagram

diagram_url = MermaidDiagram.from_process(process).url
print(diagram_url)

# Visualize the process flow from plugboard.diagram import MermaidDiagram diagram_url = MermaidDiagram.from_process(process).url print(diagram_url)

In [ ]:

# Run the simulation
async with process:
    await process.run()

# Run the simulation async with process: await process.run()

At the end of the simulation, the final cost per unit is available on the CostPerUnit component output.

In [ ]:

final_cost = process.components["cost_per_unit"].cost_per_unit
print(f"Final cost per unit: ${final_cost:.2f}")

final_cost = process.components["cost_per_unit"].cost_per_unit print(f"Final cost per unit: ${final_cost:.2f}")

Now suppose we want to build an optimisation to find the best values of the threshold settings on controller1 and controller2. Specifically we want to minimise cost_per_unit by choosing the settings on the two controllers. The easiest way to do this is to convert our model to Python code with an associated YAML config file. See:

• production_line.py; and
• production-line.yaml.

The easiest way to launch an optimisation job is via the CLI by running:

plugboard process tune production-line.yaml

This will use Optuna to explore the parameter space and report the controller thresholds that minimise cost per unit at the end of the run, for example:

Best parameters found:
Config: {'controller1.threshold': 10, 'controller2.threshold': 38} - Metrics: {'cost_per_unit.cost_per_unit': 22.491974317817014, 'timestamp': 1755022664, 
'checkpoint_dir_name': None, 'done': True, 'training_iteration': 1, 'trial_id': '287aff0b', 'date': '2025-08-12_19-17-44', 'time_this_iter_s': 2.224583864212036, 
'time_total_s': 2.224583864212036, 'pid': 94765, 'hostname': 'hostname.local', 'node_ip': '127.0.0.1', 'config': {'controller1.threshold': 10, 
'controller2.threshold': 38}, 'time_since_restore': 2.224583864212036, 'iterations_since_restore': 1, 'experiment_tag': 
'14_controller1_threshold=10,controller2_threshold=38'}