Manufacturing Simulation with SimPy: Modelling the Factory Floor
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
Build Professional Simulations
Break free from commercial software and learn how to build powerful, industry-standard simulations in Python. The Complete Simulation in Python with SimPy Bootcamp gives you everything you need.
Explore the Bootcamp