Manufacturing is where simulation proves its worth. Every minute of downtime costs money. Every bottleneck limits output. SimPy helps you find them before they cost you.

The Manufacturing Model

A typical manufacturing simulation includes:

• Machines - Process parts
• Buffers - Hold work-in-progress
• Parts - Flow through the system
• Failures - Machines break down
• Operators - People who run things

Basic Production Line

import simpy
import random

class ProductionLine:
    def __init__(self, env, num_machines, process_times, buffer_sizes):
        self.env = env
        self.machines = [
            simpy.Resource(env, capacity=1) for _ in range(num_machines)
        ]
        self.buffers = [
            simpy.Container(env, capacity=size, init=0)
            for size in buffer_sizes
        ]
        self.process_times = process_times
        self.parts_completed = 0

    def process_part(self, part_id):
        for i, (machine, process_time) in enumerate(
            zip(self.machines, self.process_times)
        ):
            # Wait for buffer space (if not first machine)
            if i > 0:
                yield self.buffers[i-1].get(1)

            # Process on machine
            with machine.request() as req:
                yield req
                yield self.env.timeout(random.expovariate(1/process_time))

            # Put in next buffer (if not last machine)
            if i < len(self.buffers):
                yield self.buffers[i].put(1)

        self.parts_completed += 1

def raw_material_source(env, line, arrival_rate):
    part_id = 0
    while True:
        yield env.timeout(random.expovariate(arrival_rate))
        env.process(line.process_part(part_id))
        part_id += 1

# Run simulation
env = simpy.Environment()
line = ProductionLine(
    env,
    num_machines=3,
    process_times=[5, 7, 4],
    buffer_sizes=[10, 10]
)
env.process(raw_material_source(env, line, arrival_rate=1/4))
env.run(until=1000)

print(f"Parts completed: {line.parts_completed}")
print(f"Throughput: {line.parts_completed / 1000:.2f} parts/time unit")

Machine with Breakdowns

class MachineWithBreakdowns:
    def __init__(self, env, name, process_time, mttf, mttr):
        self.env = env
        self.name = name
        self.process_time = process_time
        self.mttf = mttf  # Mean time to failure
        self.mttr = mttr  # Mean time to repair
        self.resource = simpy.PreemptiveResource(env, capacity=1)
        self.broken = False
        self.parts_made = 0
        self.breakdown_count = 0

        # Start breakdown process
        env.process(self.breakdown_generator())

    def breakdown_generator(self):
        while True:
            yield self.env.timeout(random.expovariate(1/self.mttf))
            if not self.broken:
                self.broken = True
                self.breakdown_count += 1
                print(f"{self.name} broke down at {self.env.now:.1f}")

                # Interrupt any current job
                if self.resource.count > 0:
                    for req in self.resource.users:
                        req.proc.interrupt("breakdown")

                # Repair
                yield self.env.timeout(random.expovariate(1/self.mttr))
                self.broken = False
                print(f"{self.name} repaired at {self.env.now:.1f}")

    def process(self, part_id):
        while True:
            try:
                with self.resource.request(priority=1) as req:
                    yield req
                    if self.broken:
                        continue  # Machine broken, try again

                    yield self.env.timeout(
                        random.expovariate(1/self.process_time)
                    )
                    self.parts_made += 1
                    return  # Success
            except simpy.Interrupt:
                pass  # Breakdown, will retry

Kanban System

Pull-based production:

class KanbanCell:
    def __init__(self, env, name, process_time, kanban_cards):
        self.env = env
        self.name = name
        self.process_time = process_time
        self.output_buffer = simpy.Store(env, capacity=kanban_cards)
        self.kanban_signal = simpy.Store(env)
        self.parts_made = 0

    def run(self, input_source):
        while True:
            # Wait for kanban signal (demand from downstream)
            yield self.kanban_signal.get()

            # Get input
            part = yield input_source.get()

            # Process
            yield self.env.timeout(
                random.expovariate(1/self.process_time)
            )

            # Output
            yield self.output_buffer.put(part)
            self.parts_made += 1

def downstream_demand(env, cell, demand_rate):
    """Simulate downstream pulling parts."""
    while True:
        yield env.timeout(random.expovariate(demand_rate))
        yield cell.kanban_signal.put("demand")
        part = yield cell.output_buffer.get()

Batch Processing

class BatchProcessor:
    def __init__(self, env, name, batch_size, process_time):
        self.env = env
        self.name = name
        self.batch_size = batch_size
        self.process_time = process_time
        self.input_buffer = simpy.Store(env)
        self.output_buffer = simpy.Store(env)
        self.batches_processed = 0

    def run(self):
        while True:
            # Collect batch
            batch = []
            for _ in range(self.batch_size):
                part = yield self.input_buffer.get()
                batch.append(part)

            # Process entire batch
            yield self.env.timeout(self.process_time)

            # Release all parts
            for part in batch:
                yield self.output_buffer.put(part)

            self.batches_processed += 1
            print(f"{self.name} processed batch at {self.env.now:.1f}")

