Factory simulation with maintenance optimisation¶

This demo simulates a factory containing 5 machines over 1 year (365 days). The simulation runs in steps of 1 day, with each machine exhibiting realistic behaviour:

Components:¶

1. Iterator (clock): Drives the simulation by emitting day numbers from 0 to 364
2. Machine (5 instances, each with different reliability characteristics):
   • Produces $10,000 of value per day when running
   • Has a probability of random breakdown modelled as a sigmoid function of days since last maintenance — near zero soon after maintenance, rising to near 1 over time
   • Stops for 5 days when it breaks down
   • Has a proactive maintenance schedule: stops for 1 day at regular intervals
3. Factory: Aggregates daily output from all machines and tracks the cumulative total value

Machine reliability profiles:¶

Machine	Sigmoid steepness	Sigmoid midpoint	Profile
1	0.10	50 days	Reliable
2	0.12	45 days	Average
3	0.15	40 days	Average
4	0.18	35 days	Fragile
5	0.20	30 days	Very fragile

We can then use the model to find optimal proactive maintenance intervals for each machine to maximise total output.

In [ ]:

# Install plugboard and dependencies for Google Colab
!pip install -q plugboard[ray]

# Install plugboard and dependencies for Google Colab !pip install -q plugboard[ray]

In [ ]:

from plugboard.connector import AsyncioConnector
from plugboard.process import LocalProcess
from plugboard.schemas import ConnectorSpec

from factory_simulation import Iterator, Machine, Factory, MACHINE_CONFIGS

from plugboard.connector import AsyncioConnector from plugboard.process import LocalProcess from plugboard.schemas import ConnectorSpec from factory_simulation import Iterator, Machine, Factory, MACHINE_CONFIGS

The components are defined in factory_simulation.py:

• Iterator emits a sequence of day numbers and closes once the configured number of days is reached.
• Machine tracks its own state (running, broken, or under maintenance). At each step it checks whether it is due for proactive maintenance, whether a random breakdown occurs (using a sigmoid probability curve), and outputs its daily production value.
• Factory sums daily values from all machines and maintains a running total.

Each machine has different sigmoid parameters that control its breakdown probability curve, making some machines more reliable than others.

Now assemble the components into a Process and connect them together. Each machine receives the day count from the clock, and sends its daily output to the factory aggregator.

In [ ]:

connect = lambda s, t: AsyncioConnector(spec=ConnectorSpec(source=s, target=t))

components = [Iterator(name="clock", num_days=365)]
connectors = []

for i, cfg in enumerate(MACHINE_CONFIGS, start=1):
    components.append(Machine(name=f"machine_{i}", maintenance_interval=30, **cfg))
    connectors.append(connect("clock.day", f"machine_{i}.day"))
    connectors.append(connect(f"machine_{i}.daily_value", f"factory.value_{i}"))

components.append(Factory(name="factory"))

process = LocalProcess(components=components, connectors=connectors)

print(f"Process has {len(process.components)} components and {len(process.connectors)} connectors")

connect = lambda s, t: AsyncioConnector(spec=ConnectorSpec(source=s, target=t)) components = [Iterator(name="clock", num_days=365)] connectors = [] for i, cfg in enumerate(MACHINE_CONFIGS, start=1): components.append(Machine(name=f"machine_{i}", maintenance_interval=30, **cfg)) connectors.append(connect("clock.day", f"machine_{i}.day")) connectors.append(connect(f"machine_{i}.daily_value", f"factory.value_{i}")) components.append(Factory(name="factory")) process = LocalProcess(components=components, connectors=connectors) 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)

Run the simulation for 365 days with a default maintenance interval of 30 days for all machines.

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, we can read the cumulative total value from the Factory component.

In [ ]:

total_value = process.components["factory"].total_value
max_possible = 365 * 5 * 10_000
print(f"Total factory output over 365 days: ${total_value:,.0f}")
print(f"Maximum possible output: ${max_possible:,.0f}")
print(f"Efficiency: {total_value / max_possible:.1%}")

total_value = process.components["factory"].total_value max_possible = 365 * 5 * 10_000 print(f"Total factory output over 365 days: ${total_value:,.0f}") print(f"Maximum possible output: ${max_possible:,.0f}") print(f"Efficiency: {total_value / max_possible:.1%}")

