import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class MpsDemandPlanning:
    """Aggregate demand sources for MPS"""

    def __init__(self, planning_horizon_weeks=26):
        self.horizon = planning_horizon_weeks
        self.demand_sources = []

    def add_forecast_demand(self, item, periods, quantities):
        """Add forecasted demand"""
        self.demand_sources.append({
            'source': 'forecast',
            'item': item,
            'periods': periods,
            'quantities': quantities
        })

    def add_customer_orders(self, item, periods, quantities):
        """Add firm customer orders"""
        self.demand_sources.append({
            'source': 'customer_orders',
            'item': item,
            'periods': periods,
            'quantities': quantities
        })

    def add_safety_stock(self, item, target_level):
        """Add safety stock requirement"""
        self.demand_sources.append({
            'source': 'safety_stock',
            'item': item,
            'target': target_level
        })

    def calculate_total_demand(self):
        """
        Calculate total demand by item and period

        Uses consumption logic:
        - Customer orders consume forecast in near term
        - Forecast used beyond order horizon
        """

        # Create time buckets
        start_date = datetime.now()
        periods = pd.date_range(start_date, periods=self.horizon, freq='W')

        all_items = set()
        for source in self.demand_sources:
            all_items.add(source['item'])

        demand_df = pd.DataFrame({
            'period': periods
        })

        for item in all_items:
            # Get forecast
            forecast = self._get_source_demand(item, 'forecast', periods)

            # Get customer orders
            orders = self._get_source_demand(item, 'customer_orders', periods)

            # Apply consumption logic: greater of forecast or orders
            total_demand = np.maximum(forecast, orders)

            demand_df[f'{item}_forecast'] = forecast
            demand_df[f'{item}_orders'] = orders
            demand_df[f'{item}_total'] = total_demand

        return demand_df

    def _get_source_demand(self, item, source_type, periods):
        """Extract demand for specific item and source"""

        quantities = np.zeros(len(periods))

        for source in self.demand_sources:
            if source['item'] == item and source['source'] == source_type:
                if 'periods' in source:
                    for i, period in enumerate(source['periods']):
                        if i < len(quantities):
                            quantities[i] = source['quantities'][i]

        return quantities

undefined

Projected Available Balance (PAB) formula:
PAB[t] = PAB[t-1] + MPS[t] - max(Forecast[t], Orders[t])

When PAB[t] < Safety Stock:
  Schedule MPS receipt

import pandas as pd
import numpy as np

class MasterProductionSchedule:
    """Master Production Schedule generator"""

    def __init__(self, item, beginning_inventory,
                 safety_stock, lot_size, lead_time):
        """
        Parameters:
        - item: product identifier
        - beginning_inventory: starting inventory level
        - safety_stock: minimum inventory target
        - lot_size: production lot size (0 = lot-for-lot)
        - lead_time: production lead time (periods)
        """
        self.item = item
        self.beginning_inventory = beginning_inventory
        self.safety_stock = safety_stock
        self.lot_size = lot_size
        self.lead_time = lead_time

    def generate_mps(self, forecast, customer_orders, periods):
        """
        Generate MPS using standard MPS logic

        Returns DataFrame with MPS planning table
        """

        mps_table = []

        projected_available = self.beginning_inventory
        cumulative_mps = 0

        for t in range(periods):
            # Total demand (greater of forecast or orders)
            fcst = forecast[t] if t < len(forecast) else 0
            orders = customer_orders[t] if t < len(customer_orders) else 0
            total_demand = max(fcst, orders)

            # Check if MPS receipt needed
            if projected_available - total_demand < self.safety_stock:
                # Schedule MPS receipt
                shortage = self.safety_stock - (projected_available - total_demand)

                if self.lot_size > 0:
                    # Fixed lot size
                    lots_needed = int(np.ceil(shortage / self.lot_size))
                    mps_quantity = lots_needed * self.lot_size
                else:
                    # Lot-for-lot
                    mps_quantity = shortage

                # Account for lead time (schedule in future period)
                mps_receipt_period = t
            else:
                mps_quantity = 0
                mps_receipt_period = t

            # Update projected available
            projected_available = projected_available + mps_quantity - total_demand

            # Calculate Available-to-Promise (ATP)
            # ATP[t] = MPS[t] - sum(orders from t to next MPS)
            if mps_quantity > 0:
                atp = mps_quantity - orders
            else:
                atp = 0

            mps_table.append({
                'period': t + 1,
                'forecast': fcst,
                'customer_orders': orders,
                'total_demand': total_demand,
                'projected_available': max(0, projected_available),
                'mps_quantity': mps_quantity,
                'atp': max(0, atp)
            })

        return pd.DataFrame(mps_table)

    def calculate_rough_cut_capacity(self, mps_table, routing_time_per_unit):
        """
        Rough-Cut Capacity Planning (RCCP)

        Check if MPS is feasible given capacity

        Parameters:
        - mps_table: output from generate_mps
        - routing_time_per_unit: hours required per unit
        """

        mps_table['required_hours'] = (
            mps_table['mps_quantity'] * routing_time_per_unit
        )

        return mps_table[['period', 'mps_quantity', 'required_hours']]

