""" PHI-Arc Engine PHM Digital Twin Monitoring System Hugging Face Spaces entry point: src/streamlit_app.py """ import subprocess import sys import os # --- AUTO-INSTALL MISSING PACKAGES --- REQUIRED_PACKAGES = { "streamlit": "streamlit>=1.28.0", "numpy": "numpy>=1.24.0", "pandas": "pandas>=2.0.0", "matplotlib": "matplotlib>=3.7.0", "plotly": "plotly>=5.15.0", "fpdf": "fpdf2>=2.7.0", "docx": "python-docx>=0.8.11", "PIL": "Pillow>=10.0.0", } def ensure_packages(): missing = [] for module, package in REQUIRED_PACKAGES.items(): try: __import__(module) except ImportError: missing.append(package) if missing: print(f"[PHI-Arc PHM] Installing missing packages: {missing}") subprocess.check_call([sys.executable, "-m", "pip", "install", *missing]) print("[PHI-Arc PHM] Packages installed successfully.") ensure_packages() # --- END AUTO-INSTALLER --- import streamlit as st import numpy as np import pandas as pd import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import plotly.express as px import plotly.graph_objects as go from plotly.subplots import make_subplots from datetime import datetime import io sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from engines.turboprop import TPE331Engine from engines.turbofan import TurbofanEngine from engines.turbojet import TurbojetEngine from engines.ramjet import RamjetEngine from engines.scramjet import ScramjetEngine from engines.rocket import RocketEngine from reports.pdf_generator import generate_pdf_report from reports.docx_generator import generate_docx_report from reports.csv_exporter import generate_csv_report, generate_trend_csv # ==================== PAGE CONFIG ==================== st.set_page_config( page_title="PHI-Arc Engine PHM | Digital Twin", page_icon="🚀", layout="wide", initial_sidebar_state="collapsed" ) def load_css(): try: css_path = os.path.join(os.path.dirname(__file__), "assets", "style.css") with open(css_path) as f: st.markdown(f"", unsafe_allow_html=True) except Exception: pass load_css() ENGINE_REGISTRY = { "Turboprop (TPE331-25)": {"class": TPE331Engine, "icon": "🛩ī¸", "desc": "Single-shaft turboprop"}, "Turbofan (CFM-LEAP Class)": {"class": TurbofanEngine, "icon": "✈ī¸", "desc": "High-bypass turbofan"}, "Turbojet (J85-GE Class)": {"class": TurbojetEngine, "icon": "🚀", "desc": "Military turbojet"}, "Ramjet (RJ43 Class)": {"class": RamjetEngine, "icon": "⚡", "desc": "Supersonic ramjet"}, "Scramjet (X-51A Class)": {"class": ScramjetEngine, "icon": "đŸ”Ĩ", "desc": "Hypersonic scramjet"}, "Rocket (RS-25 Class)": {"class": RocketEngine, "icon": "🌌", "desc": "Liquid propellant rocket"} } if 'diagnosis' not in st.session_state: st.session_state.diagnosis = None if 'engine_instance' not in st.session_state: st.session_state.engine_instance = None if 'trend_data' not in st.session_state: st.session_state.trend_data = None if 'selected_engine' not in st.session_state: st.session_state.selected_engine = "Turboprop (TPE331-25)" # ==================== HEADER ==================== st.markdown("""

PHI-Arc Engine PHM

Digital Twin Monitoring System - Prognostics & Health Management for MRO Technicians

