Manufacturing Simulation with SimPy: Modelling the Factory Floor

Published on February 13, 2026

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. <h2 id="the-manufacturing-model">The Manufacturing Model</h2> 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 <h2 id="basic-production-line">Basic Production Line</h2> <div class="code-block"><pre><code>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") </code></pre></div> <h2 id="machine-with-breakdowns">Machine with Breakdowns</h2> <div class="code-block"><pre><code>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 </code></pre></div> <h2 id="kanban-system">Kanban System</h2> Pull-based production: <div class="code-block"><pre><code>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() </code></pre></div> <h2 id="batch-processing">Batch Processing</h2> <div class="code-block"><pre><code>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}") </code></pre></div> <h2 id="quality-control">Quality Control</h2> <div class="code-block"><pre><code>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 </code></pre></div> <h2 id="shift-patterns">Shift Patterns</h2> <div class="code-block"><pre><code>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) </code></pre></div> <h2 id="complete-manufacturing-example">Complete Manufacturing Example</h2> <div class="code-block"><pre><code>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() </code></pre></div> <h2 id="summary">Summary</h2> 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. <h2 id="next-steps">Next Steps</h2> <ul> <li><a href="/blog_posts/simpy-machine-shop-simulation.html">Machine Shop Simulation</a></li> <li><a href="/blog_posts/simpy-production-line-simulation.html">Production Line Simulation</a></li> <li><a href="/blog_posts/simpy-supply-chain-simulation.html">Supply Chain Simulation</a></li> </ul>

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