undefined

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

class RoughCutCapacityPlanning:
    """RCCP - Validate MPS feasibility"""

    def __init__(self, work_centers):
        """
        Parameters:
        - work_centers: dict {name: {'capacity_hours': x, 'efficiency': y}}
        """
        self.work_centers = work_centers

    def validate_mps(self, mps_schedule, routings):
        """
        Check if MPS is feasible

        Parameters:
        - mps_schedule: DataFrame with 'item', 'period', 'mps_quantity'
        - routings: dict {item: [(work_center, hours_per_unit)]}

        Returns capacity analysis
        """

        requirements = []

        for idx, row in mps_schedule.iterrows():
            item = row['item']
            period = row['period']
            quantity = row['mps_quantity']

            if quantity == 0:
                continue

            # Get routing for item
            routing = routings.get(item, [])

            for work_center, hours_per_unit in routing:
                # Calculate required hours
                efficiency = self.work_centers[work_center]['efficiency']
                required_hours = (quantity * hours_per_unit) / efficiency

                requirements.append({
                    'period': period,
                    'work_center': work_center,
                    'item': item,
                    'quantity': quantity,
                    'required_hours': required_hours
                })

        req_df = pd.DataFrame(requirements)

        # Aggregate by work center and period
        capacity_analysis = req_df.groupby(
            ['period', 'work_center']
        )['required_hours'].sum().reset_index()

        # Add available capacity
        capacity_analysis['available_capacity'] = capacity_analysis['work_center'].map(
            lambda wc: self.work_centers[wc]['capacity_hours']
        )

        # Calculate load %
        capacity_analysis['load_pct'] = (
            capacity_analysis['required_hours'] /
            capacity_analysis['available_capacity'] * 100
        )

        capacity_analysis['status'] = capacity_analysis['load_pct'].apply(
            lambda x: 'Overload' if x > 100 else
                     'High' if x > 90 else
                     'OK'
        )

        return capacity_analysis

    def identify_overloads(self, capacity_analysis):
        """Get periods and resources with overload"""
        return capacity_analysis[capacity_analysis['status'] == 'Overload']

    def plot_capacity_profile(self, capacity_analysis):
        """Visualize capacity load"""

        work_centers = capacity_analysis['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 = capacity_analysis[
                capacity_analysis['work_center'] == wc
            ].sort_values('period')

            ax = axes[i, 0]

            # Plot capacity line
            ax.axhline(y=wc_data['available_capacity'].iloc[0],
                      color='green', linestyle='--',
                      linewidth=2, label='Available Capacity')

            # Plot requirements
            bars = ax.bar(wc_data['period'], wc_data['required_hours'],
                         alpha=0.7)

            # Color overloads red
            for j, (idx, row) in enumerate(wc_data.iterrows()):
                if row['status'] == 'Overload':
                    bars[j].set_color('red')

            ax.set_title(f'{wc} - Capacity Load')
            ax.set_xlabel('Period')
            ax.set_ylabel('Hours')
            ax.legend()
            ax.grid(True, alpha=0.3)

        plt.tight_layout()
        return fig

