diff --git a/src/calculation_budget_constrains/__init__.py b/src/calculation_budget_constrains/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/calculation_budget_constrains/router.py b/src/calculation_budget_constrains/router.py new file mode 100644 index 0000000..15744ee --- /dev/null +++ b/src/calculation_budget_constrains/router.py @@ -0,0 +1,21 @@ + +from typing import Dict, List, Optional +from fastapi import APIRouter, HTTPException, status +from fastapi.params import Query + +from .service import get_all_budget_constrains + +from src.models import StandardResponse +from src.database.core import DbSession +router = APIRouter() + + +@router.get("", response_model=StandardResponse[List[Dict]]) +async def get_target_reliability(db_session: DbSession, scope_name: Optional[str] = Query(None), cost_threshold: float = Query(100)): + """Get all scope pagination.""" + results = await get_all_budget_constrains(db_session=db_session, scope_name=scope_name, cost_threshold=cost_threshold) + + return StandardResponse( + data=results, + message="Data retrieved successfully", + ) diff --git a/src/calculation_budget_constrains/schema.py b/src/calculation_budget_constrains/schema.py new file mode 100644 index 0000000..b2a59bc --- /dev/null +++ b/src/calculation_budget_constrains/schema.py @@ -0,0 +1,71 @@ + +from datetime import datetime +from typing import Any, Dict, List, Optional +from uuid import UUID + +from pydantic import Field, BaseModel +from src.models import DefultBase, Pagination + + +class OverhaulBase(BaseModel): + pass + + +class OverhaulCriticalParts(OverhaulBase): + criticalParts: List[str] = Field(..., description="List of critical parts") + + +class OverhaulSchedules(OverhaulBase): + schedules: List[Dict[str, Any] + ] = Field(..., description="List of schedules") + + +class OverhaulSystemComponents(OverhaulBase): + systemComponents: Dict[str, + Any] = Field(..., description="List of system components") + + +class OverhaulRead(OverhaulBase): + overview: Dict[str, Any] + criticalParts: List[str] + schedules: List[Dict[str, Any]] + systemComponents: Dict[str, Any] + + +# { +# "overview": { +# "totalEquipment": 30, +# "nextSchedule": { +# "date": "2025-01-12", +# "Overhaul": "B", +# "equipmentCount": 30 +# } +# }, +# "criticalParts": [ +# "Boiler feed pump", +# "Boiler reheater system", +# "Drum Level (Right) Root Valve A", +# "BCP A Discharge Valve", +# "BFPT A EXH Press HI Root VLV" +# ], +# "schedules": [ +# { +# "date": "2025-01-12", +# "Overhaul": "B", +# "status": "upcoming" +# } +# // ... other scheduled overhauls +# ], +# "systemComponents": { +# "boiler": { +# "status": "operational", +# "lastOverhaul": "2024-06-15" +# }, +# "turbine": { +# "hpt": { "status": "operational" }, +# "ipt": { "status": "operational" }, +# "lpt": { "status": "operational" } +# } +# // ... other major components +# } +# } diff --git a/src/calculation_budget_constrains/service.py b/src/calculation_budget_constrains/service.py new file mode 100644 index 0000000..8932a56 --- /dev/null +++ b/src/calculation_budget_constrains/service.py @@ -0,0 +1,46 @@ + + +from sqlalchemy import Select, Delete +from typing import Optional + +from src.database.core import DbSession +from src.auth.service import CurrentUser +from src.overhaul_schedule.service import get_all as get_all_schedules +from src.scope.model import Scope +from src.scope_equipment.model import ScopeEquipment +from src.scope_equipment.service import get_by_scope_name + + +async def get_all_budget_constrains(*, db_session: DbSession, scope_name: str, cost_threshold: float = 100.0): + """Get all overhaul overview with EAF values that sum to 100%.""" + equipments = await get_by_scope_name(db_session=db_session, scope_name=scope_name) + + # If no equipments found, return empty list + if not equipments: + return [] + # Create result array of dictionaries + result = [ + { + 'id': equipment.id, + 'assetnum': equipment.assetnum, + 'location_tag': equipment.master_equipment.location_tag, + 'name': equipment.master_equipment.name, + 'total_cost': equipment.total_cost + } + for equipment in equipments + ] + + result.sort(key=lambda x: x['total_cost'], reverse=True) + + # Filter equipment up to threshold + cumulative_cost = 0 + filtered_result = [] + + for equipment in result: + cumulative_cost += equipment['total_cost'] + filtered_result.append(equipment) + + if cumulative_cost >= cost_threshold: + break + + return filtered_result diff --git a/src/scope_equipment_activity/schema.py b/src/scope_equipment_activity/schema.py index 79ef18e..b0788bd 100644 --- a/src/scope_equipment_activity/schema.py +++ b/src/scope_equipment_activity/schema.py @@ -22,6 +22,7 @@ class ScopeEquipmentActivityUpdate(ScopeEquipmentActivityBase): class ScopeEquipmentActivityRead(ScopeEquipmentActivityBase): + id: UUID name: str cost: float