Machine Shop Simulation with SimPy: Jobs, Machines, and Breakdowns

Published on February 19, 2026

The machine shop is a classic simulation problem. Jobs arrive. Machines process them. Breakdowns happen. The question: can you meet your deadlines? <h2 id="the-machine-shop-model">The Machine Shop Model</h2> 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 <h2 id="basic-machine-shop">Basic Machine Shop</h2> <div class="code-block"><pre><code>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.') </code></pre></div> <h2 id="job-shop-with-routing">Job Shop with Routing</h2> Jobs visit multiple machines in sequence: <div class="code-block"><pre><code>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() </code></pre></div> <h2 id="priority-scheduling">Priority Scheduling</h2> Rush jobs jump the queue: <div class="code-block"><pre><code>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)) </code></pre></div> <h2 id="setup-times">Setup Times</h2> Changing job types requires setup: <div class="code-block"><pre><code>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) </code></pre></div> <h2 id="complete-machine-shop">Complete Machine Shop</h2> <div class="code-block"><pre><code>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() </code></pre></div> <h2 id="summary">Summary</h2> 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. <h2 id="next-steps">Next Steps</h2> <ul> <li><a href="/blog_posts/simpy-manufacturing-simulation.html">Manufacturing Simulation</a></li> <li><a href="/blog_posts/simpy-production-line-simulation.html">Production Line Simulation</a></li> <li><a href="/blog_posts/simpy-preemptiveresource-explained.html">PreemptiveResource Explained</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