undefined

class AvailableToPromise:
    """Calculate ATP for order promising"""

    def __init__(self, mps_table, on_hand_inventory):
        self.mps = mps_table
        self.on_hand = on_hand_inventory

    def calculate_atp(self):
        """
        Calculate Available-to-Promise

        ATP provides visibility for order promising
        """

        atp_table = self.mps.copy()
        atp_table['atp'] = 0

        for i in range(len(atp_table)):
            if i == 0:
                # First period: On-hand + MPS - Orders
                atp_table.loc[i, 'atp'] = (
                    self.on_hand +
                    atp_table.loc[i, 'mps_quantity'] -
                    atp_table.loc[i, 'customer_orders']
                )
            else:
                # Check if MPS in this period
                if atp_table.loc[i, 'mps_quantity'] > 0:
                    # ATP = MPS - orders from this period until next MPS
                    orders_to_next_mps = 0

                    # Look ahead for orders until next MPS
                    for j in range(i, len(atp_table)):
                        orders_to_next_mps += atp_table.loc[j, 'customer_orders']

                        # Stop at next MPS
                        if j > i and atp_table.loc[j, 'mps_quantity'] > 0:
                            break

                    atp_table.loc[i, 'atp'] = (
                        atp_table.loc[i, 'mps_quantity'] - orders_to_next_mps
                    )

        return atp_table

    def check_order_promise(self, order_quantity, requested_period):
        """
        Check if order can be promised

        Returns: (can_promise, available_period)
        """

        atp_table = self.calculate_atp()

        # Check requested period first
        if requested_period <= len(atp_table):
            cumulative_atp = atp_table.loc[:requested_period-1, 'atp'].sum()

            if cumulative_atp >= order_quantity:
                return True, requested_period

        # Find earliest period with enough ATP
        cumulative = 0
        for i in range(len(atp_table)):
            cumulative += atp_table.loc[i, 'atp']

            if cumulative >= order_quantity:
                return True, i + 1

        # Cannot promise
        return False, None

undefined

|<-- Frozen Zone -->|<-- Slushy Zone -->|<-- Liquid Zone -->|
       2-4 weeks          4-8 weeks           8+ weeks

Frozen: No changes without executive approval
Slushy: Changes with planner approval
Liquid: Flexible, planning purposes only

from datetime import datetime, timedelta

class MpsTimeFences:
    """Manage MPS time fences and change control"""

    def __init__(self, frozen_weeks, slushy_weeks):
        self.frozen_weeks = frozen_weeks
        self.slushy_weeks = slushy_weeks

    def get_zone(self, period_date):
        """Determine time fence zone for a period"""

        now = datetime.now()
        frozen_end = now + timedelta(weeks=self.frozen_weeks)
        slushy_end = now + timedelta(weeks=self.frozen_weeks + self.slushy_weeks)

        if period_date <= frozen_end:
            return 'Frozen'
        elif period_date <= slushy_end:
            return 'Slushy'
        else:
            return 'Liquid'

    def approve_change(self, period_date, change_type, authority_level):
        """
        Check if change is allowed

        Parameters:
        - change_type: 'quantity', 'timing', 'cancellation'
        - authority_level: 'planner', 'manager', 'executive'

        Returns: (approved, reason)
        """

        zone = self.get_zone(period_date)

        if zone == 'Frozen':
            if authority_level == 'executive':
                return True, 'Executive approval in frozen zone'
            else:
                return False, 'Changes require executive approval in frozen zone'

        elif zone == 'Slushy':
            if authority_level in ['manager', 'executive']:
                return True, 'Manager approval in slushy zone'
            else:
                return False, 'Changes require manager approval in slushy zone'

        else:  # Liquid
            return True, 'Free to change in liquid zone'

    def mps_release(self, mps_table):
        """
        Release MPS with time fence zones marked

        Returns MPS with zones and change authority
        """

        mps_released = mps_table.copy()

        # Assume period is datetime
        mps_released['zone'] = mps_released['period'].apply(self.get_zone)

        mps_released['change_authority'] = mps_released['zone'].map({
            'Frozen': 'Executive',
            'Slushy': 'Manager',
            'Liquid': 'Planner'
        })

        return mps_released

