Compare commits
No commits in common. 'b4e952ebdacbaa3053f291fefc1cdabf3b0d02e9' and 'd26326191cc500da0354674b19c2681547e305de' have entirely different histories.
b4e952ebda
...
d26326191c
@ -1,814 +0,0 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional
|
||||
from uuid import UUID
|
||||
|
||||
import httpx
|
||||
|
||||
import src.aeros_equipment.model # noqa: F401
|
||||
from src.aeros_simulation.model import AerosSimulationCalcResult, AerosSimulationPlotResult
|
||||
from src.aeros_simulation.service import get_all_aeros_node, get_simulation_by_id
|
||||
from src.config import (
|
||||
RAMP_API_URL,
|
||||
RAMP_MODEL_ID,
|
||||
RAMP_MODEL_NAME,
|
||||
RAMP_POLL_INTERVAL_SECONDS,
|
||||
RAMP_POLL_TIMEOUT_SECONDS,
|
||||
RAMP_STALE_TIMEOUT_SECONDS,
|
||||
RAMP_TARGET_SCHEMA,
|
||||
SIMULATION_ENGINE,
|
||||
)
|
||||
from src.database.core import DbSession
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
PLANT_NODE_NAME = "- TJB - Unit 3 -"
|
||||
|
||||
|
||||
def selected_simulation_engine(sim_data: dict | None = None) -> str:
|
||||
sim_data = sim_data or {}
|
||||
engine = (
|
||||
sim_data.get("SimulationEngine")
|
||||
or sim_data.get("simulation_engine")
|
||||
or SIMULATION_ENGINE
|
||||
or "aeros"
|
||||
)
|
||||
return str(engine).strip().lower()
|
||||
|
||||
|
||||
def _simulation_api_base() -> str:
|
||||
base_url = str(RAMP_API_URL).rstrip("/")
|
||||
if base_url.endswith("/api/simulation"):
|
||||
return base_url
|
||||
if base_url.endswith("/api"):
|
||||
return f"{base_url}/simulation"
|
||||
return f"{base_url}/api/simulation"
|
||||
|
||||
|
||||
def _norm_key(value: Any) -> str:
|
||||
return str(value or "").strip().upper()
|
||||
|
||||
|
||||
def _first_present(data: dict, *keys: str):
|
||||
for key in keys:
|
||||
if key in data and data[key] is not None:
|
||||
return data[key]
|
||||
return None
|
||||
|
||||
|
||||
def _as_float(value, default: Optional[float] = None):
|
||||
if value is None:
|
||||
return default
|
||||
try:
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
|
||||
def _clean_dict(data: dict) -> dict:
|
||||
return {key: value for key, value in data.items() if value is not None}
|
||||
|
||||
|
||||
def _distribution_type(value: Any) -> str:
|
||||
raw = str(value or "Fixed").strip()
|
||||
aliases = {
|
||||
"WEIBULL2": "Weibull",
|
||||
"WEIBULL": "Weibull",
|
||||
"EXPONENTIAL2": "Exponential",
|
||||
"EXPONENTIAL": "Exponential",
|
||||
"LOGNORMAL": "LogNormal",
|
||||
"LOG_NORMAL": "LogNormal",
|
||||
"NORMAL": "Normal",
|
||||
"FIXED": "Fixed",
|
||||
"NOFAILURE": "NoFailure",
|
||||
"NO_FAILURE": "NoFailure",
|
||||
"NHPP": "NHPP",
|
||||
"NHPPTTFF": "NHPPTTFF",
|
||||
"REPAIRABLENHPP": "RepairableNHPP",
|
||||
}
|
||||
return aliases.get(raw.upper(), raw)
|
||||
|
||||
|
||||
def _distribution_from_fields(data: dict, prefix: str = "") -> dict | None:
|
||||
type_keys = [f"{prefix}type", f"{prefix}Type"] if prefix else ["type", "distribution", "dist_type", "distType"]
|
||||
dist_type = _first_present(data, *type_keys)
|
||||
if dist_type is None:
|
||||
return None
|
||||
|
||||
dist_type = _distribution_type(dist_type)
|
||||
p1 = _as_float(_first_present(data, f"{prefix}p1", f"{prefix}P1", "p1", "shape", "beta", "eta", "mean", "mu", "interval"))
|
||||
p2 = _as_float(_first_present(data, f"{prefix}p2", f"{prefix}P2", "p2", "scale", "std_dev", "stdDev", "sigma"))
|
||||
|
||||
if dist_type == "Weibull":
|
||||
return _clean_dict({
|
||||
"type": "Weibull",
|
||||
"shape": _as_float(_first_present(data, "shape", "beta"), p1),
|
||||
"scale": _as_float(_first_present(data, "scale", "eta"), p2),
|
||||
})
|
||||
if dist_type == "Exponential":
|
||||
return _clean_dict({"type": "Exponential", "eta": _as_float(_first_present(data, "eta"), p1)})
|
||||
if dist_type == "Normal":
|
||||
return _clean_dict({
|
||||
"type": "Normal",
|
||||
"mean": _as_float(_first_present(data, "mean"), p1),
|
||||
"std_dev": _as_float(_first_present(data, "std_dev", "stdDev"), p2),
|
||||
})
|
||||
if dist_type == "LogNormal":
|
||||
return _clean_dict({
|
||||
"type": "LogNormal",
|
||||
"mu": _as_float(_first_present(data, "mu"), p1),
|
||||
"sigma": _as_float(_first_present(data, "sigma"), p2),
|
||||
})
|
||||
if dist_type == "NoFailure":
|
||||
return {"type": "NoFailure"}
|
||||
return _clean_dict({"type": "Fixed", "interval": _as_float(_first_present(data, "interval"), p1)})
|
||||
|
||||
|
||||
def _model_from_aeros_style(override: dict) -> dict | None:
|
||||
level = override.get("level")
|
||||
if level == "NoFailure":
|
||||
return {"type": "NonRepairable", "distribution": {"type": "NoFailure"}}
|
||||
if level == "Out of Service":
|
||||
return {"type": "NonRepairable", "distribution": {"type": "Fixed", "interval": 0.0001}}
|
||||
|
||||
failure_rate = override.get("failure_rate")
|
||||
if isinstance(failure_rate, list):
|
||||
failure_rate = failure_rate[0] if failure_rate else None
|
||||
|
||||
if failure_rate is not None:
|
||||
rate_val = _as_float(failure_rate)
|
||||
mtbf = 1000000000.0 if not rate_val or rate_val <= 0 else 1e6 / rate_val
|
||||
dist_type = _distribution_type(override.get("distribution", "Fixed"))
|
||||
if dist_type == "Weibull":
|
||||
return {"type": "NonRepairable", "distribution": {"type": "Weibull", "shape": 1.0, "scale": mtbf}}
|
||||
if dist_type == "Exponential":
|
||||
return {"type": "NonRepairable", "distribution": {"type": "Exponential", "eta": mtbf}}
|
||||
return {"type": "NonRepairable", "distribution": {"type": "Fixed", "interval": mtbf}}
|
||||
|
||||
rel_type = _first_present(override, "rel_dis_type", "relDisType")
|
||||
if rel_type is not None:
|
||||
dist = _distribution_from_fields(
|
||||
{
|
||||
"type": rel_type,
|
||||
"p1": _first_present(override, "rel_dis_p1", "relDisP1"),
|
||||
"p2": _first_present(override, "rel_dis_p2", "relDisP2"),
|
||||
}
|
||||
)
|
||||
return {"type": "NonRepairable", "distribution": dist}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _repair_duration_from_aeros_style(override: dict):
|
||||
if "mttr" in override and override["mttr"] is not None:
|
||||
return _as_float(override["mttr"], 0)
|
||||
|
||||
cm_type = _first_present(override, "cm_dis_type", "cmDisType")
|
||||
if cm_type is None:
|
||||
return None
|
||||
|
||||
return _distribution_from_fields(
|
||||
{
|
||||
"type": cm_type,
|
||||
"p1": _first_present(override, "cm_dis_p1", "cmDisP1"),
|
||||
"p2": _first_present(override, "cm_dis_p2", "cmDisP2"),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _normalize_asset_override(override: dict) -> dict:
|
||||
normalized = {}
|
||||
allowed_root_fields = {
|
||||
"initial_age",
|
||||
"repair_restoration_factor",
|
||||
"max_capacity",
|
||||
"design_capacity",
|
||||
"buffer_duration",
|
||||
"transition_delay",
|
||||
"repair_duration",
|
||||
"pm",
|
||||
"overhaul",
|
||||
}
|
||||
|
||||
for key in allowed_root_fields:
|
||||
if key in override:
|
||||
normalized[key] = override[key]
|
||||
|
||||
if isinstance(override.get("model"), dict):
|
||||
normalized["model"] = override["model"]
|
||||
elif isinstance(override.get("distribution"), dict):
|
||||
normalized["model"] = {"type": "NonRepairable", "distribution": override["distribution"]}
|
||||
else:
|
||||
model = _model_from_aeros_style(override)
|
||||
if model:
|
||||
normalized["model"] = model
|
||||
|
||||
if "model" not in normalized:
|
||||
dist = _distribution_from_fields(override)
|
||||
if dist:
|
||||
normalized["model"] = {"type": "NonRepairable", "distribution": dist}
|
||||
|
||||
if "repair_duration" not in normalized:
|
||||
repair_duration = _repair_duration_from_aeros_style(override)
|
||||
if repair_duration is not None:
|
||||
normalized["repair_duration"] = repair_duration
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
def build_ramp_distribution_overrides(sim_data: dict) -> dict:
|
||||
raw_overrides = {}
|
||||
for source_key in ("CustomInput", "DistributionOverrides", "RampDistributionOverrides"):
|
||||
source = sim_data.get(source_key)
|
||||
if isinstance(source, dict):
|
||||
raw_overrides.update(source)
|
||||
|
||||
normalized = {}
|
||||
for asset_key, override in raw_overrides.items():
|
||||
if not isinstance(override, dict):
|
||||
continue
|
||||
|
||||
asset_override = _normalize_asset_override(override)
|
||||
if asset_override:
|
||||
normalized[asset_key] = asset_override
|
||||
|
||||
return normalized
|
||||
|
||||
|
||||
def _rbd_reliability_to_ramp_override(reliability: dict) -> dict:
|
||||
distribution = _distribution_type(reliability.get("distribution"))
|
||||
beta = _as_float(reliability.get("beta"))
|
||||
eta = _as_float(reliability.get("eta"))
|
||||
mttr = _as_float(reliability.get("mttr"))
|
||||
|
||||
override = {}
|
||||
if distribution in {"NHPPTTFF", "NHPP", "RepairableNHPP"}:
|
||||
override["model"] = _clean_dict({
|
||||
"type": "RepairableNHPP",
|
||||
"beta": beta,
|
||||
"eta": eta,
|
||||
})
|
||||
elif distribution == "Weibull":
|
||||
override["model"] = {
|
||||
"type": "NonRepairable",
|
||||
"distribution": _clean_dict({"type": "Weibull", "shape": beta, "scale": eta}),
|
||||
}
|
||||
elif distribution == "Exponential":
|
||||
exp_eta = eta if eta is not None else beta
|
||||
override["model"] = {
|
||||
"type": "NonRepairable",
|
||||
"distribution": _clean_dict({"type": "Exponential", "eta": exp_eta}),
|
||||
}
|
||||
elif distribution == "Normal":
|
||||
override["model"] = {
|
||||
"type": "NonRepairable",
|
||||
"distribution": _clean_dict({"type": "Normal", "mean": beta, "std_dev": eta}),
|
||||
}
|
||||
elif distribution == "Fixed":
|
||||
override["model"] = {
|
||||
"type": "NonRepairable",
|
||||
"distribution": _clean_dict({"type": "Fixed", "interval": beta or eta}),
|
||||
}
|
||||
|
||||
if mttr is not None:
|
||||
override["repair_duration"] = mttr
|
||||
return override
|
||||
|
||||
|
||||
def build_ramp_distribution_overrides_from_rbd_reliability(eq_update: dict | None) -> dict:
|
||||
overrides = {}
|
||||
for asset_key, reliability in (eq_update or {}).items():
|
||||
if not isinstance(reliability, dict):
|
||||
continue
|
||||
override = _rbd_reliability_to_ramp_override(reliability)
|
||||
if override:
|
||||
overrides[asset_key] = override
|
||||
return overrides
|
||||
|
||||
|
||||
def _duration_to_hours(value: Any, unit: str | None) -> int:
|
||||
duration = _as_float(value, 0) or 0
|
||||
unit_key = str(unit or "UHour").strip().upper()
|
||||
multipliers = {
|
||||
"UHOUR": 1,
|
||||
"UHOURS": 1,
|
||||
"HOUR": 1,
|
||||
"HOURS": 1,
|
||||
"UDAY": 24,
|
||||
"UDAYS": 24,
|
||||
"DAY": 24,
|
||||
"DAYS": 24,
|
||||
"UWEEK": 24 * 7,
|
||||
"UWEEKS": 24 * 7,
|
||||
"WEEK": 24 * 7,
|
||||
"WEEKS": 24 * 7,
|
||||
"UMONTH": 730,
|
||||
"UMONTHS": 730,
|
||||
"MONTH": 730,
|
||||
"MONTHS": 730,
|
||||
"UYEAR": 8760,
|
||||
"UYEARS": 8760,
|
||||
"YEAR": 8760,
|
||||
"YEARS": 8760,
|
||||
}
|
||||
return max(1, int(duration * multipliers.get(unit_key, 1)))
|
||||
|
||||
|
||||
def _availability_percent(value: Any) -> float:
|
||||
availability = _as_float(value, 0) or 0
|
||||
if 0 < abs(availability) <= 1:
|
||||
availability *= 100
|
||||
return availability
|
||||
|
||||
|
||||
def _availability_fraction(value: Any) -> float:
|
||||
availability = _as_float(value, 0) or 0
|
||||
if abs(availability) > 1:
|
||||
availability /= 100
|
||||
return availability
|
||||
|
||||
|
||||
def _get_number(data: dict, keys: list[str], default: float = 0) -> float:
|
||||
for key in keys:
|
||||
if key in data and data[key] is not None:
|
||||
parsed = _as_float(data[key])
|
||||
if parsed is not None:
|
||||
return parsed
|
||||
return default
|
||||
|
||||
|
||||
def _result_number(value: Any, default: float = 0) -> float:
|
||||
parsed = _as_float(value)
|
||||
return default if parsed is None else parsed
|
||||
|
||||
|
||||
def _ramp_model_id_from_sim_data(sim_data: dict) -> str | None:
|
||||
model_id = (
|
||||
sim_data.get("RampModelId")
|
||||
or sim_data.get("RAMPModelId")
|
||||
or sim_data.get("ramp_model_id")
|
||||
or RAMP_MODEL_ID
|
||||
)
|
||||
return str(model_id) if model_id not in (None, "") else None
|
||||
|
||||
|
||||
def _ramp_model_name_from_sim_data(sim_data: dict) -> str | None:
|
||||
model_name = sim_data.get("RampModelName") or sim_data.get("ramp_model_name") or RAMP_MODEL_NAME
|
||||
return str(model_name).strip() if model_name else None
|
||||
|
||||
|
||||
async def _resolve_ramp_model_id(client: httpx.AsyncClient, sim_data: dict) -> str:
|
||||
model_id = _ramp_model_id_from_sim_data(sim_data)
|
||||
if model_id:
|
||||
return model_id
|
||||
|
||||
model_name = _ramp_model_name_from_sim_data(sim_data)
|
||||
if not model_name:
|
||||
raise RuntimeError("RAMP model is not configured. Set RAMP_MODEL_ID, RAMP_MODEL_NAME, or pass RampModelId.")
|
||||
|
||||
response = await client.get(f"{_simulation_api_base()}/model")
|
||||
response.raise_for_status()
|
||||
for model in response.json().get("models", []):
|
||||
if _norm_key(model.get("name")) == _norm_key(model_name):
|
||||
return str(model["id"])
|
||||
|
||||
raise RuntimeError(f"RAMP model named {model_name!r} was not found.")
|
||||
|
||||
|
||||
async def _fetch_ramp_model_asset_names(client: httpx.AsyncClient, model_id: str) -> set[str]:
|
||||
"""Asset names defined in a RAMP model, normalized for comparison."""
|
||||
response = await client.get(f"{_simulation_api_base()}/model/{model_id}")
|
||||
response.raise_for_status()
|
||||
assets = (response.json().get("payload") or {}).get("assets") or []
|
||||
return {_norm_key(asset.get("name")) for asset in assets if asset.get("name")}
|
||||
|
||||
|
||||
def _find_ramp_task(tasks_payload: dict, task_id: str) -> dict | None:
|
||||
for task in tasks_payload.get("tasks", []):
|
||||
if str(task.get("id")) == str(task_id):
|
||||
return task
|
||||
return None
|
||||
|
||||
|
||||
async def _fetch_ramp_task_status(client: httpx.AsyncClient, task_id: str) -> dict:
|
||||
"""Fetch task status, falling back to the lighter task-list endpoint.
|
||||
|
||||
RAMP's status endpoint includes the result payload for completed tasks. When
|
||||
the engine is busy that endpoint can occasionally time out even for running
|
||||
tasks, while /tasks still returns the persisted progress row quickly.
|
||||
"""
|
||||
try:
|
||||
status_response = await client.get(f"{_simulation_api_base()}/status/{task_id}")
|
||||
status_response.raise_for_status()
|
||||
return status_response.json()
|
||||
except (httpx.ReadTimeout, httpx.ConnectTimeout, httpx.ConnectError) as status_error:
|
||||
log.warning("RAMP status request failed for task %s: %s", task_id, status_error)
|
||||
|
||||
tasks_response = await client.get(f"{_simulation_api_base()}/tasks")
|
||||
tasks_response.raise_for_status()
|
||||
task = _find_ramp_task(tasks_response.json(), task_id)
|
||||
if task:
|
||||
return task
|
||||
raise RuntimeError(f"RAMP task {task_id} was not found in /tasks fallback.")
|
||||
|
||||
|
||||
async def _fetch_completed_ramp_result(client: httpx.AsyncClient, task_id: str) -> dict:
|
||||
response = await client.get(f"{_simulation_api_base()}/status/{task_id}", timeout=120.0)
|
||||
response.raise_for_status()
|
||||
result = response.json().get("result")
|
||||
if not isinstance(result, dict):
|
||||
raise RuntimeError(f"RAMP task {task_id} completed without a result payload.")
|
||||
return result
|
||||
|
||||
|
||||
async def _cancel_ramp_task(client: httpx.AsyncClient, task_id: str) -> None:
|
||||
try:
|
||||
await client.post(f"{_simulation_api_base()}/cancel/{task_id}")
|
||||
except Exception as cancel_error:
|
||||
log.warning("Failed to cancel stale RAMP task %s: %s", task_id, cancel_error)
|
||||
|
||||
|
||||
def _filter_overrides_to_model(overrides: dict, asset_names: set[str]) -> tuple[dict, list[str]]:
|
||||
"""Keep only overrides whose asset exists in the RAMP model.
|
||||
|
||||
The RBD schematic is a superset of any single RAMP model, so distributions for
|
||||
assets the model does not contain are dropped rather than failing the run.
|
||||
"""
|
||||
if not asset_names:
|
||||
return dict(overrides), []
|
||||
|
||||
kept = {}
|
||||
dropped = []
|
||||
for asset_key, override in (overrides or {}).items():
|
||||
if _norm_key(asset_key) in asset_names:
|
||||
kept[asset_key] = override
|
||||
else:
|
||||
dropped.append(str(asset_key))
|
||||
return kept, sorted(dropped)
|
||||
|
||||
|
||||
def _build_ramp_execute_payload(sim_data: dict, eq_update: dict | None = None) -> dict:
|
||||
payload = {
|
||||
"simulation_duration": _duration_to_hours(sim_data.get("SimDuration"), sim_data.get("DurationUnit")),
|
||||
"simulation_replications": sim_data.get("SimNumRun"),
|
||||
"seed": sim_data.get("SimSeed"),
|
||||
"target_schema": sim_data.get("RampTargetSchema") or RAMP_TARGET_SCHEMA,
|
||||
}
|
||||
overrides = build_ramp_distribution_overrides_from_rbd_reliability(eq_update)
|
||||
overrides.update(build_ramp_distribution_overrides(sim_data))
|
||||
if overrides:
|
||||
payload["distribution_overrides"] = overrides
|
||||
return {key: value for key, value in payload.items() if value is not None}
|
||||
|
||||
|
||||
def _event_time(event: dict) -> float:
|
||||
return _get_number(event, ["cumulativeTime", "cumulative_time", "start", "start_time", "time"])
|
||||
|
||||
|
||||
def _build_ramp_timestamp_outs(ramp_result: dict, duration_hours: float) -> list[dict]:
|
||||
events = (
|
||||
ramp_result.get("system_events")
|
||||
or ramp_result.get("events")
|
||||
or ramp_result.get("system_event_log")
|
||||
or []
|
||||
)
|
||||
timestamps = [
|
||||
{
|
||||
"cumulativeTime": 0,
|
||||
"flowRate": _get_number(ramp_result, ["total_demand", "demand"], 660),
|
||||
"currentEQStatus": "Svc",
|
||||
"currentOPStatus": "Svc",
|
||||
"isTriggerNode": False,
|
||||
"storageLevel": 0,
|
||||
"currentEvent": "START",
|
||||
}
|
||||
]
|
||||
|
||||
for event in events:
|
||||
if not isinstance(event, dict):
|
||||
continue
|
||||
start_time = _event_time(event)
|
||||
end_time = _get_number(event, ["end", "end_time"], start_time)
|
||||
cause = event.get("cause") or event.get("event_type") or event.get("state") or "RAMP_EVENT"
|
||||
load_factor = _as_float(event.get("load_factor"), 0)
|
||||
flow_rate = 660 * load_factor if load_factor and load_factor <= 1 else _get_number(event, ["flowRate", "flow_rate"], 0)
|
||||
timestamps.append({
|
||||
"cumulativeTime": start_time,
|
||||
"flowRate": flow_rate,
|
||||
"currentEQStatus": "OoS",
|
||||
"currentOPStatus": "OoS",
|
||||
"isTriggerNode": True,
|
||||
"storageLevel": 0,
|
||||
"currentEvent": cause,
|
||||
})
|
||||
if end_time and end_time > start_time:
|
||||
timestamps.append({
|
||||
"cumulativeTime": end_time,
|
||||
"flowRate": _get_number(ramp_result, ["total_demand", "demand"], 660),
|
||||
"currentEQStatus": "Svc",
|
||||
"currentOPStatus": "Svc",
|
||||
"isTriggerNode": False,
|
||||
"storageLevel": 0,
|
||||
"currentEvent": "RECOVERY",
|
||||
})
|
||||
|
||||
if duration_hours:
|
||||
timestamps.append({
|
||||
"cumulativeTime": duration_hours,
|
||||
"flowRate": _get_number(ramp_result, ["total_demand", "demand"], 660),
|
||||
"currentEQStatus": "Svc",
|
||||
"currentOPStatus": "Svc",
|
||||
"isTriggerNode": False,
|
||||
"storageLevel": 0,
|
||||
"currentEvent": "END",
|
||||
})
|
||||
|
||||
return sorted(timestamps, key=lambda item: item["cumulativeTime"])
|
||||
|
||||
|
||||
def _calc_result(
|
||||
*,
|
||||
simulation_id: UUID,
|
||||
node_id: UUID,
|
||||
metrics: dict,
|
||||
reliability: dict | None = None,
|
||||
parameters: dict | None = None,
|
||||
) -> AerosSimulationCalcResult:
|
||||
total_uptime = _get_number(metrics, ["total_uptime", "system_uptime", "uptime"])
|
||||
total_downtime = _get_number(metrics, ["total_downtime", "system_downtime", "downtime"])
|
||||
period_hours = total_uptime + total_downtime
|
||||
availability_source = _first_present(metrics, "availability", "system_availability", "production_availability")
|
||||
availability = _availability_fraction(availability_source)
|
||||
if not availability and period_hours:
|
||||
availability = total_uptime / period_hours
|
||||
eaf = _availability_percent(availability)
|
||||
efor = total_downtime / period_hours * 100 if period_hours else 0
|
||||
|
||||
production = _get_number(metrics, ["production", "total_productions", "total_production", "actual_production"])
|
||||
ideal_production = _get_number(metrics, ["ideal_production", "total_demand", "demand"])
|
||||
effective_loss = max(ideal_production - production, 0) if ideal_production else 0
|
||||
efficiency = production / ideal_production * 100 if ideal_production else availability
|
||||
|
||||
reliability = reliability or {}
|
||||
return AerosSimulationCalcResult(
|
||||
aeros_simulation_id=simulation_id,
|
||||
aeros_node_id=node_id,
|
||||
total_downtime=total_downtime,
|
||||
total_uptime=total_uptime,
|
||||
num_events=int(_get_number(metrics, ["num_events", "system_failures", "failures"])),
|
||||
production=production,
|
||||
production_std=_get_number(metrics, ["production_std"]),
|
||||
ideal_production=ideal_production,
|
||||
availability=availability,
|
||||
efficiency=efficiency,
|
||||
effective_loss=effective_loss,
|
||||
num_cm=int(_get_number(metrics, ["num_cm", "failures"])),
|
||||
cm_waiting_time=_get_number(metrics, ["cm_waiting_time"]),
|
||||
total_cm_downtime=total_downtime,
|
||||
num_pm=int(_get_number(metrics, ["num_pm"])),
|
||||
total_pm_downtime=_get_number(metrics, ["total_pm_downtime"]),
|
||||
num_ip=int(_get_number(metrics, ["num_ip"])),
|
||||
total_ip_downtime=_get_number(metrics, ["total_ip_downtime"]),
|
||||
num_oh=int(_get_number(metrics, ["num_oh"])),
|
||||
total_oh_downtime=_get_number(metrics, ["total_oh_downtime"]),
|
||||
t_wait_for_crew=_get_number(metrics, ["t_wait_for_crew"]),
|
||||
t_wait_for_spare=_get_number(metrics, ["t_wait_for_spare"]),
|
||||
duration_at_full=total_uptime,
|
||||
duration_above_hh=0,
|
||||
duration_above_h=0,
|
||||
duration_below_l=0,
|
||||
duration_below_ll=0,
|
||||
duration_at_empty=total_downtime,
|
||||
stg_input=0,
|
||||
stg_output=0,
|
||||
average_level=0,
|
||||
potential_production=ideal_production,
|
||||
eaf=eaf,
|
||||
efor=efor,
|
||||
derating_hours=effective_loss,
|
||||
beta=_result_number(reliability.get("beta")),
|
||||
eta=_result_number(reliability.get("eta")),
|
||||
mttr=int(_result_number(reliability.get("mttr"))),
|
||||
parameters=parameters or reliability.get("parameters") or {},
|
||||
contribution=_get_number(metrics, ["contribution"]),
|
||||
criticality=_get_number(metrics, ["criticality"]),
|
||||
contribution_factor=_get_number(metrics, ["contribution_factor"]),
|
||||
total_mo_downtime=_get_number(metrics, ["total_mo_downtime"]),
|
||||
total_po_downtime=_get_number(metrics, ["total_po_downtime"]),
|
||||
sof=_get_number(metrics, ["sof"]),
|
||||
)
|
||||
|
||||
|
||||
async def save_ramp_simulation_result(
|
||||
*,
|
||||
db_session: DbSession,
|
||||
simulation_id: UUID,
|
||||
sim_data: dict,
|
||||
ramp_result: dict,
|
||||
eq_update: dict,
|
||||
ramp_task_id: str,
|
||||
):
|
||||
simulation = await get_simulation_by_id(db_session=db_session, simulation_id=simulation_id)
|
||||
simulation.status = "running"
|
||||
simulation.input = {
|
||||
**(simulation.input or {}),
|
||||
"engine": "ramp",
|
||||
"ramp_task_id": ramp_task_id,
|
||||
"ramp_result_id": ramp_result.get("id"),
|
||||
}
|
||||
await db_session.commit()
|
||||
|
||||
schematic_name = sim_data.get("SchematicName") or PLANT_NODE_NAME
|
||||
available_nodes = await get_all_aeros_node(db_session=db_session, schematic_name=schematic_name)
|
||||
nodes_by_key = {}
|
||||
for node in available_nodes:
|
||||
nodes_by_key[_norm_key(node.node_name)] = node
|
||||
|
||||
plant_node = nodes_by_key.get(_norm_key(schematic_name)) or nodes_by_key.get(_norm_key(PLANT_NODE_NAME))
|
||||
if not plant_node:
|
||||
raise RuntimeError(f"Plant node {schematic_name!r} was not found in RBD master nodes.")
|
||||
|
||||
root_metrics = {
|
||||
"system_availability": ramp_result.get("system_availability"),
|
||||
"production_availability": ramp_result.get("production_availability"),
|
||||
"system_downtime": ramp_result.get("system_downtime"),
|
||||
"system_failures": ramp_result.get("system_failures"),
|
||||
"system_uptime": ramp_result.get("system_uptime"),
|
||||
"total_demand": ramp_result.get("total_demand"),
|
||||
"total_productions": ramp_result.get("total_productions") or ramp_result.get("total_production"),
|
||||
}
|
||||
db_session.add(_calc_result(simulation_id=simulation_id, node_id=plant_node.id, metrics=root_metrics))
|
||||
duration_hours = _get_number(root_metrics, ["system_uptime"]) + _get_number(root_metrics, ["system_downtime"])
|
||||
timestamp_outs = _build_ramp_timestamp_outs(ramp_result, duration_hours)
|
||||
db_session.add(
|
||||
AerosSimulationPlotResult(
|
||||
aeros_simulation_id=simulation_id,
|
||||
aeros_node_id=plant_node.id,
|
||||
max_flow_rate=660,
|
||||
storage_capacity=0,
|
||||
point_availabilities=[],
|
||||
point_flowrates=[],
|
||||
timestamp_outs=timestamp_outs,
|
||||
)
|
||||
)
|
||||
|
||||
for asset in ramp_result.get("asset_results") or []:
|
||||
if not isinstance(asset, dict):
|
||||
continue
|
||||
node = nodes_by_key.get(_norm_key(asset.get("name"))) or nodes_by_key.get(_norm_key(asset.get("id")))
|
||||
if not node:
|
||||
log.warning("Skipping RAMP asset result with no matching RBD node: %s", asset.get("name") or asset.get("id"))
|
||||
continue
|
||||
reliability = eq_update.get(asset.get("name")) or eq_update.get(asset.get("id")) or {}
|
||||
db_session.add(
|
||||
_calc_result(
|
||||
simulation_id=simulation_id,
|
||||
node_id=node.id,
|
||||
metrics=asset,
|
||||
reliability=reliability,
|
||||
parameters={"engine": "ramp", "asset_id": asset.get("id")},
|
||||
)
|
||||
)
|
||||
|
||||
simulation.status = "completed"
|
||||
simulation.error = None
|
||||
simulation.completed_at = datetime.now()
|
||||
await db_session.commit()
|
||||
|
||||
|
||||
async def record_ramp_task_start(
|
||||
*,
|
||||
db_session: DbSession,
|
||||
simulation_id: UUID,
|
||||
model_id: str,
|
||||
ramp_task_id: str,
|
||||
payload: dict,
|
||||
) -> None:
|
||||
simulation = await get_simulation_by_id(db_session=db_session, simulation_id=simulation_id)
|
||||
simulation.status = "running"
|
||||
simulation.input = {
|
||||
**(simulation.input or {}),
|
||||
"engine": "ramp",
|
||||
"ramp_model_id": model_id,
|
||||
"ramp_task_id": ramp_task_id,
|
||||
"ramp_payload": {
|
||||
key: value
|
||||
for key, value in payload.items()
|
||||
if key != "distribution_overrides"
|
||||
},
|
||||
"ramp_distribution_override_count": len(payload.get("distribution_overrides") or {}),
|
||||
}
|
||||
await db_session.commit()
|
||||
|
||||
|
||||
async def execute_ramp_simulation(
|
||||
*,
|
||||
db_session: DbSession,
|
||||
simulation_id: UUID,
|
||||
sim_data: dict,
|
||||
eq_update: dict | None = None,
|
||||
) -> bool:
|
||||
timeout = httpx.Timeout(30.0, connect=10.0)
|
||||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||||
model_id = await _resolve_ramp_model_id(client, sim_data)
|
||||
payload = _build_ramp_execute_payload(sim_data, eq_update=eq_update)
|
||||
|
||||
overrides = payload.get("distribution_overrides")
|
||||
if overrides:
|
||||
asset_names = await _fetch_ramp_model_asset_names(client, model_id)
|
||||
kept, dropped = _filter_overrides_to_model(overrides, asset_names)
|
||||
if dropped:
|
||||
log.warning(
|
||||
"Dropping %s distribution override(s) absent from RAMP model %s: %s",
|
||||
len(dropped), model_id, ", ".join(dropped[:25]),
|
||||
)
|
||||
if kept:
|
||||
payload["distribution_overrides"] = kept
|
||||
else:
|
||||
payload.pop("distribution_overrides")
|
||||
log.info(
|
||||
"RAMP model %s matched %s/%s distribution overrides",
|
||||
model_id, len(kept), len(overrides),
|
||||
)
|
||||
|
||||
log.info("Executing RAMP simulation model_id=%s simulation_id=%s", model_id, simulation_id)
|
||||
|
||||
response = await client.post(f"{_simulation_api_base()}/execute/{model_id}", json=payload)
|
||||
response.raise_for_status()
|
||||
task_id = response.json().get("task_id")
|
||||
if not task_id:
|
||||
raise RuntimeError(f"RAMP execute response did not include task_id: {response.text}")
|
||||
await record_ramp_task_start(
|
||||
db_session=db_session,
|
||||
simulation_id=simulation_id,
|
||||
model_id=model_id,
|
||||
ramp_task_id=task_id,
|
||||
payload=payload,
|
||||
)
|
||||
|
||||
deadline = time.monotonic() + RAMP_POLL_TIMEOUT_SECONDS
|
||||
last_status = None
|
||||
last_progress = None
|
||||
last_updated_at = None
|
||||
last_change_at = time.monotonic()
|
||||
last_successful_poll_at = time.monotonic()
|
||||
while time.monotonic() < deadline:
|
||||
try:
|
||||
status_payload = await _fetch_ramp_task_status(client, task_id)
|
||||
last_successful_poll_at = time.monotonic()
|
||||
except httpx.HTTPError as poll_error:
|
||||
if time.monotonic() - last_successful_poll_at > RAMP_STALE_TIMEOUT_SECONDS:
|
||||
await _cancel_ramp_task(client, task_id)
|
||||
raise TimeoutError(
|
||||
f"RAMP task {task_id} status polling failed for "
|
||||
f"{RAMP_STALE_TIMEOUT_SECONDS}s: {poll_error}"
|
||||
) from poll_error
|
||||
log.warning("Transient RAMP poll failure for task %s: %s", task_id, poll_error)
|
||||
await asyncio.sleep(RAMP_POLL_INTERVAL_SECONDS)
|
||||
continue
|
||||
|
||||
previous_marker = (last_status, last_progress, last_updated_at)
|
||||
last_status = status_payload.get("status")
|
||||
progress = status_payload.get("progress")
|
||||
updated_at = status_payload.get("updated_at")
|
||||
|
||||
if (last_status, progress, updated_at) != previous_marker:
|
||||
last_change_at = time.monotonic()
|
||||
last_progress = progress
|
||||
last_updated_at = updated_at
|
||||
|
||||
if (
|
||||
last_status == "running"
|
||||
and RAMP_STALE_TIMEOUT_SECONDS > 0
|
||||
and time.monotonic() - last_change_at > RAMP_STALE_TIMEOUT_SECONDS
|
||||
):
|
||||
await _cancel_ramp_task(client, task_id)
|
||||
raise TimeoutError(
|
||||
f"RAMP task {task_id} made no progress for "
|
||||
f"{RAMP_STALE_TIMEOUT_SECONDS}s. Last progress: {progress}."
|
||||
)
|
||||
|
||||
if last_status == "completed":
|
||||
result = status_payload.get("result")
|
||||
if not isinstance(result, dict):
|
||||
result = await _fetch_completed_ramp_result(client, task_id)
|
||||
await save_ramp_simulation_result(
|
||||
db_session=db_session,
|
||||
simulation_id=simulation_id,
|
||||
sim_data=sim_data,
|
||||
ramp_result=result,
|
||||
eq_update=eq_update or build_ramp_distribution_overrides(sim_data),
|
||||
ramp_task_id=task_id,
|
||||
)
|
||||
return True
|
||||
|
||||
if last_status in {"failed", "cancelled"}:
|
||||
error = status_payload.get("error") or f"RAMP task ended with status {last_status}"
|
||||
raise RuntimeError(error)
|
||||
|
||||
await asyncio.sleep(RAMP_POLL_INTERVAL_SECONDS)
|
||||
|
||||
await _cancel_ramp_task(client, task_id)
|
||||
raise TimeoutError(f"RAMP task timed out after {RAMP_POLL_TIMEOUT_SECONDS}s. Last status: {last_status}")
|
||||
@ -1,219 +0,0 @@
|
||||
from uuid import uuid4
|
||||
|
||||
from src.aeros_simulation.ramp_engine import (
|
||||
_calc_result,
|
||||
_duration_to_hours,
|
||||
_filter_overrides_to_model,
|
||||
build_ramp_distribution_overrides_from_rbd_reliability,
|
||||
build_ramp_distribution_overrides,
|
||||
selected_simulation_engine,
|
||||
)
|
||||
from src.aeros_simulation.schema import SimulationCalc
|
||||
|
||||
|
||||
def test_selected_simulation_engine_defaults_to_aeros():
|
||||
assert selected_simulation_engine({}) == "aeros"
|
||||
|
||||
|
||||
def test_duration_to_hours_supports_aeros_units():
|
||||
assert _duration_to_hours(2, "UYear") == 17520
|
||||
assert _duration_to_hours(3, "UDay") == 72
|
||||
assert _duration_to_hours(5, "UHour") == 5
|
||||
|
||||
|
||||
def test_build_ramp_distribution_overrides_from_ahm_custom_input():
|
||||
overrides = build_ramp_distribution_overrides(
|
||||
{
|
||||
"CustomInput": {
|
||||
"EQ001": {
|
||||
"mttr": 10,
|
||||
"failure_rate": 0.5,
|
||||
"distribution": "Weibull",
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
assert overrides["EQ001"]["repair_duration"] == 10
|
||||
assert overrides["EQ001"]["model"] == {
|
||||
"type": "NonRepairable",
|
||||
"distribution": {"type": "Weibull", "shape": 1.0, "scale": 2000000.0},
|
||||
}
|
||||
|
||||
|
||||
def test_build_ramp_distribution_overrides_from_aeros_distribution_fields():
|
||||
overrides = build_ramp_distribution_overrides(
|
||||
{
|
||||
"DistributionOverrides": {
|
||||
"PUMP-A": {
|
||||
"relDisType": "Weibull2",
|
||||
"relDisP1": 1.7,
|
||||
"relDisP2": 9000,
|
||||
"cmDisType": "Normal",
|
||||
"cmDisP1": 8,
|
||||
"cmDisP2": 1.5,
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
assert overrides["PUMP-A"]["model"]["distribution"] == {
|
||||
"type": "Weibull",
|
||||
"shape": 1.7,
|
||||
"scale": 9000.0,
|
||||
}
|
||||
assert overrides["PUMP-A"]["repair_duration"] == {
|
||||
"type": "Normal",
|
||||
"mean": 8.0,
|
||||
"std_dev": 1.5,
|
||||
}
|
||||
|
||||
|
||||
def test_ramp_calc_result_uses_numeric_defaults_for_aeros_only_fields():
|
||||
calc = _calc_result(
|
||||
simulation_id=uuid4(),
|
||||
node_id=uuid4(),
|
||||
metrics={
|
||||
"system_availability": 0.95,
|
||||
"system_uptime": 95,
|
||||
"system_downtime": 5,
|
||||
"system_failures": 2,
|
||||
},
|
||||
)
|
||||
|
||||
assert calc.stg_input == 0
|
||||
assert calc.stg_output == 0
|
||||
assert calc.average_level == 0
|
||||
assert calc.availability == 0.95
|
||||
assert calc.eaf == 95.0
|
||||
assert calc.sof == 0
|
||||
assert calc.total_mo_downtime == 0
|
||||
assert calc.total_po_downtime == 0
|
||||
|
||||
|
||||
def test_simulation_calc_response_defaults_nullable_numbers_for_fe():
|
||||
payload = {
|
||||
"id": uuid4(),
|
||||
"total_downtime": 5,
|
||||
"total_uptime": 95,
|
||||
"num_events": 2,
|
||||
"production": 10,
|
||||
"production_std": 0,
|
||||
"ideal_production": 12,
|
||||
"availability": 0.95,
|
||||
"efficiency": 83.3,
|
||||
"effective_loss": 2,
|
||||
"num_cm": 2,
|
||||
"cm_waiting_time": 0,
|
||||
"total_cm_downtime": 5,
|
||||
"num_pm": 0,
|
||||
"total_pm_downtime": 0,
|
||||
"num_ip": 0,
|
||||
"total_ip_downtime": 0,
|
||||
"num_oh": 0,
|
||||
"total_oh_downtime": 0,
|
||||
"t_wait_for_crew": 0,
|
||||
"t_wait_for_spare": 0,
|
||||
"duration_at_full": 95,
|
||||
"duration_above_hh": 0,
|
||||
"duration_above_h": 0,
|
||||
"duration_below_l": 0,
|
||||
"duration_below_ll": 0,
|
||||
"duration_at_empty": 5,
|
||||
"stg_input": 0,
|
||||
"eaf": 95,
|
||||
"eaf_konkin": None,
|
||||
"efor": None,
|
||||
"derating_hours": None,
|
||||
"aeros_node": {
|
||||
"id": uuid4(),
|
||||
"node_type": "SchematicNode",
|
||||
"node_id": 1,
|
||||
"node_name": "- TJB - Unit 3 -",
|
||||
"structure_name": "- TJB - Unit 3 -",
|
||||
"schematic_name": "- TJB - Unit 3 -",
|
||||
"schematic_id": uuid4(),
|
||||
"model_image": [],
|
||||
"equipment": None,
|
||||
},
|
||||
"contribution": None,
|
||||
"criticality": None,
|
||||
"contribution_factor": None,
|
||||
"sof": None,
|
||||
}
|
||||
|
||||
calc = SimulationCalc.model_validate(payload)
|
||||
|
||||
assert calc.eaf_konkin == 0
|
||||
assert calc.efor == 0
|
||||
assert calc.derating_hours == 0
|
||||
assert calc.sof == 0
|
||||
|
||||
|
||||
def test_build_ramp_distribution_overrides_from_rbd_reliability():
|
||||
overrides = build_ramp_distribution_overrides_from_rbd_reliability(
|
||||
{
|
||||
"PUMP-A": {
|
||||
"distribution": "NHPPTTFF",
|
||||
"beta": 1.4,
|
||||
"eta": 8000,
|
||||
"mttr": 12,
|
||||
},
|
||||
"PUMP-B": {
|
||||
"distribution": "Weibull2",
|
||||
"beta": 2,
|
||||
"eta": 9000,
|
||||
"mttr": 8,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
assert overrides["PUMP-A"]["model"] == {
|
||||
"type": "RepairableNHPP",
|
||||
"beta": 1.4,
|
||||
"eta": 8000.0,
|
||||
}
|
||||
assert overrides["PUMP-A"]["repair_duration"] == 12.0
|
||||
assert overrides["PUMP-B"]["model"]["distribution"] == {
|
||||
"type": "Weibull",
|
||||
"shape": 2.0,
|
||||
"scale": 9000.0,
|
||||
}
|
||||
|
||||
|
||||
def test_build_ramp_distribution_overrides_uses_rbd_eta_for_exponential():
|
||||
overrides = build_ramp_distribution_overrides_from_rbd_reliability(
|
||||
{
|
||||
"HEATER-A": {
|
||||
"distribution": "Exponential2",
|
||||
"beta": 13034.0,
|
||||
"eta": 8760.0,
|
||||
"mttr": 6,
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
assert overrides["HEATER-A"]["model"]["distribution"] == {
|
||||
"type": "Exponential",
|
||||
"eta": 8760.0,
|
||||
}
|
||||
|
||||
|
||||
def test_filter_overrides_keeps_only_assets_present_in_model():
|
||||
overrides = {
|
||||
"3AH-AU501A": {"repair_duration": 4},
|
||||
"3bad-m531": {"repair_duration": 6},
|
||||
"00CHA-CV804": {"repair_duration": 8},
|
||||
}
|
||||
kept, dropped = _filter_overrides_to_model(overrides, {"3AH-AU501A", "3BAD-M531"})
|
||||
|
||||
assert set(kept) == {"3AH-AU501A", "3bad-m531"}
|
||||
assert dropped == ["00CHA-CV804"]
|
||||
|
||||
|
||||
def test_filter_overrides_passes_through_when_model_assets_unknown():
|
||||
overrides = {"3AH-AU501A": {"repair_duration": 4}}
|
||||
kept, dropped = _filter_overrides_to_model(overrides, set())
|
||||
|
||||
assert kept == overrides
|
||||
assert dropped == []
|
||||
Loading…
Reference in New Issue