445 lines
14 KiB
Python
445 lines
14 KiB
Python
"""
|
|
Shelf Life Manager
|
|
|
|
Manages shelf life constraints for perishable ingredients to minimize waste
|
|
and ensure food safety.
|
|
"""
|
|
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from typing import List, Dict, Optional, Tuple
|
|
from dataclasses import dataclass
|
|
import logging
|
|
import statistics
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ShelfLifeConstraint:
|
|
"""Shelf life constraints for an ingredient"""
|
|
ingredient_id: str
|
|
ingredient_name: str
|
|
shelf_life_days: int
|
|
is_perishable: bool
|
|
category: str # 'fresh', 'frozen', 'dry', 'canned'
|
|
max_order_quantity_days: Optional[int] = None # Max days worth to order at once
|
|
|
|
|
|
@dataclass
|
|
class ShelfLifeAdjustment:
|
|
"""Result of shelf life adjustment"""
|
|
original_quantity: Decimal
|
|
adjusted_quantity: Decimal
|
|
adjustment_reason: str
|
|
waste_risk: str # 'low', 'medium', 'high'
|
|
recommended_order_date: date
|
|
use_by_date: date
|
|
is_constrained: bool
|
|
|
|
|
|
class ShelfLifeManager:
|
|
"""
|
|
Manages procurement planning considering shelf life constraints.
|
|
|
|
For perishable items:
|
|
1. Don't order too far in advance (will expire)
|
|
2. Don't order too much at once (will waste)
|
|
3. Calculate optimal order timing
|
|
4. Warn about expiration risks
|
|
"""
|
|
|
|
# Category-specific defaults
|
|
CATEGORY_DEFAULTS = {
|
|
'fresh': {
|
|
'max_days_ahead': 2,
|
|
'max_order_days_supply': 3,
|
|
'waste_risk_threshold': 0.80
|
|
},
|
|
'dairy': {
|
|
'max_days_ahead': 3,
|
|
'max_order_days_supply': 5,
|
|
'waste_risk_threshold': 0.85
|
|
},
|
|
'frozen': {
|
|
'max_days_ahead': 14,
|
|
'max_order_days_supply': 30,
|
|
'waste_risk_threshold': 0.90
|
|
},
|
|
'dry': {
|
|
'max_days_ahead': 90,
|
|
'max_order_days_supply': 90,
|
|
'waste_risk_threshold': 0.95
|
|
},
|
|
'canned': {
|
|
'max_days_ahead': 180,
|
|
'max_order_days_supply': 180,
|
|
'waste_risk_threshold': 0.95
|
|
}
|
|
}
|
|
|
|
def __init__(self, waste_risk_threshold: float = 0.85):
|
|
"""
|
|
Initialize shelf life manager.
|
|
|
|
Args:
|
|
waste_risk_threshold: % of shelf life before considering waste risk
|
|
"""
|
|
self.waste_risk_threshold = waste_risk_threshold
|
|
|
|
def adjust_order_quantity_for_shelf_life(
|
|
self,
|
|
ingredient_id: str,
|
|
ingredient_name: str,
|
|
requested_quantity: Decimal,
|
|
daily_consumption_rate: float,
|
|
shelf_life_days: int,
|
|
category: str = 'dry',
|
|
is_perishable: bool = True,
|
|
delivery_date: Optional[date] = None
|
|
) -> ShelfLifeAdjustment:
|
|
"""
|
|
Adjust order quantity to prevent waste due to expiration.
|
|
|
|
Args:
|
|
ingredient_id: Ingredient ID
|
|
ingredient_name: Ingredient name
|
|
requested_quantity: Requested order quantity
|
|
daily_consumption_rate: Average daily usage
|
|
shelf_life_days: Days until expiration
|
|
category: Ingredient category
|
|
is_perishable: Whether item is perishable
|
|
delivery_date: Expected delivery date
|
|
|
|
Returns:
|
|
ShelfLifeAdjustment with adjusted quantity
|
|
"""
|
|
if not is_perishable:
|
|
# Non-perishable, no adjustment needed
|
|
return ShelfLifeAdjustment(
|
|
original_quantity=requested_quantity,
|
|
adjusted_quantity=requested_quantity,
|
|
adjustment_reason='Non-perishable item, no shelf life constraint',
|
|
waste_risk='low',
|
|
recommended_order_date=delivery_date or date.today(),
|
|
use_by_date=delivery_date + timedelta(days=365) if delivery_date else date.today() + timedelta(days=365),
|
|
is_constrained=False
|
|
)
|
|
|
|
if delivery_date is None:
|
|
delivery_date = date.today()
|
|
|
|
# Get category defaults
|
|
defaults = self.CATEGORY_DEFAULTS.get(
|
|
category.lower(),
|
|
self.CATEGORY_DEFAULTS['dry']
|
|
)
|
|
|
|
# Calculate use by date
|
|
use_by_date = delivery_date + timedelta(days=shelf_life_days)
|
|
|
|
# Calculate how many days the requested quantity will last
|
|
if daily_consumption_rate > 0:
|
|
days_supply = float(requested_quantity) / daily_consumption_rate
|
|
else:
|
|
days_supply = 0
|
|
|
|
# Calculate maximum safe quantity (using waste risk threshold)
|
|
safe_shelf_life_days = int(shelf_life_days * self.waste_risk_threshold)
|
|
max_safe_quantity = Decimal(str(daily_consumption_rate * safe_shelf_life_days))
|
|
|
|
# Check if adjustment needed
|
|
is_constrained = requested_quantity > max_safe_quantity
|
|
adjusted_quantity = requested_quantity
|
|
|
|
if is_constrained:
|
|
adjusted_quantity = max_safe_quantity
|
|
adjustment_reason = (
|
|
f"Reduced from {requested_quantity} to {adjusted_quantity} to fit within "
|
|
f"{safe_shelf_life_days}-day safe consumption window (shelf life: {shelf_life_days} days)"
|
|
)
|
|
logger.warning(
|
|
f"{ingredient_name}: Order quantity reduced due to shelf life constraint "
|
|
f"({requested_quantity} → {adjusted_quantity})"
|
|
)
|
|
else:
|
|
adjustment_reason = "Quantity within safe shelf life window"
|
|
|
|
# Calculate waste risk
|
|
waste_risk = self._calculate_waste_risk(
|
|
days_supply=days_supply,
|
|
shelf_life_days=shelf_life_days,
|
|
threshold=defaults['waste_risk_threshold']
|
|
)
|
|
|
|
return ShelfLifeAdjustment(
|
|
original_quantity=requested_quantity,
|
|
adjusted_quantity=adjusted_quantity,
|
|
adjustment_reason=adjustment_reason,
|
|
waste_risk=waste_risk,
|
|
recommended_order_date=delivery_date - timedelta(days=defaults['max_days_ahead']),
|
|
use_by_date=use_by_date,
|
|
is_constrained=is_constrained
|
|
)
|
|
|
|
def calculate_optimal_order_date(
|
|
self,
|
|
required_by_date: date,
|
|
shelf_life_days: int,
|
|
category: str = 'dry',
|
|
lead_time_days: int = 0
|
|
) -> Tuple[date, str]:
|
|
"""
|
|
Calculate optimal order date considering shelf life.
|
|
|
|
Args:
|
|
required_by_date: When item is needed
|
|
shelf_life_days: Shelf life in days
|
|
category: Ingredient category
|
|
lead_time_days: Supplier lead time
|
|
|
|
Returns:
|
|
Tuple of (optimal_order_date, reasoning)
|
|
"""
|
|
defaults = self.CATEGORY_DEFAULTS.get(
|
|
category.lower(),
|
|
self.CATEGORY_DEFAULTS['dry']
|
|
)
|
|
|
|
# Calculate delivery date accounting for lead time
|
|
delivery_date = required_by_date - timedelta(days=lead_time_days)
|
|
|
|
# For perishables, don't deliver too far in advance
|
|
max_advance_days = min(
|
|
defaults['max_days_ahead'],
|
|
int(shelf_life_days * 0.3) # Max 30% of shelf life
|
|
)
|
|
|
|
# Optimal delivery: close to required date but not too early
|
|
optimal_delivery_date = required_by_date - timedelta(days=max_advance_days)
|
|
|
|
# Optimal order date
|
|
optimal_order_date = optimal_delivery_date - timedelta(days=lead_time_days)
|
|
|
|
reasoning = (
|
|
f"Order placed {lead_time_days} days before delivery "
|
|
f"(arrives {max_advance_days} days before use to maintain freshness)"
|
|
)
|
|
|
|
return optimal_order_date, reasoning
|
|
|
|
def validate_order_timing(
|
|
self,
|
|
order_date: date,
|
|
delivery_date: date,
|
|
required_by_date: date,
|
|
shelf_life_days: int,
|
|
ingredient_name: str
|
|
) -> Tuple[bool, List[str]]:
|
|
"""
|
|
Validate order timing against shelf life constraints.
|
|
|
|
Args:
|
|
order_date: Planned order date
|
|
delivery_date: Expected delivery date
|
|
required_by_date: Date when item is needed
|
|
shelf_life_days: Shelf life in days
|
|
ingredient_name: Name of ingredient
|
|
|
|
Returns:
|
|
Tuple of (is_valid, list of warnings)
|
|
"""
|
|
warnings = []
|
|
|
|
# Check if item will arrive in time
|
|
if delivery_date > required_by_date:
|
|
warnings.append(
|
|
f"Delivery date {delivery_date} is after required date {required_by_date}"
|
|
)
|
|
|
|
# Check if item will expire before use
|
|
expiry_date = delivery_date + timedelta(days=shelf_life_days)
|
|
if expiry_date < required_by_date:
|
|
warnings.append(
|
|
f"Item will expire on {expiry_date} before required date {required_by_date}"
|
|
)
|
|
|
|
# Check if ordering too far in advance
|
|
days_in_storage = (required_by_date - delivery_date).days
|
|
if days_in_storage > shelf_life_days * 0.8:
|
|
warnings.append(
|
|
f"Item will be in storage for {days_in_storage} days "
|
|
f"(80% of {shelf_life_days}-day shelf life)"
|
|
)
|
|
|
|
is_valid = len(warnings) == 0
|
|
|
|
if not is_valid:
|
|
for warning in warnings:
|
|
logger.warning(f"{ingredient_name}: {warning}")
|
|
|
|
return is_valid, warnings
|
|
|
|
def calculate_fifo_rotation_schedule(
|
|
self,
|
|
current_inventory: List[Dict],
|
|
new_order_quantity: Decimal,
|
|
delivery_date: date,
|
|
daily_consumption: float
|
|
) -> List[Dict]:
|
|
"""
|
|
Calculate FIFO (First In First Out) rotation schedule.
|
|
|
|
Args:
|
|
current_inventory: List of existing batches with expiry dates
|
|
new_order_quantity: New order quantity
|
|
delivery_date: New order delivery date
|
|
daily_consumption: Daily consumption rate
|
|
|
|
Returns:
|
|
List of usage schedule
|
|
"""
|
|
# Combine current and new inventory
|
|
all_batches = []
|
|
|
|
for batch in current_inventory:
|
|
all_batches.append({
|
|
'quantity': batch['quantity'],
|
|
'expiry_date': batch['expiry_date'],
|
|
'is_existing': True
|
|
})
|
|
|
|
# Add new order (estimate shelf life from existing batches)
|
|
if current_inventory:
|
|
avg_shelf_life_days = statistics.mean([
|
|
(batch['expiry_date'] - date.today()).days
|
|
for batch in current_inventory
|
|
])
|
|
else:
|
|
avg_shelf_life_days = 30
|
|
|
|
all_batches.append({
|
|
'quantity': new_order_quantity,
|
|
'expiry_date': delivery_date + timedelta(days=int(avg_shelf_life_days)),
|
|
'is_existing': False
|
|
})
|
|
|
|
# Sort by expiry date (FIFO)
|
|
all_batches.sort(key=lambda x: x['expiry_date'])
|
|
|
|
# Create consumption schedule
|
|
schedule = []
|
|
current_date = date.today()
|
|
remaining_consumption = daily_consumption
|
|
|
|
for batch in all_batches:
|
|
days_until_expiry = (batch['expiry_date'] - current_date).days
|
|
batch_quantity = float(batch['quantity'])
|
|
|
|
# Calculate days to consume this batch
|
|
days_to_consume = min(
|
|
batch_quantity / daily_consumption,
|
|
days_until_expiry
|
|
)
|
|
|
|
quantity_consumed = days_to_consume * daily_consumption
|
|
waste = max(0, batch_quantity - quantity_consumed)
|
|
|
|
schedule.append({
|
|
'start_date': current_date,
|
|
'end_date': current_date + timedelta(days=int(days_to_consume)),
|
|
'quantity': batch['quantity'],
|
|
'quantity_consumed': Decimal(str(quantity_consumed)),
|
|
'quantity_wasted': Decimal(str(waste)),
|
|
'expiry_date': batch['expiry_date'],
|
|
'is_existing': batch['is_existing']
|
|
})
|
|
|
|
current_date += timedelta(days=int(days_to_consume))
|
|
|
|
return schedule
|
|
|
|
def _calculate_waste_risk(
|
|
self,
|
|
days_supply: float,
|
|
shelf_life_days: int,
|
|
threshold: float
|
|
) -> str:
|
|
"""
|
|
Calculate waste risk level.
|
|
|
|
Args:
|
|
days_supply: Days of supply ordered
|
|
shelf_life_days: Shelf life in days
|
|
threshold: Waste risk threshold
|
|
|
|
Returns:
|
|
Risk level: 'low', 'medium', 'high'
|
|
"""
|
|
if days_supply <= shelf_life_days * threshold * 0.5:
|
|
return 'low'
|
|
elif days_supply <= shelf_life_days * threshold:
|
|
return 'medium'
|
|
else:
|
|
return 'high'
|
|
|
|
def get_expiration_alerts(
|
|
self,
|
|
inventory_batches: List[Dict],
|
|
alert_days_threshold: int = 3
|
|
) -> List[Dict]:
|
|
"""
|
|
Get alerts for batches expiring soon.
|
|
|
|
Args:
|
|
inventory_batches: List of batches with expiry dates
|
|
alert_days_threshold: Days before expiry to alert
|
|
|
|
Returns:
|
|
List of expiration alerts
|
|
"""
|
|
alerts = []
|
|
today = date.today()
|
|
|
|
for batch in inventory_batches:
|
|
expiry_date = batch.get('expiry_date')
|
|
if not expiry_date:
|
|
continue
|
|
|
|
days_until_expiry = (expiry_date - today).days
|
|
|
|
if days_until_expiry <= alert_days_threshold:
|
|
alerts.append({
|
|
'ingredient_id': batch.get('ingredient_id'),
|
|
'ingredient_name': batch.get('ingredient_name'),
|
|
'quantity': batch.get('quantity'),
|
|
'expiry_date': expiry_date,
|
|
'days_until_expiry': days_until_expiry,
|
|
'severity': 'critical' if days_until_expiry <= 1 else 'high'
|
|
})
|
|
|
|
if alerts:
|
|
logger.warning(f"Found {len(alerts)} batches expiring within {alert_days_threshold} days")
|
|
|
|
return alerts
|
|
|
|
def export_to_dict(self, adjustment: ShelfLifeAdjustment) -> Dict:
|
|
"""
|
|
Export adjustment to dictionary for API response.
|
|
|
|
Args:
|
|
adjustment: ShelfLifeAdjustment
|
|
|
|
Returns:
|
|
Dictionary representation
|
|
"""
|
|
return {
|
|
'original_quantity': float(adjustment.original_quantity),
|
|
'adjusted_quantity': float(adjustment.adjusted_quantity),
|
|
'adjustment_reason': adjustment.adjustment_reason,
|
|
'waste_risk': adjustment.waste_risk,
|
|
'recommended_order_date': adjustment.recommended_order_date.isoformat(),
|
|
'use_by_date': adjustment.use_by_date.isoformat(),
|
|
'is_constrained': adjustment.is_constrained
|
|
}
|