Machine Shop Simulation with SimPy: Jobs, Machines, and Breakdowns
The machine shop is a classic simulation problem. Jobs arrive. Machines process them. Breakdowns happen. The question: can you meet your deadlines?
The Machine Shop Model
This is SimPy's official example, expanded and explained.
Key elements: - Jobs - Work to be done - Machines - Process jobs - Repairmen - Fix breakdowns - Priorities - Some jobs matter more
Basic Machine Shop
import simpy
import random
RANDOM_SEED = 42
PT_MEAN = 10.0 # Mean processing time
PT_SIGMA = 2.0 # Processing time std dev
MTTF = 300.0 # Mean time to failure
REPAIR_TIME = 30.0 # Time to repair
JOB_DURATION = 30.0 # Time between job arrivals
NUM_MACHINES = 10
SIM_TIME = 4 * 7 * 24 * 60 # 4 weeks in minutes
def time_per_part():
"""Return actual processing time for a part."""
return max(0, random.normalvariate(PT_MEAN, PT_SIGMA))
def time_to_failure():
"""Return time until next failure."""
return random.expovariate(1/MTTF)
class Machine:
def __init__(self, env, name, repairman):
self.env = env
self.name = name
self.parts_made = 0
self.broken = False
self.repairman = repairman
self.process = env.process(self.working())
env.process(self.break_machine())
def working(self):
"""Produce parts as long as the simulation runs."""
while True:
done_in = time_per_part()
while done_in:
try:
start = self.env.now
yield self.env.timeout(done_in)
done_in = 0 # Part complete
except simpy.Interrupt:
self.broken = True
done_in -= self.env.now - start # Time remaining
# Request repair
with self.repairman.request(priority=1) as req:
yield req
yield self.env.timeout(REPAIR_TIME)
self.broken = False
self.parts_made += 1
def break_machine(self):
"""Break the machine periodically."""
while True:
yield self.env.timeout(time_to_failure())
if not self.broken:
self.process.interrupt()
# Run simulation
random.seed(RANDOM_SEED)
env = simpy.Environment()
repairman = simpy.PreemptiveResource(env, capacity=1)
machines = [Machine(env, f'Machine {i}', repairman) for i in range(NUM_MACHINES)]
env.run(until=SIM_TIME)
print('Machine shop results:')
for machine in machines:
print(f'{machine.name} made {machine.parts_made} parts.')
Job Shop with Routing
Jobs visit multiple machines in sequence:
class Job:
def __init__(self, job_id, routing, due_date):
self.id = job_id
self.routing = routing # List of (machine, time) tuples
self.due_date = due_date
self.start_time = None
self.end_time = None
class JobShop:
def __init__(self, env, machine_names):
self.env = env
self.machines = {
name: simpy.Resource(env, capacity=1)
for name in machine_names
}
self.completed_jobs = []
def process_job(self, job):
job.start_time = self.env.now
for machine_name, process_time in job.routing:
machine = self.machines[machine_name]
with machine.request() as req:
yield req
yield self.env.timeout(process_time)
job.end_time = self.env.now
self.completed_jobs.append(job)
lateness = max(0, job.end_time - job.due_date)
if lateness > 0:
print(f"Job {job.id} late by {lateness:.1f}")
# Example usage
env = simpy.Environment()
shop = JobShop(env, ['lathe', 'mill', 'drill', 'grinder'])
# Create jobs with different routings
jobs = [
Job(1, [('lathe', 10), ('mill', 15), ('drill', 5)], due_date=50),
Job(2, [('mill', 20), ('grinder', 10)], due_date=40),
Job(3, [('lathe', 8), ('drill', 12), ('grinder', 8)], due_date=60),
]
for job in jobs:
env.process(shop.process_job(job))
env.run()
Priority Scheduling
Rush jobs jump the queue:
class PriorityJobShop:
def __init__(self, env, machine_names):
self.env = env
self.machines = {
name: simpy.PriorityResource(env, capacity=1)
for name in machine_names
}
def process_job(self, job, priority):
"""Lower priority number = higher priority."""
for machine_name, process_time in job.routing:
machine = self.machines[machine_name]
with machine.request(priority=priority) as req:
yield req
yield self.env.timeout(process_time)
# Rush job (priority 1) vs normal job (priority 5)
env.process(shop.process_job(rush_job, priority=1))
env.process(shop.process_job(normal_job, priority=5))
Setup Times
Changing job types requires setup:
class MachineWithSetup:
def __init__(self, env, name, setup_time):
self.env = env
self.name = name
self.setup_time = setup_time
self.resource = simpy.Resource(env, capacity=1)
self.current_job_type = None
def process(self, job_type, process_time):
with self.resource.request() as req:
yield req
# Setup if job type changed
if job_type != self.current_job_type:
print(f"{self.name}: Setup for {job_type}")
yield self.env.timeout(self.setup_time)
self.current_job_type = job_type
# Process
yield self.env.timeout(process_time)
Complete Machine Shop
import simpy
import random
import numpy as np
class MachineShopSimulation:
def __init__(self, env, config):
self.env = env
self.config = config
# Machines
self.machines = {}
for name, count in config['machines'].items():
self.machines[name] = simpy.PreemptiveResource(env, capacity=count)
# Repairmen
self.repairmen = simpy.PriorityResource(env, capacity=config['repairmen'])
# Stats
self.jobs_completed = []
self.breakdowns = []
def machine_process(self, machine_name, job_id, process_time, priority):
"""Process a job on a machine, handling breakdowns."""
remaining = process_time
while remaining > 0:
try:
with self.machines[machine_name].request(priority=priority) as req:
yield req
start = self.env.now
yield self.env.timeout(remaining)
remaining = 0
except simpy.Interrupt as interrupt:
remaining -= (self.env.now - start)
# Handle breakdown
yield from self.repair(machine_name)
def repair(self, machine_name):
"""Request repair."""
with self.repairmen.request(priority=1) as req:
yield req
repair_time = random.expovariate(1/self.config['mean_repair_time'])
yield self.env.timeout(repair_time)
def breakdown_generator(self, machine_name):
"""Generate random breakdowns for a machine."""
while True:
ttf = random.expovariate(1/self.config['mttf'])
yield self.env.timeout(ttf)
# Interrupt current job if machine is busy
machine = self.machines[machine_name]
if machine.count > 0:
for req in machine.users:
if hasattr(req, 'proc') and req.proc.is_alive:
req.proc.interrupt(f"breakdown at {machine_name}")
self.breakdowns.append({
'machine': machine_name,
'time': self.env.now
})
def job(self, job_id, routing, due_date, priority):
"""Process a complete job through its routing."""
arrival = self.env.now
for machine_name, process_time in routing:
yield from self.machine_process(machine_name, job_id, process_time, priority)
completion = self.env.now
self.jobs_completed.append({
'job_id': job_id,
'arrival': arrival,
'completion': completion,
'due_date': due_date,
'flow_time': completion - arrival,
'lateness': max(0, completion - due_date)
})
def job_arrivals(self):
"""Generate incoming jobs."""
job_id = 0
machine_names = list(self.machines.keys())
while True:
yield self.env.timeout(random.expovariate(self.config['job_rate']))
# Random routing
num_ops = random.randint(2, 4)
routing = [
(random.choice(machine_names), random.uniform(5, 20))
for _ in range(num_ops)
]
# Due date
total_time = sum(t for _, t in routing)
due_date = self.env.now + total_time * random.uniform(1.5, 3)
# Priority (lower = more urgent)
priority = random.choices([1, 3, 5], weights=[10, 30, 60])[0]
self.env.process(self.job(job_id, routing, due_date, priority))
job_id += 1
def run(self, duration):
# Start job arrivals
self.env.process(self.job_arrivals())
# Start breakdown generators
for machine_name in self.machines:
self.env.process(self.breakdown_generator(machine_name))
self.env.run(until=duration)
def report(self):
print("\n=== Machine Shop Report ===")
print(f"Jobs completed: {len(self.jobs_completed)}")
print(f"Breakdowns: {len(self.breakdowns)}")
if self.jobs_completed:
flow_times = [j['flow_time'] for j in self.jobs_completed]
lateness = [j['lateness'] for j in self.jobs_completed]
on_time = sum(1 for j in self.jobs_completed if j['lateness'] == 0)
print(f"\nFlow time:")
print(f" Mean: {np.mean(flow_times):.1f}")
print(f" Max: {max(flow_times):.1f}")
print(f"\nOn-time delivery: {on_time}/{len(self.jobs_completed)} ({on_time/len(self.jobs_completed):.1%})")
print(f"Mean lateness (when late): {np.mean([l for l in lateness if l > 0]):.1f}")
# Config
config = {
'machines': {'lathe': 2, 'mill': 2, 'drill': 1, 'grinder': 1},
'repairmen': 1,
'mttf': 200,
'mean_repair_time': 20,
'job_rate': 0.1
}
random.seed(42)
env = simpy.Environment()
sim = MachineShopSimulation(env, config)
sim.run(duration=480)
sim.report()
Summary
Machine shop simulation covers: - Multiple machine types - Job routing and sequencing - Breakdowns and repairs - Priority scheduling - Setup times and changeovers
The machine shop is simulation's proving ground. Master it.
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