AI Bridge and Tunnel Health Monitoring System Development

Development of an AI system for monitoring the condition of bridges and tunnels

A bridge or tunnel shows no signs of deterioration before collapse—unless monitored. Structural Health Monitoring (SHM) with ML replaces infrequent scheduled inspections with continuous analysis of vibrations, deformations, and acoustic signals.

SHM sensor network

Sensor Types and Placement:

shm_sensors = {
    'accelerometers': {
        'location': 'главные пролёты, пилоны, подвески',
        'frequency': '100-200 Hz',
        'measures': 'динамический отклик на транспорт и ветер'
    },
    'strain_gauges': {
        'location': 'критические узлы балок, сварные швы',
        'frequency': '1-10 Hz',
        'measures': 'деформация (микрострейн, мкм/м)'
    },
    'displacement_sensors': {
        'location': 'опорные части моста',
        'frequency': '1 Hz',
        'measures': 'вертикальное и горизонтальное смещение'
    },
    'crack_gauges': {
        'location': 'известные трещины, сварные швы',
        'frequency': '0.1 Hz',
        'measures': 'раскрытие трещины в мм'
    },
    'temperature': {
        'location': 'по длине пролёта',
        'frequency': '1/60 Hz',
        'measures': 'температурные деформации (компенсация)'
    }
}

Modal Analysis

Extracting the natural frequencies of the structure:

import numpy as np
from scipy import signal
from scipy.linalg import svd

def extract_modal_frequencies(acceleration_data: np.ndarray,
                               sampling_rate: float = 100) -> dict:
    """
    OMA (Operational Modal Analysis) — идентификация мод из рабочей вибрации.
    Изменение собственной частоты = изменение жёсткости = повреждение.
    """
    n_sensors = acceleration_data.shape[1]

    # Кросс-спектральная матрица
    freqs, Sxy = signal.csd(
        acceleration_data[:, 0], acceleration_data[:, 1],
        fs=sampling_rate, nperseg=2048
    )

    # SVD для выделения мод (FDD — Frequency Domain Decomposition)
    S_matrices = []
    for i in range(len(freqs)):
        row_data = acceleration_data  # упрощённо
        S_matrices.append(np.outer(row_data[i], row_data[i].conj()))

    # Пики сингулярных значений = резонансные частоты
    singular_values = []
    for S in S_matrices:
        U, s, Vh = svd(S)
        singular_values.append(s[0])

    peaks, properties = signal.find_peaks(
        singular_values,
        height=np.mean(singular_values) * 3,
        distance=5
    )

    modal_frequencies = freqs[peaks].tolist()

    return {
        'modal_frequencies': modal_frequencies,
        'dominant_mode': freqs[peaks[np.argmax(properties['peak_heights'])]] if len(peaks) > 0 else None
    }

Monitoring frequency changes:

def detect_structural_change(current_freqs: list, baseline_freqs: list,
                              tolerance_pct: float = 3.0) -> dict:
    """
    Снижение собственной частоты на >5% = потеря жёсткости = возможное повреждение.
    Компенсируем температурный эффект (зимой частота выше из-за жёсткости металла).
    """
    changes = []
    for i, (curr, base) in enumerate(zip(current_freqs, baseline_freqs)):
        change_pct = (curr - base) / base * 100
        if abs(change_pct) > tolerance_pct:
            changes.append({
                'mode': i + 1,
                'baseline_hz': round(base, 3),
                'current_hz': round(curr, 3),
                'change_pct': round(change_pct, 2),
                'direction': 'decrease' if change_pct < 0 else 'increase'
            })

    severity = 'none'
    if changes:
        max_change = max(abs(c['change_pct']) for c in changes)
        severity = 'critical' if max_change > 10 else ('warning' if max_change > 5 else 'notice')

    return {'structural_changes': changes, 'severity': severity}

Deformation analysis and fatigue calculation

