347 lines
11 KiB
Python
347 lines
11 KiB
Python
# ================================================================
|
|
# services/production/app/schemas/production.py
|
|
# ================================================================
|
|
"""
|
|
Pydantic schemas for production service
|
|
"""
|
|
|
|
from pydantic import BaseModel, Field, validator
|
|
from typing import Optional, List, Dict, Any, Union
|
|
from datetime import datetime, date
|
|
from uuid import UUID
|
|
from enum import Enum
|
|
|
|
|
|
class ProductionStatusEnum(str, Enum):
|
|
"""Production batch status enumeration for API"""
|
|
PENDING = "PENDING"
|
|
IN_PROGRESS = "IN_PROGRESS"
|
|
COMPLETED = "COMPLETED"
|
|
CANCELLED = "CANCELLED"
|
|
ON_HOLD = "ON_HOLD"
|
|
QUALITY_CHECK = "QUALITY_CHECK"
|
|
FAILED = "FAILED"
|
|
|
|
|
|
class ProductionPriorityEnum(str, Enum):
|
|
"""Production priority levels for API"""
|
|
LOW = "LOW"
|
|
MEDIUM = "MEDIUM"
|
|
HIGH = "HIGH"
|
|
URGENT = "URGENT"
|
|
|
|
|
|
|
|
|
|
# ================================================================
|
|
# PRODUCTION BATCH SCHEMAS
|
|
# ================================================================
|
|
|
|
class ProductionBatchBase(BaseModel):
|
|
"""Base schema for production batch"""
|
|
product_id: UUID
|
|
product_name: str = Field(..., min_length=1, max_length=255)
|
|
recipe_id: Optional[UUID] = None
|
|
planned_start_time: datetime
|
|
planned_end_time: datetime
|
|
planned_quantity: float = Field(..., gt=0)
|
|
planned_duration_minutes: int = Field(..., gt=0)
|
|
priority: ProductionPriorityEnum = ProductionPriorityEnum.MEDIUM
|
|
is_rush_order: bool = False
|
|
is_special_recipe: bool = False
|
|
production_notes: Optional[str] = None
|
|
|
|
@validator('planned_end_time')
|
|
def validate_end_time_after_start(cls, v, values):
|
|
if 'planned_start_time' in values and v <= values['planned_start_time']:
|
|
raise ValueError('planned_end_time must be after planned_start_time')
|
|
return v
|
|
|
|
|
|
class ProductionBatchCreate(ProductionBatchBase):
|
|
"""Schema for creating a production batch"""
|
|
batch_number: Optional[str] = Field(None, max_length=50)
|
|
order_id: Optional[UUID] = None
|
|
forecast_id: Optional[UUID] = None
|
|
equipment_used: Optional[List[str]] = None
|
|
staff_assigned: Optional[List[str]] = None
|
|
station_id: Optional[str] = Field(None, max_length=50)
|
|
|
|
|
|
class ProductionBatchUpdate(BaseModel):
|
|
"""Schema for updating a production batch"""
|
|
product_name: Optional[str] = Field(None, min_length=1, max_length=255)
|
|
planned_start_time: Optional[datetime] = None
|
|
planned_end_time: Optional[datetime] = None
|
|
planned_quantity: Optional[float] = Field(None, gt=0)
|
|
planned_duration_minutes: Optional[int] = Field(None, gt=0)
|
|
actual_quantity: Optional[float] = Field(None, ge=0)
|
|
priority: Optional[ProductionPriorityEnum] = None
|
|
equipment_used: Optional[List[str]] = None
|
|
staff_assigned: Optional[List[str]] = None
|
|
station_id: Optional[str] = Field(None, max_length=50)
|
|
production_notes: Optional[str] = None
|
|
|
|
|
|
class ProductionBatchStatusUpdate(BaseModel):
|
|
"""Schema for updating production batch status"""
|
|
status: ProductionStatusEnum
|
|
actual_quantity: Optional[float] = Field(None, ge=0)
|
|
notes: Optional[str] = None
|
|
|
|
|
|
class ProductionBatchResponse(BaseModel):
|
|
"""Schema for production batch response"""
|
|
id: UUID
|
|
tenant_id: UUID
|
|
batch_number: str
|
|
product_id: UUID
|
|
product_name: str
|
|
recipe_id: Optional[UUID]
|
|
planned_start_time: datetime
|
|
planned_end_time: datetime
|
|
planned_quantity: float
|
|
planned_duration_minutes: int
|
|
actual_start_time: Optional[datetime]
|
|
actual_end_time: Optional[datetime]
|
|
actual_quantity: Optional[float]
|
|
actual_duration_minutes: Optional[int]
|
|
status: ProductionStatusEnum
|
|
priority: ProductionPriorityEnum
|
|
estimated_cost: Optional[float]
|
|
actual_cost: Optional[float]
|
|
yield_percentage: Optional[float]
|
|
quality_score: Optional[float]
|
|
equipment_used: Optional[List[str]]
|
|
staff_assigned: Optional[List[str]]
|
|
station_id: Optional[str]
|
|
order_id: Optional[UUID]
|
|
forecast_id: Optional[UUID]
|
|
is_rush_order: bool
|
|
is_special_recipe: bool
|
|
production_notes: Optional[str]
|
|
quality_notes: Optional[str]
|
|
delay_reason: Optional[str]
|
|
cancellation_reason: Optional[str]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
completed_at: Optional[datetime]
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
# ================================================================
|
|
# PRODUCTION SCHEDULE SCHEMAS
|
|
# ================================================================
|
|
|
|
class ProductionScheduleBase(BaseModel):
|
|
"""Base schema for production schedule"""
|
|
schedule_date: date
|
|
shift_start: datetime
|
|
shift_end: datetime
|
|
total_capacity_hours: float = Field(..., gt=0)
|
|
planned_capacity_hours: float = Field(..., gt=0)
|
|
staff_count: int = Field(..., gt=0)
|
|
equipment_capacity: Optional[Dict[str, Any]] = None
|
|
station_assignments: Optional[Dict[str, Any]] = None
|
|
schedule_notes: Optional[str] = None
|
|
|
|
@validator('shift_end')
|
|
def validate_shift_end_after_start(cls, v, values):
|
|
if 'shift_start' in values and v <= values['shift_start']:
|
|
raise ValueError('shift_end must be after shift_start')
|
|
return v
|
|
|
|
@validator('planned_capacity_hours')
|
|
def validate_planned_capacity(cls, v, values):
|
|
if 'total_capacity_hours' in values and v > values['total_capacity_hours']:
|
|
raise ValueError('planned_capacity_hours cannot exceed total_capacity_hours')
|
|
return v
|
|
|
|
|
|
class ProductionScheduleCreate(ProductionScheduleBase):
|
|
"""Schema for creating a production schedule"""
|
|
pass
|
|
|
|
|
|
class ProductionScheduleUpdate(BaseModel):
|
|
"""Schema for updating a production schedule"""
|
|
shift_start: Optional[datetime] = None
|
|
shift_end: Optional[datetime] = None
|
|
total_capacity_hours: Optional[float] = Field(None, gt=0)
|
|
planned_capacity_hours: Optional[float] = Field(None, gt=0)
|
|
staff_count: Optional[int] = Field(None, gt=0)
|
|
overtime_hours: Optional[float] = Field(None, ge=0)
|
|
equipment_capacity: Optional[Dict[str, Any]] = None
|
|
station_assignments: Optional[Dict[str, Any]] = None
|
|
schedule_notes: Optional[str] = None
|
|
|
|
|
|
class ProductionScheduleResponse(BaseModel):
|
|
"""Schema for production schedule response"""
|
|
id: UUID
|
|
tenant_id: UUID
|
|
schedule_date: date
|
|
shift_start: datetime
|
|
shift_end: datetime
|
|
total_capacity_hours: float
|
|
planned_capacity_hours: float
|
|
actual_capacity_hours: Optional[float]
|
|
overtime_hours: Optional[float]
|
|
staff_count: int
|
|
equipment_capacity: Optional[Dict[str, Any]]
|
|
station_assignments: Optional[Dict[str, Any]]
|
|
total_batches_planned: int
|
|
total_batches_completed: Optional[int]
|
|
total_quantity_planned: float
|
|
total_quantity_produced: Optional[float]
|
|
is_finalized: bool
|
|
is_active: bool
|
|
efficiency_percentage: Optional[float]
|
|
utilization_percentage: Optional[float]
|
|
on_time_completion_rate: Optional[float]
|
|
schedule_notes: Optional[str]
|
|
schedule_adjustments: Optional[Dict[str, Any]]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
finalized_at: Optional[datetime]
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
# ================================================================
|
|
# QUALITY CHECK SCHEMAS
|
|
# ================================================================
|
|
|
|
class QualityCheckBase(BaseModel):
|
|
"""Base schema for quality check"""
|
|
batch_id: UUID
|
|
check_type: str = Field(..., min_length=1, max_length=50)
|
|
check_time: datetime
|
|
quality_score: float = Field(..., ge=1, le=10)
|
|
pass_fail: bool
|
|
defect_count: int = Field(0, ge=0)
|
|
defect_types: Optional[List[str]] = None
|
|
check_notes: Optional[str] = None
|
|
|
|
|
|
class QualityCheckCreate(QualityCheckBase):
|
|
"""Schema for creating a quality check"""
|
|
checker_id: Optional[str] = Field(None, max_length=100)
|
|
measured_weight: Optional[float] = Field(None, gt=0)
|
|
measured_temperature: Optional[float] = None
|
|
measured_moisture: Optional[float] = Field(None, ge=0, le=100)
|
|
measured_dimensions: Optional[Dict[str, float]] = None
|
|
target_weight: Optional[float] = Field(None, gt=0)
|
|
target_temperature: Optional[float] = None
|
|
target_moisture: Optional[float] = Field(None, ge=0, le=100)
|
|
tolerance_percentage: Optional[float] = Field(None, ge=0, le=100)
|
|
corrective_actions: Optional[List[str]] = None
|
|
|
|
|
|
class QualityCheckResponse(BaseModel):
|
|
"""Schema for quality check response"""
|
|
id: UUID
|
|
tenant_id: UUID
|
|
batch_id: UUID
|
|
check_type: str
|
|
check_time: datetime
|
|
checker_id: Optional[str]
|
|
quality_score: float
|
|
pass_fail: bool
|
|
defect_count: int
|
|
defect_types: Optional[List[str]]
|
|
measured_weight: Optional[float]
|
|
measured_temperature: Optional[float]
|
|
measured_moisture: Optional[float]
|
|
measured_dimensions: Optional[Dict[str, float]]
|
|
target_weight: Optional[float]
|
|
target_temperature: Optional[float]
|
|
target_moisture: Optional[float]
|
|
tolerance_percentage: Optional[float]
|
|
within_tolerance: Optional[bool]
|
|
corrective_action_needed: bool
|
|
corrective_actions: Optional[List[str]]
|
|
check_notes: Optional[str]
|
|
photos_urls: Optional[List[str]]
|
|
certificate_url: Optional[str]
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
|
|
|
|
# ================================================================
|
|
# DASHBOARD AND ANALYTICS SCHEMAS
|
|
# ================================================================
|
|
|
|
class ProductionDashboardSummary(BaseModel):
|
|
"""Schema for production dashboard summary"""
|
|
active_batches: int
|
|
todays_production_plan: List[Dict[str, Any]]
|
|
capacity_utilization: float
|
|
on_time_completion_rate: float
|
|
average_quality_score: float
|
|
total_output_today: float
|
|
efficiency_percentage: float
|
|
|
|
|
|
class DailyProductionRequirements(BaseModel):
|
|
"""Schema for daily production requirements"""
|
|
date: date
|
|
production_plan: List[Dict[str, Any]]
|
|
total_capacity_needed: float
|
|
available_capacity: float
|
|
capacity_gap: float
|
|
urgent_items: int
|
|
recommended_schedule: Optional[Dict[str, Any]]
|
|
|
|
|
|
class ProductionMetrics(BaseModel):
|
|
"""Schema for production metrics"""
|
|
period_start: date
|
|
period_end: date
|
|
total_batches: int
|
|
completed_batches: int
|
|
completion_rate: float
|
|
average_yield_percentage: float
|
|
on_time_completion_rate: float
|
|
total_production_cost: float
|
|
average_quality_score: float
|
|
efficiency_trends: List[Dict[str, Any]]
|
|
|
|
|
|
# ================================================================
|
|
# REQUEST/RESPONSE WRAPPERS
|
|
# ================================================================
|
|
|
|
class ProductionBatchListResponse(BaseModel):
|
|
"""Schema for production batch list response"""
|
|
batches: List[ProductionBatchResponse]
|
|
total_count: int
|
|
page: int
|
|
page_size: int
|
|
|
|
|
|
class ProductionScheduleListResponse(BaseModel):
|
|
"""Schema for production schedule list response"""
|
|
schedules: List[ProductionScheduleResponse]
|
|
total_count: int
|
|
page: int
|
|
page_size: int
|
|
|
|
|
|
class QualityCheckListResponse(BaseModel):
|
|
"""Schema for quality check list response"""
|
|
quality_checks: List[QualityCheckResponse]
|
|
total_count: int
|
|
page: int
|
|
page_size: int
|
|
|
|
|