"""
flightgear_phm_dashboard.py
============================
FlightGear LPT/HPT Live PHM Dashboard
Works with both B787-8 (GEnx-1B) and A320neo (CFM LEAP-1A)
KEY FIXES in this version:
1. Cold/dark state is NOW VISIBLE on the dashboard (big amber banner)
instead of silently blocking in a background thread.
2. Raw packet values shown in real-time even before engine start —
so you can confirm data is arriving.
3. Aircraft selector at the top (B787 vs A320neo) adjusts remapping
ranges automatically — no code edits needed.
4. Fault tokens properly removed from inference (were causing CRITICAL).
5. Self-calibrating health baseline (50-frame window after engine start).
6. Port-in-use check is race-condition-free.
7. FAULT INJECTOR & CSV LOGGER ADDED.
AGTF30 COMPATIBILITY NOTE:
AGTF30 is a generic high-bypass turbofan research model. It is NOT
aircraft-specific. Both GEnx-1B (B787) and CFM LEAP-1A (A320neo) are
high-bypass turbofans with the same thermodynamic architecture. The
model's predictions are valid as relative health indicators for either
aircraft — only the physical operating ranges in FG_REF differ.
LAUNCH ORDER:
1. python flightgear_phm_dashboard.py ← start first
2. Copy and run the fgfs command printed in terminal
3. In FlightGear: Equipment → Autostart
(or: Engines panel → set throttle → wait for N2 > 20%)
Author: Mohammed Bello Sani
"""
import sys, os, socket, threading, time, csv, queue # <--- ADDED queue
from collections import deque
import numpy as np
import torch
import tkinter as tk
from tkinter import font as tkfont
import matplotlib
matplotlib.use('TkAgg')
import matplotlib.pyplot as plt
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import math
sys.path.insert(0, r'C:\Users\User\Desktop\Steph')
from cmapss_model import CNNBiLSTMAttention
# ─────────────────────────────────────────────────────────
# PATHS — edit these if yours differ
# ─────────────────────────────────────────────────────────
BASE = r'C:\Users\User\Desktop\Steph\archive\AGTF30\AGTF30'
MODEL_PT = os.path.join(BASE, 'lpt_hpt_blade_model.pt')
SCALER_F = os.path.join(BASE, 'lpt_hpt_scaler_params.npz')
FG_ROOT = r'C:\Users\User\FlightGear\Downloads\fgdata_2024_1'
FG_AIRCRAFT_DIR = (r'C:\Users\User\FlightGear\Downloads\Aircraft'
r'\org.flightgear.fgaddon.stable_2024\Aircraft')
UDP_HOST = '127.0.0.1'
UDP_PORT = 5500
FG_HZ = 10
WINDOW_SIZE = 30
RUL_CAP = 125.0
MC_SAMPLES = 50
N_CAL = 50 # frames to collect for baseline calibration
DEVICE = torch.device('cpu')
# Raw FG N1% threshold to consider engine running (before any remapping)
N1_RUN_MIN = 15.0
# ─────────────────────────────────────────────────────────
# AIRCRAFT PROFILES
# FG_REF: [fg_raw_min, fg_raw_max, agtf30_dst_min, agtf30_dst_max]
# Operating ranges differ between GEnx-1B and LEAP-1A but the
# AGTF30 model destination range is always the same.
# ─────────────────────────────────────────────────────────
AIRCRAFT_PROFILES = {
'B787-8 (GEnx-1B)': {
'aircraft_flag' : '787-8',
'protocol_name' : 'b787_protocol',
'protocol_file' : os.path.join(FG_ROOT, 'Protocol', 'b787_protocol.xml'),
'airport' : 'EGLL',
'fg_ref': {
# EGT in B787/GEnx-1B can reach ~1700 degF at full power
'T45' : [300, 1700, 800, 1400],
'Pt45' : [50, 100, 20, 60 ],
'T25' : [300, 1700, 500, 1100],
'Pt25' : [20, 100, 15, 50 ],
'T3' : [300, 1700, 900, 1600],
'Pt3' : [0, 90000, 200, 800 ],
'Ps3' : [0, 90000, 150, 650 ],
'N2' : [50, 100, 8000, 16000],
'N1' : [20, 100, 2000, 6000 ],
'N3' : [20, 100, 1500, 4500 ],
'Fnet' : [0, 90000, 0, 30000],
},
},
'A320neo (CFM LEAP-1A)': {
'aircraft_flag' : 'A320neo',
'protocol_name' : 'a320neo_protocol',
'protocol_file' : os.path.join(FG_ROOT, 'Protocol', 'a320neo_protocol.xml'),
'airport' : 'EDDM', # Munich — typical A320neo test airport
'fg_ref': {
# CFM LEAP-1A: slightly lower EGT ceiling than GEnx (~1600 degF max)
'T45' : [300, 1600, 800, 1400],
'Pt45' : [50, 100, 20, 60 ],
'T25' : [300, 1600, 500, 1100],
'Pt25' : [20, 100, 15, 50 ],
'T3' : [300, 1600, 900, 1600],
'Pt3' : [0, 60000, 200, 800 ],
'Ps3' : [0, 60000, 150, 650 ],
'N2' : [50, 100, 8000, 16000],
'N1' : [20, 100, 2000, 6000 ],
'N3' : [20, 100, 1500, 4500 ],
'Fnet' : [0, 60000, 0, 30000],
},
},
}
SENSOR_NAMES = ['T45','Pt45','T25','Pt25','T3','Pt3','Ps3','N2','N1','N3','Fnet']
N_SENSORS = 11
# Advisory thresholds — health_pct = current_rul / baseline_rul * 100
ADV_MAP = {
'NOMINAL' : ('#10B981', '#072D15',
'✓ Engine health nominal — all parameters within baseline limits'),
'ADVISORY' : ('#F59E0B', '#2D2207',
'⚠ Early degradation signal — monitor closely, schedule inspection'),
'URGENT' : ('#F97316', '#2D1607',
'⚡ Significant degradation trend — maintenance required after landing'),
'CRITICAL' : ('#EF4444', '#2D0707',
'🚨 Severe degradation vs baseline — REDUCE THRUST, NOTIFY ATC'),
'COLD' : ('#6B7280', '#0C1220',
'❄ Engine cold / dark — start engines then apply throttle'),
'CALIBRATE': ('#38BDF8', '#0C1F2D',
'⏳ Calibrating baseline health — hold steady throttle...'),
}
def get_advisory(health_pct):
if health_pct <= 50: return 'CRITICAL'
elif health_pct <= 70: return 'URGENT'
elif health_pct <= 85: return 'ADVISORY'
else: return 'NOMINAL'
# ─────────────────────────────────────────────────────────
# MODEL + SCALER
# ─────────────────────────────────────────────────────────
def load_model():
print(f'[MODEL] Loading: {MODEL_PT}')
ckpt = torch.load(MODEL_PT, map_location=DEVICE, weights_only=False)
hp = ckpt['hp']
model = CNNBiLSTMAttention(hp).to(DEVICE)
model.load_state_dict(ckpt['model_state'])
n_feat = hp.get('n_features', N_SENSORS)
val_r = float(ckpt.get('val_rmse', -1))
print(f'[MODEL] n_features={n_feat} val_rmse={val_r:.3f}')
return model, val_r, n_feat
def load_scaler():
print(f'[SCALER] Loading: {SCALER_F}')
d = np.load(SCALER_F, allow_pickle=True)
return d['data_min'], d['data_max']
def remap(val, s0, s1, d0, d1):
frac = (val - s0) / max(s1 - s0, 1e-9)
return float(np.clip(d0 + frac * (d1 - d0), d0, d1))
def fg_to_agtf30(raw_11, fg_ref):
out = np.zeros(N_SENSORS, dtype=np.float32)
for i, name in enumerate(SENSOR_NAMES):
s0, s1, d0, d1 = fg_ref[name]
out[i] = remap(raw_11[i], s0, s1, d0, d1)
return out
def predict_rul(model, window, d_min, d_max, n_feat):
"""
Pure inference — no fault tokens injected.
Fault tokens were causing permanent CRITICAL in previous versions.
If model expects >11 features, neutral zeros are padded.
"""
sc = np.clip((window - d_min) / (d_max - d_min + 1e-9), 0, 1)
if n_feat > N_SENSORS:
pad = np.zeros((WINDOW_SIZE, n_feat - N_SENSORS), dtype=np.float32)
sc = np.concatenate([sc, pad], axis=1)
model.train()
t = torch.tensor(sc[np.newaxis], dtype=torch.float32)
preds = []
with torch.no_grad():
for _ in range(MC_SAMPLES):
preds.append(model(t).item())
p = np.array(preds)
mean = float(np.clip(p.mean(), 0, RUL_CAP))
std = float(p.std())
lo = float(np.clip(mean - 1.645*std, 0, RUL_CAP))
hi = float(np.clip(mean + 1.645*std, 0, RUL_CAP))
return mean, std, lo, hi
# ─────────────────────────────────────────────────────────
# UDP RECEIVER
# ─────────────────────────────────────────────────────────
class FGReceiver(threading.Thread):
def __init__(self, host, port, callback):
super().__init__(daemon=True)
self.host = host
self.port = port
self.callback = callback
self.running = True
self.pkt_count = 0
self.fault_injected = False # <-- ADDED FOR FAULT INJECTOR
def run(self):
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((self.host, self.port))
sock.settimeout(1.0)
print(f'[UDP] Listening on {self.host}:{self.port}')
warn_count = 0
while self.running:
try:
data, addr = sock.recvfrom(4096)
line = data.decode('ascii', errors='ignore').strip()
if 'comma' in line.lower():
warn_count += 1
if warn_count <= 3:
print('[WARN] FG sent literal "comma" — XML has wrong separator tag.')
print(' Check , in your XML.')
continue
warn_count = 0
parts = [p.strip() for p in line.split(',')]
if self.pkt_count < 3:
self.pkt_count += 1
print(f'[PKT #{self.pkt_count}] {len(parts)} fields | {line[:80]}')
for i, (p, name) in enumerate(
zip(parts[:11], SENSOR_NAMES)):
print(f' [{i:02d}] {name:<6} raw={p}')
if len(parts) < 11:
continue
try:
raw = [float(p) for p in parts[:11]]
except ValueError:
continue
# --- FAULT INJECTOR ADDITION START ---
self.pkt_count += 1
if self.pkt_count > 1800:
if not self.fault_injected:
print("\n" + "!"*60)
print("🚨 INJECTING SIMULATED HPT FAULT (EGT SPIKE, N2 DROP) 🚨")
print("!"*60 + "\n")
self.fault_injected = True
# Apply the physics-based fault signature
# HPT Degradation: EGT (raw[0]) spikes up, Core Speed N2 (raw[7]) drops
raw[0] *= 1.08 # 8% spike in EGT
raw[7] *= 0.96 # 4% drop in N2 spool speed
# --- FAULT INJECTOR ADDITION END ---
ctx = {}
if len(parts) >= 15:
try:
ctx = {
'altitude': float(parts[11]),
'airspeed': float(parts[12]),
'heading': float(parts[13]),
'elapsed': float(parts[14]),
}
except ValueError:
pass
self.callback(raw, ctx)
except socket.timeout:
pass
except OSError as e:
if self.running:
print(f'[UDP] OSError: {e}')
break
except Exception as e:
print(f'[UDP] error: {e}')
sock.close()
print('[UDP] Socket closed.')
# ─────────────────────────────────────────────────────────
# TURBINE VISUALISER
# ─────────────────────────────────────────────────────────
class TurbineViz:
def __init__(self, parent, label, color, size=140):
self.color = color
self.angle = 0.0
self.n_pct = 0.0
self.size = size
self.cx = self.cy = size // 2
self.r = size // 2 - 6
self.n_blades = 12
frm = tk.Frame(parent, bg='#0A0E1A')
frm.pack(side='left', padx=10)
tk.Label(frm, text=label, fg=color, bg='#0A0E1A',
font=tkfont.Font(family='Consolas', size=9,
weight='bold')).pack()
self.cv = tk.Canvas(frm, width=size, height=size,
bg='#0A0E1A', highlightthickness=0)
self.cv.pack()
self.n_lbl = tk.Label(frm, text='0.0%', fg=color, bg='#0A0E1A',
font=tkfont.Font(family='Consolas', size=9))
self.n_lbl.pack()
self.t_lbl = tk.Label(frm, text='T: --', fg='#9CA3AF', bg='#0A0E1A',
font=tkfont.Font(family='Consolas', size=8))
self.t_lbl.pack()
def draw(self, angle):
self.cv.delete('all')
cx, cy, r = self.cx, self.cy, self.r
heat = min(1.0, self.n_pct / 100.0)
self.cv.create_oval(cx-r, cy-r, cx+r, cy+r,
fill='#1F2937', outline=self.color, width=2)
self.cv.create_oval(cx-12, cy-12, cx+12, cy+12,
fill=self.color, outline='')
hc = self._heatcol(heat)
for i in range(self.n_blades):
th = angle + i*(2*math.pi/self.n_blades)
x1 = cx + 13*math.cos(th); y1 = cy + 13*math.sin(th)
x2 = cx + (r-4)*math.cos(th); y2 = cy + (r-4)*math.sin(th)
dx = 4*math.sin(th); dy = -4*math.cos(th)
self.cv.create_polygon(
x1+dx, y1+dy, x2+dx/3, y2+dy/3,
x2-dx/3, y2-dy/3, x1-dx, y1-dy,
fill=hc, outline='', smooth=True)
if heat > 0.6:
for a in range(1, int((heat-0.6)/0.4*3)+2):
self.cv.create_oval(cx-r-a*2, cy-r-a*2,
cx+r+a*2, cy+r+a*2,
outline='#FF4500', width=1, dash=(3,5))
def _heatcol(self, f):
if f < 0.5:
rv = int(100 + f*2*155)
return f'#{rv:02X}8080'
g = int(255 - (f-0.5)*2*200)
return f'#FF{max(0,g):02X}00'
def update(self, n_pct, temp=None):
self.n_pct = n_pct
self.n_lbl.configure(text=f'{n_pct:.1f}%')
if temp is not None:
tc = ('#EF4444' if temp>1500 else
'#F59E0B' if temp>1200 else '#10B981')
self.t_lbl.configure(text=f'T: {temp:.0f}', fg=tc)
def spin_step(self):
self.angle += self.n_pct / 100.0 * 0.25
self.draw(self.angle)
# ─────────────────────────────────────────────────────────
# MAIN DASHBOARD
# ─────────────────────────────────────────────────────────
class PHMDashboard(tk.Tk):
def __init__(self, model, val_rmse, d_min, d_max, n_feat):
super().__init__()
self.model = model
self.val_rmse = val_rmse
self.d_min = d_min
self.d_max = d_max
self.n_feat = n_feat
self.running = True
# Aircraft selection — set before UDP starts
self.aircraft_var = tk.StringVar(value='A320neo (CFM LEAP-1A)')
self.fg_ref = AIRCRAFT_PROFILES[
self.aircraft_var.get()]['fg_ref']
# Data state
self.window_buf = deque(maxlen=WINDOW_SIZE)
self.healthy = None
self.frame_count = 0
self.engine_state = 'COLD' # 'COLD' | 'CALIBRATE' | 'RUNNING'
# Calibration
self.cal_samples = []
self.baseline_rul = None
self.health_pct = 100.0
# Histories
self.h_frame = deque(maxlen=500)
self.h_rul = deque(maxlen=500)
self.h_std = deque(maxlen=500)
self.h_n1 = deque(maxlen=500)
self.h_n2 = deque(maxlen=500)
self.h_t45 = deque(maxlen=500)
self.h_t25 = deque(maxlen=500)
# Raw latest data (written by UDP thread, read by infer thread)
self.latest_agtf30 = None
self.latest_raw = None # raw 11-element FG values
self.latest_raw_n1 = 0.0
self.latest_ctx = {}
self.data_lock = threading.Lock()
self.fill_ci = None
# --- THREAD-SAFE LOGGER QUEUE ---
self.log_queue = queue.Queue()
self.log_thread = threading.Thread(target=self._file_writer_thread, daemon=True)
self.log_thread.start()
self._build_ui()
self._start_udp()
self._start_inference()
self._animate()
# ─── AIRCRAFT SWITCHER ───────────────────────────────
def _on_aircraft_change(self, *args):
profile = AIRCRAFT_PROFILES[self.aircraft_var.get()]
self.fg_ref = profile['fg_ref']
# Reset calibration
self.cal_samples = []
self.baseline_rul = None
self.health_pct = 100.0
self.engine_state = 'COLD'
self.window_buf.clear()
self.healthy = None
self._print_launch_command()
self.status_lbl.configure(
text=f'Aircraft changed to {self.aircraft_var.get()} — '
f'relaunch FlightGear with new command.')
def _print_launch_command(self):
profile = AIRCRAFT_PROFILES[self.aircraft_var.get()]
print(f'\n[LAUNCH] Copy this command to launch FlightGear:')
print(f' fgfs.exe --aircraft={profile["aircraft_flag"]}')
print(f' --airport={profile["airport"]}')
print(f' --fg-root="{FG_ROOT}"')
print(f' --fg-aircraft="{FG_AIRCRAFT_DIR}"')
print(f' --generic=socket,out,{FG_HZ},'
f'{UDP_HOST},{UDP_PORT},udp,{profile["protocol_name"]}')
print(f' Protocol XML: {profile["protocol_file"]}\n')
# ─── UI BUILD ─────────────────────────────────────────
def _build_ui(self):
self.title('FlightGear A320neo/B787 — LPT/HPT PHM Dashboard')
self.configure(bg='#030712')
self.geometry('1520x960')
SF = tkfont.Font(family='Consolas', size=9)
SF7 = tkfont.Font(family='Consolas', size=7)
SF8 = tkfont.Font(family='Consolas', size=8)
BF = tkfont.Font(family='Consolas', size=10, weight='bold')
HF = tkfont.Font(family='Consolas', size=13, weight='bold')
GF = tkfont.Font(family='Consolas', size=30, weight='bold')
# ── HEADER ─────────────────────────────────────────
hdr = tk.Frame(self, bg='#0C1220', pady=8)
hdr.pack(fill='x')
tk.Label(hdr,
text='LPT / HPT ◆ LIVE TURBINE HEALTH MONITOR ◆ AGTF30 PHM',
font=HF, bg='#0C1220', fg='#38BDF8').pack(side='left', padx=14)
# Aircraft selector on the right of header
sel_frm = tk.Frame(hdr, bg='#0C1220')
sel_frm.pack(side='right', padx=14)
tk.Label(sel_frm, text='Aircraft:', font=SF8, bg='#0C1220',
fg='#9CA3AF').pack(side='left')
ac_menu = tk.OptionMenu(sel_frm, self.aircraft_var,
*AIRCRAFT_PROFILES.keys())
ac_menu.configure(bg='#1F2937', fg='#E5E7EB', font=SF8,
activebackground='#374151',
highlightthickness=0, bd=0)
ac_menu['menu'].configure(bg='#1F2937', fg='#E5E7EB', font=SF8)
ac_menu.pack(side='left', padx=6)
self.aircraft_var.trace('w', self._on_aircraft_change)
# ── ENGINE STATE BANNER ────────────────────────────
# This is the BIG VISIBLE state that was missing before
self.state_banner = tk.Frame(self, bg='#0C1220', pady=10)
self.state_banner.pack(fill='x', padx=8, pady=(4,0))
self.state_icon = tk.Label(self.state_banner,
text='❄ ENGINE COLD / DARK',
font=tkfont.Font(family='Consolas',
size=15, weight='bold'),
bg='#0C1220', fg='#6B7280')
self.state_icon.pack(side='left', padx=14)
self.state_msg = tk.Label(self.state_banner,
text='Start engines: Equipment → Autostart',
font=SF8, bg='#0C1220', fg='#9CA3AF')
self.state_msg.pack(side='left')
self.cal_cv = tk.Canvas(self.state_banner, width=200, height=14,
bg='#1F2937', highlightthickness=0)
self.cal_cv.pack(side='right', padx=14)
# ── TURBINE SPINNERS + RAW DATA ────────────────────
spin_row = tk.Frame(self, bg='#030712', pady=6)
spin_row.pack(fill='x')
spin_inner = tk.Frame(spin_row, bg='#030712')
spin_inner.pack()
self.lpt_viz = TurbineViz(spin_inner, 'LPT N1', '#4ECDC4', 130)
self.hpt_viz = TurbineViz(spin_inner, 'HPT N2', '#FF6B35', 130)
# Live raw packet panel (always visible)
raw_frm = tk.Frame(spin_inner, bg='#0C1220', padx=12, pady=6)
raw_frm.pack(side='left', padx=14, fill='y')
tk.Label(raw_frm, text='LIVE RAW TELEMETRY',
font=SF7, bg='#0C1220', fg='#374151').grid(
row=0, column=0, columnspan=2, sticky='w')
self.raw_lbls = {}
show_raw = ['N1','N2','T45 (EGT)','Fnet','Alt','IAS']
for i, name in enumerate(show_raw):
tk.Label(raw_frm, text=f'{name}:', font=SF7, bg='#0C1220',
fg='#6B7280', width=10, anchor='w').grid(
row=i+1, column=0, sticky='w')
lbl = tk.Label(raw_frm, text='---', font=SF7, bg='#0C1220',
fg='#38BDF8', width=12, anchor='e')
lbl.grid(row=i+1, column=1, sticky='e')
self.raw_lbls[name] = lbl
# ── MAIN ROW ───────────────────────────────────────
row = tk.Frame(self, bg='#030712')
row.pack(fill='both', expand=True, padx=8, pady=(4,0))
left = tk.Frame(row, bg='#030712', width=440)
left.pack(side='left', fill='y', padx=(0,6))
left.pack_propagate(False)
self._build_left(left, GF, BF, SF, SF7, SF8)
right = tk.Frame(row, bg='#030712')
right.pack(side='left', fill='both', expand=True)
self._build_plots(right)
self._build_full_table(right, SF7)
# ── STATUS BAR ─────────────────────────────────────
self.status_lbl = tk.Label(self,
text='Dashboard ready — waiting for FlightGear.',
font=SF7, bg='#0C1220', fg='#6B7280',
anchor='w')
self.status_lbl.pack(fill='x', side='bottom', padx=8, pady=2)
def _build_left(self, parent, GF, BF, SF, SF7, SF8):
# Advisory panel
self.adv_frame = tk.Frame(parent, bg='#072D15', pady=14, padx=8)
self.adv_frame.pack(fill='x', pady=(0,6))
self.adv_lbl = tk.Label(self.adv_frame,
text='◉ NOMINAL',
font=tkfont.Font(family='Consolas',
size=14, weight='bold'),
bg='#072D15', fg='#10B981')
self.adv_lbl.pack()
self.adv_msg = tk.Label(self.adv_frame,
text=ADV_MAP['NOMINAL'][2],
font=SF8, bg='#072D15', fg='#D1FAE5',
wraplength=420)
self.adv_msg.pack(pady=(4,0))
# RUL gauge
rul_p = tk.Frame(parent, bg='#111827', pady=8)
rul_p.pack(fill='x', pady=(0,5))
tk.Label(rul_p, text='PREDICTED RUL (cycles)',
font=SF7, bg='#111827', fg='#6B7280').pack()
self.rul_lbl = tk.Label(rul_p, text='---', font=GF,
bg='#111827', fg='#10B981')
self.rul_lbl.pack()
self.rul_sub = tk.Label(rul_p, text='± -- cycles',
font=SF8, bg='#111827', fg='#6B7280')
self.rul_sub.pack()
self.ci_lbl = tk.Label(rul_p, text='90% CI: [ -- — -- ]',
font=SF8, bg='#111827', fg='#6B7280')
self.ci_lbl.pack()
# Health bar
hb_p = tk.Frame(parent, bg='#111827', pady=6, padx=6)
hb_p.pack(fill='x', pady=(0,5))
tk.Label(hb_p, text='ENGINE HEALTH vs BASELINE',
font=SF7, bg='#111827', fg='#6B7280').pack(anchor='w')
self.health_cv = tk.Canvas(hb_p, width=410, height=26,
bg='#111827', highlightthickness=0)
self.health_cv.pack(pady=2)
self.health_txt = tk.Label(hb_p, text='Calibrating...',
font=SF8, bg='#111827', fg='#6B7280')
self.health_txt.pack()
# Context panel
ctx_p = tk.Frame(parent, bg='#111827', pady=5, padx=10)
ctx_p.pack(fill='x', pady=(0,5))
self.ctx_lbls = {}
for key, label in [('alt','Alt'),('ias','IAS'),
('hdg','Hdg'),('elapsed','Elapsed')]:
r = tk.Frame(ctx_p, bg='#111827')
r.pack(fill='x')
tk.Label(r, text=f'{label}:', font=SF7, bg='#111827',
fg='#6B7280', width=8, anchor='w').pack(side='left')
v = tk.Label(r, text='---', font=SF7, bg='#111827',
fg='#E5E7EB', anchor='w')
v.pack(side='left')
self.ctx_lbls[key] = v
# Key sensor delta table
sd_p = tk.Frame(parent, bg='#111827', padx=8, pady=5)
sd_p.pack(fill='x', pady=(0,4))
tk.Label(sd_p, text='SENSOR DELTAS vs FIRST HEALTHY READING',
font=SF7, bg='#111827', fg='#6B7280').pack(anchor='w')
self.sens_w = {}
for sname in ['T45','T25','N1','N2','Fnet']:
acc = '#FF6B35' if sname in ('T45','N2') else '#4ECDC4'
r = tk.Frame(sd_p, bg='#111827')
r.pack(fill='x', pady=1)
tk.Label(r, text=f'{sname:<5}', font=SF7, bg='#111827',
fg=acc, width=5, anchor='w').pack(side='left')
vl = tk.Label(r, text='-------', font=SF7, bg='#111827',
fg='#E5E7EB', width=11, anchor='e')
vl.pack(side='left')
pl = tk.Label(r, text='------', font=SF7, bg='#111827',
fg='#6B7280', width=10, anchor='e')
pl.pack(side='left')
bar = tk.Canvas(r, width=90, height=10, bg='#111827',
highlightthickness=0)
bar.pack(side='left', padx=4)
self.sens_w[sname] = (vl, pl, bar)
# Footer info
info = tk.Frame(parent, bg='#030712')
info.pack(fill='x', pady=3)
tk.Label(info,
text=f'val-RMSE: {self.val_rmse:.3f} n_feat: {self.n_feat} '
f'MC samples: {MC_SAMPLES} window: {WINDOW_SIZE}',
font=SF7, bg='#030712', fg='#374151').pack(side='left', padx=4)
self.rmse_lbl = tk.Label(info, text='Live σ: --',
font=SF7, bg='#030712', fg='#374151')
self.rmse_lbl.pack(side='left')
def _build_plots(self, parent):
fig, (ax1, ax2, ax3) = plt.subplots(1, 3, figsize=(11, 3.0))
fig.patch.set_facecolor('#030712')
for ax in [ax1, ax2, ax3]:
ax.set_facecolor('#111827')
ax.tick_params(colors='#6B7280', labelsize=6)
for sp in ax.spines.values():
sp.set_color('#374151')
ax1.set_title('Predicted RUL | 90% CI Band',
color='#9CA3AF', fontsize=7)
ax1.set_ylabel('RUL (cycles)', color='#6B7280', fontsize=6)
ax1.set_xlabel('Frame', color='#6B7280', fontsize=6)
self.ln_rul, = ax1.plot([], [], color='#38BDF8', lw=1.5,
label='Predicted RUL')
ax1.set_ylim(0, RUL_CAP + 10)
ax1.legend(fontsize=6, facecolor='#111827', labelcolor='#9CA3AF')
self.ax1 = ax1
ax2.set_title('N1 (LPT) & N2 (HPT) Shaft Speeds (%)',
color='#9CA3AF', fontsize=7)
ax2.set_ylabel('N (%)', color='#6B7280', fontsize=6)
ax2.set_xlabel('Frame', color='#6B7280', fontsize=6)
self.ln_n1, = ax2.plot([], [], color='#4ECDC4', lw=1.2, label='N1 LPT')
self.ln_n2, = ax2.plot([], [], color='#FF6B35', lw=1.2, label='N2 HPT')
ax2.set_ylim(0, 105)
ax2.legend(fontsize=6, facecolor='#111827', labelcolor='#9CA3AF')
self.ax2 = ax2
ax3.set_title('Temperature Trends EGT (T45) & T25 (norm)',
color='#9CA3AF', fontsize=7)
ax3.set_xlabel('Frame', color='#6B7280', fontsize=6)
self.ln_t45, = ax3.plot([], [], color='#FF6B35', lw=1.2, label='T45 HPT')
self.ln_t25, = ax3.plot([], [], color='#4ECDC4', lw=1.2, label='T25 LPT')
ax3.legend(fontsize=6, facecolor='#111827', labelcolor='#9CA3AF')
self.ax3 = ax3
self.fig = fig
self.canvas = FigureCanvasTkAgg(fig, master=parent)
self.canvas.get_tk_widget().pack(fill='x')
def _build_full_table(self, parent, SF7):
frm = tk.Frame(parent, bg='#0C1220')
frm.pack(fill='x', pady=(3,0))
tk.Label(frm,
text=' REMAPPED AGTF30 SENSOR VALUES (FG → turbofan units)',
font=SF7, bg='#0C1220', fg='#374151').grid(
row=0, column=0, columnspan=N_SENSORS, sticky='w')
self.full_lbl = {}
for c, name in enumerate(SENSOR_NAMES):
col = '#FF6B35' if name in ('T45','Pt45','T3','Pt3','N2') \
else '#4ECDC4'
tk.Label(frm, text=name, font=SF7, bg='#0C1220',
fg=col, width=7).grid(row=1, column=c, padx=2)
lbl = tk.Label(frm, text='---', font=SF7, bg='#0C1220',
fg='#9CA3AF', width=8)
lbl.grid(row=2, column=c, padx=2)
self.full_lbl[name] = lbl
# ─── UDP CALLBACK ─────────────────────────────────────
def on_fg_data(self, raw_11, ctx):
agtf30 = fg_to_agtf30(raw_11, self.fg_ref)
with self.data_lock:
self.latest_raw = raw_11
self.latest_agtf30 = agtf30
self.latest_raw_n1 = raw_11[8] # field[8] = N1%
self.latest_ctx = ctx
# ─── INFERENCE THREAD ─────────────────────────────────
def _start_inference(self):
t = threading.Thread(target=self._infer_loop, daemon=True)
t.start()
def _infer_loop(self):
while self.running:
with self.data_lock:
sensors = (self.latest_agtf30.copy()
if self.latest_agtf30 is not None else None)
raw_11 = list(self.latest_raw) if self.latest_raw else None
raw_n1 = self.latest_raw_n1
ctx = dict(self.latest_ctx)
if sensors is None:
time.sleep(0.15)
continue
# Always update raw panel (visible even cold/dark)
self.after(0, lambda r=raw_11, c=ctx:
self._update_raw_panel(r, c))
# ── COLD / DARK ───────────────────────────────
if raw_n1 < N1_RUN_MIN:
self.engine_state = 'COLD'
self.after(0, self._show_cold_state)
time.sleep(0.2)
continue
# ── ENGINE RUNNING — build window ─────────────
if self.healthy is None:
self.healthy = sensors.copy()
self.window_buf.append(sensors)
if len(self.window_buf) < WINDOW_SIZE:
n = len(self.window_buf)
self.engine_state = 'COLD' # still warming up
self.after(0, lambda n=n:
self.state_icon.configure(
text=f'⏳ Building window: {n}/{WINDOW_SIZE}',
fg='#38BDF8'))
time.sleep(0.05)
continue
# ── INFERENCE ────────────────────────────────
arr = np.array(list(self.window_buf), dtype=np.float32)
mean, std, lo, hi = predict_rul(
self.model, arr, self.d_min, self.d_max, self.n_feat)
# ── CALIBRATION ──────────────────────────────
if len(self.cal_samples) < N_CAL:
self.cal_samples.append(mean)
n_done = len(self.cal_samples)
cal_pct = int(n_done / N_CAL * 100)
self.engine_state = 'CALIBRATE'
self.after(0, lambda n=n_done, p=cal_pct, m=mean:
self._show_calibrating(n, p, m))
time.sleep(0.08)
continue
if self.baseline_rul is None:
self.baseline_rul = float(np.median(self.cal_samples))
print(f'[CAL] Baseline RUL = {self.baseline_rul:.1f} cycles')
# ── HEALTH SCORE ─────────────────────────────
health_pct = min(100.0,
(mean / max(1.0, self.baseline_rul)) * 100)
self.engine_state = 'RUNNING'
self.frame_count += 1
self.h_frame.append(self.frame_count)
self.h_rul.append(mean)
self.h_std.append(std)
n1_idx = SENSOR_NAMES.index('N1')
n2_idx = SENSOR_NAMES.index('N2')
t45_idx = SENSOR_NAMES.index('T45')
t25_idx = SENSOR_NAMES.index('T25')
n1_pct = np.clip(sensors[n1_idx] / 6000.0, 0, 1) * 100
n2_pct = np.clip(sensors[n2_idx] / 16000.0, 0, 1) * 100
self.h_n1.append(n1_pct)
self.h_n2.append(n2_pct)
self.h_t45.append(np.clip((sensors[t45_idx]-800)/600.0, 0, 1))
self.h_t25.append(np.clip((sensors[t25_idx]-500)/600.0, 0, 1))
# --- NON-BLOCKING LOGGING (thread-safe queue) ---
self.log_queue.put([self.frame_count, mean, std, health_pct,
n1_pct, n2_pct, raw_11[0] if raw_11 else 0])
self.after(0, lambda
m=mean, s=std, l=lo, h=hi, hp=health_pct,
sens=sensors.copy(), c=ctx,
n1=n1_pct, n2=n2_pct, t45=sensors[t45_idx]:
self.update_ui(m, s, l, h, hp, sens, c, n1, n2, t45))
time.sleep(0.08)
def _file_writer_thread(self):
"""Background thread that writes log data from the queue to CSV."""
with open('flightgear_phm_log.csv', 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['frame', 'rul_mean', 'rul_std', 'health_pct',
'n1_pct', 'n2_pct', 'egt_raw'])
while self.running:
try:
data = self.log_queue.get(timeout=1)
writer.writerow(data)
f.flush()
except queue.Empty:
continue
# ─── ENGINE STATE PANELS ──────────────────────────────
def _show_cold_state(self):
col, bg, msg = ADV_MAP['COLD'][0], ADV_MAP['COLD'][1], ADV_MAP['COLD'][2]
self.state_banner.configure(bg=bg)
self.state_icon.configure(
text='❄ ENGINE COLD / DARK', fg=col, bg=bg)
self.state_msg.configure(text=msg, bg=bg, fg='#9CA3AF')
self.cal_cv.delete('all')
self.adv_frame.configure(bg=bg)
self.adv_lbl.configure(text='❄ COLD', fg=col, bg=bg)
self.adv_msg.configure(text='Engines not started. Use Equipment → Autostart.',
bg=bg)
def _show_calibrating(self, n_done, pct, cur_rul):
col = ADV_MAP['CALIBRATE'][0]
bg = ADV_MAP['CALIBRATE'][1]
self.state_banner.configure(bg=bg)
self.state_icon.configure(
text=f'⏳ CALIBRATING {n_done}/{N_CAL} ({pct}%)',
fg=col, bg=bg)
self.state_msg.configure(
text=f'Hold steady throttle — current RUL estimate: {cur_rul:.1f}',
bg=bg, fg='#93C5FD')
# Progress bar
self.cal_cv.delete('all')
w = 200
f = int(w * pct / 100)
self.cal_cv.configure(bg='#1F2937')
self.cal_cv.create_rectangle(0,0,f,14, fill=col, outline='')
self.adv_frame.configure(bg=bg)
self.adv_lbl.configure(text='⏳ CALIBRATING', fg=col, bg=bg)
self.adv_msg.configure(
text=ADV_MAP['CALIBRATE'][2], bg=bg, fg='#BFDBFE')
def _update_raw_panel(self, raw_11, ctx):
if raw_11 is None:
return
# Map raw values to human-readable panel
n1_raw = raw_11[8] # N1%
n2_raw = raw_11[1] # N2%
egt_raw = raw_11[0] # EGT degF
thr_raw = raw_11[10] # thrust lbf
alt = ctx.get('altitude', -9999)
ias = ctx.get('airspeed', 0)
col_n1 = '#10B981' if n1_raw > N1_RUN_MIN else '#6B7280'
self.raw_lbls['N1'].configure(
text=f'{n1_raw:.2f} %', fg=col_n1)
self.raw_lbls['N2'].configure(
text=f'{n2_raw:.2f} %',
fg='#FF6B35' if n2_raw > 20 else '#6B7280')
self.raw_lbls['T45 (EGT)'].configure(
text=f'{egt_raw:.1f} °F',
fg='#EF4444' if egt_raw > 1200 else
'#F59E0B' if egt_raw > 600 else '#4ECDC4')
self.raw_lbls['Fnet'].configure(
text=f'{thr_raw:.0f} lbf')
# Altitude: -9999 = not valid yet (FG init)
alt_str = f'{alt:.0f} ft' if alt > -100 else 'INIT'
self.raw_lbls['Alt'].configure(text=alt_str)
self.raw_lbls['IAS'].configure(text=f'{ias:.1f} kt')
# ─── MAIN UI UPDATE ───────────────────────────────────
def update_ui(self, mean, std, lo, hi, health_pct,
sensors, ctx, n1_pct, n2_pct, t45_raw):
advisory = get_advisory(health_pct)
adv_col, adv_bg, adv_msg_text = ADV_MAP[advisory]
# State banner — running
self.state_banner.configure(bg='#030712')
self.state_icon.configure(
text='▶ ENGINE RUNNING', fg='#10B981', bg='#030712')
self.state_msg.configure(
text=f'Health: {health_pct:.1f}% | '
f'Baseline: {self.baseline_rul:.1f} cycles',
bg='#030712', fg='#6B7280')
self.cal_cv.delete('all')
# Advisory box
self.adv_frame.configure(bg=adv_bg)
self.adv_lbl.configure(fg=adv_col, bg=adv_bg,
text=f'◉ {advisory}')
self.adv_msg.configure(text=adv_msg_text, bg=adv_bg)
# RUL
self.rul_lbl.configure(text=f'{mean:.0f}', fg=adv_col)
self.rul_sub.configure(text=f'± {std:.1f} cycles')
self.ci_lbl.configure(
text=f'90% CI: [ {lo:.1f} — {hi:.1f} ]')
# Health bar
self.health_cv.delete('all')
w = self.health_cv.winfo_width() or 400
f = max(0, min(w-4, int(health_pct/100.0*(w-4))))
self.health_cv.create_rectangle(2,2,w-2,24,fill='#1F2937',outline='')
self.health_cv.create_rectangle(2,2,2+f, 24,fill=adv_col, outline='')
self.health_cv.create_text(
w//2, 13, fill='#E5E7EB',
font=tkfont.Font(family='Consolas',size=8),
text=f'{health_pct:.1f}% of baseline '
f'({self.baseline_rul:.0f} reference cycles)')
self.health_txt.configure(
text=f'Health: {health_pct:.1f}% | '
f'RUL: {mean:.0f} / {self.baseline_rul:.0f} baseline',
fg=adv_col)
# Context
if ctx:
self.ctx_lbls['alt'].configure(
text=f'{ctx.get("altitude",0):.0f} ft')
self.ctx_lbls['ias'].configure(
text=f'{ctx.get("airspeed",0):.1f} kt')
self.ctx_lbls['hdg'].configure(
text=f'{ctx.get("heading",0):.1f}°')
self.ctx_lbls['elapsed'].configure(
text=f'{ctx.get("elapsed",0):.0f} s')
# Turbine spinners
self.lpt_viz.update(n1_pct, sensors[SENSOR_NAMES.index('T25')])
self.hpt_viz.update(n2_pct, t45_raw)
# Sensor delta rows
for sname, (vl, pl, bar) in self.sens_w.items():
idx = SENSOR_NAMES.index(sname)
val = sensors[idx]
vl.configure(text=f'{val:>10.2f}')
if self.healthy is not None and abs(self.healthy[idx]) > 1e-6:
delta = (val - self.healthy[idx]) / abs(self.healthy[idx]) * 100
arrow = '↑' if delta > 0 else '↓'
dc = ('#EF4444' if abs(delta)>2 else
'#F59E0B' if abs(delta)>0.5 else '#10B981')
pl.configure(text=f'{arrow}{delta:+.3f}%', fg=dc)
bw = bar.winfo_width() or 90
bar.delete('all')
bar.create_rectangle(0,2,bw,8, fill='#1F2937',outline='')
bar.create_rectangle(
0,2,min(bw,int(abs(delta)/3.0*bw)),8,
fill=dc, outline='')
# Full sensor table
for name, lbl in self.full_lbl.items():
idx = SENSOR_NAMES.index(name)
lbl.configure(text=f'{sensors[idx]:.1f}')
# RMSE
if len(self.h_std) > 5:
self.rmse_lbl.configure(
text=f'Live σ: {np.mean(list(self.h_std)):.2f}')
# Plots
frames = list(self.h_frame)
if len(frames) < 2:
return
self.ln_rul.set_data(frames, list(self.h_rul))
if self.fill_ci:
try: self.fill_ci.remove()
except: pass
self.fill_ci = None
p = np.array(list(self.h_rul))
s = np.array(list(self.h_std))
lo_b = np.clip(p - 1.645*s, 0, RUL_CAP)
hi_b = np.clip(p + 1.645*s, 0, RUL_CAP)
self.fill_ci = self.ax1.fill_between(
frames, lo_b, hi_b, alpha=0.15, color='#38BDF8')
self.ln_n1.set_data(frames, list(self.h_n1))
self.ln_n2.set_data(frames, list(self.h_n2))
self.ln_t45.set_data(frames, list(self.h_t45))
self.ln_t25.set_data(frames, list(self.h_t25))
for ax in [self.ax1, self.ax2, self.ax3]:
ax.set_xlim(max(0, frames[-1]-200), frames[-1]+5)
ax.relim()
ax.autoscale_view(scalex=False)
self.canvas.draw_idle()
title_str = ('🚨 CRITICAL ENGINE WARNING | B787 PHM'
if advisory == 'CRITICAL'
else f'FlightGear {self.aircraft_var.get()} — PHM Dashboard')
self.title(title_str)
self.status_lbl.configure(
text=f'Frame {self.frame_count} | '
f'Win {len(self.window_buf)}/{WINDOW_SIZE} | '
f'N1={n1_pct:.1f}% N2={n2_pct:.1f}% | '
f'RUL={mean:.0f} Health={health_pct:.1f}% [{advisory}] | '
f'σ={std:.1f}')
def _animate(self):
self.lpt_viz.spin_step()
self.hpt_viz.spin_step()
self.after(50, self._animate)
def _start_udp(self):
self.receiver = FGReceiver(UDP_HOST, UDP_PORT, self.on_fg_data)
self.receiver.start()
def on_close(self):
self.running = False
self.receiver.running = False
# Wait a little for the file writer to flush remaining items
time.sleep(0.2)
self.destroy()
# ─────────────────────────────────────────────────────────
# MAIN
# ─────────────────────────────────────────────────────────
def main():
print('[INFO] Loading CNN-BiLSTM model...')
model, val_rmse, n_feat = load_model()
d_min, d_max = load_scaler()
# Default to A320neo
profile = AIRCRAFT_PROFILES['A320neo (CFM LEAP-1A)']
print(f'\n[PROTO] Place the XML at:')
print(f' {profile["protocol_file"]}')
ok = os.path.exists(profile['protocol_file'])
print(f' {"FOUND ✓" if ok else "NOT FOUND — copy a320neo_protocol.xml there"}')
print(f'\n[LAUNCH] Copy this to a terminal to start FlightGear:')
print(f' fgfs.exe --aircraft={profile["aircraft_flag"]}')
print(f' --airport={profile["airport"]}')
print(f' --fg-root="{FG_ROOT}"')
print(f' --fg-aircraft="{FG_AIRCRAFT_DIR}"')
print(f' --generic=socket,out,{FG_HZ},{UDP_HOST},{UDP_PORT},udp,'
f'{profile["protocol_name"]}')
print()
print('[STEPS ONCE FG IS LOADED]:')
print(' 1. Wait for FG to fully load (cockpit visible)')
print(' 2. Equipment → Autostart (wait for N2 > 20%)')
print(' 3. Set throttle to ~30-50% on the virtual throttle lever')
print(' 4. Dashboard will show ⏳ CALIBRATING for 50 frames')
print(' 5. After calibration: health=100% is YOUR baseline')
print(' 6. Throttle changes, altitude changes = normal health fluctuation')
print(' 7. CRITICAL = dropped below 50% of your OWN baseline\n')
# Port check
chk = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
chk.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
chk.bind((UDP_HOST, UDP_PORT))
except OSError:
chk.close()
print(f'[ERROR] Port {UDP_PORT} already in use.')
print(f' netstat -ano | findstr :{UDP_PORT}')
print(f' taskkill /PID /F')
sys.exit(1)
chk.close()
app = PHMDashboard(model, val_rmse, d_min, d_max, n_feat)
app.protocol('WM_DELETE_WINDOW', app.on_close)
app.mainloop()
if __name__ == '__main__':
main()