Quality Control

def quality_check(env, part, inspection_time, defect_rate):
    """Inspect part, reject defects."""
    yield env.timeout(inspection_time)

    if random.random() < defect_rate:
        return "reject"
    return "pass"

def manufacturing_with_qc(env, machines, qc_station, rework_station):
    part_id = 0
    while True:
        part = {'id': part_id, 'rework_count': 0}

        # Main processing
        for machine in machines:
            with machine.request() as req:
                yield req
                yield env.timeout(random.expovariate(1/5))

        # Quality check
        result = yield from quality_check(env, part, 2, 0.1)

        if result == "reject":
            if part['rework_count'] < 2:
                part['rework_count'] += 1
                # Send to rework (simplified)
                yield env.timeout(10)
            else:
                print(f"Part {part_id} scrapped")

        part_id += 1

Shift Patterns

class ShiftManager:
    def __init__(self, env, machines, shift_hours, break_duration):
        self.env = env
        self.machines = machines
        self.shift_hours = shift_hours
        self.break_duration = break_duration
        env.process(self.run())

    def run(self):
        while True:
            # Work period
            yield self.env.timeout(self.shift_hours / 2)

            # Break - reduce capacity
            print(f"Break starts at {self.env.now}")
            for m in self.machines:
                m.capacity = 0

            yield self.env.timeout(self.break_duration)

            # Back to work
            for m in self.machines:
                m.capacity = 1
            print(f"Break ends at {self.env.now}")

            yield self.env.timeout(self.shift_hours / 2)

Complete Manufacturing Example

import simpy
import random
import pandas as pd

class ManufacturingSimulation:
    def __init__(self, config):
        self.config = config
        self.env = simpy.Environment()
        self.stats = {
            'parts_completed': 0,
            'parts_scrapped': 0,
            'machine_busy_time': {},
            'buffer_levels': []
        }

        # Create resources
        self.machines = []
        for i, (name, rate) in enumerate(config['machines']):
            machine = simpy.Resource(self.env, capacity=1)
            self.machines.append({'name': name, 'resource': machine, 'rate': rate})
            self.stats['machine_busy_time'][name] = 0

        self.buffers = [
            simpy.Container(self.env, capacity=cap, init=0)
            for cap in config['buffer_sizes']
        ]

    def part_flow(self, part_id):
        """Part flows through all machines."""
        for i, machine_info in enumerate(self.machines):
            machine = machine_info['resource']
            rate = machine_info['rate']

            with machine.request() as req:
                yield req
                process_time = random.expovariate(rate)
                self.stats['machine_busy_time'][machine_info['name']] += process_time
                yield self.env.timeout(process_time)

            # Buffer management (simplified)
            if i < len(self.buffers):
                yield self.buffers[i].put(1)

        self.stats['parts_completed'] += 1

    def arrivals(self):
        part_id = 0
        while True:
            yield self.env.timeout(
                random.expovariate(self.config['arrival_rate'])
            )
            self.env.process(self.part_flow(part_id))
            part_id += 1

    def monitor(self, interval=10):
        while True:
            levels = [b.level for b in self.buffers]
            self.stats['buffer_levels'].append({
                'time': self.env.now,
                'levels': levels
            })
            yield self.env.timeout(interval)

    def run(self, duration):
        self.env.process(self.arrivals())
        self.env.process(self.monitor())
        self.env.run(until=duration)

    def report(self):
        print(f"\n=== Manufacturing Simulation Report ===")
        print(f"Duration: {self.env.now}")
        print(f"Parts completed: {self.stats['parts_completed']}")
        print(f"Throughput: {self.stats['parts_completed']/self.env.now:.3f}/unit")

        print("\nMachine Utilisation:")
        for name, busy in self.stats['machine_busy_time'].items():
            util = busy / self.env.now
            print(f"  {name}: {util:.1%}")

# Run
config = {
    'machines': [('Lathe', 1/5), ('Mill', 1/7), ('Drill', 1/4)],
    'buffer_sizes': [20, 20],
    'arrival_rate': 1/4
}

random.seed(42)
sim = ManufacturingSimulation(config)
sim.run(1000)
sim.report()

Summary

Manufacturing simulation captures:

• Machine processing and breakdowns
• Buffer dynamics and blocking
• Quality control and rework
• Shift patterns and capacity
• Bottleneck identification

Simulate before you build. Fix before you fail.

Next Steps

• Machine Shop Simulation
• Production Line Simulation
• Supply Chain Simulation