401 lines
17 KiB
Python
401 lines
17 KiB
Python
"""
|
|
Production Service
|
|
Main business logic for production operations
|
|
"""
|
|
|
|
from typing import Optional, List, Dict, Any
|
|
from datetime import datetime, date, timedelta
|
|
from uuid import UUID
|
|
import structlog
|
|
|
|
from shared.database.transactions import transactional
|
|
from shared.clients import get_inventory_client, get_sales_client
|
|
from shared.clients.orders_client import OrdersServiceClient
|
|
from shared.clients.recipes_client import RecipesServiceClient
|
|
from shared.config.base import BaseServiceSettings
|
|
|
|
from app.repositories.production_batch_repository import ProductionBatchRepository
|
|
from app.repositories.production_schedule_repository import ProductionScheduleRepository
|
|
from app.repositories.production_capacity_repository import ProductionCapacityRepository
|
|
from app.repositories.quality_check_repository import QualityCheckRepository
|
|
from app.models.production import ProductionBatch, ProductionStatus, ProductionPriority
|
|
from app.schemas.production import (
|
|
ProductionBatchCreate, ProductionBatchUpdate, ProductionBatchStatusUpdate,
|
|
DailyProductionRequirements, ProductionDashboardSummary, ProductionMetrics
|
|
)
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class ProductionService:
|
|
"""Main production service with business logic"""
|
|
|
|
def __init__(self, database_manager, config: BaseServiceSettings):
|
|
self.database_manager = database_manager
|
|
self.config = config
|
|
|
|
# Initialize shared clients
|
|
self.inventory_client = get_inventory_client(config, "production")
|
|
self.orders_client = OrdersServiceClient(config)
|
|
self.recipes_client = RecipesServiceClient(config)
|
|
self.sales_client = get_sales_client(config, "production")
|
|
|
|
async def calculate_daily_requirements(
|
|
self,
|
|
tenant_id: UUID,
|
|
target_date: date
|
|
) -> DailyProductionRequirements:
|
|
"""Calculate production requirements using shared client pattern"""
|
|
try:
|
|
# 1. Get demand requirements from Orders Service
|
|
demand_data = await self.orders_client.get_demand_requirements(
|
|
str(tenant_id),
|
|
target_date.isoformat()
|
|
)
|
|
|
|
# 2. Get current stock levels from Inventory Service
|
|
stock_levels = await self.inventory_client.get_stock_levels(str(tenant_id))
|
|
|
|
# 3. Get recipe requirements from Recipes Service
|
|
recipe_data = await self.recipes_client.get_recipe_requirements(str(tenant_id))
|
|
|
|
# 4. Get capacity information
|
|
async with self.database_manager.get_session() as session:
|
|
capacity_repo = ProductionCapacityRepository(session)
|
|
available_capacity = await self._calculate_available_capacity(
|
|
capacity_repo, tenant_id, target_date
|
|
)
|
|
|
|
# 5. Apply production planning business logic
|
|
production_plan = await self._calculate_production_plan(
|
|
tenant_id, target_date, demand_data, stock_levels, recipe_data, available_capacity
|
|
)
|
|
|
|
return production_plan
|
|
|
|
except Exception as e:
|
|
logger.error("Error calculating daily production requirements",
|
|
error=str(e), tenant_id=str(tenant_id), date=target_date.isoformat())
|
|
raise
|
|
|
|
@transactional
|
|
async def create_production_batch(
|
|
self,
|
|
tenant_id: UUID,
|
|
batch_data: ProductionBatchCreate
|
|
) -> ProductionBatch:
|
|
"""Create a new production batch"""
|
|
try:
|
|
async with self.database_manager.get_session() as session:
|
|
batch_repo = ProductionBatchRepository(session)
|
|
|
|
# Prepare batch data
|
|
batch_dict = batch_data.model_dump()
|
|
batch_dict["tenant_id"] = tenant_id
|
|
|
|
# Validate recipe exists if provided
|
|
if batch_data.recipe_id:
|
|
recipe_details = await self.recipes_client.get_recipe_by_id(
|
|
str(tenant_id), str(batch_data.recipe_id)
|
|
)
|
|
if not recipe_details:
|
|
raise ValueError(f"Recipe {batch_data.recipe_id} not found")
|
|
|
|
# Check ingredient availability
|
|
if batch_data.recipe_id:
|
|
ingredient_requirements = await self.recipes_client.calculate_ingredients_for_quantity(
|
|
str(tenant_id), str(batch_data.recipe_id), batch_data.planned_quantity
|
|
)
|
|
|
|
if ingredient_requirements:
|
|
availability_check = await self.inventory_client.check_availability(
|
|
str(tenant_id), ingredient_requirements.get("requirements", [])
|
|
)
|
|
|
|
if not availability_check or not availability_check.get("all_available", True):
|
|
logger.warning("Insufficient ingredients for batch",
|
|
batch_data=batch_dict, availability=availability_check)
|
|
|
|
# Create the batch
|
|
batch = await batch_repo.create_batch(batch_dict)
|
|
|
|
logger.info("Production batch created",
|
|
batch_id=str(batch.id), tenant_id=str(tenant_id))
|
|
|
|
return batch
|
|
|
|
except Exception as e:
|
|
logger.error("Error creating production batch",
|
|
error=str(e), tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
@transactional
|
|
async def update_batch_status(
|
|
self,
|
|
tenant_id: UUID,
|
|
batch_id: UUID,
|
|
status_update: ProductionBatchStatusUpdate
|
|
) -> ProductionBatch:
|
|
"""Update production batch status"""
|
|
try:
|
|
async with self.database_manager.get_session() as session:
|
|
batch_repo = ProductionBatchRepository(session)
|
|
|
|
# Update batch status
|
|
batch = await batch_repo.update_batch_status(
|
|
batch_id,
|
|
status_update.status,
|
|
status_update.actual_quantity,
|
|
status_update.notes
|
|
)
|
|
|
|
# Update inventory if batch is completed
|
|
if status_update.status == ProductionStatus.COMPLETED and status_update.actual_quantity:
|
|
await self._update_inventory_on_completion(
|
|
tenant_id, batch, status_update.actual_quantity
|
|
)
|
|
|
|
logger.info("Updated batch status",
|
|
batch_id=str(batch_id),
|
|
new_status=status_update.status.value,
|
|
tenant_id=str(tenant_id))
|
|
|
|
return batch
|
|
|
|
except Exception as e:
|
|
logger.error("Error updating batch status",
|
|
error=str(e), batch_id=str(batch_id), tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
@transactional
|
|
async def get_dashboard_summary(self, tenant_id: UUID) -> ProductionDashboardSummary:
|
|
"""Get production dashboard summary data"""
|
|
try:
|
|
async with self.database_manager.get_session() as session:
|
|
batch_repo = ProductionBatchRepository(session)
|
|
|
|
# Get active batches
|
|
active_batches = await batch_repo.get_active_batches(str(tenant_id))
|
|
|
|
# Get today's production plan
|
|
today = date.today()
|
|
todays_batches = await batch_repo.get_batches_by_date_range(
|
|
str(tenant_id), today, today
|
|
)
|
|
|
|
# Calculate metrics
|
|
todays_plan = [
|
|
{
|
|
"product_name": batch.product_name,
|
|
"planned_quantity": batch.planned_quantity,
|
|
"status": batch.status.value,
|
|
"completion_time": batch.planned_end_time.isoformat() if batch.planned_end_time else None
|
|
}
|
|
for batch in todays_batches
|
|
]
|
|
|
|
# Get metrics for last 7 days
|
|
week_ago = today - timedelta(days=7)
|
|
weekly_metrics = await batch_repo.get_production_metrics(
|
|
str(tenant_id), week_ago, today
|
|
)
|
|
|
|
return ProductionDashboardSummary(
|
|
active_batches=len(active_batches),
|
|
todays_production_plan=todays_plan,
|
|
capacity_utilization=85.0, # TODO: Calculate from actual capacity data
|
|
on_time_completion_rate=weekly_metrics.get("on_time_completion_rate", 0),
|
|
average_quality_score=8.5, # TODO: Get from quality checks
|
|
total_output_today=sum(b.actual_quantity or 0 for b in todays_batches),
|
|
efficiency_percentage=weekly_metrics.get("average_yield_percentage", 0)
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting dashboard summary",
|
|
error=str(e), tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
@transactional
|
|
async def get_production_requirements(
|
|
self,
|
|
tenant_id: UUID,
|
|
target_date: Optional[date] = None
|
|
) -> Dict[str, Any]:
|
|
"""Get production requirements for procurement planning"""
|
|
try:
|
|
if not target_date:
|
|
target_date = date.today()
|
|
|
|
# Get planned batches for the date
|
|
async with self.database_manager.get_session() as session:
|
|
batch_repo = ProductionBatchRepository(session)
|
|
planned_batches = await batch_repo.get_batches_by_date_range(
|
|
str(tenant_id), target_date, target_date, ProductionStatus.PENDING
|
|
)
|
|
|
|
# Calculate ingredient requirements
|
|
total_requirements = {}
|
|
for batch in planned_batches:
|
|
if batch.recipe_id:
|
|
requirements = await self.recipes_client.calculate_ingredients_for_quantity(
|
|
str(tenant_id), str(batch.recipe_id), batch.planned_quantity
|
|
)
|
|
|
|
if requirements and "requirements" in requirements:
|
|
for req in requirements["requirements"]:
|
|
ingredient_id = req.get("ingredient_id")
|
|
quantity = req.get("quantity", 0)
|
|
|
|
if ingredient_id in total_requirements:
|
|
total_requirements[ingredient_id]["quantity"] += quantity
|
|
else:
|
|
total_requirements[ingredient_id] = {
|
|
"ingredient_id": ingredient_id,
|
|
"ingredient_name": req.get("ingredient_name"),
|
|
"quantity": quantity,
|
|
"unit": req.get("unit"),
|
|
"priority": "medium"
|
|
}
|
|
|
|
return {
|
|
"date": target_date.isoformat(),
|
|
"total_batches": len(planned_batches),
|
|
"ingredient_requirements": list(total_requirements.values()),
|
|
"estimated_start_time": "06:00:00",
|
|
"estimated_duration_hours": sum(b.planned_duration_minutes for b in planned_batches) / 60
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting production requirements",
|
|
error=str(e), tenant_id=str(tenant_id))
|
|
raise
|
|
|
|
async def _calculate_production_plan(
|
|
self,
|
|
tenant_id: UUID,
|
|
target_date: date,
|
|
demand_data: Optional[Dict[str, Any]],
|
|
stock_levels: Optional[Dict[str, Any]],
|
|
recipe_data: Optional[Dict[str, Any]],
|
|
available_capacity: Dict[str, Any]
|
|
) -> DailyProductionRequirements:
|
|
"""Apply production planning business logic"""
|
|
|
|
# Default production plan structure
|
|
production_plan = []
|
|
total_capacity_needed = 0.0
|
|
urgent_items = 0
|
|
|
|
if demand_data and "demand_items" in demand_data:
|
|
for item in demand_data["demand_items"]:
|
|
product_id = item.get("product_id")
|
|
demand_quantity = item.get("quantity", 0)
|
|
current_stock = 0
|
|
|
|
# Find current stock for this product
|
|
if stock_levels and "stock_levels" in stock_levels:
|
|
for stock in stock_levels["stock_levels"]:
|
|
if stock.get("product_id") == product_id:
|
|
current_stock = stock.get("available_quantity", 0)
|
|
break
|
|
|
|
# Calculate production need
|
|
production_needed = max(0, demand_quantity - current_stock)
|
|
|
|
if production_needed > 0:
|
|
# Determine urgency
|
|
urgency = "high" if demand_quantity > current_stock * 2 else "medium"
|
|
if urgency == "high":
|
|
urgent_items += 1
|
|
|
|
# Estimate capacity needed (simplified)
|
|
estimated_time_hours = production_needed * 0.5 # 30 minutes per unit
|
|
total_capacity_needed += estimated_time_hours
|
|
|
|
production_plan.append({
|
|
"product_id": product_id,
|
|
"product_name": item.get("product_name", f"Product {product_id}"),
|
|
"current_inventory": current_stock,
|
|
"demand_forecast": demand_quantity,
|
|
"pre_orders": item.get("pre_orders", 0),
|
|
"recommended_production": production_needed,
|
|
"urgency": urgency
|
|
})
|
|
|
|
return DailyProductionRequirements(
|
|
date=target_date,
|
|
production_plan=production_plan,
|
|
total_capacity_needed=total_capacity_needed,
|
|
available_capacity=available_capacity.get("total_hours", 8.0),
|
|
capacity_gap=max(0, total_capacity_needed - available_capacity.get("total_hours", 8.0)),
|
|
urgent_items=urgent_items,
|
|
recommended_schedule=None
|
|
)
|
|
|
|
async def _calculate_available_capacity(
|
|
self,
|
|
capacity_repo: ProductionCapacityRepository,
|
|
tenant_id: UUID,
|
|
target_date: date
|
|
) -> Dict[str, Any]:
|
|
"""Calculate available production capacity for a date"""
|
|
try:
|
|
# Get capacity entries for the date
|
|
equipment_capacity = await capacity_repo.get_available_capacity(
|
|
str(tenant_id), "equipment", target_date, 0
|
|
)
|
|
|
|
staff_capacity = await capacity_repo.get_available_capacity(
|
|
str(tenant_id), "staff", target_date, 0
|
|
)
|
|
|
|
# Calculate total available hours (simplified)
|
|
total_equipment_hours = sum(c.remaining_capacity_units for c in equipment_capacity)
|
|
total_staff_hours = sum(c.remaining_capacity_units for c in staff_capacity)
|
|
|
|
# Capacity is limited by the minimum of equipment or staff
|
|
effective_hours = min(total_equipment_hours, total_staff_hours) if total_staff_hours > 0 else total_equipment_hours
|
|
|
|
return {
|
|
"total_hours": effective_hours,
|
|
"equipment_hours": total_equipment_hours,
|
|
"staff_hours": total_staff_hours,
|
|
"utilization_percentage": 0 # To be calculated
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error calculating available capacity", error=str(e))
|
|
# Return default capacity if calculation fails
|
|
return {
|
|
"total_hours": 8.0,
|
|
"equipment_hours": 8.0,
|
|
"staff_hours": 8.0,
|
|
"utilization_percentage": 0
|
|
}
|
|
|
|
async def _update_inventory_on_completion(
|
|
self,
|
|
tenant_id: UUID,
|
|
batch: ProductionBatch,
|
|
actual_quantity: float
|
|
):
|
|
"""Update inventory when a batch is completed"""
|
|
try:
|
|
# Add the produced quantity to inventory
|
|
update_result = await self.inventory_client.update_stock_level(
|
|
str(tenant_id),
|
|
str(batch.product_id),
|
|
actual_quantity,
|
|
f"Production batch {batch.batch_number} completed"
|
|
)
|
|
|
|
logger.info("Updated inventory after production completion",
|
|
batch_id=str(batch.id),
|
|
product_id=str(batch.product_id),
|
|
quantity_added=actual_quantity,
|
|
update_result=update_result)
|
|
|
|
except Exception as e:
|
|
logger.error("Error updating inventory on batch completion",
|
|
error=str(e), batch_id=str(batch.id))
|
|
# Don't raise - inventory update failure shouldn't prevent batch completion |