Rainflow fatigue cycle counting:

def rainflow_fatigue_analysis(strain_history: np.ndarray,
                               material_sn_curve: dict) -> dict:
    """
    Rainflow counting: подсчёт амплитуд нагрузочных циклов.
    SN-кривая материала: N циклов до разрушения при амплитуде S.
    Правило Майнера: D = Σ(ni/Ni) — суммарное повреждение.
    """
    # Упрощённый алгоритм (реальный через fatpack или rainflow пакет)
    import rainflow

    cycles = list(rainflow.count_cycles(strain_history))

    damage = 0.0
    for amplitude, mean, count, i_start, i_end in cycles:
        # Интерполяция по SN-кривой
        cycles_to_failure = material_sn_curve['coefficient'] / (amplitude ** material_sn_curve['exponent'])
        damage += count / cycles_to_failure

    return {
        'miner_damage_ratio': damage,
        'remaining_fatigue_life_pct': max(0, (1 - damage) * 100),
        'alert': damage > 0.8  # 80% исчерпание ресурса
    }

Detection of abnormal loads

Monitoring of excess design loads:

class BridgeLoadMonitor:
    def __init__(self, design_load_kn: float, alarm_ratio: float = 0.85):
        self.design_load = design_load_kn
        self.alarm_ratio = alarm_ratio
        self.event_log = []

    def analyze_strain_event(self, timestamp, strain_data: np.ndarray,
                              section_modulus: float) -> dict:
        """
        Из деформации → напряжение → эквивалентная нагрузка
        Elastic modulus для стали: 200 ГПа
        """
        max_strain = np.max(np.abs(strain_data))
        E_steel = 200e9  # Па
        max_stress_mpa = max_strain * E_steel * 1e-6  # МПа

        # Момент → нагрузка (упрощение)
        equivalent_load = max_stress_mpa * section_modulus

        event = {
            'timestamp': timestamp,
            'max_strain_microstrain': float(max_strain * 1e6),
            'max_stress_mpa': float(max_stress_mpa),
            'load_utilization': equivalent_load / self.design_load
        }

        if event['load_utilization'] > self.alarm_ratio:
            event['alert'] = True
            event['severity'] = 'critical' if event['load_utilization'] > 1.0 else 'warning'

        self.event_log.append(event)
        return event

Long-term trend and forecast

Regression on historical data:

from sklearn.linear_model import HuberRegressor
from sklearn.preprocessing import PolynomialFeatures

def fit_degradation_trend(monthly_health_scores: pd.DataFrame) -> dict:
    """
    Линейный или полиномиальный тренд Health Score.
    Huber регрессия — устойчивость к аномальным зимним точкам.
    """
    X = monthly_health_scores['month_num'].values.reshape(-1, 1)
    y = monthly_health_scores['health_score'].values

    model = HuberRegressor()
    model.fit(X, y)

    # Экстраполяция: когда health score достигнет 0.6 (порог вмешательства)
    threshold = 0.6
    if model.coef_[0] < 0:  # деградация
        months_to_threshold = (threshold - model.intercept_) / model.coef_[0]
        current_month = monthly_health_scores['month_num'].max()
        months_remaining = months_to_threshold - current_month
    else:
        months_remaining = None  # улучшение тренда

    return {
        'slope_per_month': model.coef_[0],
        'months_to_intervention': months_remaining,
        'intervention_year': pd.Timestamp.today() + pd.DateOffset(months=int(months_remaining or 999))
    }

Integration with CAD and standards: The system exports data in formats compatible with BIM platforms (Autodesk Revit via IFC, OpenBridge). Reports comply with Russian standards: SP 35.13330.2011 (bridges), GOST R 56353.

Timeframe: Sensor installation + basic deformation monitoring + load overshoot alerts — 4-5 weeks. Modal analysis, fatigue rainflow, long-term trend analysis, BIM integration — 3-4 months.