capacity-planner

name	capacity-planner
description	Capacity requirements planning skill with demand-capacity analysis and strategic capacity decisions.
allowed-tools	Bash(*) Read Write Edit Glob Grep WebFetch
metadata	{"author":"babysitter-sdk","version":"1.0.0","category":"production-planning","backlog-id":"SK-IE-028"}
graph	{"domains":["domain:industrial-engineering"],"skillAreas":["skill-area:statistical-analysis","skill-area:organizational-design","skill-area:data-analysis"],"roles":["role:operations-analyst","role:research-engineer"]}

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'] > ][].(),
        : (gap_df[gap_df[] < ][].()),
        : gap_df[gap_df[] == ][].unique().tolist(),
        : gap_df[gap_df[] == ][].unique().tolist()
    }

     {
        : gap_df,
        : 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,
            : load / available *   available >   (),
            : load <= available
        })

     {
        : rccp_df,
        : resource_load,
        : pd.DataFrame(capacity_status),
        : [r  r  capacity_status   r[]]
    }

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,
        : ,
        : 
    }

    
    lag_capacity = [current_capacity]
    lag_additions = []
     t  (, horizon):
         demand[t-] > lag_capacity[-]:  
            lag_capacity.append(lag_capacity[-] + capacity_increment)
            lag_additions.append(t)
        :
            lag_capacity.append(lag_capacity[-])

    strategies[] = {
        : ,
        : lag_capacity,
        : lag_additions,
        : np.mean([(, d / c)  d, c  (demand, lag_capacity)]) * ,
        : ,
        : 
    }

    
    match_capacity = [current_capacity]
    match_additions = []
     t  (, horizon):
        target_util = 
        needed = demand[t] / target_util
         needed > match_capacity[-]:
            additions_needed = (np.ceil((needed - match_capacity[-]) / capacity_increment))
            match_capacity.append(match_capacity[-] + additions_needed * capacity_increment)
            match_additions.append(t)
        :
            match_capacity.append(match_capacity[-])

    strategies[] = {
        : ,
        : match_capacity,
        : match_additions,
        : ,  
        : ,
        : 
    }

     {
        : demand,
        : strategies,
        : recommend_strategy(strategies, demand_growth)
    }

 ():
     growth_rate > :
         {: , : }
     growth_rate < :
         {: , : }
    :
         {: , : }

5. Make vs Buy Analysis

def make_vs_buy_analysis(product: str, requirements: dict, options: dict):
    """
    Analyze make vs buy decision for capacity

    requirements: {'annual_volume': x, 'quality_spec': y}
    options: {'make_cost': {}, 'buy_cost': {}, 'strategic_factors': {}}
    """
    make_cost = options['make_cost']
    buy_cost = options['buy_cost']
    volume = requirements['annual_volume']

    # Calculate total costs
    make_total = (make_cost['fixed_annual'] +
                 make_cost['variable_per_unit'] * volume)
    buy_total = buy_cost['price_per_unit'] * volume

    # Break-even analysis
    if make_cost['variable_per_unit'] < buy_cost['price_per_unit']:
        breakeven_volume = make_cost['fixed_annual'] / \
                          (buy_cost['price_per_unit'] - make_cost['variable_per_unit'])
    else:
        breakeven_volume = float('inf')

    # Strategic factors
    strategic = options.get('strategic_factors', {})
    make_score = sum([
        strategic.get('core_competency', 0) * 10,
        strategic.get('quality_control', 0) * 8,
        strategic.get('lead_time_control', 0) * 6,
        strategic.get('capacity_utilization', 0) * 4
    ])
    buy_score = sum([
        strategic.get('flexibility', ) * ,
        strategic.get(, ) * ,
        strategic.get(, ) * ,
        strategic.get(, ) * 
    ])

    
    cost_advantage =   make_total < buy_total  
    strategic_advantage =   make_score > buy_score  

     {
        : {
            : make_total,
            : buy_total,
            : buy_total - make_total,
            : breakeven_volume
        },
        : {
            : make_score,
            : buy_score
        },
        : {
            : cost_advantage,
            : strategic_advantage,
            : cost_advantage  cost_advantage == strategic_advantage  
        }
    }

6. Capacity Investment Justification

def justify_capacity_investment(investment: dict, benefits: dict, financial_params: dict):
    """
    Build financial justification for capacity investment
    """
    initial_cost = investment['initial_cost']
    annual_costs = investment.get('annual_costs', 0)
    lifetime = investment.get('lifetime_years', 10)

    annual_benefits = benefits.get('annual_revenue_increase', 0) + \
                     benefits.get('annual_cost_savings', 0)

    discount_rate = financial_params.get('discount_rate', 0.10)
    tax_rate = financial_params.get('tax_rate', 0.25)

    # Net cash flows
    cash_flows = [-initial_cost]
    for year in range(1, lifetime + 1):
        net_benefit = (annual_benefits - annual_costs) * (1 - tax_rate)
        # Add depreciation tax shield
        depreciation = initial_cost / lifetime
        tax_shield = depreciation * tax_rate
        cash_flows.append(net_benefit + tax_shield)

    # NPV
    npv = sum(cf / (1 + discount_rate) ** t for t, cf in enumerate(cash_flows))

    # IRR (simplified - using numerical method)
    def npv_at_rate(rate):
        return sum(cf / (1 + rate) ** t for t, cf in (cash_flows))

    
    low, high = , 
     high - low > :
        mid = (low + high) / 
         npv_at_rate(mid) > :
            low = mid
        :
            high = mid
    irr = mid

    
    cumulative = 
    payback = lifetime
     t, cf  (cash_flows):
        cumulative += cf
         cumulative >=   t > :
            payback = t -  + (cumulative - cf) / cf
            

     {
        : {
            : initial_cost,
            : annual_benefits,
            : lifetime
        },
        : {
            : (npv, ),
            : (irr * , ),
            : (payback, ),
            : (npv + initial_cost) / initial_cost
        },
        :   npv >   irr > discount_rate  ,
        : {
            : initial_cost / lifetime / ( - tax_rate)
        }
    }

Process Integration

This skill integrates with the following processes:

• capacity-planning-analysis.js
• production-scheduling-optimization.js
• discrete-event-simulation-modeling.js

Output Format

{
  "capacity_requirements": {
    "total_required_hours": 4500,
    "by_work_center": {"Assembly": 2000, "Testing": 1500}
  },
  "gap_analysis": {
    "bottlenecks": ["Assembly"],
    "overcapacity_hours": 300
  },
  "strategy_recommendation": {
    "strategy": "match",
    "reason": "Moderate growth"
  },
  "investment_justification": {
    "NPV": 125000,
    "IRR": 18.5,
     
  

Best Practices

1. Use realistic efficiency - Account for losses
2. Consider variability - Buffer for uncertainty
3. Plan for flexibility - Don't over-specialize
4. Review regularly - Update as conditions change
5. Integrate with S&OP - Align with business planning
6. Validate assumptions - Test capacity models

Constraints

• Demand forecasts have uncertainty
• Lead times for capacity additions
• Skill requirements for new capacity
• Financial approval processes