From f4ebdf2d6019575b556753efa3c3c96e14de2320 Mon Sep 17 00:00:00 2001 From: Cizz22 Date: Fri, 10 Apr 2026 11:41:38 +0700 Subject: [PATCH] refactor: replace external API call with direct database queries for asset batch data retrieval --- src/aeros_equipment/service.py | 186 ++++++++++++++++++++------------- 1 file changed, 112 insertions(+), 74 deletions(-) diff --git a/src/aeros_equipment/service.py b/src/aeros_equipment/service.py index e3875e1..6ff5de2 100644 --- a/src/aeros_equipment/service.py +++ b/src/aeros_equipment/service.py @@ -226,55 +226,71 @@ async def save_default_equipment(*, db_session: DbSession, project_name: str): def get_asset_batch(location_tags: List[str], nr_location_tags: List[str], + db_session: DbSession, base_url: str = RELIABILITY_SERVICE_API, timeout: int = 600)-> dict: """ - Get asset batch data using GET request with JSON body. + Get asset batch data from database tables. Args: - location_tags: List of location tag strings - base_url: Base URL for the API - timeout: Request timeout in seconds + location_tags: List of location tag strings for repairable equipment + nr_location_tags: List of location tag strings for non-repairable equipment + db_session: Database session + base_url: Base URL for the API (kept for compatibility) + timeout: Request timeout in seconds (kept for compatibility) Returns: - Dictionary with response data or None if error + Dictionary with response data """ - url = f"{base_url}/asset/batch" + from sqlalchemy import text + import json + results = defaultdict(dict) - payload = { - "location_tags": location_tags - } - - headers = { - 'Content-Type': 'application/json' - } - - try: - response = requests.get( - url, - json=payload, - headers=headers, - timeout=timeout - ) - - response.raise_for_status() # Raises an HTTPError for bad responses - data = response.json() - reliabiility_data = data.get("data", []) - - for item in reliabiility_data: - location_tag = item.get("location_tag", None) + # Query repairable equipment data + if location_tags: + repairable_query = text(""" + SELECT location_tag, parameters + FROM rp_repairable_results + WHERE location_tag = ANY(:location_tags) + """) + + repairable_result = db_session.execute(repairable_query, {"location_tags": location_tags}) + + for row in repairable_result: + location_tag = row[0] + parameters = json.loads(row[1]) if row[1] else {} - if not location_tag: - print("Error processing item" + str(item)) - continue + try: + distribution, reldisp1, reldisp2 = get_distribution_from_db(parameters, is_repairable=True) + + results[location_tag]["cmDisType"] = "Normal" + results[location_tag]["cmDisP1"] = 6 + results[location_tag]["cmDisP2"] = 3 + results[location_tag]["relDisType"] = distribution + results[location_tag]["relDisP1"] = reldisp1 + results[location_tag]["relDisP2"] = reldisp2 + results[location_tag]["parameters"] = parameters + except Exception as e: + raise Exception(f"Error processing repairable item {location_tag}: {e}") + + # Query non-repairable equipment data + if nr_location_tags: + non_repairable_query = text(""" + SELECT location_tag, parameters, best_distribution + FROM rp_non_repairable_results + WHERE location_tag = ANY(:nr_location_tags) + """) + + non_repairable_result = db_session.execute(non_repairable_query, {"nr_location_tags": nr_location_tags}) + + for row in non_repairable_result: + location_tag = row[0] + parameters = json.loads(row[1]) if row[1] else {} + best_distribution = row[2] try: - # is_nr = item["distribution"] != "NHPP" - # mtbf = item["mtbf"] - # mttr = item["mttr"] - print(item) - distribution, reldisp1, reldisp2 = get_distribution(item) + distribution, reldisp1, reldisp2 = get_distribution_from_db(parameters, is_repairable=False, best_distribution=best_distribution) results[location_tag]["cmDisType"] = "Normal" results[location_tag]["cmDisP1"] = 6 @@ -282,48 +298,70 @@ def get_asset_batch(location_tags: List[str], nr_location_tags: List[str], results[location_tag]["relDisType"] = distribution results[location_tag]["relDisP1"] = reldisp1 results[location_tag]["relDisP2"] = reldisp2 - results[location_tag]["parameters"] = item.get("parameters", {}) + results[location_tag]["parameters"] = parameters except Exception as e: - raise Exception(f"Error processing item {location_tag}: {e}") - return results - + raise Exception(f"Error processing non-repairable item {location_tag}: {e}") - except Exception as e: - raise HTTPException( - status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail="Failed to fetch asset batch data " + str(e) - ) + return results -def get_distribution(item): - name = item.get("distribution", None) - # Map distribution names to the expected format - if name == "Weibull-2P": - beta = item["parameters"].get("beta", 0) - alpha = item["parameters"].get("alpha", 0) - return "Weibull2", beta, alpha - elif name == "Weibull-3P": - beta = item["parameters"].get("beta", 0) - alpha = item["parameters"].get("alpha", 0) - return "Weibull3", beta, alpha - elif name == "Exponential-2P": - lambda_ = item["parameters"].get("Lambda", 0) - gamma = item["parameters"].get("gamma", 0) - return "Exponential2", lambda_, gamma - elif name == "NHPP": - beta = item["parameters"].get("beta", 0) - eta = item["parameters"].get("eta", 0) - return "NHPPTTFF", beta, eta - elif name == "Lognormal": - mu = item["parameters"].get("mu", 0) - sigma = item["parameters"].get("sigma", 0) - return "Lognormal", mu, sigma - elif name == "Normal": - mu = item["parameters"].get("mu", 0) - sigma = item["parameters"].get("sigma", 0) - return "Normal", mu, 1000 +def get_distribution_from_db(parameters: dict, is_repairable: bool = True, best_distribution: str = None): + """ + Map distribution parameters from database to the expected format. + + Args: + parameters: Parameters JSON from database + is_repairable: Whether this is repairable equipment + best_distribution: Best distribution name for non-repairable + + Returns: + tuple: (distribution_name, param1, param2) + """ + if is_repairable: + # For repairable equipment, method is in parameters["method"] + method = parameters.get("method", "NHPP") + + if method == "NHPP": + beta = parameters.get("beta", 1.0) + eta = parameters.get("eta", 100000) + return "NHPPTTFF", beta, eta + elif method == "Weibull-2P": + beta = parameters.get("beta", 1.0) + eta = parameters.get("eta", 100000) + return "Weibull2", beta, eta + elif method == "Weibull-3P": + beta = parameters.get("beta", 1.0) + eta = parameters.get("eta", 100000) + return "Weibull3", beta, eta + elif method == "Exponential-2P": + lambda_val = parameters.get("lambda_value", 0.00001) + gamma = parameters.get("gamma", 0) + return "Exponential2", lambda_val, gamma + else: + # Default to NHPP + return "NHPPTTFF", 1.0, 100000 else: - return "NHPPTTFF", 1, 100000 + # For non-repairable equipment, use best_distribution column + distribution = best_distribution or "Normal" + + if distribution == "Lognormal": + mu = parameters.get("mu", 0) + sigma = parameters.get("sigma", 1) + return "Lognormal", mu, sigma + elif distribution == "Normal": + mu = parameters.get("mu", 0) + sigma = parameters.get("sigma", 1) + return "Normal", mu, sigma + elif distribution == "Weibull": + # Assuming Weibull parameters are available + beta = parameters.get("beta", 1.0) + eta = parameters.get("eta", 100000) + return "Weibull2", beta, eta + else: + # Default to Normal + mu = parameters.get("mu", 0) + sigma = parameters.get("sigma", 1) + return "Normal", mu, sigma async def update_oh_interval_offset(*, aeros_db_session, overhaul_offset, overhaul_interval, project_name): query = text(""" @@ -404,7 +442,7 @@ async def update_equipment_for_simulation( reliability_data = get_asset_batch( reg_nodes, non_repairable_location_tags, - RELIABILITY_SERVICE_API + db_session ) # Normalize custom input keys once