capacity-planner
You are capacity-planner - a specialized skill for capacity requirements planning and strategic capacity decisions.
Overview
This skill enables AI-powered capacity planning including:
- Capacity requirements calculation from demand
- Resource capacity documentation
- Capacity gap analysis
- Rough-cut capacity planning (RCCP)
- Detailed capacity requirements planning (CRP)
- Lead, lag, match strategy evaluation
- Make vs buy analysis
- Capacity investment justification
Capabilities
1. Capacity Requirements Calculation
import pandas as pd
import numpy as np
def calculate_capacity_requirements(demand_forecast: pd.DataFrame,
product_routing: dict,
work_center_data: dict):
"""
Calculate capacity requirements from demand
demand_forecast: DataFrame with columns ['period', 'product', 'quantity']
product_routing: {product: [(work_center, time_per_unit), ...]}
work_center_data: {work_center: {'available_hours': x, 'efficiency': y}}
"""
requirements = []
for _, row in demand_forecast.iterrows():
product = row['product']
quantity = row['quantity']
period = row['period']
if product in product_routing:
for work_center, time_per_unit in product_routing[product]:
required_hours = quantity * time_per_unit / 60 # Convert to hours
requirements.append({
'period': period,
'product': product,
'work_center': work_center,
'required_hours': required_hours
})
req_df = pd.DataFrame(requirements)
# Aggregate by work center and period
summary = req_df.groupby(['period', 'work_center'])['required_hours'].sum().reset_index()
return {
'detailed_requirements': req_df,
'summary': summary
}
2. Capacity Gap Analysis
def analyze_capacity_gaps(requirements: pd.DataFrame, capacity: dict):
"""
Identify capacity gaps between requirements and available capacity
"""
gaps = []
for _, row in requirements.iterrows():
wc = row['work_center']
required = row['required_hours']
if wc in capacity:
available = capacity[wc]['available_hours']
efficiency = capacity[wc].get('efficiency', 1.0)
effective_capacity = available * efficiency
gap = required - effective_capacity
utilization = required / effective_capacity * 100 if effective_capacity > 0 else float('inf')
gaps.append({
'period': row['period'],
'work_center': wc,
'required_hours': required,
'available_hours': available,
'effective_capacity': effective_capacity,
'gap_hours': gap,
'utilization_percent': utilization,
'status': 'overcapacity' if gap > 0 else 'undercapacity' if gap < -effective_capacity * 0.2 else 'balanced'
})
gap_df = pd.DataFrame(gaps)
# Summary
summary = {
'total_overcapacity_hours': gap_df[gap_df['gap_hours'] > 0]['gap_hours'].sum(),
'total_undercapacity_hours': abs(gap_df[gap_df['gap_hours'] < 0]['gap_hours'].sum()),
'bottleneck_work_centers': gap_df[gap_df['status'] == 'overcapacity']['work_center'].unique().tolist(),
'underutilized_work_centers': gap_df[gap_df['status'] == 'undercapacity']['work_center'].unique().tolist()
}
return {
'gaps': gap_df,
'summary': summary
}
3. Rough-Cut Capacity Planning (RCCP)
def rough_cut_capacity_plan(mps: pd.DataFrame, bill_of_resources: dict,
available_capacity: dict):
"""
Rough-Cut Capacity Planning
mps: Master Production Schedule with ['period', 'product', 'quantity']
bill_of_resources: {product: {resource: hours_per_unit}}
available_capacity: {resource: hours_per_period}
"""
rccp_results = []
for _, row in mps.iterrows():
product = row['product']
quantity = row['quantity']
period = row['period']
if product in bill_of_resources:
for resource, hours_per_unit in bill_of_resources[product].items():
load = quantity * hours_per_unit
rccp_results.append({
'period': period,
'product': product,
'resource': resource,
'load_hours': load
})
rccp_df = pd.DataFrame(rccp_results)
# Aggregate by resource and period
resource_load = rccp_df.groupby(['period', 'resource'])['load_hours'].sum().reset_index()
# Compare to available
capacity_status = []
for _, row in resource_load.iterrows():
resource = row['resource']
load = row['load_hours']
available = available_capacity.get(resource, 0)
capacity_status.append({
'period': row['period'],
'resource': resource,
'load': load,
'available': available,
'utilization': load / available * 100 if available > 0 else float('inf'),
'feasible': load <= available
})
return {
'rccp_detail': rccp_df,
'resource_load': resource_load,
'capacity_status': pd.DataFrame(capacity_status),
'infeasible_periods': [r for r in capacity_status if not r['feasible']]
}
4. Capacity Strategy Evaluation
def evaluate_capacity_strategies(demand_growth: float, current_capacity: float,
capacity_increment: float, options: dict):
"""
Evaluate Lead, Lag, and Match capacity strategies
demand_growth: annual growth rate
current_capacity: current capacity units
capacity_increment: size of capacity additions
options: {'planning_horizon': years, 'discount_rate': rate}
"""
horizon = options.get('planning_horizon', 5)
discount_rate = options.get('discount_rate', 0.10)
strategies = {}
# Project demand
demand = [current_capacity * (1 + demand_growth) ** t for t in range(horizon)]
# LEAD strategy - add capacity before needed
lead_capacity = [current_capacity]
lead_additions = []
for t in range(1, horizon):
if demand[t] > lead_capacity[-1] * 0.8: # Add at 80% utilization
lead_capacity.append(lead_capacity[-1] + capacity_increment)
lead_additions.append(t)
else:
lead_capacity.append(lead_capacity[-1])
strategies['lead'] = {
'description': 'Add capacity before demand reaches current capacity',
'capacity_timeline': lead_capacity,
'additions': lead_additions,
'average_utilization': np.mean([d / c for d, c in zip(demand, lead_capacity)]) * 100,
'lost_sales_risk': 'Low',
'capital_efficiency': 'Low'
}
# LAG strategy - add capacity after demand exceeds
lag_capacity = [current_capacity]
lag_additions = []
for t in range(1, horizon):
if demand[t-1] > lag_capacity[-1]: # Add after exceeding
lag_capacity.append(lag_capacity[-1] + capacity_increment)
lag_additions.append(t)
else:
lag_capacity.append(lag_capacity[-1])
strategies['lag'] = {
'description': 'Add capacity after demand exceeds current capacity',
'capacity_timeline': lag_capacity,
'additions': lag_additions,
'average_utilization': np.mean([min(1, d / c) for d, c in zip(demand, lag_capacity)]) * 100,
'lost_sales_risk': 'High',
'capital_efficiency': 'High'
}
# MATCH strategy - add capacity to match demand
match_capacity = [current_capacity]
match_additions = []
for t in range(1, horizon):
target_util = 0.85
needed = demand[t] / target_util
if