Warehouse Simulation with SimPy: From Receiving to Shipping
Warehouses are logistics engines. Goods flow in, get stored, get picked, and flow out. Simulation reveals the bottlenecks between receive and ship.
The Warehouse Model
Key processes: - Receiving - Goods arrive - Putaway - Storage in locations - Storage - Inventory holding - Picking - Order fulfillment - Packing - Preparation for shipping - Shipping - Goods leave
Basic Warehouse
import simpy
import random
class Warehouse:
def __init__(self, env, config):
self.env = env
self.receiving_docks = simpy.Resource(env, capacity=config['receiving_docks'])
self.forklifts = simpy.Resource(env, capacity=config['forklifts'])
self.pickers = simpy.Resource(env, capacity=config['pickers'])
self.packing_stations = simpy.Resource(env, capacity=config['packing_stations'])
self.shipping_docks = simpy.Resource(env, capacity=config['shipping_docks'])
self.inventory = simpy.Container(env, capacity=config['storage_capacity'], init=config['initial_inventory'])
self.stats = {'received': 0, 'shipped': 0, 'orders': []}
def receive_shipment(self, shipment_id, quantity):
"""Process incoming shipment."""
arrival = self.env.now
# Dock
with self.receiving_docks.request() as dock:
yield dock
yield self.env.timeout(random.uniform(15, 30)) # Unload
# Putaway with forklift
with self.forklifts.request() as forklift:
yield forklift
yield self.env.timeout(quantity * 0.5) # Time per pallet
yield self.inventory.put(quantity)
self.stats['received'] += quantity
def process_order(self, order_id, items):
"""Pick, pack, and ship an order."""
order_start = self.env.now
record = {'id': order_id, 'items': items, 'start': order_start}
# Pick items
with self.pickers.request() as picker:
yield picker
# Travel and pick time
pick_time = sum(random.uniform(0.5, 2) for _ in range(items))
yield self.env.timeout(pick_time)
yield self.inventory.get(items)
record['picked'] = self.env.now
# Pack
with self.packing_stations.request() as station:
yield station
yield self.env.timeout(random.uniform(2, 5))
record['packed'] = self.env.now
# Ship
with self.shipping_docks.request() as dock:
yield dock
yield self.env.timeout(random.uniform(1, 3))
record['shipped'] = self.env.now
record['total_time'] = self.env.now - order_start
self.stats['orders'].append(record)
self.stats['shipped'] += items
# Run simulation
env = simpy.Environment()
warehouse = Warehouse(env, {
'receiving_docks': 3,
'forklifts': 5,
'pickers': 10,
'packing_stations': 6,
'shipping_docks': 4,
'storage_capacity': 10000,
'initial_inventory': 5000
})
Zone-Based Picking
class ZonedWarehouse:
def __init__(self, env, zone_config):
self.env = env
self.zones = {}
for zone_name, config in zone_config.items():
self.zones[zone_name] = {
'inventory': simpy.Container(env, capacity=config['capacity'],
init=config['initial']),
'pickers': simpy.Resource(env, capacity=config['pickers'])
}
def pick_order(self, order_id, items_by_zone):
"""Pick items from multiple zones."""
picked_items = []
for zone_name, items in items_by_zone.items():
zone = self.zones[zone_name]
with zone['pickers'].request() as picker:
yield picker
yield zone['inventory'].get(items)
pick_time = items * random.uniform(0.3, 0.8)
yield self.env.timeout(pick_time)
picked_items.extend([zone_name] * items)
return picked_items
# Zone configuration
zone_config = {
'A': {'capacity': 3000, 'initial': 2000, 'pickers': 4}, # Fast movers
'B': {'capacity': 5000, 'initial': 3000, 'pickers': 3}, # Medium
'C': {'capacity': 8000, 'initial': 5000, 'pickers': 2}, # Slow movers
}
Wave Planning
class WaveBasedWarehouse:
def __init__(self, env, config):
self.env = env
self.config = config
self.order_queue = simpy.Store(env)
self.pickers = simpy.Resource(env, capacity=config['pickers'])
self.waves_completed = 0
def receive_order(self, order):
yield self.order_queue.put(order)
def run_waves(self, wave_interval, orders_per_wave):
"""Release orders in waves."""
while True:
yield self.env.timeout(wave_interval)
# Collect orders for this wave
wave_orders = []
while len(wave_orders) < orders_per_wave:
try:
order = yield self.order_queue.get()
wave_orders.append(order)
except:
break
if wave_orders:
# Process wave
yield self.env.process(self.process_wave(wave_orders))
self.waves_completed += 1
def process_wave(self, orders):
"""Process all orders in a wave."""
# Sort by zone to minimize travel
orders.sort(key=lambda o: o.get('zone', 'A'))
for order in orders:
self.env.process(self.pick_order(order))
# Wait for all picks to complete
yield self.env.timeout(0)
Conveyor System
class ConveyorSystem:
def __init__(self, env, speed, length):
self.env = env
self.speed = speed # Items per minute
self.length = length # Sections
self.sections = [simpy.Store(env, capacity=10) for _ in range(length)]
def put_item(self, item):
"""Put item on conveyor at start."""
yield self.sections[0].put(item)
def run(self):
"""Move items along conveyor."""
while True:
yield self.env.timeout(1 / self.speed)
# Move items from end to start
for i in range(self.length - 1, 0, -1):
if self.sections[i-1].items:
item = yield self.sections[i-1].get()
yield self.sections[i].put(item)
def get_item(self):
"""Get item from end of conveyor."""
return self.sections[-1].get()
Complete Warehouse Simulation
import simpy
import random
import numpy as np
class FullWarehouseSimulation:
def __init__(self, env, config):
self.env = env
self.config = config
# Resources
self.receiving = simpy.Resource(env, capacity=config['receiving_docks'])
self.putaway_crew = simpy.Resource(env, capacity=config['putaway_workers'])
self.inventory = simpy.Container(env, capacity=config['storage'],
init=config['initial_stock'])
self.pickers = simpy.Resource(env, capacity=config['pickers'])
self.packers = simpy.Resource(env, capacity=config['packers'])
self.shipping = simpy.Resource(env, capacity=config['shipping_docks'])
# Stats
self.order_stats = []
self.inbound_stats = []
def inbound_shipment(self, shipment_id, pallets):
"""Process inbound delivery."""
arrival = self.env.now
record = {'id': shipment_id, 'pallets': pallets, 'arrival': arrival}
# Receive
with self.receiving.request() as dock:
yield dock
unload_time = pallets * random.uniform(2, 4)
yield self.env.timeout(unload_time)
record['unloaded'] = self.env.now
# Putaway
with self.putaway_crew.request() as worker:
yield worker
putaway_time = pallets * random.uniform(3, 6)
yield self.env.timeout(putaway_time)
yield self.inventory.put(pallets * 50) # 50 units per pallet
record['stored'] = self.env.now
record['total_time'] = self.env.now - arrival
self.inbound_stats.append(record)
def outbound_order(self, order_id, units):
"""Process customer order."""
arrival = self.env.now
record = {'id': order_id, 'units': units, 'arrival': arrival}
# Wait for inventory if needed
yield self.inventory.get(units)
record['allocated'] = self.env.now
# Pick
with self.pickers.request() as picker:
yield picker
lines = max(1, units // 5)
pick_time = lines * random.uniform(1, 3)
yield self.env.timeout(pick_time)
record['picked'] = self.env.now
# Pack
with self.packers.request() as packer:
yield packer
pack_time = random.uniform(2, 5)
yield self.env.timeout(pack_time)
record['packed'] = self.env.now
# Ship
with self.shipping.request() as dock:
yield dock
yield self.env.timeout(random.uniform(1, 3))
record['shipped'] = self.env.now
record['total_time'] = self.env.now - arrival
self.order_stats.append(record)
def inbound_arrivals(self):
"""Generate inbound deliveries."""
shipment_id = 0
while True:
yield self.env.timeout(random.expovariate(self.config['inbound_rate']))
pallets = random.randint(10, 30)
self.env.process(self.inbound_shipment(shipment_id, pallets))
shipment_id += 1
def order_arrivals(self):
"""Generate customer orders."""
order_id = 0
while True:
# Time-varying order rate
hour = (self.env.now / 60) % 24
if 9 <= hour <= 17:
rate = self.config['peak_order_rate']
else:
rate = self.config['off_peak_order_rate']
yield self.env.timeout(random.expovariate(rate))
units = random.randint(5, 50)
self.env.process(self.outbound_order(order_id, units))
order_id += 1
def inventory_monitor(self, interval=60):
"""Track inventory levels."""
self.inventory_log = []
while True:
self.inventory_log.append({
'time': self.env.now,
'level': self.inventory.level
})
yield self.env.timeout(interval)
def run(self, duration):
self.env.process(self.inbound_arrivals())
self.env.process(self.order_arrivals())
self.env.process(self.inventory_monitor())
self.env.run(until=duration)
def report(self):
print("\n=== Warehouse Simulation Report ===")
print(f"Duration: {self.env.now / 60:.1f} hours")
print(f"\nInbound:")
print(f" Shipments: {len(self.inbound_stats)}")
if self.inbound_stats:
times = [s['total_time'] for s in self.inbound_stats]
print(f" Avg dock-to-stock: {np.mean(times):.1f} min")
print(f"\nOutbound:")
print(f" Orders: {len(self.order_stats)}")
if self.order_stats:
times = [s['total_time'] for s in self.order_stats]
print(f" Avg order cycle time: {np.mean(times):.1f} min")
print(f" 90th percentile: {np.percentile(times, 90):.1f} min")
print(f"\nInventory:")
levels = [l['level'] for l in self.inventory_log]
print(f" Avg level: {np.mean(levels):.0f} units")
print(f" Min level: {min(levels):.0f} units")
# Config
config = {
'receiving_docks': 3,
'putaway_workers': 4,
'storage': 50000,
'initial_stock': 25000,
'pickers': 12,
'packers': 6,
'shipping_docks': 4,
'inbound_rate': 0.02,
'peak_order_rate': 0.5,
'off_peak_order_rate': 0.1
}
random.seed(42)
env = simpy.Environment()
sim = FullWarehouseSimulation(env, config)
sim.run(duration=480) # 8 hours
sim.report()
Summary
Warehouse simulation captures: - Inbound and outbound flows - Resource constraints (docks, workers, equipment) - Inventory dynamics - Order cycle times - Zone-based operations
Simulate before you build. Optimise 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