You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
156 lines
9.9 KiB
Python
156 lines
9.9 KiB
Python
|
|
import sys
|
|
|
|
with open('/home/atra/Development/be-optimumoh/src/calculation_time_constrains/service.py', 'r') as f:
|
|
lines = f.readlines()
|
|
|
|
# Find the point where it broke
|
|
# Line 159 is ' \'month_index\': i + 1,\n'
|
|
# We want to keep up to line 159 (index 158)
|
|
|
|
new_lines = lines[:159]
|
|
new_lines.append(" 'source': source\n")
|
|
new_lines.append(" }\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" return monthly_data\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" async def get_simulation_results(self, simulation_id: str = \"default\"):\n")
|
|
new_lines.append(" \"\"\"Get simulation results for Birnbaum importance calculations\"\"\"\n")
|
|
new_lines.append(" headers = {\n")
|
|
new_lines.append(" \"Authorization\": f\"Bearer {self.token}\",\n")
|
|
new_lines.append(" \"Content-Type\": \"application/json\"\n")
|
|
new_lines.append(" }\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" calc_result_url = f\"{self.api_base_url}/aeros/simulation/result/calc/{simulation_id}?nodetype=RegularNode\"\n")
|
|
new_lines.append(" plant_result_url = f\"{self.api_base_url}/aeros/simulation/result/calc/{simulation_id}/plant\"\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" async with httpx.AsyncClient(timeout=300.0) as client:\n")
|
|
new_lines.append(" calc_task = client.get(calc_result_url, headers=headers)\n")
|
|
new_lines.append(" plant_task = client.get(plant_result_url, headers=headers)\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" calc_response, plant_response = await asyncio.gather(calc_task, plant_task)\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" calc_response.raise_for_status()\n")
|
|
new_lines.append(" plant_response.raise_for_status()\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" calc_data = calc_response.json()[\"data\"]\n")
|
|
new_lines.append(" plant_data = plant_response.json()[\"data\"]\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" return {\n")
|
|
new_lines.append(" \"calc_result\": calc_data,\n")
|
|
new_lines.append(" \"plant_result\": plant_data\n")
|
|
new_lines.append(" }\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" def _calculate_equipment_costs_with_spareparts(self, failures_prediction: Dict, birnbaum_importance: float,\n")
|
|
new_lines.append(" preventive_cost: float, failure_replacement_cost: float, ecs,\n")
|
|
new_lines.append(" location_tag: str, planned_overhauls: List = None, loss_production_permonth=0) -> List[Dict]:\n")
|
|
new_lines.append(" \"\"\"Calculate costs for each month including sparepart costs and availability\"\"\"\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" if not failures_prediction:\n")
|
|
new_lines.append(" self.logger.warning(f\"No failure prediction data for {location_tag}\")\n")
|
|
new_lines.append(" return []\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" results = []\n")
|
|
new_lines.append(" months = list(failures_prediction.keys())\n")
|
|
new_lines.append(" num_months = len(months)\n")
|
|
new_lines.append(" failure_counts = []\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" monthly_risk_cost_per_failure = 0\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" if ecs:\n")
|
|
new_lines.append(" is_trip = 1 if ecs.get(\"Diskripsi Operasional Akibat Equip. Failure\") == \"Trip\" else 0\n")
|
|
new_lines.append(" is_series = 0 if not birnbaum_importance else math.floor(birnbaum_importance)\n")
|
|
new_lines.append(" if is_trip:\n")
|
|
new_lines.append(" downtime = ecs.get(\"Estimasi Waktu Maint. / Downtime / Gangguan (Jam)\")\n")
|
|
new_lines.append(" monthly_risk_cost_per_failure = 660 * 1000000 * is_trip * downtime * is_series\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" for month_key in months:\n")
|
|
new_lines.append(" data = failures_prediction[month_key]\n")
|
|
new_lines.append(" failure_counts.append(data['cumulative_failures'])\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" for i in range(num_months):\n")
|
|
new_lines.append(" month_index = i + 1\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # Use only months within the analysis window\n")
|
|
new_lines.append(" if month_index > self.time_window_months:\n")
|
|
new_lines.append(" continue\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" # Check sparepart availability for this month\n")
|
|
new_lines.append(" sparepart_analysis = self._analyze_sparepart_availability(\n")
|
|
new_lines.append(" location_tag, month_index - 1, planned_overhauls or []\n")
|
|
new_lines.append(" )\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # THEORY: Total Expected Cost per Unit Time (CPUT)\n")
|
|
new_lines.append(" # Reference: Maintenance Optimization Models (Age/Block Replacement)\n")
|
|
new_lines.append(" # C(T) = [Total Preventive Cost + Total Expected Corrective Cost(T)] / T\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # 1. Total Expected Corrective Cost until month_index (Expected number of failures * cost per failure)\n")
|
|
new_lines.append(" # In NHPP model, Expected Failures E[N(T)] = Cumulative Failures\n")
|
|
new_lines.append(" total_expected_failure_cost = failure_counts[i] * (failure_replacement_cost + monthly_risk_cost_per_failure)\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # 2. Total Preventive Cost (One-time cost at month_index)\n")
|
|
new_lines.append(" # Includes labor, materials, and procurement delays\n")
|
|
new_lines.append(" procurement_cost = sparepart_analysis['total_procurement_cost']\n")
|
|
new_lines.append(" total_preventive_cost = preventive_cost + procurement_cost\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # 3. Expected Total Cycle Cost\n")
|
|
new_lines.append(" total_cycle_cost = total_expected_failure_cost + total_preventive_cost\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # 4. Expected Cost Per Unit Time (Optimization Target)\n")
|
|
new_lines.append(" # This value forms the U-shaped curve\n")
|
|
new_lines.append(" cput = total_cycle_cost / month_index\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # Store both absolute and amortized components for visualization\n")
|
|
new_lines.append(" results.append({\n")
|
|
new_lines.append(" 'month': month_index,\n")
|
|
new_lines.append(" 'number_of_failures': failure_counts[i],\n")
|
|
new_lines.append(" 'is_actual': failures_prediction[months[i]].get('source') == 'actual',\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # Amortized components (for the \"U\" chart)\n")
|
|
new_lines.append(" 'failure_cost': total_expected_failure_cost / month_index,\n")
|
|
new_lines.append(" 'preventive_cost': preventive_cost / month_index,\n")
|
|
new_lines.append(" 'procurement_cost': procurement_cost / month_index,\n")
|
|
new_lines.append(" 'total_cost': cput,\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # Absolute values (for breakdown analysis)\n")
|
|
new_lines.append(" 'absolute_failure_cost': total_expected_failure_cost,\n")
|
|
new_lines.append(" 'absolute_overhaul_cost': preventive_cost,\n")
|
|
new_lines.append(" 'absolute_procurement_cost': procurement_cost,\n")
|
|
new_lines.append(" 'total_cycle_cost': total_cycle_cost,\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" 'is_after_planned_oh': month_index > self.planned_oh_months,\n")
|
|
new_lines.append(" 'delay_months': max(0, month_index - self.planned_oh_months),\n")
|
|
new_lines.append(" 'procurement_details': sparepart_analysis,\n")
|
|
new_lines.append(" 'sparepart_available': sparepart_analysis['available'],\n")
|
|
new_lines.append(" 'sparepart_status': sparepart_analysis['message'],\n")
|
|
new_lines.append(" 'can_proceed': sparepart_analysis['can_proceed_with_delays']\n")
|
|
new_lines.append(" })\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" return results\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" def _analyze_sparepart_availability(self, equipment_tag: str, target_month: int, \n")
|
|
new_lines.append(" planned_overhauls: List) -> Dict:\n")
|
|
new_lines.append(" \"\"\"Analyze sparepart availability for equipment at target month\"\"\"\n")
|
|
new_lines.append(" if not self.sparepart_manager:\n")
|
|
new_lines.append(" return {\n")
|
|
new_lines.append(" 'available': True,\n")
|
|
new_lines.append(" 'message': 'Sparepart manager not initialized',\n")
|
|
new_lines.append(" 'total_procurement_cost': 0,\n")
|
|
new_lines.append(" 'procurement_needed': [],\n")
|
|
new_lines.append(" 'can_proceed_with_delays': True\n")
|
|
new_lines.append(" }\n")
|
|
new_lines.append(" \n")
|
|
new_lines.append(" # Convert planned overhauls to format expected by sparepart manager\n")
|
|
new_lines.append(" other_overhauls = [(eq_tag, month) for eq_tag, month in planned_overhauls\n")
|
|
new_lines.append(" if eq_tag != equipment_tag and month <= target_month]\n")
|
|
new_lines.append("\n")
|
|
new_lines.append(" return self.sparepart_manager.check_sparepart_availability(\n")
|
|
new_lines.append(" equipment_tag, target_month, other_overhauls\n")
|
|
new_lines.append(" )\n")
|
|
new_lines.append("\n")
|
|
|
|
new_lines.extend(lines[159:])
|
|
|
|
with open('/home/atra/Development/be-optimumoh/src/calculation_time_constrains/service.py', 'w') as f:
|
|
f.writelines(new_lines)
|