diff --git a/.env b/.env index 36970e3..90a6f47 100644 --- a/.env +++ b/.env @@ -23,4 +23,8 @@ TEMPORAL_URL=http://192.168.1.86:7233 VAULT_URL=https://192.168.1.82:8200 AEROS_SECRET_PATH=s54%gT1x2jcF7s+pj9a5 ROLE_ID=19a5e4c5-8090-9f51-02bd-0b15625f2360 -SECRET_ID=eefd1c58-0509-f598-bf37-d1dd73215e2b \ No newline at end of file +SECRET_ID=eefd1c58-0509-f598-bf37-d1dd73215e2b + +# RAMP simulation engine backend +RAMP_API_URL=http://192.168.1.82:3027 +RAMP_MODEL_ID=22 diff --git a/.env.example b/.env.example index 313f871..09240c4 100644 --- a/.env.example +++ b/.env.example @@ -33,3 +33,8 @@ VAULT_URL= ROLE_ID= SECRET_ID= AEROS_SECRET_PATH= + +# RAMP simulation engine backend +SIMULATION_ENGINE=aeros +RAMP_API_URL=http://192.168.1.82:3027 +RAMP_MODEL_ID=22 diff --git a/run_worker.py b/run_worker.py index 5e88d6d..125c32b 100644 --- a/run_worker.py +++ b/run_worker.py @@ -3,7 +3,7 @@ import os from temporalio.client import Client from temporalio.worker import Worker -from temporal.activity import calculate_plant_eaf_activity, execute_simulation_activity, update_contribution_bulk_mappings_activity, update_equipment_for_simulation_activity, call_callback_ahm, acquire_lock, release_lock,execute_update_node +from temporal.activity import calculate_plant_eaf_activity, collect_rbd_reliability_for_ramp_activity, execute_ramp_simulation_activity, execute_simulation_activity, get_simulation_engine_activity, update_contribution_bulk_mappings_activity, update_equipment_for_simulation_activity, call_callback_ahm, acquire_lock, release_lock,execute_update_node, handle_workflow_failure from temporal.workflow import SimulationWorkflow TEMPORAL_URL = os.environ.get("TEMPORAL_URL", "http://192.168.1.86:7233") @@ -18,13 +18,17 @@ async def main(): workflows=[SimulationWorkflow], activities=[ update_equipment_for_simulation_activity, + get_simulation_engine_activity, + collect_rbd_reliability_for_ramp_activity, execute_simulation_activity, + execute_ramp_simulation_activity, calculate_plant_eaf_activity, update_contribution_bulk_mappings_activity, call_callback_ahm, acquire_lock, release_lock, - execute_update_node + execute_update_node, + handle_workflow_failure ], max_concurrent_workflow_tasks=50, max_concurrent_activities=12 diff --git a/src/aeros_equipment/service.py b/src/aeros_equipment/service.py index 3cbb598..366b22c 100644 --- a/src/aeros_equipment/service.py +++ b/src/aeros_equipment/service.py @@ -416,6 +416,109 @@ async def update_oh_interval_offset(*, aeros_db_session, overhaul_offset, overha log.info("Updated OH Interval/Offset for project %s. Rows affected: %s", project_name, result.rowcount) +async def collect_rbd_reliability_for_ramp( + *, + db_session: DbSession, + schematic_name: str, + custom_input: Optional[dict] = None, +): + aeros_schematic = await get_aeros_schematic_by_name( + db_session=db_session, + schematic_name=schematic_name + ) + + equipments = ( + Select(AerosEquipment) + .where(AerosEquipment.location_tag.isnot(None)) + .join(AerosEquipment.aeros_node) + .filter( + AerosNode.aeros_schematic_id == aeros_schematic.id, + AerosNode.node_type == "RegularNode" + ) + ) + rbd_group = Select(AerosEquipmentGroup) + + equipment_nodes = await db_session.execute(equipments) + rbd_group_nodes = await db_session.execute(rbd_group) + + group_nodes = [g.group_name for g in rbd_group_nodes.scalars().all()] + reg_nodes = [n.location_tag for n in equipment_nodes.scalars().all()] + reg_nodes.extend(group_nodes) + + non_repairable = await db_session.execute( + Select(ReliabilityPredictNonRepairable) + ) + non_repairable_location_tags = [ + t.location_tag for t in non_repairable.scalars().all() + ] + + reliability_data = await get_asset_batch( + reg_nodes, + non_repairable_location_tags, + db_session + ) + + results = {} + for location_tag in reg_nodes: + if not location_tag: + continue + reliability = reliability_data.get(location_tag.strip().upper(), {}) + if not reliability: + continue + + results[location_tag] = { + "mttr": reliability.get("cmDisP1", 0), + "distribution": reliability.get("relDisType", "Fixed"), + "beta": reliability.get("relDisP1", 0), + "eta": reliability.get("relDisP2", 0), + "parameters": reliability.get("parameters", {}), + } + + if custom_input: + for location_tag, custom_param in custom_input.items(): + if not isinstance(custom_param, dict): + continue + + mttr = custom_param.get("mttr") + failure_rate = custom_param.get("failure_rate") + distribution = custom_param.get("distribution", "Fixed") + + if custom_param.get("level") == "NoFailure": + results[location_tag] = { + "mttr": 0, + "distribution": "NoFailure", + "beta": 0, + "eta": 0, + "parameters": custom_param, + } + continue + + if custom_param.get("level") == "Out of Service": + results[location_tag] = { + "mttr": 1000000000.0, + "distribution": "Fixed", + "beta": 0.0001, + "eta": 0, + "parameters": custom_param, + } + continue + + if failure_rate is not None: + if isinstance(failure_rate, list): + failure_rate = failure_rate[0] if failure_rate else None + rate_val = float(failure_rate or 0) + mtbf = 1000000000.0 if rate_val <= 0 else 1e6 / rate_val + results[location_tag] = { + "mttr": mttr or 0, + "distribution": distribution, + "beta": 1.0 if distribution != "Fixed" else mtbf, + "eta": mtbf if distribution != "Fixed" else 0, + "parameters": custom_param, + } + + return results + + async def update_equipment_for_simulation( *, db_session: DbSession, diff --git a/src/aeros_simulation/model.py b/src/aeros_simulation/model.py index bf23a8b..4bb05ab 100644 --- a/src/aeros_simulation/model.py +++ b/src/aeros_simulation/model.py @@ -2,6 +2,7 @@ from sqlalchemy import JSON, UUID, Boolean, Column, DateTime, Float, ForeignKey, from sqlalchemy.orm import relationship from src.database.core import Base +from src.aeros_equipment.model import AerosEquipment, MasterEquipment from src.models import DefaultMixin import uuid diff --git a/src/aeros_simulation/ramp_engine.py b/src/aeros_simulation/ramp_engine.py new file mode 100644 index 0000000..5b4cad7 --- /dev/null +++ b/src/aeros_simulation/ramp_engine.py @@ -0,0 +1,685 @@ +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_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 beta and beta > 0: + exp_eta = 1 / 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 _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 _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 _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 = _availability_percent(_first_present(metrics, "availability", "system_availability", "production_availability")) + if not availability and period_hours: + availability = total_uptime / period_hours * 100 + 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=availability, + efor=efor, + derating_hours=effective_loss, + beta=reliability.get("beta"), + eta=reliability.get("eta"), + mttr=reliability.get("mttr"), + parameters=parameters or reliability.get("parameters"), + contribution=_get_number(metrics, ["contribution"]), + criticality=_get_number(metrics, ["criticality"]), + contribution_factor=_get_number(metrics, ["contribution_factor"]), + ) + + +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.completed_at = datetime.now() + 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}") + + deadline = time.monotonic() + RAMP_POLL_TIMEOUT_SECONDS + last_status = None + while time.monotonic() < deadline: + status_response = await client.get(f"{_simulation_api_base()}/status/{task_id}") + status_response.raise_for_status() + status_payload = status_response.json() + last_status = status_payload.get("status") + + if last_status == "completed": + result = status_payload.get("result") + if not isinstance(result, dict): + raise RuntimeError(f"RAMP task {task_id} completed without a result payload.") + 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) + + raise TimeoutError(f"RAMP task timed out after {RAMP_POLL_TIMEOUT_SECONDS}s. Last status: {last_status}") diff --git a/src/aeros_simulation/router.py b/src/aeros_simulation/router.py index 05b2328..5ff8d0a 100644 --- a/src/aeros_simulation/router.py +++ b/src/aeros_simulation/router.py @@ -390,6 +390,13 @@ async def get_critical_equipment(db_session:DbSession, simulation_id: str): equipment_list = [eq.id for eq in equipment] results.extend(equipment_list) + + if not results: + return { + "data": [], + "status": "success", + "message": "Success", + } query = (select(AerosSimulationCalcResult).filter( AerosSimulationCalcResult.aeros_simulation_id == simulation_id)).filter(AerosSimulationCalcResult.aeros_node_id.in_(results)) @@ -789,4 +796,3 @@ async def calculate_contribution( raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e) ) - diff --git a/src/aeros_simulation/schema.py b/src/aeros_simulation/schema.py index 7804daa..f8533bb 100644 --- a/src/aeros_simulation/schema.py +++ b/src/aeros_simulation/schema.py @@ -18,6 +18,12 @@ class SimulationInput(DefultBase): SimNumRun: int = 1 SimulationName: str = Field("DefaultSimulation", max_length=100, pattern=r"^[a-zA-Z0-9_\-\s]+$") CustomInput: dict = {} + DistributionOverrides: dict = {} + RampDistributionOverrides: dict = {} + SimulationEngine: Optional[str] = Field(None, max_length=20) + RampModelId: Optional[int] = None + RampModelName: Optional[str] = Field(None, max_length=100) + RampTargetSchema: Optional[str] = Field(None, max_length=100) IsDefault:bool = False Konkin_offset: Optional[int] = 0 OffSet: Optional[int] = 0 @@ -146,4 +152,4 @@ class SimulationQueryModel(CommonParams): class SimulationCalcResultQuery(DefultBase): schematic_name: Optional[str] = Field(None, max_length=100) - node_type: Optional[str] = Field(None, alias="nodetype", max_length=50) \ No newline at end of file + node_type: Optional[str] = Field(None, alias="nodetype", max_length=50) diff --git a/src/config.py b/src/config.py index d767e6f..f38826a 100644 --- a/src/config.py +++ b/src/config.py @@ -102,6 +102,14 @@ CLAMAV_PORT = config("CLAMAV_PORT", cast=int, default=3310) AEROS_LICENSE_ID = config("AEROS_LICENSE_ID", default=None) AEROS_LICENSE_SECRET = config("AEROS_LICENSE_SECRET", default=None) +SIMULATION_ENGINE = config("SIMULATION_ENGINE", default="aeros") +RAMP_API_URL = config("RAMP_API_URL", default="http://localhost:8000") +RAMP_MODEL_ID = config("RAMP_MODEL_ID", default="22") +RAMP_MODEL_NAME = config("RAMP_MODEL_NAME", default=None) +RAMP_TARGET_SCHEMA = config("RAMP_TARGET_SCHEMA", default=None) +RAMP_POLL_INTERVAL_SECONDS = config("RAMP_POLL_INTERVAL_SECONDS", cast=int, default=10) +RAMP_POLL_TIMEOUT_SECONDS = config("RAMP_POLL_TIMEOUT_SECONDS", cast=int, default=7200) + VAULT_URL=config('VAULT_URL', default=None) ROLE_ID=config('ROLE_ID', default=None) SECRET_ID=config('SECRET_ID', default=None) @@ -110,4 +118,4 @@ AEROS_SECRET_PATH=config('AEROS_SECRET_PATH', default=None) USE_LICENSE_APP = config("USE_LICENSE_APP", cast=bool, default=True) AHM_BASE_URL = config("AHM_BASE_URL", default="http://192.168.1.82:8000/ahm") -AHM_SIMULATION_CALLBACK_URL = config("AHM_SIMULATION_CALLBACK_URL", default="/api/v1/simulations/rbd/callback") \ No newline at end of file +AHM_SIMULATION_CALLBACK_URL = config("AHM_SIMULATION_CALLBACK_URL", default="/api/v1/simulations/rbd/callback") diff --git a/src/middleware.py b/src/middleware.py index 4eeffd4..99479a1 100644 --- a/src/middleware.py +++ b/src/middleware.py @@ -16,9 +16,10 @@ ALLOWED_MULTI_PARAMS = { } ALLOWED_DATA_PARAMS = { - "AerosData", "AhmJobId", "CustomInput", "DurationUnit", "IsDefault", "Konkin_offset", - "MaintenanceOutages", "MasterData", "OffSet", "OverhaulDuration", "OverhaulInterval", - "PlannedOutages", "SchematicName", "SecretStr", "SimDuration", "SimNumRun", "SimSeed", + "AerosData", "AhmJobId", "CustomInput", "DistributionOverrides", "DurationUnit", + "IsDefault", "Konkin_offset", "MaintenanceOutages", "MasterData", "OffSet", "OverhaulDuration", "OverhaulInterval", + "PlannedOutages", "RampModelId", "RampModelName", "RampTargetSchema", "SchematicName", + "SecretStr", "SimDuration", "SimNumRun", "SimSeed", "SimulationEngine", "RampDistributionOverrides", "SimulationName", "actual_shutdown", "aeros_node", "all_params", "aro_file", "aro_file_path", "availability", "baseline_simulation_id", "calc_results", "cm_dis_p1", "cm_dis_p2", "cm_dis_p3", "cm_dis_type", "cm_dis_unit_code", diff --git a/temporal/activity.py b/temporal/activity.py index f58d16e..448ee6b 100644 --- a/temporal/activity.py +++ b/temporal/activity.py @@ -1,10 +1,17 @@ import asyncio +from datetime import datetime +from email.message import EmailMessage +import json from temporalio import activity import os +import requests import redis.asyncio as redis +import smtplib +from uuid import UUID REDIS_HOST = os.getenv("REDIS_HOST", "192.168.1.82") REDIS_PORT = int(os.getenv("REDIS_PORT", 6379)) +ERROR_NOTIFICATION_TIMEOUT_SECONDS = int(os.getenv("ERROR_NOTIFICATION_TIMEOUT_SECONDS", "10")) redis_client = redis.Redis( @@ -15,6 +22,59 @@ redis_client = redis.Redis( ) +def _shorten(value, limit=2000): + text = str(value) + return text if len(text) <= limit else f"{text[:limit]}... [truncated]" + + +def _send_error_notification(subject: str, message: str, context: dict) -> dict: + payload = { + "service": os.getenv("SERVICE_NAME", "be-rbd"), + "subject": subject, + "message": _shorten(message), + "context": context, + "timestamp": datetime.utcnow().isoformat() + "Z", + } + results = {} + + webhook_url = os.getenv("ERROR_NOTIFICATION_WEBHOOK_URL") + if webhook_url: + try: + response = requests.post(webhook_url, json=payload, timeout=ERROR_NOTIFICATION_TIMEOUT_SECONDS) + response.raise_for_status() + results["webhook"] = "sent" + except Exception as e: + results["webhook"] = f"failed: {_shorten(e, 300)}" + + email_to = os.getenv("ERROR_NOTIFICATION_EMAIL_TO") + smtp_host = os.getenv("SMTP_HOST") + if email_to and smtp_host: + try: + msg = EmailMessage() + msg["Subject"] = subject + msg["From"] = os.getenv("SMTP_FROM") or os.getenv("SMTP_USERNAME") or "noreply@localhost" + msg["To"] = email_to + msg.set_content(json.dumps(payload, indent=2, default=str)) + + port = int(os.getenv("SMTP_PORT", "587")) + with smtplib.SMTP(smtp_host, port, timeout=ERROR_NOTIFICATION_TIMEOUT_SECONDS) as smtp: + if os.getenv("SMTP_USE_TLS", "true").lower() == "true": + smtp.starttls() + username = os.getenv("SMTP_USERNAME") + password = os.getenv("SMTP_PASSWORD") + if username and password: + smtp.login(username, password) + smtp.send_message(msg) + results["email"] = "sent" + except Exception as e: + results["email"] = f"failed: {_shorten(e, 300)}" + + if not results: + results["notification"] = "skipped: configure ERROR_NOTIFICATION_WEBHOOK_URL or SMTP_HOST + ERROR_NOTIFICATION_EMAIL_TO" + + return {"status": True, "channels": results, "payload": payload} + + @activity.defn async def update_equipment_for_simulation_activity(params: dict): # ✅ Import inside the activity function @@ -36,6 +96,26 @@ async def update_equipment_for_simulation_activity(params: dict): ) +@activity.defn +async def get_simulation_engine_activity(params: dict) -> str: + from src.aeros_simulation.ramp_engine import selected_simulation_engine + + return selected_simulation_engine(params) + + +@activity.defn +async def collect_rbd_reliability_for_ramp_activity(params: dict): + from src.aeros_equipment.service import collect_rbd_reliability_for_ramp + from src.database.core import async_session + + async with async_session() as db_session: + return await collect_rbd_reliability_for_ramp( + db_session=db_session, + schematic_name=params["SchematicName"], + custom_input=params.get("CustomInput"), + ) + + @activity.defn async def execute_simulation_activity(params: dict): from src.database.core import async_session @@ -49,6 +129,21 @@ async def execute_simulation_activity(params: dict): eq_update=params["eq_update"] ) + +@activity.defn +async def execute_ramp_simulation_activity(params: dict): + from src.database.core import async_session + from src.aeros_simulation.ramp_engine import build_ramp_distribution_overrides, execute_ramp_simulation + + async with async_session() as db_session: + return await execute_ramp_simulation( + db_session=db_session, + simulation_id=params["sim_data"]["HubCnnId"], + sim_data=params["sim_data"], + eq_update=params.get("eq_update") or build_ramp_distribution_overrides(params["sim_data"]), + ) + + @activity.defn async def execute_update_node(params:dict): from src.database.core import async_session @@ -101,10 +196,43 @@ async def call_callback_ahm(params): except Exception as e: activity.logger.error(f"Callback to AHM failed: {str(e)}") raise e + + +@activity.defn +async def handle_workflow_failure(params: dict) -> dict: + import src.aeros_equipment.model # noqa: F401 + from src.aeros_simulation.model import AerosSimulation + from src.database.core import async_session + + simulation_id = params.get("simulation_id") + error = params.get("error", "Unknown workflow error") + workflow_stage = params.get("stage", "workflow") + + async with async_session() as db_session: + simulation = await db_session.get(AerosSimulation, UUID(str(simulation_id))) + if simulation: + simulation.status = "failed" + simulation.error = { + "stage": workflow_stage, + "message": _shorten(error), + } + simulation.completed_at = datetime.now() + await db_session.commit() + + notification = _send_error_notification( + subject="RBD simulation workflow failed", + message=error, + context={ + "simulation_id": simulation_id, + "stage": workflow_stage, + "ahm_job_id": params.get("ahm_job_id"), + }, + ) + return {"status": "failed_recorded", "notification": notification} LOCK_KEY = "lock:aeros_process" -LOCK_TIMEOUT = 864000 # seconds +LOCK_TIMEOUT = int(os.getenv("SIMULATION_LOCK_TIMEOUT_SECONDS", "7200")) # 2 hours @activity.defn async def acquire_lock(job_id: str) -> dict: @@ -119,7 +247,7 @@ async def acquire_lock(job_id: str) -> dict: if acquired: # Lock acquired - redis_client.hset( + await redis_client.hset( f"process:{job_id}", mapping={ "job_id": job_id, @@ -134,7 +262,7 @@ async def acquire_lock(job_id: str) -> dict: } # Someone else holds the lock - redis_client.hset( + await redis_client.hset( f"process:{job_id}", mapping={ "job_id": job_id, @@ -148,7 +276,7 @@ async def acquire_lock(job_id: str) -> dict: except Exception: - redis_client.hset( + await redis_client.hset( f"process:{job_id}", mapping={ "job_id": job_id, @@ -162,11 +290,11 @@ async def acquire_lock(job_id: str) -> dict: async def release_lock(job_id: str): await redis_client.delete(LOCK_KEY) - redis_client.hset( + await redis_client.hset( f"process:{job_id}", mapping={ "job_id": job_id, "status": "Done" } ) - await redis_client.expire(f"process:{job_id}", 300) \ No newline at end of file + await redis_client.expire(f"process:{job_id}", 300) diff --git a/temporal/workflow.py b/temporal/workflow.py index 5445a45..72f7120 100644 --- a/temporal/workflow.py +++ b/temporal/workflow.py @@ -1,12 +1,20 @@ from datetime import timedelta from temporalio import workflow from temporalio.client import Client +from temporalio.common import RetryPolicy import os with workflow.unsafe.imports_passed_through(): - from temporal.activity import calculate_plant_eaf_activity, execute_simulation_activity, update_contribution_bulk_mappings_activity, update_equipment_for_simulation_activity,call_callback_ahm, acquire_lock, release_lock,execute_update_node + from temporal.activity import calculate_plant_eaf_activity, collect_rbd_reliability_for_ramp_activity, execute_ramp_simulation_activity, execute_simulation_activity, get_simulation_engine_activity, update_contribution_bulk_mappings_activity, update_equipment_for_simulation_activity,call_callback_ahm, acquire_lock, release_lock,execute_update_node, handle_workflow_failure TEMPORAL_URL = os.environ.get("TEMPORAL_URL", "http://192.168.1.86:7233") +_RETRY_POLICY = RetryPolicy( + initial_interval=timedelta(seconds=10), + backoff_coefficient=2.0, + maximum_interval=timedelta(minutes=2), + maximum_attempts=3, +) +_NO_RETRY_POLICY = RetryPolicy(maximum_attempts=1) @workflow.defn class SimulationWorkflow: @@ -25,74 +33,141 @@ class SimulationWorkflow: @workflow.run async def run(self, sim_data: dict) -> str: self.status = "QUEUE" - - await workflow.execute_activity( - acquire_lock, - sim_data["HubCnnId"], - start_to_close_timeout=timedelta(days=10) - ) - - self.status = "RUNNING" - - # 1. Update equipment for simulation - results = await workflow.execute_activity( - update_equipment_for_simulation_activity, - {**sim_data}, # ✅ one positional argument - start_to_close_timeout=timedelta(days=1) - ) + lock_acquired = False + workflow_failed = False - await workflow.execute_activity( - execute_update_node, - {"projectName": sim_data['projectName'], "equipment_nodes": results['equipment_nodes']}, - start_to_close_timeout=timedelta(days=1) - ) + try: + await workflow.execute_activity( + acquire_lock, + sim_data["HubCnnId"], + start_to_close_timeout=timedelta(hours=2, minutes=5), + retry_policy=_NO_RETRY_POLICY, + ) + lock_acquired = True + + self.status = "RUNNING" + + engine = await workflow.execute_activity( + get_simulation_engine_activity, + {**sim_data}, + start_to_close_timeout=timedelta(seconds=30), + retry_policy=_NO_RETRY_POLICY, + ) - # 2. Execute simulation - await workflow.execute_activity( - execute_simulation_activity, - {"sim_data": sim_data, "eq_update": results['reliability']}, - start_to_close_timeout=timedelta(days=1) - ) + if engine == "aeros": + self.status = "UPDATING_AEROS_EQUIPMENT" + results = await workflow.execute_activity( + update_equipment_for_simulation_activity, + {**sim_data}, + start_to_close_timeout=timedelta(minutes=30), + retry_policy=_RETRY_POLICY, + ) - # 3. Calculate EAF - await workflow.execute_activity( - calculate_plant_eaf_activity, - sim_data, - start_to_close_timeout=timedelta(days=1) - ) - - await workflow.execute_activity( - update_contribution_bulk_mappings_activity, - sim_data["HubCnnId"], - start_to_close_timeout=timedelta(days=1) - ) + await workflow.execute_activity( + execute_update_node, + {"projectName": sim_data['projectName'], "equipment_nodes": results['equipment_nodes']}, + start_to_close_timeout=timedelta(minutes=30), + retry_policy=_RETRY_POLICY, + ) + self.status = "RUNNING_AEROS_SIMULATION" + await workflow.execute_activity( + execute_simulation_activity, + {"sim_data": sim_data, "eq_update": results['reliability']}, + start_to_close_timeout=timedelta(hours=2), + retry_policy=_NO_RETRY_POLICY, + ) - if sim_data.get("AhmJobId"): + await workflow.execute_activity( + calculate_plant_eaf_activity, + sim_data, + start_to_close_timeout=timedelta(minutes=30), + retry_policy=_RETRY_POLICY, + ) + elif engine == "ramp": + self.status = "BUILDING_RAMP_DISTRIBUTION_OVERRIDES" + results = await workflow.execute_activity( + collect_rbd_reliability_for_ramp_activity, + {**sim_data}, + start_to_close_timeout=timedelta(minutes=30), + retry_policy=_RETRY_POLICY, + ) + + self.status = "RUNNING_RAMP_SIMULATION" + await workflow.execute_activity( + execute_ramp_simulation_activity, + {"sim_data": sim_data, "eq_update": results}, + start_to_close_timeout=timedelta(hours=2), + retry_policy=_NO_RETRY_POLICY, + ) + else: + raise ValueError(f"Unsupported simulation engine: {engine}") + await workflow.execute_activity( - call_callback_ahm, - { - "simulation_id": sim_data["HubCnnId"], - "job_id": sim_data["AhmJobId"] - }, - start_to_close_timeout=timedelta(days=1) + update_contribution_bulk_mappings_activity, + sim_data["HubCnnId"], + start_to_close_timeout=timedelta(minutes=30), + retry_policy=_RETRY_POLICY, ) - self.status = "COMPLETED" - # etc… - self.isDone = True + if sim_data.get("AhmJobId"): + await workflow.execute_activity( + call_callback_ahm, + { + "simulation_id": sim_data["HubCnnId"], + "job_id": sim_data["AhmJobId"] + }, + start_to_close_timeout=timedelta(minutes=5), + retry_policy=_RETRY_POLICY, + ) - await workflow.execute_activity( - release_lock, - sim_data["HubCnnId"], - start_to_close_timeout=timedelta(minutes=10), - ) - - if sim_data.get("CallbackWorkflowId"): - handle = workflow.get_external_workflow_handle( - sim_data["CallbackWorkflowId"] - ) - await handle.signal("notify_done") + self.status = "COMPLETED" + self.isDone = True + + if sim_data.get("CallbackWorkflowId"): + handle = workflow.get_external_workflow_handle( + sim_data["CallbackWorkflowId"] + ) + await handle.signal("notify_done") + return sim_data["HubCnnId"] - return sim_data["HubCnnId"] # simulation_id + except Exception as e: + failed_stage = self.status + self.status = "FAILED" + workflow_failed = True + await workflow.execute_activity( + handle_workflow_failure, + { + "simulation_id": sim_data["HubCnnId"], + "stage": failed_stage, + "error": str(e), + "ahm_job_id": sim_data.get("AhmJobId"), + }, + start_to_close_timeout=timedelta(seconds=30), + retry_policy=_NO_RETRY_POLICY, + ) + raise + finally: + if lock_acquired: + try: + await workflow.execute_activity( + release_lock, + sim_data["HubCnnId"], + start_to_close_timeout=timedelta(minutes=10), + retry_policy=_RETRY_POLICY, + ) + except Exception as release_error: + await workflow.execute_activity( + handle_workflow_failure, + { + "simulation_id": sim_data["HubCnnId"], + "stage": "release_lock", + "error": f"Failed to release simulation lock: {release_error}", + "ahm_job_id": sim_data.get("AhmJobId"), + }, + start_to_close_timeout=timedelta(seconds=30), + retry_policy=_NO_RETRY_POLICY, + ) + if not workflow_failed: + raise diff --git a/tests/unit/test_ramp_engine.py b/tests/unit/test_ramp_engine.py new file mode 100644 index 0000000..8b840a6 --- /dev/null +++ b/tests/unit/test_ramp_engine.py @@ -0,0 +1,137 @@ +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, +) + + +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 == 95.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_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 == []