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.