CIzz22
/
rbd-app
1
0
Fork 1
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: replace external API call with direct database queries for asset batch data retrieval

Browse Source
main
Cizz22 3 months ago
parent 21e1564400
commit f4ebdf2d60
1 changed files with 112 additions and 74 deletions
Show Stats Download Patch File Download Diff File

176
src/aeros_equipment/service.py
Unescape Escape View File

@ -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], def get_asset_batch(location_tags: List[str], nr_location_tags: List[str],
db_session: DbSession,
base_url: str = RELIABILITY_SERVICE_API, base_url: str = RELIABILITY_SERVICE_API,
timeout: int = 600)-> dict: timeout: int = 600)-> dict:
""" """
Get asset batch data using GET request with JSON body. Get asset batch data from database tables.
Args: Args:
location_tags: List of location tag strings location_tags: List of location tag strings for repairable equipment
base_url: Base URL for the API nr_location_tags: List of location tag strings for non-repairable equipment
timeout: Request timeout in seconds db_session: Database session
base_url: Base URL for the API (kept for compatibility)
timeout: Request timeout in seconds (kept for compatibility)
Returns: 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) results = defaultdict(dict)
payload = { # Query repairable equipment data
"location_tags": location_tags if location_tags:
} repairable_query = text("""
SELECT location_tag, parameters
FROM rp_repairable_results
WHERE location_tag = ANY(:location_tags)
""")
headers = { repairable_result = db_session.execute(repairable_query, {"location_tags": location_tags})
'Content-Type': 'application/json'
}
try: for row in repairable_result:
response = requests.get( location_tag = row[0]
url, parameters = json.loads(row[1]) if row[1] else {}
json=payload,
headers=headers,
timeout=timeout
)
response.raise_for_status() # Raises an HTTPError for bad responses try:
data = response.json() distribution, reldisp1, reldisp2 = get_distribution_from_db(parameters, is_repairable=True)
reliabiility_data = data.get("data", [])
for item in reliabiility_data: results[location_tag]["cmDisType"] = "Normal"
location_tag = item.get("location_tag", None) 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}")
if not location_tag: # Query non-repairable equipment data
print("Error processing item" + str(item)) if nr_location_tags:
continue 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: try:
# is_nr = item["distribution"] != "NHPP" distribution, reldisp1, reldisp2 = get_distribution_from_db(parameters, is_repairable=False, best_distribution=best_distribution)
# mtbf = item["mtbf"]
# mttr = item["mttr"]
print(item)
distribution, reldisp1, reldisp2 = get_distribution(item)
results[location_tag]["cmDisType"] = "Normal" results[location_tag]["cmDisType"] = "Normal"
results[location_tag]["cmDisP1"] = 6 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]["relDisType"] = distribution
results[location_tag]["relDisP1"] = reldisp1 results[location_tag]["relDisP1"] = reldisp1
results[location_tag]["relDisP2"] = reldisp2 results[location_tag]["relDisP2"] = reldisp2
results[location_tag]["parameters"] = item.get("parameters", {}) results[location_tag]["parameters"] = parameters
except Exception as e: except Exception as e:
raise Exception(f"Error processing item {location_tag}: {e}") raise Exception(f"Error processing non-repairable item {location_tag}: {e}")
return results
return results
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Failed to fetch asset batch data " + str(e)
)
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
def get_distribution(item): Returns:
name = item.get("distribution", None) tuple: (distribution_name, param1, param2)
# Map distribution names to the expected format """
if name == "Weibull-2P": if is_repairable:
beta = item["parameters"].get("beta", 0) # For repairable equipment, method is in parameters["method"]
alpha = item["parameters"].get("alpha", 0) method = parameters.get("method", "NHPP")
return "Weibull2", beta, alpha
elif name == "Weibull-3P": if method == "NHPP":
beta = item["parameters"].get("beta", 0) beta = parameters.get("beta", 1.0)
alpha = item["parameters"].get("alpha", 0) eta = parameters.get("eta", 100000)
return "Weibull3", beta, alpha return "NHPPTTFF", beta, eta
elif name == "Exponential-2P": elif method == "Weibull-2P":
lambda_ = item["parameters"].get("Lambda", 0) beta = parameters.get("beta", 1.0)
gamma = item["parameters"].get("gamma", 0) eta = parameters.get("eta", 100000)
return "Exponential2", lambda_, gamma return "Weibull2", beta, eta
elif name == "NHPP": elif method == "Weibull-3P":
beta = item["parameters"].get("beta", 0) beta = parameters.get("beta", 1.0)
eta = item["parameters"].get("eta", 0) eta = parameters.get("eta", 100000)
return "NHPPTTFF", beta, eta return "Weibull3", beta, eta
elif name == "Lognormal": elif method == "Exponential-2P":
mu = item["parameters"].get("mu", 0) lambda_val = parameters.get("lambda_value", 0.00001)
sigma = item["parameters"].get("sigma", 0) gamma = parameters.get("gamma", 0)
return "Lognormal", mu, sigma return "Exponential2", lambda_val, gamma
elif name == "Normal": else:
mu = item["parameters"].get("mu", 0) # Default to NHPP
sigma = item["parameters"].get("sigma", 0) return "NHPPTTFF", 1.0, 100000
return "Normal", mu, 1000
else: 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): async def update_oh_interval_offset(*, aeros_db_session, overhaul_offset, overhaul_interval, project_name):
query = text(""" query = text("""
@ -404,7 +442,7 @@ async def update_equipment_for_simulation(
reliability_data = get_asset_batch( reliability_data = get_asset_batch(
reg_nodes, reg_nodes,
non_repairable_location_tags, non_repairable_location_tags,
RELIABILITY_SERVICE_API db_session
) )
# Normalize custom input keys once # Normalize custom input keys once

Write Preview
Loading…
Cancel
Save