Multi-Engine - Physics-Informed - Certification-Linked
""", unsafe_allow_html=True) # ==================== ENGINE SELECTOR ==================== st.markdown('

Select Engine Type

', unsafe_allow_html=True) cols = st.columns(6) engine_keys = list(ENGINE_REGISTRY.keys()) for i, (name, info) in enumerate(ENGINE_REGISTRY.items()): with cols[i]: selected = st.session_state.selected_engine == name border_color = "#38bdf8" if selected else "rgba(148,163,184,0.1)" bg_color = "linear-gradient(145deg, #1e293b, #0c4a6e)" if selected else "linear-gradient(145deg, #1e293b, #0f172a)" st.markdown(f"""
{info['icon']}
{name.split('(')[0].strip()}
{info['desc']}
""", unsafe_allow_html=True) if st.button(f"Select", key=f"btn_{i}", use_container_width=True): st.session_state.selected_engine = name st.session_state.diagnosis = None st.rerun() selected_engine = st.session_state.selected_engine engine_class = ENGINE_REGISTRY[selected_engine]["class"] engine = engine_class() st.session_state.engine_instance = engine # ==================== INPUT PANEL ==================== st.markdown('

Engine Parameters

', unsafe_allow_html=True) input_fields = engine.get_input_fields() flight_conditions = {} measured_params = {} flight_keys = ["altitude", "mach", "shaft_power", "thrust", "antiice", "fuel_type", "chamber_pressure", "of_ratio"] col1, col2 = st.columns(2) with col1: st.subheader("Flight / Operating Conditions") for field in input_fields: if field["name"] in flight_keys: key = field["name"] if field["type"] == "number": val = st.number_input( f"{field['label']} ({field.get('unit', '')})", value=float(field["default"]), key=f"fc_{key}" ) flight_conditions[key] = val elif field["type"] == "select": val = st.selectbox( field["label"], options=field["options"], index=field["options"].index(field["default"]) if field["default"] in field["options"] else 0, key=f"fc_{key}" ) flight_conditions[key] = val elif field["type"] == "checkbox": val = st.checkbox(field["label"], value=field["default"], key=f"fc_{key}") flight_conditions[key] = val with col2: st.subheader("Measured Telemetry (ECU / FDR / Ground Test)") for field in input_fields: if field["name"] not in flight_keys: key = field["name"] val = st.number_input( f"{field['label']} ({field.get('unit', '')})", value=float(field["default"]), key=f"mv_{key}" ) measured_params[key] = val # Unit conversions if "altitude" in flight_conditions: flight_conditions["altitude_m"] = flight_conditions["altitude"] * 0.3048 if "shaft_power" in flight_conditions: flight_conditions["shaft_power_w"] = flight_conditions["shaft_power"] * 745.7 if "thrust" in flight_conditions and selected_engine != "Rocket (RS-25 Class)": flight_conditions["thrust_n"] = flight_conditions["thrust"] * 1000 if "chamber_pressure" in flight_conditions: flight_conditions["chamber_pressure_mpa"] = flight_conditions["chamber_pressure"] if "FF" in measured_params: measured_params["FF"] = measured_params["FF"] / 3600.0 # ==================== DIAGNOSE BUTTON ==================== st.divider() col_btn1, col_btn2, col_btn3 = st.columns([1, 1, 1]) with col_btn1: if st.button("🔍 RUN DIAGNOSIS", type="primary", use_container_width=True): with st.spinner("Computing physics-informed baseline and fault signatures..."): try: baseline = engine.compute_healthy_baseline(flight_conditions) fault_sigs = engine.compute_fault_signatures(baseline, flight_conditions) diagnosis = engine.classify_fault(measured_params, baseline, fault_sigs) st.session_state.diagnosis = diagnosis st.success("Diagnosis complete!") except Exception as e: st.error(f"Diagnosis error: {str(e)}") with col_btn2: uploaded_csv = st.file_uploader("📁 Upload Trend CSV", type=["csv"], key="trend_uploader") if uploaded_csv is not None: try: df = pd.read_csv(uploaded_csv) st.session_state.trend_data = df st.info(f"Loaded {len(df)} cycles") except Exception as e: st.error(f"CSV Error: {e}") with col_btn3: if st.session_state.diagnosis: timestamp = datetime.now().strftime("%Y-%m-%d %H:%M UTC") diag = st.session_state.diagnosis try: pdf_bytes = generate_pdf_report( selected_engine, engine.engine_type, diag, flight_conditions, measured_params, engine.cert_standards, timestamp ) st.download_button( label="📄 PDF Report", data=pdf_bytes, file_name=f"PHI_Arc_PHM_{engine.engine_type}_{datetime.now().strftime('%Y%m%d_%H%M')}.pdf", mime="application/pdf", use_container_width=True ) except Exception as e: st.error(f"PDF generation failed: {e}") try: docx_bytes = generate_docx_report( selected_engine, engine.engine_type, diag, flight_conditions, measured_params, engine.cert_standards, timestamp ) st.download_button( label="📝 DOCX Report", data=docx_bytes, file_name=f"PHI_Arc_PHM_{engine.engine_type}_{datetime.now().strftime('%Y%m%d_%H%M')}.docx", mime="application/vnd.openxmlformats-officedocument.wordprocessingml.document", use_container_width=True ) except Exception as e: st.error(f"DOCX generation failed: {e}") try: csv_bytes = generate_csv_report(diag, selected_engine, timestamp) st.download_button( label="📊 CSV Data", data=csv_bytes, file_name=f"PHI_Arc_PHM_{engine.engine_type}_{datetime.now().strftime('%Y%m%d_%H%M')}.csv", mime="text/csv", use_container_width=True ) except Exception as e: st.error(f"CSV generation failed: {e}") # ==================== RESULTS DISPLAY ==================== if st.session_state.diagnosis: diag = st.session_state.diagnosis baseline = diag["base"] scores = diag["scores"] status = diag["status"] st.divider() # Status Banner status_colors = { "HEALTHY": "#22c55e", "INDETERMINATE": "#eab308", "FAULT_DETECTED": "#ef4444" } status_color = status_colors.get(status, "#64748b") st.markdown(f"""