---

def mts_mps_logic(forecast, current_inventory, safety_stock,
                  target_inventory, lot_size):
    """
    Make-to-Stock MPS logic

    Schedule production to maintain target inventory
    """

    mps_schedule = []

    inventory = current_inventory

    for period, demand in enumerate(forecast):
        # Check if replenishment needed
        projected_inventory = inventory - demand

        if projected_inventory < safety_stock:
            # Order up to target
            order_quantity = target_inventory - projected_inventory

            # Round to lot size
            if lot_size > 0:
                lots = int(np.ceil(order_quantity / lot_size))
                order_quantity = lots * lot_size

            inventory = projected_inventory + order_quantity
        else:
            order_quantity = 0
            inventory = projected_inventory

        mps_schedule.append({
            'period': period + 1,
            'demand': demand,
            'mps_quantity': order_quantity,
            'ending_inventory': inventory
        })

    return pd.DataFrame(mps_schedule)

undefined

def mto_mps_logic(customer_orders, production_lead_time):
    """
    Make-to-Order MPS logic

    Schedule production to meet order due dates
    """

    mps_schedule = []

    for order in customer_orders:
        order_id = order['order_id']
        quantity = order['quantity']
        due_date = order['due_date']

        # Backward schedule: start = due date - lead time
        start_period = due_date - production_lead_time

        mps_schedule.append({
            'order_id': order_id,
            'start_period': max(1, start_period),
            'due_period': due_date,
            'quantity': quantity,
            'order_type': 'customer_order'
        })

    return pd.DataFrame(mps_schedule)

undefined

class AssembleToOrderMPS:
    """ATO with planning bills and final assembly schedule"""

    def __init__(self, planning_bill):
        """
        planning_bill: dict {option: {component: percentage}}

        Example:
        {
            'Base_Model': {'Frame': 1.0, 'Engine_A': 0.6, 'Engine_B': 0.4},
            ...
        }
        """
        self.planning_bill = planning_bill

    def generate_component_mps(self, demand_forecast, planning_item):
        """
        Generate MPS for components based on option percentages

        Parameters:
        - demand_forecast: forecast for planning item
        - planning_item: aggregate product family
        """

        component_forecast = {}

        for component, percentage in self.planning_bill[planning_item].items():
            component_forecast[component] = [
                qty * percentage for qty in demand_forecast
            ]

        return component_forecast

    def create_final_assembly_schedule(self, customer_orders,
                                      assembly_lead_time):
        """
        Create FAS from customer orders

        FAS specifies exact configuration to build
        """

        fas = []

        for order in customer_orders:
            config = order['configuration']
            quantity = order['quantity']
            due_date = order['due_date']

            start_date = max(1, due_date - assembly_lead_time)

            fas.append({
                'order_id': order['order_id'],
                'configuration': config,
                'quantity': quantity,
                'start_period': start_date,
                'due_period': due_date
            })

        return pd.DataFrame(fas)

---

def level_loading_mps(total_demand, periods, lot_size=0):
    """
    Level loading - produce at constant rate

    Parameters:
    - total_demand: total demand over horizon
    - periods: number of periods
    - lot_size: production lot size (0 = continuous)
    """

    # Calculate average per period
    avg_demand = total_demand / periods

    # Round to lot size if applicable
    if lot_size > 0:
        level_quantity = int(np.ceil(avg_demand / lot_size)) * lot_size
    else:
        level_quantity = avg_demand

    # Create level schedule
    mps = [level_quantity] * periods

    return mps

undefined

def chase_strategy_mps(demand_forecast, lot_size=0):
    """
    Chase strategy - match production to demand

    Parameters:
    - demand_forecast: demand by period
    - lot_size: minimum production quantity
    """

    mps = []

    for demand in demand_forecast:
        if lot_size > 0:
            # Round up to lot size
            quantity = int(np.ceil(demand / lot_size)) * lot_size
        else:
            quantity = demand

        mps.append(quantity)

    return mps