When the user wants to plan production or distribution capacity, analyze capacity requirements, optimize resource utilization, or balance capacity with demand. Also use when the user mentions "capacity analysis," "resource planning," "bottleneck analysis," "capacity expansion," "load balancing," "throughput planning," "utilization optimization," or "capacity modeling." For production scheduling, see master-production-scheduling. For long-term network capacity, see network-design.
You are an expert in capacity planning and resource optimization. Your goal is to help organizations match capacity with demand, optimize resource utilization, identify bottlenecks, and make cost-effective capacity investment decisions.
Before developing capacity plans, understand:
Planning Context
Demand Profile
Current State
Constraints & Requirements
1. Long-Term Strategic Capacity Planning
2. Medium-Term Tactical Capacity Planning
3. Short-Term Operational Capacity Planning
Leading Strategy
Lagging Strategy
Matching Strategy
Cushion Strategy
Production Capacity Metrics:
1. Design Capacity
2. Effective Capacity
3. Actual Output
Key Formulas:
def calculate_capacity_metrics(design_capacity, effective_capacity, actual_output):
"""
Calculate capacity utilization and efficiency
Returns:
- utilization: actual / design capacity
- efficiency: actual / effective capacity
"""
utilization = (actual_output / design_capacity) * 100
efficiency = (actual_output / effective_capacity) * 100
return {
'utilization': utilization,
'efficiency': efficiency,
'design_capacity': design_capacity,
'effective_capacity': effective_capacity,
'actual_output': actual_output
}
# Example
design = 10000 # units per month
effective = 8500 # accounting for maintenance, breaks
actual = 7500 # what's produced
metrics = calculate_capacity_metrics(design, effective, actual)
print(f"Utilization: {metrics['utilization']:.1f}%") # 75%
print(f"Efficiency: {metrics['efficiency']:.1f}%") # 88.2%
Overall Equipment Effectiveness (OEE):
def calculate_oee(availability, performance, quality):
"""
Calculate OEE (Overall Equipment Effectiveness)
Parameters:
- availability: uptime / planned production time
- performance: actual output / theoretical output at 100% speed
- quality: good units / total units produced
World-class OEE: > 85%
"""
oee = availability * performance * quality * 100
return {
'oee': oee,
'availability': availability * 100,
'performance': performance * 100,
'quality': quality * 100
}
# Example
availability = 0.90 # 90% uptime
performance = 0.85 # 85% of theoretical speed
quality = 0.95 # 95% good units
oee_metrics = calculate_oee(availability, performance, quality)
print(f"OEE: {oee_metrics['oee']:.1f}%") # 72.7%
Theory of Constraints (TOC):
import pandas as pd
import numpy as np
def identify_bottleneck(process_steps):
"""
Identify bottleneck in production process
Parameters:
- process_steps: list of dicts with 'name', 'capacity_per_hour', 'hours_available'
Returns bottleneck step and throughput
"""
df = pd.DataFrame(process_steps)
# Calculate total capacity per period
df['total_capacity'] = df['capacity_per_hour'] * df['hours_available']
# Identify bottleneck (minimum capacity)
bottleneck_idx = df['total_capacity'].idxmin()
bottleneck = df.loc[bottleneck_idx]
# System throughput limited by bottleneck
system_throughput = bottleneck['total_capacity']
# Calculate utilization based on bottleneck
df['utilization'] = (system_throughput / df['total_capacity']) * 100
return {
'bottleneck_step': bottleneck['name'],
'system_throughput': system_throughput,
'bottleneck_capacity': bottleneck['total_capacity'],
'process_analysis': df
}
# Example: Manufacturing process
process = [
{'name': 'Cutting', 'capacity_per_hour': 100, 'hours_available': 160},
{'name': 'Assembly', 'capacity_per_hour': 80, 'hours_available': 160},
{'name': 'Testing', 'capacity_per_hour': 120, 'hours_available': 160},
{'name': 'Packaging', 'capacity_per_hour': 90, 'hours_available': 160}
]
bottleneck_analysis = identify_bottleneck(process)
print(f"Bottleneck: {bottleneck_analysis['bottleneck_step']}")
print(f"System Throughput: {bottleneck_analysis['system_throughput']:,.0f} units/month")
print("\nProcess Analysis:")
print(bottleneck_analysis['process_analysis'])
Drum-Buffer-Rope (DBR) Scheduling:
class DrumBufferRope:
"""
Theory of Constraints scheduling method
- Drum: Bottleneck sets the pace
- Buffer: Protect bottleneck from disruptions
- Rope: Pull mechanism to control material release
"""
def __init__(self, bottleneck_capacity, buffer_time_days=3):
self.bottleneck_capacity = bottleneck_capacity
self.buffer_time = buffer_time_days
def calculate_schedule(self, demand, lead_times):
"""
Create production schedule based on DBR
Parameters:
- demand: array of daily demand
- lead_times: dict of process step lead times
"""
schedule = []
for day, daily_demand in enumerate(demand):
# Bottleneck sets the pace (DRUM)
bottleneck_output = min(daily_demand, self.bottleneck_capacity)
# Buffer: Start production earlier to protect bottleneck
buffer_start_day = max(0, day - self.buffer_time)
# Rope: Material release tied to bottleneck schedule
material_release = bottleneck_output
schedule.append({
'day': day,
'demand': daily_demand,
'bottleneck_output': bottleneck_output,
'buffer_start': buffer_start_day,
'material_release': material_release
})
return pd.DataFrame(schedule)
# Example usage
dbr = DrumBufferRope(bottleneck_capacity=800, buffer_time_days=3)
# Daily demand for next 10 days
demand = np.array([750, 850, 800, 900, 700, 800, 950, 800, 850, 800])
lead_times = {'cutting': 1, 'assembly': 2, 'testing': 1}
schedule = dbr.calculate_schedule(demand, lead_times)
print(schedule)
Objective: Minimize total costs while meeting demand
Decision Variables:
Costs:
from pulp import *
import pandas as pd
import numpy as np
def aggregate_planning(demand, costs, constraints, periods=12):
"""
Aggregate production planning optimization
Parameters:
- demand: array of demand by period
- costs: dict with cost parameters
- constraints: dict with capacity constraints
- periods: planning horizon
Returns optimal plan
"""
# Create problem
prob = LpProblem("Aggregate_Planning", LpMinimize)
# Decision variables
P = LpVariable.dicts("Production", range(periods), lowBound=0)
W = LpVariable.dicts("Workforce", range(periods), lowBound=0, cat='Integer')
O = LpVariable.dicts("Overtime", range(periods), lowBound=0)
I = LpVariable.dicts("Inventory", range(periods), lowBound=0)
H = LpVariable.dicts("Hire", range(periods), lowBound=0, cat='Integer')
F = LpVariable.dicts("Fire", range(periods), lowBound=0, cat='Integer')
B = LpVariable.dicts("Backorder", range(periods), lowBound=0)
S = LpVariable.dicts("Subcontract", range(periods), lowBound=0)
# Objective function
prob += lpSum([
# Regular production cost
costs['regular_cost'] * P[t] +
# Workforce cost
costs['labor_cost'] * W[t] +
# Overtime cost
costs['overtime_cost'] * O[t] +
# Inventory holding cost
costs['holding_cost'] * I[t] +
# Hiring cost
costs['hiring_cost'] * H[t] +
# Firing cost
costs['firing_cost'] * F[t] +
# Backorder cost
costs['backorder_cost'] * B[t] +
# Subcontracting cost
costs['subcontract_cost'] * S[t]
for t in range(periods)
])
# Constraints
# Initial conditions
initial_workforce = constraints['initial_workforce']
initial_inventory = constraints['initial_inventory']
for t in range(periods):
# Production capacity constraint
prob += P[t] <= W[t] * constraints['units_per_worker'], f"Capacity_{t}"
# Overtime capacity
prob += O[t] <= W[t] * constraints['overtime_per_worker'], f"Overtime_{t}"
# Subcontracting capacity
prob += S[t] <= constraints['max_subcontract'], f"Subcontract_{t}"
# Workforce balance
if t == 0:
prob += W[t] == initial_workforce + H[t] - F[t], f"Workforce_{t}"
else:
prob += W[t] == W[t-1] + H[t] - F[t], f"Workforce_{t}"
# Inventory balance
if t == 0:
prob += I[t] == initial_inventory + P[t] + O[t] + S[t] - demand[t] + B[t], f"Inventory_{t}"
else:
prob += I[t] == I[t-1] + P[t] + O[t] + S[t] - demand[t] + B[t] - B[t-1], f"Inventory_{t}"
# Minimum service level (max backorder)
prob += B[t] <= demand[t] * constraints['max_backorder_pct'], f"Service_{t}"
# Solve
prob.solve(PULP_CBC_CMD(msg=0))
# Extract results
results = {
'status': LpStatus[prob.status],
'total_cost': value(prob.objective),
'production': [P[t].varValue for t in range(periods)],
'workforce': [W[t].varValue for t in range(periods)],
'overtime': [O[t].varValue for t in range(periods)],
'inventory': [I[t].varValue for t in range(periods)],
'hired': [H[t].varValue for t in range(periods)],
'fired': [F[t].varValue for t in range(periods)],
'backorders': [B[t].varValue for t in range(periods)],
'subcontract': [S[t].varValue for t in range(periods)]
}
# Create summary DataFrame
df = pd.DataFrame({
'Period': range(1, periods + 1),
'Demand': demand,
'Production': results['production'],
'Workforce': results['workforce'],
'Overtime': results['overtime'],
'Inventory': results['inventory'],
'Backorders': results['backorders'],
'Subcontract': results['subcontract']
})
results['summary'] = df
return results
# Example usage
demand = np.array([1000, 1200, 1500, 1800, 2000, 1800,
1600, 1400, 1300, 1200, 1100, 1000])
costs = {
'regular_cost': 50,
'labor_cost': 2000, # per worker per period
'overtime_cost': 75,
'holding_cost': 5,
'hiring_cost': 1000,
'firing_cost': 1500,
'backorder_cost': 100,
'subcontract_cost': 80
}
constraints = {
'initial_workforce': 40,
'initial_inventory': 500,
'units_per_worker': 30,
'overtime_per_worker': 10,
'max_subcontract': 500,
'max_backorder_pct': 0.10
}
plan = aggregate_planning(demand, costs, constraints)
print(f"Optimal Total Cost: ${plan['total_cost']:,.0f}")
print("\nProduction Plan:")
print(plan['summary'])
Process:
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
class CapacityRequirementsPlanning:
"""
CRP - Calculate and analyze capacity requirements
based on production schedule and routings
"""
def __init__(self, work_centers):
"""
Parameters:
- work_centers: dict {name: {'capacity': hours, 'efficiency': 0-1}}
"""
self.work_centers = work_centers
def calculate_requirements(self, schedule, routings):
"""
Calculate capacity requirements
Parameters:
- schedule: DataFrame with 'product', 'period', 'quantity'
- routings: dict {product: [(work_center, hours_per_unit)]}
Returns DataFrame with requirements by work center and period
"""
requirements = []
for idx, row in schedule.iterrows():
product = row['product']
period = row['period']
quantity = row['quantity']
# Get routing for product
routing = routings.get(product, [])
for work_center, hours_per_unit in routing:
total_hours = quantity * hours_per_unit
# Adjust for efficiency
efficiency = self.work_centers[work_center]['efficiency']
required_hours = total_hours / efficiency
requirements.append({
'period': period,
'work_center': work_center,
'product': product,
'quantity': quantity,
'hours_required': required_hours
})
df = pd.DataFrame(requirements)
# Aggregate by work center and period
summary = df.groupby(['period', 'work_center'])['hours_required'].sum().reset_index()
# Add capacity and utilization
summary['capacity'] = summary['work_center'].map(
lambda wc: self.work_centers[wc]['capacity']
)
summary['utilization'] = (summary['hours_required'] / summary['capacity']) * 100
summary['variance'] = summary['capacity'] - summary['hours_required']
return summary
def identify_overloads(self, requirements, threshold=100):
"""
Identify periods/work centers with overload
Parameters:
- requirements: output from calculate_requirements
- threshold: utilization % threshold
Returns overloaded resources
"""
overloads = requirements[requirements['utilization'] > threshold].copy()
overloads = overloads.sort_values(['period', 'work_center'])
return overloads
def plot_capacity_profile(self, requirements):
"""Visualize capacity requirements vs. available"""
work_centers = requirements['work_center'].unique()
fig, axes = plt.subplots(len(work_centers), 1,
figsize=(12, 4 * len(work_centers)),
squeeze=False)
for i, wc in enumerate(work_centers):
wc_data = requirements[requirements['work_center'] == wc]
ax = axes[i, 0]
# Plot capacity line
ax.axhline(y=wc_data['capacity'].iloc[0],
color='green', linestyle='--',
linewidth=2, label='Capacity')
# Plot requirements
ax.bar(wc_data['period'], wc_data['hours_required'],
alpha=0.7, label='Required')
# Highlight overloads
overload = wc_data[wc_data['utilization'] > 100]
if not overload.empty:
ax.bar(overload['period'], overload['hours_required'],
color='red', alpha=0.7, label='Overload')
ax.set_title(f'{wc} - Capacity Profile')
ax.set_xlabel('Period')
ax.set_ylabel('Hours')
ax.legend()
ax.grid(True, alpha=0.3)
plt.tight_layout()
return fig
# Example usage
work_centers = {
'Cutting': {'capacity': 160, 'efficiency': 0.90},
'Welding': {'capacity': 160, 'efficiency': 0.85},
'Assembly': {'capacity': 160, 'efficiency': 0.92},
'Inspection': {'capacity': 160, 'efficiency': 0.95}
}
crp = CapacityRequirementsPlanning(work_centers)
# Production schedule
schedule = pd.DataFrame({
'product': ['A', 'A', 'B', 'B', 'C', 'C'] * 3,
'period': [1, 2, 1, 2, 1, 2] * 3,
'quantity': [100, 120, 80, 90, 60, 70] * 3
})
# Routings: hours per unit at each work center
routings = {
'A': [('Cutting', 0.5), ('Welding', 0.8), ('Assembly', 1.0), ('Inspection', 0.3)],
'B': [('Cutting', 0.6), ('Assembly', 1.2), ('Inspection', 0.4)],
'C': [('Cutting', 0.4), ('Welding', 1.0), ('Assembly', 0.8), ('Inspection', 0.2)]
}
# Calculate requirements
requirements = crp.calculate_requirements(schedule, routings)
print("Capacity Requirements:")
print(requirements)
# Identify overloads
overloads = crp.identify_overloads(requirements)
if not overloads.empty:
print("\nOverloaded Resources:")
print(overloads[['period', 'work_center', 'utilization', 'variance']])