import math from typing import Optional, List from dataclasses import dataclass from sqlalchemy import Delete, Select import httpx from src.auth.service import CurrentUser from src.config import RBD_SERVICE_API from src.contribution_util import calculate_contribution, calculate_contribution_accurate from src.database.core import DbSession, CollectorDbSession from datetime import datetime, timedelta import random from .utils import generate_down_periods from src.overhaul_scope.service import get as get_overhaul from bisect import bisect_left from collections import defaultdict import asyncio from .schema import AssetWeight,MaintenanceScenario,OptimizationResult from src.overhaul_activity.service import get_standard_scope_by_session_id client = httpx.AsyncClient(timeout=300.0) async def run_rbd_simulation(*, sim_hours: int, token): sim_data = { "SimulationName": f"Simulasi TR OH {sim_hours}", "SchematicName": "- TJB - Unit 3 -", "SimSeed": 1, "SimDuration": sim_hours, "OverhaulInterval": sim_hours - 1201, "DurationUnit": "UHour", "SimNumRun": 1, "IsDefault": False, "OverhaulDuration": 1200 } headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } rbd_simulation_url = f"{RBD_SERVICE_API}/aeros/simulation/run" async with httpx.AsyncClient(timeout=300.0) as client: response = await client.post(rbd_simulation_url, json=sim_data, headers=headers) response.raise_for_status() return response.json() async def get_simulation_results(*, simulation_id: str, token: str): headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json" } calc_result_url = f"{RBD_SERVICE_API}/aeros/simulation/result/calc/{simulation_id}?nodetype=RegularNode" # plot_result_url = f"{RBD_SERVICE_API}/aeros/simulation/result/plot/{simulation_id}?nodetype=RegularNode" calc_plant_result = f"{RBD_SERVICE_API}/aeros/simulation/result/calc/{simulation_id}/plant" async with httpx.AsyncClient(timeout=300.0) as client: calc_task = client.get(calc_result_url, headers=headers) # plot_task = client.get(plot_result_url, headers=headers) plant_task = client.get(calc_plant_result, headers=headers) # Run all three requests concurrently calc_response, plant_response = await asyncio.gather(calc_task, plant_task) calc_response.raise_for_status() # plot_response.raise_for_status() plant_response.raise_for_status() calc_data = calc_response.json()["data"] # plot_data = plot_response.json()["data"] plant_data = plant_response.json()["data"] return { "calc_result": calc_data, # "plot_result": plot_data, "plant_result": plant_data } def calculate_asset_eaf_contributions(plant_result, eq_results, standard_scope, eaf_gap, scheduled_outage): """ Calculate each asset's contribution to plant EAF with realistic, fair improvement allocation. The total EAF gap is distributed among assets proportionally to their contribution potential. Automatically skips equipment with no unplanned downtime (only scheduled outages). """ eaf_gap_fraction = eaf_gap / 100.0 if eaf_gap > 1.0 else eaf_gap total_hours = plant_result.get("total_uptime") + plant_result.get("total_downtime") plant_operating_fraction = (total_hours - scheduled_outage) / total_hours REALISTIC_MAX_TECHNICAL = 0.995 REALISTIC_MAX_AVAILABILITY = REALISTIC_MAX_TECHNICAL * plant_operating_fraction MIN_IMPROVEMENT_PERCENT = 0.0001 min_improvement_fraction = MIN_IMPROVEMENT_PERCENT / 100.0 EPSILON = 0.001 # 1 ms or a fraction of an hour for comparison tolerance results = [] weighted_assets = [] # Step 1: Collect eligible assets and their weights for asset in eq_results: node = asset.get("aeros_node") if not node: continue asset_name = node.get("node_name") num_of_events = asset.get("num_events", 0) if asset_name not in standard_scope: continue contribution_factor = asset.get("contribution_factor", 0.0) birbaum = asset.get("contribution", 0.0) current_availability = asset.get("availability", 0.0) downtime = asset.get("total_downtime", 0.0) # --- NEW: Skip equipment with no failures and near-maximum availability --- if ( num_of_events < 2 # no unplanned events or contribution_factor <= 0 ): # This equipment has nothing to improve realistically continue # --- Compute realistic possible improvement --- if REALISTIC_MAX_AVAILABILITY > current_availability: max_possible_improvement = REALISTIC_MAX_AVAILABILITY - current_availability else: max_possible_improvement = 0.0 # No improvement possible # Compute weighted importance (Birnbaum × FV) raw_weight = birbaum weight = math.sqrt(max(raw_weight, 0.0)) weighted_assets.append((asset, weight, 0)) # Step 2: Compute total weight total_weight = sum(w for _, w, _ in weighted_assets) or 1.0 # Step 3: Distribute improvement proportionally to weight for asset, weight, max_possible_improvement in weighted_assets: node = asset.get("aeros_node") contribution_factor = asset.get("contribution_factor", 0.0) birbaum = asset.get("contribution", 0.0) current_availability = asset.get("availability", 0.0) downtime = asset.get("total_downtime", 0.0) required_improvement = eaf_gap_fraction * (weight/total_weight) required_improvement = min(required_improvement, max_possible_improvement) required_improvement = max(required_improvement, min_improvement_fraction) improvement_impact = required_improvement * contribution_factor efficiency = birbaum / downtime if downtime > 0 else birbaum contribution = AssetWeight( node=node, availability=current_availability, contribution=contribution_factor, required_improvement=required_improvement, improvement_impact=improvement_impact, num_of_failures=asset.get("num_events", 0), down_time=downtime, efficiency=efficiency, birbaum=birbaum, ) results.append(contribution) # Step 4: Sort by Birnbaum importance results.sort(key=lambda x: x.birbaum, reverse=True) return results def project_eaf_improvement(asset: AssetWeight, improvement_factor: float = 0.3) -> float: """ Project EAF improvement after maintenance This is a simplified model - you should replace with your actual prediction logic """ current_downtime_pct = 100 - asset.eaf improved_downtime_pct = current_downtime_pct * (1 - improvement_factor) projected_eaf = 100 - improved_downtime_pct return min(projected_eaf, 99.9) # Cap at 99.9% async def identify_worst_eaf_contributors( *, simulation_result, target_eaf: float, db_session: DbSession, oh_session_id: str, collector_db: CollectorDbSession, simulation_id: str, duration: int, po_duration: int, cut_hours: float = 0, # new optional parameter: how many hours of planned outage user wants to cut ): """ Identify equipment that contributes most to plant EAF reduction, evaluate if target EAF is physically achievable, and optionally calculate the additional improvement if user cuts scheduled outage. """ calc_result = simulation_result["calc_result"] plant_result = simulation_result["plant_result"] eq_results = calc_result if isinstance(calc_result, list) else [calc_result] # Base parameters current_plant_eaf = plant_result.get("eaf", 0) total_hours = duration scheduled_outage = int(po_duration) reduced_outage = max(scheduled_outage - cut_hours, 0) max_eaf_possible = (total_hours - reduced_outage) / total_hours * 100 # Improvement purely from outage reduction (global) scheduled_eaf_gain = (cut_hours / total_hours) * 100 if cut_hours > 0 else 0.0 # Target feasibility check warning_message = None if target_eaf > max_eaf_possible: impossible_gap = target_eaf - max_eaf_possible required_scheduled_hours = total_hours * (1 - target_eaf / 100) required_reduction = reduced_outage - required_scheduled_hours # Build dynamic phrase for clarity if cut_hours > 0: reduction_phrase = f" even after reducing planned outage by {cut_hours}h" else: reduction_phrase = "" warning_message = ( f"⚠️ Target EAF {target_eaf:.2f}% exceeds theoretical maximum {max_eaf_possible:.2f}%" f"{reduction_phrase}.\n" f"To achieve it, planned outage must be further reduced by approximately " f"{required_reduction:.1f} hours (from {reduced_outage:.0f}h → {required_scheduled_hours:.0f}h)." ) # Cap target EAF to max achievable for calculation target_eaf = max_eaf_possible eaf_gap = (target_eaf - current_plant_eaf) / 100.0 if eaf_gap <= 0: return OptimizationResult( current_plant_eaf=current_plant_eaf, target_plant_eaf=target_eaf, possible_plant_eaf=current_plant_eaf, eaf_gap=0, warning_message=warning_message or "Target already achieved or exceeded.", asset_contributions=[], optimization_success=True, simulation_id=simulation_id, eaf_improvement_text="" ) # Get standard scope (equipment in OH) standard_scope = await get_standard_scope_by_session_id( db_session=db_session, overhaul_session_id=oh_session_id, collector_db=collector_db, ) standard_scope_location_tags = [tag.location_tag for tag in standard_scope] # Compute contributions for reliability improvements asset_contributions = calculate_asset_eaf_contributions( plant_result, eq_results, standard_scope_location_tags, eaf_gap, reduced_outage ) # Greedy improvement allocation project_eaf_improvement_total = 0.0 selected_eq = [] for asset in asset_contributions: if project_eaf_improvement_total >= eaf_gap: break if (project_eaf_improvement_total + asset.improvement_impact) <= eaf_gap: selected_eq.append(asset) project_eaf_improvement_total += asset.improvement_impact else: continue # Total EAF after improvements + optional outage cut possible_eaf_plant = current_plant_eaf + project_eaf_improvement_total * 100 + scheduled_eaf_gain possible_eaf_plant = min(possible_eaf_plant, max_eaf_possible) selected_eq.sort(key=lambda x: x.birbaum, reverse=True) required_cut_hours = 0 # --- 2. Optimization feasible but cannot reach target (underperformance case) --- if possible_eaf_plant < target_eaf: # Calculate shortfall performance_gap = target_eaf - possible_eaf_plant # Estimate how many scheduled outage hours must be reduced to close the remaining gap # Each hour reduced adds (1 / total_hours) * 100 % to plant EAF required_cut_hours = (performance_gap / 100) * total_hours reliability_limit_msg = ( f"⚠️ Optimization was unable to reach target EAF {target_eaf:.2f}%.\n" f"The best achievable EAF based on current reliability is " f"{possible_eaf_plant:.2f}% (short by {performance_gap:.2f}%)." ) # Add actionable recommendation recommendation_msg = ( f"To achieve the target EAF, consider reducing planned outage by approximately " f"{required_cut_hours:.1f} hours or {int(required_cut_hours/24)} days (from {reduced_outage:.0f}h → {reduced_outage - required_cut_hours:.0f}h)." ) if warning_message: warning_message = warning_message + "\n\n" + reliability_limit_msg + "\n" + recommendation_msg else: warning_message = reliability_limit_msg + "\n" + recommendation_msg # --- EAF improvement reporting --- eaf_improvement_points = (possible_eaf_plant - current_plant_eaf) # Express as text for user readability if eaf_improvement_points > 0: improvement_text = f"{eaf_improvement_points:.6f} percentage points increase" else: improvement_text = "No measurable improvement achieved" # Build result return OptimizationResult( current_plant_eaf=current_plant_eaf, target_plant_eaf=target_eaf, possible_plant_eaf=possible_eaf_plant, eaf_gap=eaf_gap, warning_message=warning_message, # numeric eaf_improvement_text=improvement_text, recommended_reduced_outage=required_cut_hours, asset_contributions=[ { "node": asset.node, "availability": asset.availability, "contribution": asset.contribution, "sensitivy": asset.birbaum, "required_improvement": asset.required_improvement, "system_impact": asset.improvement_impact, "num_of_failures": asset.num_of_failures, "down_time": asset.down_time, "efficiency": asset.efficiency, } for asset in selected_eq ], outage_reduction_hours=cut_hours, optimization_success=(current_plant_eaf + project_eaf_improvement_total * 100 + scheduled_eaf_gain) >= target_eaf, simulation_id=simulation_id, )