AI Aviation Safety Analytics System

AI-based analytics for aviation safety

An aviation incident rarely occurs due to a single cause. ICAO describes this as a "Heinrich chain": the final event is just the tip of a pyramid of preceding anomalies. An AI safety analysis system catches the links in this chain before they become complete.

Data sources and types of anomalies

Aviation security operates through several streams:

• FDR/QAR data (Flight Data Recorder / Quick Access Recorder) - thousands of parameters with a frequency of 4-64 Hz: speed, altitude, engine modes, rudder deflections, acceleration
• ACARS messages — aircraft-to-ground technical messages, including fault codes
• ATC negotiations — transcripts of negotiations with the dispatcher (NLP)
• FOQA (Flight Operational Quality Assurance) — flight quality monitoring programs
• Maintenance data - MRO records, component failure history

Typical anomalies for automatic detection:

Flight parametric data analysis algorithm

FOQA Event Detection via Sliding Windows:

import numpy as np
import pandas as pd
from scipy.signal import savgol_filter
from typing import List, Dict

def detect_flight_exceedances(
    fdr_df: pd.DataFrame,
    aircraft_type: str,
    phase_of_flight: str
) -> List[Dict]:
    """
    fdr_df: DataFrame с временными рядами параметров полёта
    Возвращает список событий-нарушений с временными метками.
    """
    exceedances = []

    # Сглаживание шума датчиков
    fdr_df['vsi_smooth'] = savgol_filter(
        fdr_df['vertical_speed_fpm'], window_length=11, polyorder=3
    )

    # Жёсткая посадка
    if phase_of_flight == 'landing':
        touchdown_mask = (
            (fdr_df['radio_altitude_ft'] < 5) &
            (fdr_df['radio_altitude_ft'].shift(1) >= 5)
        )
        touchdown_idx = fdr_df[touchdown_mask].index

        for idx in touchdown_idx:
            if idx in fdr_df.index:
                nz = fdr_df.loc[idx, 'normal_acceleration_g']
                if nz > 1.8:
                    severity = 'hard' if nz > 2.6 else 'firm'
                    exceedances.append({
                        'event_type': 'hard_landing',
                        'timestamp': fdr_df.loc[idx, 'timestamp'],
                        'value': round(nz, 3),
                        'threshold': 2.6,
                        'severity': severity,
                        'requires_inspection': nz > 2.6
                    })

    # Нестабилизированный заход: скорость вне коридора на 500 ft
    if phase_of_flight == 'approach':
        gate_mask = (
            (fdr_df['radio_altitude_ft'].between(480, 520)) &
            (fdr_df['flap_position'] < 30)  # не полный флап
        )
        gate_points = fdr_df[gate_mask]

        for idx, row in gate_points.iterrows():
            # Скорость должна быть Vapp ± 10 kt
            vapp = row.get('vapp_kt', 145)  # из FMS
            if abs(row['ias_kt'] - vapp) > 10:
                exceedances.append({
                    'event_type': 'unstabilized_approach_speed',
                    'timestamp': row['timestamp'],
                    'value': round(row['ias_kt'], 1),
                    'expected': vapp,
                    'deviation_kt': round(row['ias_kt'] - vapp, 1),
                    'severity': 'go_around_required'
                })

    return exceedances


def engine_health_anomaly(
    engine_params: pd.DataFrame,
    baseline_egt_delta: float = 0.0
) -> Dict:
    """
    EGT Margin — разница между фактическим и предельным EGT.
    Тренд снижения margin указывает на деградацию двигателя.
    """
    # Нормализация на условия (OAT, давление, N1)
    engine_params['egt_corrected'] = (
        engine_params['egt_c'] - engine_params['oat_c'] * 0.95
        - engine_params['altitude_ft'] * 0.002
    )

    # EGT margin trend за последние 50 циклов
    recent_cycles = engine_params.tail(50)
    x = np.arange(len(recent_cycles))
    slope, intercept = np.polyfit(x, recent_cycles['egt_corrected'], 1)

    # Тренд >0.5°C/цикл — признак деградации hot section
    degradation_rate = slope  # °C за цикл

    return {
        'egt_trend_per_cycle': round(degradation_rate, 3),
        'current_margin_c': round(
            engine_params['egt_limit_c'].iloc[-1] - engine_params['egt_corrected'].iloc[-1], 1
        ),
        'alert': degradation_rate > 0.5,
        'cycles_to_limit': (
            int((engine_params['egt_limit_c'].iloc[-1] - engine_params['egt_corrected'].iloc[-1]) / degradation_rate)
            if degradation_rate > 0 else None
        )
    }

NLP analysis of ATC negotiations

Negotiation transcripts are an undervalued source of warning signs. Typical risk patterns include non-standard instructions, repeated requests for clearance, and frequency switching during critical phases.

from transformers import pipeline

# Классификатор на дообученном авиационном корпусе
atc_safety_classifier = pipeline(
    "text-classification",
    model="aviation-safety/atc-risk-classifier-bert",
    device=0
)

risk_patterns = [
    "say again",           # непонимание инструкции
    "unable",              # невозможность выполнения
    "traffic alert",       # TCAS
    "go around",           # уход на второй круг
    "emergency",
    "minimum fuel",
    "expedite"
]

def analyze_atc_transcript(transcript: str) -> dict:
    risk_keywords_found = [p for p in risk_patterns if p.lower() in transcript.lower()]
    ml_result = atc_safety_classifier(transcript[:512])[0]

    return {
        'risk_keywords': risk_keywords_found,
        'ml_risk_label': ml_result['label'],
        'ml_risk_score': round(ml_result['score'], 3),
        'requires_review': len(risk_keywords_found) > 0 or ml_result['label'] == 'HIGH_RISK'
    }

Case: FOQA monitoring of a fleet of 24 aircraft

An airline operating 24 B737NG/A320 aircraft. Before the system's implementation, FOQA analyzed a selective sample of 5% of flights manually. After automation, 100% of flights were analyzed, with eight event types. During the first three months, 340 unstabilized approaches were identified (including 38 with significant deviations), seven hard landings above the inspection threshold, and EGT margin degradation on two engines predicted 60 cycles before the scheduled hot section replacement. The system ordered an unscheduled removal of one engine, as cracks in the compressor blade were detected.

Stack

Deadlines: Basic FOQA parametric event analyzer – 6-8 weeks. Full stack with NLP, predictive engine maintenance, and dashboard – 4-5 months.