367 lines
10 KiB
Python
367 lines
10 KiB
Python
"""
|
|
Lead Time Planner
|
|
|
|
Calculates order dates based on supplier lead times to ensure timely delivery.
|
|
"""
|
|
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from typing import List, Dict, Optional, Tuple
|
|
from dataclasses import dataclass
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class LeadTimeRequirement:
|
|
"""Requirement with lead time information"""
|
|
ingredient_id: str
|
|
ingredient_name: str
|
|
required_quantity: Decimal
|
|
required_by_date: date
|
|
supplier_id: Optional[str] = None
|
|
lead_time_days: int = 0
|
|
buffer_days: int = 1
|
|
|
|
|
|
@dataclass
|
|
class LeadTimePlan:
|
|
"""Planned order with dates"""
|
|
ingredient_id: str
|
|
ingredient_name: str
|
|
order_quantity: Decimal
|
|
order_date: date
|
|
delivery_date: date
|
|
required_by_date: date
|
|
lead_time_days: int
|
|
buffer_days: int
|
|
is_urgent: bool
|
|
urgency_reason: Optional[str] = None
|
|
supplier_id: Optional[str] = None
|
|
|
|
|
|
class LeadTimePlanner:
|
|
"""
|
|
Plans order dates based on supplier lead times.
|
|
|
|
Ensures that:
|
|
1. Orders are placed early enough for on-time delivery
|
|
2. Buffer days are added for risk mitigation
|
|
3. Urgent orders are identified
|
|
4. Weekend/holiday adjustments can be applied
|
|
"""
|
|
|
|
def __init__(self, default_buffer_days: int = 1):
|
|
"""
|
|
Initialize lead time planner.
|
|
|
|
Args:
|
|
default_buffer_days: Default buffer days to add
|
|
"""
|
|
self.default_buffer_days = default_buffer_days
|
|
|
|
def calculate_order_date(
|
|
self,
|
|
required_by_date: date,
|
|
lead_time_days: int,
|
|
buffer_days: Optional[int] = None
|
|
) -> date:
|
|
"""
|
|
Calculate when order should be placed.
|
|
|
|
Order Date = Required Date - Lead Time - Buffer
|
|
|
|
Args:
|
|
required_by_date: Date when item is needed
|
|
lead_time_days: Supplier lead time in days
|
|
buffer_days: Additional buffer days (uses default if None)
|
|
|
|
Returns:
|
|
Order date
|
|
"""
|
|
buffer = buffer_days if buffer_days is not None else self.default_buffer_days
|
|
total_days = lead_time_days + buffer
|
|
|
|
order_date = required_by_date - timedelta(days=total_days)
|
|
|
|
return order_date
|
|
|
|
def calculate_delivery_date(
|
|
self,
|
|
order_date: date,
|
|
lead_time_days: int
|
|
) -> date:
|
|
"""
|
|
Calculate expected delivery date.
|
|
|
|
Delivery Date = Order Date + Lead Time
|
|
|
|
Args:
|
|
order_date: Date when order is placed
|
|
lead_time_days: Supplier lead time in days
|
|
|
|
Returns:
|
|
Expected delivery date
|
|
"""
|
|
return order_date + timedelta(days=lead_time_days)
|
|
|
|
def is_urgent(
|
|
self,
|
|
order_date: date,
|
|
today: date,
|
|
urgency_threshold_days: int = 2
|
|
) -> Tuple[bool, Optional[str]]:
|
|
"""
|
|
Determine if order is urgent.
|
|
|
|
Args:
|
|
order_date: Calculated order date
|
|
today: Current date
|
|
urgency_threshold_days: Days threshold for urgency
|
|
|
|
Returns:
|
|
Tuple of (is_urgent, reason)
|
|
"""
|
|
days_until_order = (order_date - today).days
|
|
|
|
if days_until_order < 0:
|
|
return True, f"Order should have been placed {abs(days_until_order)} days ago"
|
|
elif days_until_order <= urgency_threshold_days:
|
|
return True, f"Order must be placed within {days_until_order} days"
|
|
else:
|
|
return False, None
|
|
|
|
def plan_requirements(
|
|
self,
|
|
requirements: List[LeadTimeRequirement],
|
|
today: Optional[date] = None
|
|
) -> List[LeadTimePlan]:
|
|
"""
|
|
Plan order dates for multiple requirements.
|
|
|
|
Args:
|
|
requirements: List of requirements with lead time info
|
|
today: Current date (defaults to today)
|
|
|
|
Returns:
|
|
List of lead time plans
|
|
"""
|
|
if today is None:
|
|
today = date.today()
|
|
|
|
plans = []
|
|
|
|
for req in requirements:
|
|
# Calculate order date
|
|
order_date = self.calculate_order_date(
|
|
required_by_date=req.required_by_date,
|
|
lead_time_days=req.lead_time_days,
|
|
buffer_days=req.buffer_days if hasattr(req, 'buffer_days') else None
|
|
)
|
|
|
|
# Calculate delivery date
|
|
delivery_date = self.calculate_delivery_date(
|
|
order_date=order_date,
|
|
lead_time_days=req.lead_time_days
|
|
)
|
|
|
|
# Check urgency
|
|
is_urgent, urgency_reason = self.is_urgent(
|
|
order_date=order_date,
|
|
today=today
|
|
)
|
|
|
|
# Create plan
|
|
plan = LeadTimePlan(
|
|
ingredient_id=req.ingredient_id,
|
|
ingredient_name=req.ingredient_name,
|
|
order_quantity=req.required_quantity,
|
|
order_date=max(order_date, today), # Can't order in the past
|
|
delivery_date=delivery_date,
|
|
required_by_date=req.required_by_date,
|
|
lead_time_days=req.lead_time_days,
|
|
buffer_days=self.default_buffer_days,
|
|
is_urgent=is_urgent,
|
|
urgency_reason=urgency_reason,
|
|
supplier_id=req.supplier_id
|
|
)
|
|
|
|
plans.append(plan)
|
|
|
|
if is_urgent:
|
|
logger.warning(
|
|
f"URGENT: {req.ingredient_name} - {urgency_reason}"
|
|
)
|
|
|
|
# Sort by order date (urgent first)
|
|
plans.sort(key=lambda p: (not p.is_urgent, p.order_date))
|
|
|
|
return plans
|
|
|
|
def adjust_for_working_days(
|
|
self,
|
|
target_date: date,
|
|
non_working_days: List[int] = None
|
|
) -> date:
|
|
"""
|
|
Adjust date to skip non-working days (e.g., weekends).
|
|
|
|
Args:
|
|
target_date: Original date
|
|
non_working_days: List of weekday numbers (0=Monday, 6=Sunday)
|
|
|
|
Returns:
|
|
Adjusted date
|
|
"""
|
|
if non_working_days is None:
|
|
non_working_days = [5, 6] # Saturday, Sunday
|
|
|
|
adjusted = target_date
|
|
|
|
# Move backwards to previous working day
|
|
while adjusted.weekday() in non_working_days:
|
|
adjusted -= timedelta(days=1)
|
|
|
|
return adjusted
|
|
|
|
def consolidate_orders_by_date(
|
|
self,
|
|
plans: List[LeadTimePlan],
|
|
consolidation_window_days: int = 3
|
|
) -> Dict[date, List[LeadTimePlan]]:
|
|
"""
|
|
Group orders that can be placed together.
|
|
|
|
Args:
|
|
plans: List of lead time plans
|
|
consolidation_window_days: Days within which to consolidate
|
|
|
|
Returns:
|
|
Dictionary mapping order date to list of plans
|
|
"""
|
|
if not plans:
|
|
return {}
|
|
|
|
# Sort plans by order date
|
|
sorted_plans = sorted(plans, key=lambda p: p.order_date)
|
|
|
|
consolidated: Dict[date, List[LeadTimePlan]] = {}
|
|
current_date = None
|
|
current_batch = []
|
|
|
|
for plan in sorted_plans:
|
|
if current_date is None:
|
|
current_date = plan.order_date
|
|
current_batch = [plan]
|
|
else:
|
|
days_diff = (plan.order_date - current_date).days
|
|
|
|
if days_diff <= consolidation_window_days:
|
|
# Within consolidation window
|
|
current_batch.append(plan)
|
|
else:
|
|
# Save current batch
|
|
consolidated[current_date] = current_batch
|
|
|
|
# Start new batch
|
|
current_date = plan.order_date
|
|
current_batch = [plan]
|
|
|
|
# Save last batch
|
|
if current_batch:
|
|
consolidated[current_date] = current_batch
|
|
|
|
logger.info(
|
|
f"Consolidated {len(plans)} orders into {len(consolidated)} order dates"
|
|
)
|
|
|
|
return consolidated
|
|
|
|
def calculate_coverage_window(
|
|
self,
|
|
order_date: date,
|
|
delivery_date: date,
|
|
required_by_date: date
|
|
) -> Dict[str, int]:
|
|
"""
|
|
Calculate time windows for an order.
|
|
|
|
Args:
|
|
order_date: When order is placed
|
|
delivery_date: When order arrives
|
|
required_by_date: When item is needed
|
|
|
|
Returns:
|
|
Dictionary with time windows
|
|
"""
|
|
return {
|
|
"order_to_delivery_days": (delivery_date - order_date).days,
|
|
"delivery_to_required_days": (required_by_date - delivery_date).days,
|
|
"total_lead_time_days": (delivery_date - order_date).days,
|
|
"buffer_time_days": (required_by_date - delivery_date).days
|
|
}
|
|
|
|
def validate_plan(
|
|
self,
|
|
plan: LeadTimePlan,
|
|
today: Optional[date] = None
|
|
) -> Tuple[bool, List[str]]:
|
|
"""
|
|
Validate a lead time plan for feasibility.
|
|
|
|
Args:
|
|
plan: Lead time plan to validate
|
|
today: Current date
|
|
|
|
Returns:
|
|
Tuple of (is_valid, list of issues)
|
|
"""
|
|
if today is None:
|
|
today = date.today()
|
|
|
|
issues = []
|
|
|
|
# Check if order date is in the past
|
|
if plan.order_date < today:
|
|
issues.append(f"Order date {plan.order_date} is in the past")
|
|
|
|
# Check if delivery date is before required date
|
|
if plan.delivery_date > plan.required_by_date:
|
|
days_late = (plan.delivery_date - plan.required_by_date).days
|
|
issues.append(
|
|
f"Delivery will be {days_late} days late (arrives {plan.delivery_date}, needed {plan.required_by_date})"
|
|
)
|
|
|
|
# Check if lead time is reasonable
|
|
if plan.lead_time_days > 90:
|
|
issues.append(f"Lead time of {plan.lead_time_days} days seems unusually long")
|
|
|
|
# Check if order quantity is valid
|
|
if plan.order_quantity <= 0:
|
|
issues.append(f"Order quantity {plan.order_quantity} is invalid")
|
|
|
|
is_valid = len(issues) == 0
|
|
|
|
return is_valid, issues
|
|
|
|
def get_urgent_orders(
|
|
self,
|
|
plans: List[LeadTimePlan]
|
|
) -> List[LeadTimePlan]:
|
|
"""
|
|
Filter to only urgent orders.
|
|
|
|
Args:
|
|
plans: List of lead time plans
|
|
|
|
Returns:
|
|
List of urgent plans
|
|
"""
|
|
urgent = [p for p in plans if p.is_urgent]
|
|
|
|
if urgent:
|
|
logger.warning(f"Found {len(urgent)} urgent orders requiring immediate attention")
|
|
|
|
return urgent
|