Now suppose we want to find the optimal proactive maintenance interval for each machine. Maintaining too frequently wastes productive days, but maintaining too rarely leads to expensive breakdowns (5 days of downtime vs 1 day for scheduled maintenance).

We can set up an optimisation to maximise total_value by varying the maintenance_interval argument on each machine. First, we build a ProcessSpec that describes the simulation programmatically.

In [ ]:

from plugboard.schemas import ProcessArgsSpec, ProcessSpec

spec_components = [
    {"type": "factory_simulation.Iterator", "args": {"name": "clock", "num_days": 365}},
]
spec_connectors = []

for i, cfg in enumerate(MACHINE_CONFIGS, start=1):
    spec_components.append(
        {
            "type": "factory_simulation.Machine",
            "args": {"name": f"machine_{i}", "maintenance_interval": 30, **cfg},
        }
    )
    spec_connectors.append({"source": "clock.day", "target": f"machine_{i}.day"})
    spec_connectors.append({"source": f"machine_{i}.daily_value", "target": f"factory.value_{i}"})

spec_components.append({"type": "factory_simulation.Factory", "args": {"name": "factory"}})

spec = ProcessSpec(
    args=ProcessArgsSpec(components=spec_components, connectors=spec_connectors),
    type="plugboard.process.LocalProcess",
)

from plugboard.schemas import ProcessArgsSpec, ProcessSpec spec_components = [ {"type": "factory_simulation.Iterator", "args": {"name": "clock", "num_days": 365}}, ] spec_connectors = [] for i, cfg in enumerate(MACHINE_CONFIGS, start=1): spec_components.append( { "type": "factory_simulation.Machine", "args": {"name": f"machine_{i}", "maintenance_interval": 30, **cfg}, } ) spec_connectors.append({"source": "clock.day", "target": f"machine_{i}.day"}) spec_connectors.append({"source": f"machine_{i}.daily_value", "target": f"factory.value_{i}"}) spec_components.append({"type": "factory_simulation.Factory", "args": {"name": "factory"}}) spec = ProcessSpec( args=ProcessArgsSpec(components=spec_components, connectors=spec_connectors), type="plugboard.process.LocalProcess", )

Now configure the Tuner to search for the best maintenance_interval for each machine (between 5 and 60 days), maximising the factory's total_value.

In [ ]:

from plugboard.schemas import IntParameterSpec, ObjectiveSpec
from plugboard.tune import Tuner

tuner = Tuner(
    objective=ObjectiveSpec(
        object_type="component",
        object_name="factory",
        field_type="field",
        field_name="total_value",
    ),
    parameters=[
        IntParameterSpec(
            object_type="component",
            object_name=f"machine_{i}",
            field_type="arg",
            field_name="maintenance_interval",
            lower=5,
            upper=60,
        )
        for i in range(1, 6)
    ],
    num_samples=40,
    max_concurrent=4,
    mode="max",
)

result = tuner.run(spec=spec)

print("=== Optimisation Results ===")
for i in range(1, 6):
    key = f"component.machine_{i}.arg.maintenance_interval"
    print(f"  Machine {i} optimal maintenance interval: {result.config[key]} days")
print(f"  Maximum total value: ${result.metrics['component.factory.field.total_value']:,.0f}")

from plugboard.schemas import IntParameterSpec, ObjectiveSpec from plugboard.tune import Tuner tuner = Tuner( objective=ObjectiveSpec( object_type="component", object_name="factory", field_type="field", field_name="total_value", ), parameters=[ IntParameterSpec( object_type="component", object_name=f"machine_{i}", field_type="arg", field_name="maintenance_interval", lower=5, upper=60, ) for i in range(1, 6) ], num_samples=40, max_concurrent=4, mode="max", ) result = tuner.run(spec=spec) print("=== Optimisation Results ===") for i in range(1, 6): key = f"component.machine_{i}.arg.maintenance_interval" print(f" Machine {i} optimal maintenance interval: {result.config[key]} days") print(f" Maximum total value: ${result.metrics['component.factory.field.total_value']:,.0f}")

Since each machine has different reliability characteristics, the optimal maintenance schedule will differ — more fragile machines benefit from more frequent maintenance, while reliable machines can run longer between stops.

Alternatively, the same optimisation can be launched via the CLI using the YAML config in factory-simulation.yaml:

plugboard process tune factory-simulation.yaml