""" Replenishment Planning Service Main orchestrator for advanced procurement planning that integrates: - Lead time planning - Inventory projection - Safety stock calculation - Shelf life management """ from datetime import date, timedelta from decimal import Decimal from typing import List, Dict, Optional, Tuple from dataclasses import dataclass, asdict import logging import uuid from .lead_time_planner import LeadTimePlanner, LeadTimeRequirement, LeadTimePlan from .inventory_projector import ( InventoryProjector, DailyDemand, ScheduledReceipt, IngredientProjection ) from .safety_stock_calculator import SafetyStockCalculator, SafetyStockResult from .shelf_life_manager import ShelfLifeManager, ShelfLifeAdjustment logger = logging.getLogger(__name__) @dataclass class IngredientRequirement: """Complete requirement for one ingredient""" ingredient_id: str ingredient_name: str required_quantity: Decimal required_by_date: date supplier_id: Optional[str] = None lead_time_days: int = 3 shelf_life_days: Optional[int] = None is_perishable: bool = False category: str = 'dry' unit_of_measure: str = 'kg' current_stock: Decimal = Decimal('0') daily_consumption_rate: float = 0.0 demand_std_dev: float = 0.0 @dataclass class ReplenishmentPlanItem: """Single item in replenishment plan""" id: str ingredient_id: str ingredient_name: str # Quantities base_quantity: Decimal safety_stock_quantity: Decimal shelf_life_adjusted_quantity: Decimal final_order_quantity: Decimal # Dates order_date: date delivery_date: date required_by_date: date # Metadata lead_time_days: int is_urgent: bool urgency_reason: Optional[str] waste_risk: str stockout_risk: str supplier_id: Optional[str] # Calculation details safety_stock_calculation: Dict shelf_life_adjustment: Dict inventory_projection: Optional[Dict] @dataclass class ReplenishmentPlan: """Complete replenishment plan""" plan_id: str tenant_id: str planning_date: date projection_horizon_days: int items: List[ReplenishmentPlanItem] # Summary statistics total_items: int urgent_items: int high_risk_items: int total_estimated_cost: Decimal # Metadata created_at: date class ReplenishmentPlanningService: """ Orchestrates advanced replenishment planning. Workflow: 1. Project inventory levels (InventoryProjector) 2. Identify coverage gaps and required quantities 3. Calculate safety stock (SafetyStockCalculator) 4. Adjust for shelf life (ShelfLifeManager) 5. Calculate order dates (LeadTimePlanner) 6. Generate complete replenishment plan """ def __init__( self, projection_horizon_days: int = 7, default_service_level: float = 0.95, default_buffer_days: int = 1 ): """ Initialize replenishment planning service. Args: projection_horizon_days: Days to project ahead default_service_level: Default target service level default_buffer_days: Default buffer days for orders """ self.projection_horizon_days = projection_horizon_days # Initialize sub-services self.inventory_projector = InventoryProjector(projection_horizon_days) self.safety_stock_calculator = SafetyStockCalculator(default_service_level) self.shelf_life_manager = ShelfLifeManager() self.lead_time_planner = LeadTimePlanner(default_buffer_days) async def generate_replenishment_plan( self, tenant_id: str, requirements: List[IngredientRequirement], forecast_id: Optional[str] = None, production_schedule_id: Optional[str] = None ) -> ReplenishmentPlan: """ Generate complete replenishment plan. Args: tenant_id: Tenant ID requirements: List of ingredient requirements forecast_id: Optional reference to forecast production_schedule_id: Optional reference to production schedule Returns: Complete replenishment plan """ plan_id = str(uuid.uuid4()) planning_date = date.today() logger.info( f"Generating replenishment plan {plan_id} for {len(requirements)} ingredients" ) plan_items = [] for req in requirements: try: item = await self._plan_ingredient_replenishment(req) plan_items.append(item) except Exception as e: logger.error( f"Failed to plan replenishment for {req.ingredient_name}: {e}" ) # Continue with other ingredients # Calculate summary statistics total_items = len(plan_items) urgent_items = sum(1 for item in plan_items if item.is_urgent) high_risk_items = sum( 1 for item in plan_items if item.stockout_risk in ['high', 'critical'] ) # Estimate total cost (placeholder - need price data) total_estimated_cost = sum( item.final_order_quantity for item in plan_items ) plan = ReplenishmentPlan( plan_id=plan_id, tenant_id=tenant_id, planning_date=planning_date, projection_horizon_days=self.projection_horizon_days, items=plan_items, total_items=total_items, urgent_items=urgent_items, high_risk_items=high_risk_items, total_estimated_cost=total_estimated_cost, created_at=planning_date ) logger.info( f"Replenishment plan generated: {total_items} items, " f"{urgent_items} urgent, {high_risk_items} high risk" ) return plan async def _plan_ingredient_replenishment( self, req: IngredientRequirement ) -> ReplenishmentPlanItem: """ Plan replenishment for a single ingredient. Args: req: Ingredient requirement Returns: Replenishment plan item """ # Step 1: Project inventory to identify needs projection = await self._project_ingredient_inventory(req) # Step 2: Calculate base quantity needed base_quantity = self._calculate_base_quantity(req, projection) # Step 3: Calculate safety stock safety_stock_result = self._calculate_safety_stock(req) safety_stock_quantity = safety_stock_result.safety_stock_quantity # Step 4: Adjust for shelf life total_quantity = base_quantity + safety_stock_quantity shelf_life_adjustment = self._adjust_for_shelf_life( req, total_quantity ) # Step 5: Calculate order dates lead_time_plan = self._calculate_order_dates( req, shelf_life_adjustment.adjusted_quantity ) # Create plan item item = ReplenishmentPlanItem( id=str(uuid.uuid4()), ingredient_id=req.ingredient_id, ingredient_name=req.ingredient_name, base_quantity=base_quantity, safety_stock_quantity=safety_stock_quantity, shelf_life_adjusted_quantity=shelf_life_adjustment.adjusted_quantity, final_order_quantity=shelf_life_adjustment.adjusted_quantity, order_date=lead_time_plan.order_date, delivery_date=lead_time_plan.delivery_date, required_by_date=req.required_by_date, lead_time_days=req.lead_time_days, is_urgent=lead_time_plan.is_urgent, urgency_reason=lead_time_plan.urgency_reason, waste_risk=shelf_life_adjustment.waste_risk, stockout_risk=projection.stockout_risk if projection else 'unknown', supplier_id=req.supplier_id, safety_stock_calculation=self.safety_stock_calculator.export_to_dict(safety_stock_result), shelf_life_adjustment=self.shelf_life_manager.export_to_dict(shelf_life_adjustment), inventory_projection=self.inventory_projector.export_projection_to_dict(projection) if projection else None ) return item async def _project_ingredient_inventory( self, req: IngredientRequirement ) -> Optional[IngredientProjection]: """ Project inventory for ingredient. Args: req: Ingredient requirement Returns: Inventory projection """ try: # Build daily demand forecast daily_demand = [] if req.daily_consumption_rate > 0: for i in range(self.projection_horizon_days): demand_date = date.today() + timedelta(days=i) daily_demand.append( DailyDemand( ingredient_id=req.ingredient_id, date=demand_date, quantity=Decimal(str(req.daily_consumption_rate)) ) ) # No scheduled receipts for now (could add future POs here) scheduled_receipts = [] projection = self.inventory_projector.project_inventory( ingredient_id=req.ingredient_id, ingredient_name=req.ingredient_name, current_stock=req.current_stock, unit_of_measure=req.unit_of_measure, daily_demand=daily_demand, scheduled_receipts=scheduled_receipts ) return projection except Exception as e: logger.error(f"Failed to project inventory for {req.ingredient_name}: {e}") return None def _calculate_base_quantity( self, req: IngredientRequirement, projection: Optional[IngredientProjection] ) -> Decimal: """ Calculate base quantity needed. Args: req: Ingredient requirement projection: Inventory projection Returns: Base quantity """ if projection: # Use projection to calculate need required = self.inventory_projector.calculate_required_order_quantity( projection, target_coverage_days=self.projection_horizon_days ) return max(required, req.required_quantity) else: # Fallback to required quantity return req.required_quantity def _calculate_safety_stock( self, req: IngredientRequirement ) -> SafetyStockResult: """ Calculate safety stock. Args: req: Ingredient requirement Returns: Safety stock result """ if req.demand_std_dev > 0: # Use statistical method return self.safety_stock_calculator.calculate_safety_stock( demand_std_dev=req.demand_std_dev, lead_time_days=req.lead_time_days ) elif req.daily_consumption_rate > 0: # Use percentage method return self.safety_stock_calculator.calculate_using_fixed_percentage( average_demand=req.daily_consumption_rate, lead_time_days=req.lead_time_days, percentage=0.20 ) else: # No safety stock return SafetyStockResult( safety_stock_quantity=Decimal('0'), service_level=0.0, z_score=0.0, demand_std_dev=0.0, lead_time_days=req.lead_time_days, calculation_method='none', confidence='low', reasoning='ERROR:INSUFFICIENT_DATA' # Error code for i18n translation ) def _adjust_for_shelf_life( self, req: IngredientRequirement, quantity: Decimal ) -> ShelfLifeAdjustment: """ Adjust quantity for shelf life constraints. Args: req: Ingredient requirement quantity: Proposed quantity Returns: Shelf life adjustment """ if not req.is_perishable or not req.shelf_life_days: # No shelf life constraint return ShelfLifeAdjustment( original_quantity=quantity, adjusted_quantity=quantity, adjustment_reason='Non-perishable or no shelf life data', waste_risk='low', recommended_order_date=date.today(), use_by_date=date.today() + timedelta(days=365), is_constrained=False ) return self.shelf_life_manager.adjust_order_quantity_for_shelf_life( ingredient_id=req.ingredient_id, ingredient_name=req.ingredient_name, requested_quantity=quantity, daily_consumption_rate=req.daily_consumption_rate, shelf_life_days=req.shelf_life_days, category=req.category, is_perishable=req.is_perishable, delivery_date=req.required_by_date - timedelta(days=req.lead_time_days) ) def _calculate_order_dates( self, req: IngredientRequirement, quantity: Decimal ) -> LeadTimePlan: """ Calculate order and delivery dates. Args: req: Ingredient requirement quantity: Order quantity Returns: Lead time plan """ lead_time_req = LeadTimeRequirement( ingredient_id=req.ingredient_id, ingredient_name=req.ingredient_name, required_quantity=quantity, required_by_date=req.required_by_date, supplier_id=req.supplier_id, lead_time_days=req.lead_time_days ) plans = self.lead_time_planner.plan_requirements([lead_time_req]) return plans[0] if plans else LeadTimePlan( ingredient_id=req.ingredient_id, ingredient_name=req.ingredient_name, order_quantity=quantity, order_date=date.today(), delivery_date=date.today() + timedelta(days=req.lead_time_days), required_by_date=req.required_by_date, lead_time_days=req.lead_time_days, buffer_days=1, is_urgent=False, supplier_id=req.supplier_id ) def export_plan_to_dict(self, plan: ReplenishmentPlan) -> Dict: """ Export plan to dictionary for API response. Args: plan: Replenishment plan Returns: Dictionary representation """ return { 'plan_id': plan.plan_id, 'tenant_id': plan.tenant_id, 'planning_date': plan.planning_date.isoformat(), 'projection_horizon_days': plan.projection_horizon_days, 'total_items': plan.total_items, 'urgent_items': plan.urgent_items, 'high_risk_items': plan.high_risk_items, 'total_estimated_cost': float(plan.total_estimated_cost), 'created_at': plan.created_at.isoformat(), 'items': [ { 'id': item.id, 'ingredient_id': item.ingredient_id, 'ingredient_name': item.ingredient_name, 'base_quantity': float(item.base_quantity), 'safety_stock_quantity': float(item.safety_stock_quantity), 'shelf_life_adjusted_quantity': float(item.shelf_life_adjusted_quantity), 'final_order_quantity': float(item.final_order_quantity), 'order_date': item.order_date.isoformat(), 'delivery_date': item.delivery_date.isoformat(), 'required_by_date': item.required_by_date.isoformat(), 'lead_time_days': item.lead_time_days, 'is_urgent': item.is_urgent, 'urgency_reason': item.urgency_reason, 'waste_risk': item.waste_risk, 'stockout_risk': item.stockout_risk, 'supplier_id': item.supplier_id, 'safety_stock_calculation': item.safety_stock_calculation, 'shelf_life_adjustment': item.shelf_life_adjustment, 'inventory_projection': item.inventory_projection } for item in plan.items ] }