kishorkukreja/awesome-supply-chain/capacity-planning

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%

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%

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'])

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)

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