""" PHI-Arc Engine PHM — TPE331-25 Turboprop Engine Model Based on: FAA TCDS E2WE Rev.8, Kurzke & Halliwell (2018) Aircraft: Dornier 228 / Honeywell TPE331-25 Series """ import numpy as np from typing import Dict, List from .base_engine import BaseEngine # ==================== PHYSICS CONSTANTS ==================== GAMMA_C = 1.40 CP_C = 1005.0 GAMMA_H = 1.333 CP_H = 1148.0 EXPC = (GAMMA_C - 1) / GAMMA_C EXPT = (GAMMA_H - 1) / GAMMA_H P_SL = 101325.0 T_ISA_SL = 288.15 LAPSE = 0.0065 LHV_JP4 = 42.80e6 LHV_JETA = 43.10e6 OPR_REF = 9.0 ETA_C_REF = 0.82 ETA_T_REF = 0.87 ETA_B_REF = 0.995 ETA_MECH_REF = 0.98 MDOT_REF = 2.2 DP_COMB = 0.05 EGT_MAX_CONT = 535.0 EGT_WARN = 520.0 EGT_TAKEOFF = 571.0 EGT_START = 815.0 class TPE331Engine(BaseEngine): def __init__(self): super().__init__( name="TPE331-25", engine_type="turboprop", ata_chapters={ "72-30": "Compressor Section", "72-50": "Turbine Section", "73-10": "Fuel Distribution", "75-30": "Compressor Bleed Air Control", "72-60": "Accessory Gearbox", "79-20": "Oil Distribution" } ) self._build_fault_library() self.cert_standards = [ "FAA TCDS E2WE Rev.8 — Honeywell TPE331", "EASA CS-E / FAR Part 33 — Engine Certification", "SAE AIR 1168/4 — Gas Turbine Performance Deterioration", "Dornier 228-100/200 AMM, ATA Spec 100" ] def _build_fault_library(self): self.fault_library = [ { "id": "F1", "name": "F1 — Compressor Fouling", "ata": "ATA 72-30", "mechanism": "Dirt/salt deposits on centrifugal impeller blades reduce camber, increasing incidence losses. Compressor OPR and airflow capacity both decline, raising turbine inlet temperature to maintain shaft power.", "leading": "CDP drop (leading) → EGT rise (lagging)", "color": "#c9a227", "levels": [[0.99,0.99,1.00,0.00,0.00],[0.97,0.98,1.00,0.00,0.00],[0.95,0.96,1.00,0.00,0.00]], "actions": [ "Compressor wash procedure (AMM 72-30-00-200-801)", "Compressor borescope inspection (AMM 72-30-00-200-802)", "Engine removal for workshop overhaul (AMM 72-00-00-720-801)" ], "cert_requirements": [ "FAR 33.201 — Fire prevention: compressor deposits must not create fire hazard", "CS-E 510 — Compressor discharge pressure must remain within declared limits", "SAE AIR 1168/4 — Fouling index must not exceed 15% OPR degradation" ] }, { "id": "F2", "name": "F2 — Turbine Blade Erosion", "ata": "ATA 72-50", "mechanism": "Sand/particle ingestion erodes turbine blade tip and trailing edges, increasing tip clearance and reducing aerodynamic efficiency. Turbine extracts less work per unit temperature drop; TIT must rise to maintain shaft power.", "leading": "EGT↑↑ with CDP normal (EGT is leading indicator)", "color": "#d9534f", "levels": [[1.00,1.00,0.99,0.00,0.00],[1.00,1.00,0.97,0.00,0.00],[1.00,1.00,0.95,0.00,0.00]], "actions": [ "Hot section borescope inspection (AMM 72-50-00-200-801)", "Hot section borescope inspection, repeat in 50 cycles (AMM 72-50-00-200-801)", "Turbine module removal and replacement (AMM 72-50-00-720-801)" ], "cert_requirements": [ "FAR 33.27 — Turbine blades must withstand declared erosion tolerance", "CS-E 800 — Turbine cooling flow must remain within design margins", "TCDS E2WE Note 1 — EGT limits: 535°C continuous, 571°C takeoff" ] }, { "id": "F3", "name": "F3 — Fuel Nozzle Fouling", "ata": "ATA 73-10", "mechanism": "Carbon deposits partially block nozzle orifice, distorting the fuel spray cone and increasing droplet size. Combustion completeness drops; more fuel must be injected to achieve the required TIT, with an unsteady EGT signature.", "leading": "FF:EGT ratio increase (FF leads EGT rise)", "color": "#9b59b6", "levels": [[0.99,0.99,1.00,0.00,0.00],[0.98,0.98,1.00,0.00,0.00],[0.97,0.97,1.00,0.00,0.00]], "eta_b_mult": [0.990, 0.975, 0.960], "actions": [ "Fuel nozzle inspection and flow test (AMM 73-10-00-200-801)", "Fuel nozzle removal and cleaning/replacement (AMM 73-10-00-400-801)", "Restrict engine operation; borescope + engine removal if confirmed" ], "cert_requirements": [ "FAR 33.33 — Fuel system must provide declared flow distribution", "CS-E 560 — Combustor pattern factor must remain within 0.25", "SAE ARP 1536 — Fuel nozzle flow tolerance ±4% of nominal" ] }, { "id": "F4", "name": "F4 — Bleed Valve Stuck Open", "ata": "ATA 75-30", "mechanism": "Anti-ice or customer bleed valve jams open, dumping a fraction of compressor delivery air overboard. Net airflow to combustor drops causing a dramatic CDP fall; the propeller governor increases fuel to recover shaft speed, driving N1 and EGT upward.", "leading": "CDP↓↓ + N1↑ simultaneously (unique bleed signature)", "color": "#3498db", "levels": [[1.00,1.00,1.00,0.03,0.00],[1.00,1.00,1.00,0.06,0.00],[1.00,1.00,1.00,0.10,0.00]], "actions": [ "Bleed valve operational test (AMM 75-30-00-040-801)", "Bleed valve removal and replacement (AMM 75-30-00-400-801)", "Ground aircraft; bleed valve replacement (AMM 75-30-00-400-801)" ], "cert_requirements": [ "FAR 33.66 — Bleed air system must not affect declared performance", "CS-E 650 — Bleed valve must close completely at declared conditions", "TCDS E2WE Note 6 — Bleed flow fractions: 5% (normal), 3% (anti-ice)" ] }, { "id": "F5", "name": "F5 — Bearing / Gearbox Wear", "ata": "ATA 72-60 / 79-20", "mechanism": "Reduction-gearbox bearing wear generates metallic debris detectable in oil (SOAP analysis) and raises friction torque. Gearbox mechanical efficiency drops; additional fuel is needed to maintain shaft power output, elevating EGT and FF while CDP and N1 remain largely normal.", "leading": "FF↑↑ with near-normal CDP and N1 (confirm via oil analysis)", "color": "#1abc9c", "levels": [[1.00,1.00,1.00,0.00,0.010],[1.00,1.00,1.00,0.00,0.020],[1.00,1.00,1.00,0.00,0.035]], "actions": [ "Oil SOAP/spectrometric analysis (AMM 79-20-00-200-801)", "Gearbox chip detector inspection (AMM 72-60-00-200-801)", "Engine removal for gearbox bearing replacement (AMM 72-00-00-720-801)" ], "cert_requirements": [ "FAR 33.62 — Oil system must prevent bearing damage propagation", "CS-E 840 — Oil analysis must detect bearing degradation before failure", "MIL-PRF-23699 — Oil contamination limits for turbine engines" ] } ] def isa_conditions(self, alt_m): T = T_ISA_SL - LAPSE * alt_m P = P_SL * (T / T_ISA_SL) ** 5.2561 return T, P def stagnation(self, T_amb, P_amb, M): T_t2 = T_amb * (1 + 0.5 * (GAMMA_C - 1) * M * M) P_t2 = P_amb * (T_t2 / T_amb) ** (GAMMA_C / (GAMMA_C - 1)) return T_t2, P_t2 def compute_state(self, OPR_in, eta_c_in, eta_t_in, mdot_in, eta_b_in, T_t2, P_t2, P_amb, P_shaft_in, f_bleed_in, LHV): T_t3 = T_t2 + T_t2 * (OPR_in ** EXPC - 1) / eta_c_in P_t3 = P_t2 * OPR_in CDP = P_t3 / 1000.0 P_t4 = P_t3 * (1 - DP_COMB) PR_t = P_t4 / P_amb tau_t = 1 - PR_t ** (-EXPT) mdot_net = mdot_in * (1 - f_bleed_in) W_comp = mdot_in * CP_C * (T_t3 - T_t2) W_req = W_comp + P_shaft_in / ETA_MECH_REF TIT = W_req / (mdot_net * CP_H * eta_t_in * tau_t) T_t5 = TIT * (1 - eta_t_in * tau_t) EGT = T_t5 - 273.15 FF = max(mdot_net * CP_H * (TIT - T_t3) / (eta_b_in * LHV), 0.001) N1 = (mdot_in / 2.2) ** 0.5 * (OPR_in / 9.0) ** 0.25 * 100 return {"EGT": EGT, "FF": FF, "N1": N1, "CDP": CDP, "TIT": TIT} def compute_healthy_baseline(self, flight_conditions: Dict) -> Dict[str, float]: alt_m = flight_conditions.get("altitude_m", 3048) M = flight_conditions.get("mach", 0.35) shp_w = flight_conditions.get("shaft_power_w", 500 * 745.7) antiice = flight_conditions.get("antiice", False) fuel = flight_conditions.get("fuel_type", "Jet-A") T, P = self.isa_conditions(alt_m) T_t2, P_t2 = self.stagnation(T, P, M) LHV = LHV_JETA if fuel == "Jet-A" else LHV_JP4 f_bleed_no = 0.05 / 1.05 f_bleed_anti = 0.03 / 1.03 f_bleed = f_bleed_anti if antiice else f_bleed_no result = self.compute_state(OPR_REF, ETA_C_REF, ETA_T_REF, MDOT_REF, ETA_B_REF, T_t2, P_t2, P, shp_w, f_bleed, LHV) result["T_amb"] = T result["P_amb"] = P result["T_t2"] = T_t2 result["P_t2"] = P_t2 return result def compute_fault_signatures(self, baseline: Dict, flight_conditions: Dict) -> List[List[Dict]]: alt_m = flight_conditions.get("altitude_m", 3048) M = flight_conditions.get("mach", 0.35) shp_w = flight_conditions.get("shaft_power_w", 500 * 745.7) antiice = flight_conditions.get("antiice", False) fuel = flight_conditions.get("fuel_type", "Jet-A") T, P = self.isa_conditions(alt_m) T_t2, P_t2 = self.stagnation(T, P, M) LHV = LHV_JETA if fuel == "Jet-A" else LHV_JP4 f_bleed_no = 0.05 / 1.05 f_bleed_anti = 0.03 / 1.03 f_bleed = f_bleed_anti if antiice else f_bleed_no results = [] for f in self.fault_library: lvl_vals = [] for lv in range(3): m = f["levels"][lv] ec = ETA_C_REF * m[0] md = MDOT_REF * m[1] et = ETA_T_REF * m[2] bl = f_bleed + m[3] ml = m[4] eb = ETA_B_REF if "eta_b_mult" in f: eb = ETA_B_REF * f["eta_b_mult"][lv] P_adj = shp_w / (ETA_MECH_REF - ml) s = self.compute_state(OPR_REF * m[1], ec, et, md, eb, T_t2, P_t2, P, P_adj, bl, LHV) lvl_vals.append(s) results.append(lvl_vals) return results def classify_fault(self, measured: Dict, baseline: Dict, fault_sigs: List[List[Dict]]) -> Dict: egt_c = measured.get("EGT", baseline["EGT"]) ff_kgps = measured.get("FF", baseline["FF"]) n1_pct = measured.get("N1", baseline["N1"]) cdp_kpa = measured.get("CDP", baseline["CDP"]) dev_EGT = (egt_c - baseline["EGT"]) / (EGT_MAX_CONT - baseline["EGT"]) dev_FF = (ff_kgps - baseline["FF"]) / (baseline["FF"] * 0.10) dev_N1 = (n1_pct - baseline["N1"]) / (baseline["N1"] * 0.05) dev_CDP = (cdp_kpa - baseline["CDP"]) / (baseline["CDP"] * 0.10) user_sig = np.array([dev_EGT, dev_FF, dev_N1, dev_CDP]) scores = [] for fi, f in enumerate(self.fault_library): best_score = -1e9 best_lvl = 0 for lv in range(3): fv = fault_sigs[fi][lv] dEGT = (fv["EGT"] - baseline["EGT"]) / (EGT_MAX_CONT - baseline["EGT"]) dFF = (fv["FF"] - baseline["FF"]) / (baseline["FF"] * 0.10) dN1 = (fv["N1"] - baseline["N1"]) / (baseline["N1"] * 0.05) dCDP = (fv["CDP"] - baseline["CDP"]) / (baseline["CDP"] * 0.10) sig = np.array([dEGT, dFF, dN1, dCDP]) denom = (np.linalg.norm(sig) * np.linalg.norm(user_sig)) + 1e-9 score = np.dot(sig, user_sig) / denom if score > best_score: best_score = score best_lvl = lv sev = max(0.0, min(1.0, best_score)) scores.append({"fi": fi, "f": f, "sev": sev, "lvl": best_lvl, "sig": user_sig}) scores.sort(key=lambda x: x["sev"], reverse=True) # Determine overall health status magnitude = np.linalg.norm(user_sig) if magnitude < 0.12: status = "HEALTHY" elif scores[0]["sev"] < 0.30: status = "INDETERMINATE" else: status = "FAULT_DETECTED" # Compute thresholds thresh = { "EGT_warn": EGT_WARN, "EGT_maint": EGT_MAX_CONT, "FF_warn": baseline["FF"] * 1.03, "FF_maint": baseline["FF"] * 1.06, "N1_warn": baseline["N1"] * 0.985, "N1_maint": baseline["N1"] * 0.970, "CDP_warn": baseline["CDP"] * 0.970, "CDP_maint": baseline["CDP"] * 0.940 } return { "base": baseline, "faults": fault_sigs, "scores": scores, "user_sig": user_sig, "status": status, "thresholds": thresh, "deviations": {"EGT": dev_EGT, "FF": dev_FF, "N1": dev_N1, "CDP": dev_CDP} } def get_parameter_labels(self) -> List[str]: return ["EGT", "Fuel Flow", "N1", "CDP"] def get_parameter_units(self) -> Dict[str, str]: return {"EGT": "°C", "FF": "kg/s", "N1": "%", "CDP": "kPa"} def get_input_fields(self) -> List[Dict]: return [ {"name": "altitude", "label": "Altitude", "default": 10000, "unit": "ft", "type": "number"}, {"name": "mach", "label": "Mach / Airspeed", "default": 0.35, "unit": "Mach", "type": "number"}, {"name": "shaft_power", "label": "Shaft Power", "default": 500, "unit": "SHP", "type": "number"}, {"name": "antiice", "label": "Anti-Icing", "default": False, "type": "checkbox"}, {"name": "fuel_type", "label": "Fuel Type", "default": "Jet-A", "options": ["Jet-A", "JP-4"], "type": "select"}, {"name": "EGT", "label": "EGT", "default": 520.0, "unit": "°C", "type": "number"}, {"name": "FF", "label": "Fuel Flow", "default": 140.0, "unit": "kg/hr", "type": "number"}, {"name": "N1", "label": "N1 Speed", "default": 99.5, "unit": "%", "type": "number"}, {"name": "CDP", "label": "CDP", "default": 628.0, "unit": "kPa", "type": "number"} ]