System Status: {status}

Physics-informed baseline computed for current flight conditions

""", unsafe_allow_html=True) # ==================== DYNAMIC PARAMETER SETUP (EARLY) ==================== param_labels = engine.get_parameter_labels() param_units = engine.get_parameter_units() param_keys = list(param_units.keys()) n_params = min(len(param_keys), 4) # Build generic baseline values dict (no hardcoded egt_h, ff_h, etc.) baseline_vals = {k: baseline.get(k, 0) for k in param_keys} # ==================== EXACT MATLAB-STYLE TELEMETRY OUTPUT ==================== st.markdown('

Computed Healthy Baseline & Telemetry

', unsafe_allow_html=True) # Build telemetry display dynamically telemetry_lines = [] for i, k in enumerate(param_keys[:4]): val = baseline_vals.get(k, 0) unit = param_units.get(k, "") label = param_labels[i] if i < len(param_labels) else k telemetry_lines.append(f"{label} = {val:.2f} {unit}") # Also get measured values measured_lines = [] for i, k in enumerate(param_keys[:4]): val = measured_params.get(k, baseline_vals.get(k, 0)) unit = param_units.get(k, "") label = param_labels[i] if i < len(param_labels) else k measured_lines.append(f"{label} = {val:.2f} {unit}") col_t1, col_t2 = st.columns(2) with col_t1: st.markdown("**Healthy Baseline**") st.code("\n".join(telemetry_lines), language="text") with col_t2: st.markdown("**Measured Telemetry (Your Input)**") st.code("\n".join(measured_lines), language="text") # Parameter Deviations Table (generic, no hardcoded keys) st.markdown("**Parameter Deviations from Healthy Baseline**") dev_data = [] if scores: for fi in range(min(5, len(scores))): f = scores[fi]["f"] fv = diag["faults"][fi][2] if fi < len(diag["faults"]) else {} row = {"Fault": f["name"]} for k in param_keys[:4]: unit = param_units.get(k, "") label = k base_v = baseline_vals.get(k, 0) fault_v = fv.get(k, base_v) row[f"d{label} ({unit})"] = f"{fault_v - base_v:+.2f}" dev_data.append(row) if dev_data: st.dataframe(pd.DataFrame(dev_data), use_container_width=True) # Threshold Summary (generic) st.markdown("**Threshold Summary**") egt_warn = getattr(engine, 'EGT_WARN', 520) egt_maint = getattr(engine, 'EGT_MAX_CONT', 535) st.code(f"""EGT Warning : >= {egt_warn:.0f} deg C | EGT Maintenance : >= {egt_maint:.0f} deg C FF Warning : +3% above baseline | FF Maintenance : +6% N1 Warning : -1.5% below baseline | N1 Maintenance : -3.0% CDP Warning : -3% below baseline | CDP Maintenance : -6%""", language="text") # ==================== METRIC CARDS ==================== st.markdown('

Baseline vs Measured Comparison

', unsafe_allow_html=True) metric_cols = st.columns(n_params) for i in range(n_params): key = param_keys[i] base_val = baseline.get(key, 0) meas_val = measured_params.get(key, base_val) unit = param_units.get(key, "") label = param_labels[i] if i < len(param_labels) else key delta = meas_val - base_val delta_pct = (delta / base_val * 100) if base_val != 0 else 0 color = "#ef4444" if delta > 0 else "#22c55e" with metric_cols[i]: st.markdown(f"""
{label}
{meas_val:.2f} {unit}
{'+' if delta > 0 else ''}{delta:.2f} ({delta_pct:+.1f}%)
Baseline: {base_val:.2f}
""", unsafe_allow_html=True) # ==================== FAULT RANKINGS ==================== st.markdown('

Fault Signature Match Results

', unsafe_allow_html=True) if scores: top = scores[0] tier_names = ["Healthy / Within Limits", "Advisory", "Warning", "Critical / Maintenance Required"] for s in scores[:5]: tier = tier_names[s["lvl"]] if s["lvl"] < 4 else "Unknown" color = s["f"]["color"] st.markdown(f"""
{s['f']['name']} {tier} - {s['sev']*100:.1f}% match
""", unsafe_allow_html=True) st.progress(float(s["sev"]), text=f"{s['f']['id']} - {s['f']['leading']}") # Primary Diagnosis st.markdown('

Primary Diagnosis & Maintenance Decision

', unsafe_allow_html=True) col_d1, col_d2 = st.columns(2) with col_d1: st.markdown(f""" **Fault:** {top['f']['name']} **ATA Reference:** {top['f']['ata']} **Leading Indicator:** {top['f']['leading']} **Confidence:** {top['sev']*100:.1f}% """) st.markdown("**Mechanism:**") st.info(top['f']['mechanism']) with col_d2: st.markdown("**Recommended Maintenance Actions:**") actions = top['f']['actions'] tiers = ["Advisory", "Warning", "Critical"] colors = ["#eab308", "#f97316", "#ef4444"] for i, act in enumerate(actions): st.markdown(f'[{tiers[i]}] {act}', unsafe_allow_html=True) st.markdown("**Certification Requirements:**") cert_reqs = top['f'].get('cert_requirements', engine.cert_standards) for req in cert_reqs: st.markdown(f"- {req}") # ==================== ALL 4 MATLAB-STYLE FIGURES ==================== st.markdown('

Diagnostic Charts

', unsafe_allow_html=True) tab1, tab2, tab3, tab4 = st.tabs([ "Fig 1: Steady-State Comparison", "Fig 2: Trends F1 & F2 (Compressor/Turbine)", "Fig 3: Trends F3, F4 & F5 (Nozzle/Bleed/Bearing)", "Fig 4: Fault Signature Matrix" ]) params = param_keys[:4] labels = param_labels[:4] healthy_vals = [baseline.get(p, 0) for p in params] user_vals = [measured_params.get(p, baseline.get(p, 0)) for p in params] # ---- FIGURE 1: Steady-State Bar Comparison ---- with tab1: try: fig1, axes1 = plt.subplots(2, 2, figsize=(14, 10)) fig1.suptitle( f"{selected_engine.split('(')[0].strip()} Engine Health Monitoring - Steady-State Parameter Comparison\n" f"Level 3 Fault Scenarios vs Healthy Baseline | {selected_engine}\n" f"Source: {getattr(engine, 'cert_standards', ['Certified'])[0]}", fontsize=12, fontweight='bold', color='white', y=0.98 ) x_labels = ['Healthy', 'F1 Comp.\nFouling', 'F2 Turb.\nErosion', 'F3 Fuel\nNozzle', 'F4 Bleed\nValve', 'F5 Bearing\nWear'] n_bars = 6 bar_x = np.arange(n_bars) palette = [ [0.22, 0.65, 0.30], # Healthy - green [0.95, 0.60, 0.07], # F1 - amber [0.90, 0.20, 0.10], # F2 - red [0.60, 0.10, 0.80], # F3 - purple [0.10, 0.45, 0.90], # F4 - blue [0.30, 0.65, 0.75], # F5 - teal ] for idx, (param, label) in enumerate(zip(params, labels)): ax = axes1.flat[idx] vals = [healthy_vals[idx]] for fi in range(min(5, len(diag["faults"]))): vals.append(diag["faults"][fi][2].get(param, healthy_vals[idx])) vals.append(user_vals[idx]) plot_palette = palette + [[0.10, 0.10, 0.10]] plot_labels = x_labels + ['Your\nReading'] bars = ax.bar(range(len(vals)), vals, edgecolor='black', linewidth=0.7) for k, bar in enumerate(bars): bar.set_color(plot_palette[k]) ax.set_xticks(range(len(vals))) ax.set_xticklabels(plot_labels, rotation=25, ha='right', fontsize=8) ax.set_ylabel(f"{label} ({param_units.get(param, '')})", fontsize=9, color='white') ax.set_title(f"{label}", fontweight='bold', color='white', fontsize=10) ax.tick_params(colors='white') ax.xaxis.label.set_color('white') ax.yaxis.label.set_color('white') ax.set_facecolor('#0f172a') for spine in ax.spines.values(): spine.set_color('#334155') ax.grid(axis='y', alpha=0.3, color='#334155') if idx == 0 and hasattr(engine, 'EGT_WARN'): ax.axhline(engine.EGT_WARN, color='gold', linestyle='--', linewidth=1.5, label='Warn') ax.axhline(engine.EGT_MAX_CONT if hasattr(engine, 'EGT_MAX_CONT') else 535, color='red', linestyle='-', linewidth=1.8, label='Limit') ax.legend(fontsize=7, loc='upper left') fig1.patch.set_facecolor('#0f172a') plt.tight_layout(rect=[0, 0, 1, 0.94]) st.pyplot(fig1) st.markdown("""
DIAGNOSTIC KEY: F1: CDP down + EGT up (fouling) | F2: EGT up-up + CDP~0 (erosion) | F3: FF up-up + EGT up (nozzle) | F4: CDP down-down + N1 up (bleed valve) | F5: FF up-up + N1~0 (bearing)
""", unsafe_allow_html=True) except Exception as e: st.error(f"Fig 1 error: {e}") # ---- FIGURE 2: Trends F1 & F2 ---- with tab2: try: if st.session_state.trend_data is not None: df = st.session_state.trend_data st.write(f"Trend data: {len(df)} cycles loaded") fig2, axes2 = plt.subplots(2, 2, figsize=(14, 10)) fig2.suptitle( f"{selected_engine.split('(')[0].strip()} Health Parameter Trends - F1: Compressor Fouling vs F2: Turbine Erosion\n" f"Fault Onset at Cycle 20 - 100 Flight Cycle Simulation", fontsize=12, fontweight='bold', color='white', y=0.98 ) trend_params = params[:4] if len(params) >= 4 else params + ['EGT', 'FF', 'N1', 'CDP'][:4-len(params)] ylabels_t = [f"{labels[i]} ({param_units.get(params[i], '')})" if i < len(labels) else params[i] for i in range(4)] base_val_arr = [baseline_vals.get(p, 0) for p in trend_params] fault_colors = { 0: [0.95, 0.60, 0.07], # F1 - amber 1: [0.90, 0.20, 0.10], # F2 - red } N_cycles = 100 ONSET_CY = 20 t = np.arange(1, N_cycles + 1) for p_idx, (param, ylabel) in enumerate(zip(trend_params, ylabels_t)): ax = axes2.flat[p_idx] ax.set_facecolor('#0f172a') for spine in ax.spines.values(): spine.set_color('#334155') ax.tick_params(colors='white') ax.xaxis.label.set_color('white') ax.yaxis.label.set_color('white') ax.set_title(f"{ylabel} Trend - F1 vs F2", fontweight='bold', color='white', fontsize=10) ax.set_ylabel(ylabel, fontsize=9, color='white') ax.set_xlabel('Flight Cycles', fontsize=9, color='white') ax.axhline(base_val_arr[p_idx], color=[0.22, 0.65, 0.30], linestyle=':', linewidth=1.5, label='Healthy') for fi in [0, 1]: if fi < len(scores): base_v = base_val_arr[p_idx] l3_v = diag["faults"][fi][2].get(param, base_v) if fi < len(diag["faults"]) else base_v y_trend = np.full(N_cycles, base_v) if ONSET_CY <= N_cycles: y_trend[ONSET_CY-1:] = np.linspace(base_v, l3_v, N_cycles - ONSET_CY + 1) np.random.seed(42) noise = np.random.randn(N_cycles) * (base_v * 0.01) y_trend = y_trend + noise ax.plot(t, y_trend, '-', color=fault_colors[fi], linewidth=2.0, label=scores[fi]['f']['name'] if fi < len(scores) else f"F{fi+1}") ax.axvline(ONSET_CY, color='white', linestyle='--', linewidth=1.2, alpha=0.7, label='Fault Onset') if p_idx == 0 and hasattr(engine, 'EGT_WARN'): ax.axhline(engine.EGT_WARN, color='gold', linestyle='--', linewidth=1.5, alpha=0.7, label='Warning') ax.axhline(engine.EGT_MAX_CONT if hasattr(engine, 'EGT_MAX_CONT') else 535, color='red', linestyle='-', linewidth=1.8, alpha=0.7, label='Maint. Limit') ax.fill_between([1, N_cycles], engine.EGT_WARN, engine.EGT_MAX_CONT if hasattr(engine, 'EGT_MAX_CONT') else 535, alpha=0.08, color='yellow') ax.fill_between([1, N_cycles], engine.EGT_MAX_CONT if hasattr(engine, 'EGT_MAX_CONT') else 535, (engine.EGT_MAX_CONT if hasattr(engine, 'EGT_MAX_CONT') else 535) + 80, alpha=0.08, color='red') ax.legend(loc='best', fontsize=7) ax.grid(True, alpha=0.25, color='#334155') ax.set_xlim(1, N_cycles) fig2.patch.set_facecolor('#0f172a') plt.tight_layout(rect=[0, 0, 1, 0.94]) st.pyplot(fig2) except Exception as e: st.error(f"Fig 2 error: {e}") # ---- FIGURE 3: Trends F3, F4, F5 ---- with tab3: try: fig3, axes3 = plt.subplots(2, 2, figsize=(14, 10)) fig3.suptitle( f"{selected_engine.split('(')[0].strip()} Health Parameter Trends - F3: Fuel Nozzle | F4: Bleed Valve | F5: Bearing Wear\n" f"Fault Onset at Cycle 20 - 100 Flight Cycle Simulation", fontsize=12, fontweight='bold', color='white', y=0.98 ) fault_colors_3 = { 2: [0.60, 0.10, 0.80], # F3 - purple 3: [0.10, 0.45, 0.90], # F4 - blue 4: [0.30, 0.65, 0.75], # F5 - teal } for p_idx, (param, ylabel) in enumerate(zip(trend_params, ylabels_t)): ax = axes3.flat[p_idx] ax.set_facecolor('#0f172a') for spine in ax.spines.values(): spine.set_color('#334155') ax.tick_params(colors='white') ax.xaxis.label.set_color('white') ax.yaxis.label.set_color('white') ax.set_title(f"{ylabel} Trend - F3/F4/F5", fontweight='bold', color='white', fontsize=10) ax.set_ylabel(ylabel, fontsize=9, color='white') ax.set_xlabel('Flight Cycles', fontsize=9, color='white') ax.axhline(base_val_arr[p_idx], color=[0.22, 0.65, 0.30], linestyle=':', linewidth=1.5, label='Healthy') for fi in [2, 3, 4]: if fi < len(scores): base_v = base_val_arr[p_idx] l3_v = diag["faults"][fi][2].get(param, base_v) if fi < len(diag["faults"]) else base_v y_trend = np.full(N_cycles, base_v) if ONSET_CY <= N_cycles: y_trend[ONSET_CY-1:] = np.linspace(base_v, l3_v, N_cycles - ONSET_CY + 1) np.random.seed(42 + fi) noise = np.random.randn(N_cycles) * (base_v * 0.01) y_trend = y_trend + noise ax.plot(t, y_trend, '-', color=fault_colors_3[fi], linewidth=2.0, label=scores[fi]['f']['name'] if fi < len(scores) else f"F{fi+1}") ax.axvline(ONSET_CY, color='white', linestyle='--', linewidth=1.2, alpha=0.7, label='Fault Onset') if p_idx == 0 and hasattr(engine, 'EGT_WARN'): ax.axhline(engine.EGT_WARN, color='gold', linestyle='--', linewidth=1.5, alpha=0.7, label='Warning') ax.axhline(engine.EGT_MAX_CONT if hasattr(engine, 'EGT_MAX_CONT') else 535, color='red', linestyle='-', linewidth=1.8, alpha=0.7, label='Maint. Limit') ax.fill_between([1, N_cycles], engine.EGT_WARN, engine.EGT_MAX_CONT if hasattr(engine, 'EGT_MAX_CONT') else 535, alpha=0.08, color='yellow') ax.legend(loc='best', fontsize=7) ax.grid(True, alpha=0.25, color='#334155') ax.set_xlim(1, N_cycles) fig3.patch.set_facecolor('#0f172a') plt.tight_layout(rect=[0, 0, 1, 0.94]) st.pyplot(fig3) except Exception as e: st.error(f"Fig 3 error: {e}") # ---- FIGURE 4: Fault Signature Matrix (Heatmap) ---- with tab4: try: nF = min(5, len(scores)) nP = len(params) matrix = np.zeros((nF + 1, nP)) for fi in range(nF): for pi, p in enumerate(params): base_val = baseline.get(p, 1) if base_val == 0: base_val = 1 if hasattr(engine, 'EGT_REDLINE') and pi == 0: denom = engine.EGT_REDLINE - base_val else: denom = base_val * 0.10 if denom == 0: denom = 1 matrix[fi, pi] = (diag["faults"][fi][2].get(p, base_val) - base_val) / denom for pi, p in enumerate(params): matrix[nF, pi] = diag["user_sig"][pi] if pi < len(diag["user_sig"]) else 0 row_labels = [s['f']['name'] for s in scores[:nF]] + ['YOUR READING'] fig4, ax4 = plt.subplots(figsize=(10, 6)) ax4.set_facecolor('#0f172a') fig4.patch.set_facecolor('#0f172a') im = ax4.imshow(matrix, cmap='RdBu_r', vmin=-1.5, vmax=1.5, aspect='auto') for fi in range(nF + 1): for pi in range(nP): val = matrix[fi, pi] text_color = 'white' if abs(val) > 0.5 else 'black' ax4.text(pi, fi, f"{val:+.2f}", ha='center', va='center', color=text_color, fontweight='bold', fontsize=10) ax4.set_xticks(range(nP)) ax4.set_xticklabels(labels, color='white', fontsize=11) ax4.set_yticks(range(nF + 1)) ax4.set_yticklabels(row_labels, color='white', fontsize=10) ax4.set_xlabel('Health Parameter', fontsize=11, color='white') ax4.set_ylabel('Fault Scenario', fontsize=11, color='white') ax4.set_title( f"{selected_engine.split('(')[0].strip()} Fault Signature Matrix - Normalised Parameter Deviations (Level 3)\n" f"Colour: Blue = Below Baseline (down) | Red = Above Baseline (up)\n" f"Use this matrix to differentiate fault modes from sensor data", fontsize=11, fontweight='bold', color='white', pad=20 ) cbar = plt.colorbar(im, ax=ax4, label='Normalised Deviation from Healthy Baseline') cbar.ax.yaxis.label.set_color('white') cbar.ax.tick_params(colors='white') ax4.set_xticks(np.arange(nP) - 0.5, minor=True) ax4.set_yticks(np.arange(nF + 1) - 0.5, minor=True) ax4.grid(which='minor', color='#334155', linestyle='-', linewidth=0.5) plt.tight_layout() st.pyplot(fig4) st.markdown("""
DIAGNOSTIC KEY: F1: CDP down + EGT up (fouling) | F2: EGT up-up + CDP~0 (erosion) | F3: FF up-up + EGT up (nozzle) | F4: CDP down-down + N1 up (bleed valve) | F5: FF up-up + N1~0 (bearing)
""", unsafe_allow_html=True) except Exception as e: st.error(f"Fig 4 error: {e}") # ==================== FOOTER ==================== st.markdown("""
PHI-Arc Engine PHM Digital Twin - Physics-Informed Prognostics for Turbofan, Turbojet, Turboprop, Ramjet, Scramjet & Rocket.
Built for MRO technicians and engine health monitoring professionals.
Š 2024 PHI-Arc Systems. All rights reserved.
""", unsafe_allow_html=True)