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
    }

**Overall Equipment Effectiveness (OEE):**

```python
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
    }

undefined

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
    }

**Drum-Buffer-Rope (DBR) Scheduling:**

```python
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)

---

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

undefined

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

---

import numpy as np

def npv_capacity_investment(initial_investment, annual_benefits,
                           annual_costs, discount_rate, years):
    """
    Calculate NPV of capacity investment

    Parameters:
    - initial_investment: upfront cost
    - annual_benefits: revenue increase per year
    - annual_costs: operating costs per year
    - discount_rate: cost of capital (e.g., 0.10 for 10%)
    - years: investment horizon
    """

    cash_flows = [-initial_investment]

    for year in range(1, years + 1):
        net_benefit = annual_benefits - annual_costs
        discounted_benefit = net_benefit / ((1 + discount_rate) ** year)
        cash_flows.append(discounted_benefit)

    npv = sum(cash_flows)

    # Calculate IRR (Internal Rate of Return)
    irr = np.irr([-initial_investment] + [annual_benefits - annual_costs] * years)

    # Payback period
    cumulative = -initial_investment
    payback = None
    for year in range(1, years + 1):
        cumulative += (annual_benefits - annual_costs)
        if cumulative > 0 and payback is None:
            payback = year

    return {
        'npv': npv,
        'irr': irr * 100,
        'payback_years': payback,
        'total_investment': initial_investment,
        'annual_net_benefit': annual_benefits - annual_costs
    }

undefined

import matplotlib.pyplot as plt
import numpy as np

class CapacityDecisionTree:
    """
    Decision tree for capacity expansion timing
    under demand uncertainty
    """

    def __init__(self):
        self.scenarios = []

    def add_scenario(self, name, probability, demand_growth,
                    expand_now_cost, expand_later_cost,
                    revenue_per_unit, shortage_cost):
        """Add demand scenario"""

        self.scenarios.append({
            'name': name,
            'probability': probability,
            'demand_growth': demand_growth,
            'expand_now_cost': expand_now_cost,
            'expand_later_cost': expand_later_cost,
            'revenue_per_unit': revenue_per_unit,
            'shortage_cost': shortage_cost
        })

    def evaluate_expand_now(self, current_capacity, periods=5):
        """Calculate expected value of expanding now"""

        ev = 0

        for scenario in self.scenarios:
            # Cost of expanding now
            cost = -scenario['expand_now_cost']

            # Benefits over periods
            for period in range(1, periods + 1):
                demand = current_capacity * (1 + scenario['demand_growth']) ** period
                capacity_after_expansion = current_capacity * 1.5  # Assume 50% expansion

                # Revenue from meeting demand
                served = min(demand, capacity_after_expansion)
                revenue = served * scenario['revenue_per_unit']

                # Discount
                discounted_revenue = revenue / (1.10 ** period)
                cost += discounted_revenue

            # Weight by probability
            ev += cost * scenario['probability']

        return ev

    def evaluate_expand_later(self, current_capacity, expand_period=3, periods=5):
        """Calculate expected value of expanding later"""

        ev = 0

        for scenario in self.scenarios:
            cost = 0

            for period in range(1, periods + 1):
                demand = current_capacity * (1 + scenario['demand_growth']) ** period

                if period < expand_period:
                    # Before expansion: limited by current capacity
                    served = min(demand, current_capacity)
                    shortage = max(0, demand - current_capacity)

                    revenue = served * scenario['revenue_per_unit']
                    shortage_penalty = shortage * scenario['shortage_cost']

                    discounted_value = (revenue - shortage_penalty) / (1.10 ** period)

                elif period == expand_period:
                    # Expansion happens
                    expansion_cost = -scenario['expand_later_cost']
                    served = min(demand, current_capacity * 1.5)
                    revenue = served * scenario['revenue_per_unit']

                    discounted_value = (expansion_cost + revenue) / (1.10 ** period)

                else:
                    # After expansion
                    served = min(demand, current_capacity * 1.5)
                    revenue = served * scenario['revenue_per_unit']

                    discounted_value = revenue / (1.10 ** period)

                cost += discounted_value

            # Weight by probability
            ev += cost * scenario['probability']

        return ev

    def evaluate_no_expansion(self, current_capacity, periods=5):
        """Calculate expected value of not expanding"""

        ev = 0

        for scenario in self.scenarios:
            cost = 0

            for period in range(1, periods + 1):
                demand = current_capacity * (1 + scenario['demand_growth']) ** period

                # Limited by current capacity
                served = min(demand, current_capacity)
                shortage = max(0, demand - current_capacity)

                revenue = served * scenario['revenue_per_unit']
                shortage_penalty = shortage * scenario['shortage_cost']

                discounted_value = (revenue - shortage_penalty) / (1.10 ** period)
                cost += discounted_value

            # Weight by probability
            ev += cost * scenario['probability']

        return ev

    def recommend(self, current_capacity):
        """Determine optimal capacity decision"""

        ev_now = self.evaluate_expand_now(current_capacity)
        ev_later = self.evaluate_expand_later(current_capacity)
        ev_no = self.evaluate_no_expansion(current_capacity)

        results = {
            'expand_now': ev_now,
            'expand_later': ev_later,
            'no_expansion': ev_no
        }

        best_option = max(results, key=results.get)

        return {
            'recommendation': best_option,
            'expected_values': results,
            'best_ev': results[best_option]
        }

---

def value_of_flexibility(demand_scenarios, fixed_capacity,
                        flexible_capacity, flexibility_cost):
    """
    Calculate value of flexible capacity using real options approach

    Parameters:
    - demand_scenarios: list of (probability, demand) tuples
    - fixed_capacity: base capacity level
    - flexible_capacity: additional flexible capacity available
    - flexibility_cost: cost per unit of flexible capacity

    Returns value of flexibility option
    """

    # Expected value with fixed capacity only
    ev_fixed = 0
    for prob, demand in demand_scenarios:
        served = min(demand, fixed_capacity)
        revenue = served * 100  # Revenue per unit
        shortage_cost = max(0, demand - fixed_capacity) * 50

        ev_fixed += prob * (revenue - shortage_cost)

    # Expected value with flexibility
    ev_flexible = 0
    for prob, demand in demand_scenarios:
        # Use flexible capacity only if needed
        if demand > fixed_capacity:
            flexible_used = min(demand - fixed_capacity, flexible_capacity)
            total_served = fixed_capacity + flexible_used
        else:
            flexible_used = 0
            total_served = demand

        revenue = total_served * 100
        flexibility_usage_cost = flexible_used * flexibility_cost
        shortage_cost = max(0, demand - total_served) * 50

        ev_flexible += prob * (revenue - flexibility_usage_cost - shortage_cost)

    value_of_flexibility = ev_flexible - ev_fixed

    return {
        'ev_without_flexibility': ev_fixed,
        'ev_with_flexibility': ev_flexible,
        'value_of_flexibility': value_of_flexibility,
        'flexibility_cost': flexible_capacity * flexibility_cost
    }