CIzz22
/
be-optimumoh
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix

Browse Source
main
Cizz22 3 months ago
parent 7bbec54477
commit 8822c24d56
3 changed files with 116 additions and 46 deletions
Show Stats Download Patch File Download Diff File

5
src/calculation_target_reliability/router.py
Unescape Escape View File

@ -41,7 +41,7 @@ async def get_target_reliability(
eaf_input: float = Query(99.8),
duration: int = Query(17520),
simulation_id: Optional[str] = Query(None),
po_duration = Query(1200)
cut_hours = Query(0)
):
"""Get all scope pagination."""
if not oh_session_id:
@ -85,7 +85,8 @@ async def get_target_reliability(
collector_db=collector_db,
simulation_id=simulation_id,
duration=duration,
po_duration=po_duration
po_duration=1200,
cut_hours=float(cut_hours)
)

1
src/calculation_target_reliability/schema.py
Unescape Escape View File

@ -56,6 +56,7 @@ class OptimizationResult(OverhaulBase):
target_plant_eaf: float
possible_plant_eaf:float
eaf_gap: float
eaf_improvement_text:str
warning_message:Optional[str]
asset_contributions: List[dict]
optimization_success: bool = False

156
src/calculation_target_reliability/service.py
Unescape Escape View File

@ -82,61 +82,84 @@ def calculate_asset_eaf_contributions(plant_result, eq_results, standard_scope,
"""
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_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:
asset_name = asset.get("aeros_node").get("node_name")
num_of_events = asset.get("num_events")
node = asset.get("aeros_node")
if not node:
continue
if asset_name not in standard_scope or num_of_events < 2:
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)
max_possible_improvement = REALISTIC_MAX_AVAILABILITY - current_availability if REALISTIC_MAX_AVAILABILITY > current_availability else REALISTIC_MAX_TECHNICAL
# --- 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))
raw_weight = birbaum * contribution_factor
weight = math.sqrt(raw_weight)
weighted_assets.append((asset, weight, max_possible_improvement))
# 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:
asset_name = asset.get("aeros_node").get("node_name")
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)
# Proportional improvement share
required_improvement = eaf_gap_fraction * (weight / total_weight)
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
# Secondary metric: efficiency
efficiency = birbaum / downtime if downtime > 0 else birbaum
contribution = AssetWeight(
node=asset.get("aeros_node"),
node=node,
availability=current_availability,
contribution=contribution_factor,
required_improvement=required_improvement,
@ -144,7 +167,7 @@ def calculate_asset_eaf_contributions(plant_result, eq_results, standard_scope,
num_of_failures=asset.get("num_events", 0),
down_time=downtime,
efficiency=efficiency,
birbaum=birbaum
birbaum=birbaum,
)
results.append(contribution)
@ -154,8 +177,6 @@ def calculate_asset_eaf_contributions(plant_result, eq_results, standard_scope,
return results
def project_eaf_improvement(asset: AssetWeight, improvement_factor: float = 0.3) -> float:
"""
Project EAF improvement after maintenance
@ -166,7 +187,6 @@ def project_eaf_improvement(asset: AssetWeight, improvement_factor: float = 0.3)
projected_eaf = 100 - improved_downtime_pct
return min(projected_eaf, 99.9) # Cap at 99.9%
async def identify_worst_eaf_contributors(
*,
simulation_result,
@ -175,37 +195,49 @@ async def identify_worst_eaf_contributors(
oh_session_id: str,
collector_db: CollectorDbSession,
simulation_id: str,
duration:int,
po_duration: int
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
and evaluate if target EAF is physically achievable.
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.
"""
# Extract results
calc_result = simulation_result["calc_result"]
plant_result = simulation_result["plant_result"]
eq_results = calc_result if isinstance(calc_result, list) else [calc_result]
# Current plant EAF and gap
# Base parameters
current_plant_eaf = plant_result.get("eaf", 0)
total_hours = duration
scheduled_outage = int(po_duration)
max_eaf_possible = (total_hours - scheduled_outage) / total_hours * 100
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
# Check if target EAF exceeds theoretical maximum
# 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 = scheduled_outage - required_scheduled_hours
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}%.\n"
f"To achieve it, planned outage must be reduced by approximately "
f"{required_reduction:.1f} hours (from {scheduled_outage:.0f}h → {required_scheduled_hours:.0f}h)."
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
@ -222,9 +254,10 @@ async def identify_worst_eaf_contributors(
asset_contributions=[],
optimization_success=True,
simulation_id=simulation_id,
eaf_improvement_text=""
)
# Get standard scope
# 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,
@ -232,16 +265,12 @@ async def identify_worst_eaf_contributors(
)
standard_scope_location_tags = [tag.location_tag for tag in standard_scope]
# Compute contributions
# Compute contributions for reliability improvements
asset_contributions = calculate_asset_eaf_contributions(
plant_result,
eq_results,
standard_scope_location_tags,
eaf_gap,
scheduled_outage
plant_result, eq_results, standard_scope_location_tags, eaf_gap, reduced_outage
)
# Greedy selection to fill EAF gap
# Greedy improvement allocation
project_eaf_improvement_total = 0.0
selected_eq = []
@ -255,18 +284,55 @@ async def identify_worst_eaf_contributors(
else:
continue
possible_eaf_plant = current_plant_eaf + project_eaf_improvement_total * 100
# 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)
# --- 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}%)."
)
# Final return
# Add actionable recommendation
recommendation_msg = (
f"To achieve the target EAF, consider reducing planned outage by approximately "
f"{required_cut_hours:.1f} hours (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,
warning_message=warning_message, # numeric
eaf_improvement_text=improvement_text, # human-readable text
asset_contributions=[
{
"node": asset.node,
@ -281,6 +347,8 @@ async def identify_worst_eaf_contributors(
}
for asset in selected_eq
],
optimization_success=(current_plant_eaf + project_eaf_improvement_total * 100) >= target_eaf,
outage_reduction_hours=cut_hours,
optimization_success=(current_plant_eaf + project_eaf_improvement_total * 100 + scheduled_eaf_gain)
>= target_eaf,
simulation_id=simulation_id,
)
)

Write Preview
Loading…
Cancel
Save