Разработка AI-системы предиктивного обслуживания энергосетей

Разработка AI-системы предиктивного обслуживания энергосетей

Энергетическое оборудование — трансформаторы, кабели, выключатели — выходит из строя с предшествующими сигналами деградации за недели и месяцы до аварии. Переход от планово-предупредительного ТО к предиктивному снижает затраты на 20-30% и сокращает незапланированные отключения.

Диагностика силовых трансформаторов

DGA (Dissolved Gas Analysis) — главный индикатор:

import numpy as np

# Ключи диагностики: растворённые газы в масле трансформатора
dga_thresholds = {
    'hydrogen_H2': {'warning': 150, 'alarm': 500},       # ppm
    'methane_CH4': {'warning': 75, 'alarm': 200},
    'ethylene_C2H4': {'warning': 60, 'alarm': 150},
    'ethane_C2H6': {'warning': 100, 'alarm': 200},
    'acetylene_C2H2': {'warning': 2, 'alarm': 30},       # ацетилен = дуга!
    'carbon_monoxide_CO': {'warning': 700, 'alarm': 1500},
    'carbon_dioxide_CO2': {'warning': 10000, 'alarm': 15000}
}

def diagnose_transformer_dga(gas_ppm: dict) -> dict:
    """
    Треугольник Дюваля + Rogers ratios для классификации дефекта.
    """
    # Проверка порогов
    alarm_gases = []
    for gas, value in gas_ppm.items():
        if gas in dga_thresholds:
            if value > dga_thresholds[gas]['alarm']:
                alarm_gases.append({'gas': gas, 'value': value, 'level': 'alarm'})
            elif value > dga_thresholds[gas]['warning']:
                alarm_gases.append({'gas': gas, 'value': value, 'level': 'warning'})

    # Rogers ratios (IEEE C57.104)
    ch4_h2 = gas_ppm.get('methane_CH4', 0) / (gas_ppm.get('hydrogen_H2', 1) + 1e-9)
    c2h4_c2h6 = gas_ppm.get('ethylene_C2H4', 0) / (gas_ppm.get('ethane_C2H6', 1) + 1e-9)
    c2h2_c2h4 = gas_ppm.get('acetylene_C2H2', 0) / (gas_ppm.get('ethylene_C2H4', 1) + 1e-9)

    # Диагностика дефекта
    fault_type = 'normal'
    if gas_ppm.get('acetylene_C2H2', 0) > 5:
        fault_type = 'arc_discharge'        # электрическая дуга — КРИТИЧНО
    elif c2h4_c2h6 > 1.0 and ch4_h2 > 0.1:
        fault_type = 'thermal_fault_high'   # перегрев > 700°C
    elif ch4_h2 > 0.1 and c2h4_c2h6 < 1.0:
        fault_type = 'thermal_fault_medium' # перегрев 300-700°C
    elif gas_ppm.get('hydrogen_H2', 0) > 200 and c2h4_c2h6 < 0.1:
        fault_type = 'partial_discharge'    # частичные разряды

    urgency = {
        'arc_discharge': 'immediate_shutdown',
        'thermal_fault_high': 'urgent_inspection',
        'thermal_fault_medium': 'schedule_maintenance',
        'partial_discharge': 'enhanced_monitoring',
        'normal': 'routine'
    }

    return {
        'fault_type': fault_type,
        'urgency': urgency[fault_type],
        'alarm_gases': alarm_gases,
        'rogers_ratios': {
            'CH4/H2': round(ch4_h2, 3),
            'C2H4/C2H6': round(c2h4_c2h6, 3),
            'C2H2/C2H4': round(c2h2_c2h4, 3)
        }
    }

Тренд газов — прогноз деградации:

from sklearn.linear_model import LinearRegression
import pandas as pd

def forecast_gas_trend(dga_history: pd.DataFrame, gas: str,
                        forecast_days: int = 30) -> dict:
    """
    Экспоненциальный рост газов = ускорение дефекта.
    Линейный тренд на log(concentration) для обнаружения ускорения.
    """
    data = dga_history[['days_ago', gas]].dropna()
    data = data[data[gas] > 0]

    if len(data) < 3:
        return {'status': 'insufficient_data'}

    X = data['days_ago'].values.reshape(-1, 1)
    y = np.log(data[gas].values)

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

    # Прогноз
    future_x = np.array([[-forecast_days]])  # отрицательный = будущее
    predicted_log = model.predict(future_x)[0]
    predicted_concentration = np.exp(predicted_log)

    doubling_time = np.log(2) / abs(model.coef_[0]) if model.coef_[0] > 0 else None

    return {
        'current': data[gas].iloc[-1],
        f'forecast_{forecast_days}d': round(predicted_concentration, 1),
        'growth_rate_per_day': model.coef_[0],
        'doubling_time_days': round(doubling_time, 0) if doubling_time else None,
        'threshold_breach_days': estimate_days_to_threshold(
            data[gas].iloc[-1], predicted_concentration,
            dga_thresholds.get(gas, {}).get('alarm', float('inf')),
            forecast_days
        )
    }

Мониторинг кабельных линий

Частичные разряды в кабеле:

def analyze_partial_discharge(pd_measurements: dict) -> dict:
    """
    PD-мониторинг высоковольтных кабелей.
    pC (пикокулоны) — величина разряда.
    PDIV (Partial Discharge Inception Voltage) — напряжение начала ПР.
    """
    max_pd_pc = pd_measurements.get('max_pd_magnitude_pc', 0)
    pd_rate_per_minute = pd_measurements.get('pd_pulse_rate', 0)
    pd_location_m = pd_measurements.get('pd_location_tdr_m', None)  # TDR локация

    # Классификация по IEC 60270
    if max_pd_pc > 1000:
        severity = 'critical'
        action = 'immediate_cable_replacement'
    elif max_pd_pc > 300:
        severity = 'major'
        action = 'schedule_replacement_3months'
    elif max_pd_pc > 100:
        severity = 'minor'
        action = 'enhanced_monitoring'
    else:
        severity = 'normal'
        action = 'routine_monitoring'

    return {
        'max_pd_pc': max_pd_pc,
        'pd_rate': pd_rate_per_minute,
        'pd_location_m': pd_location_m,
        'severity': severity,
        'action': action
    }

Предиктивная модель для воздушных линий

Прогнозирование провисания провода:

def calculate_sag_risk(weather_data: dict, line_parameters: dict) -> dict:
    """
    Нагрев провода током + температура → провисание → опасное приближение к земле.
    Допустимый нагрев AAAC провода: до 90°C (аварийный режим).
    """
    I_amps = line_parameters['current_a']
    R_ohm_per_km = line_parameters['resistance_ohm_per_km']
    length_km = line_parameters['length_km']
    ambient_temp = weather_data['temperature_c']
    wind_speed = weather_data['wind_speed_ms']
    solar_radiation = weather_data.get('solar_irradiance_wm2', 0)

    # Упрощённая тепловая модель (IEEE 738)
    joule_heating = I_amps**2 * R_ohm_per_km / 1000  # Вт/м
    convective_cooling = (0.7 + 0.5 * wind_speed) * (80 - ambient_temp)  # Вт/м (упрощение)
    solar_heating = solar_radiation * 0.015  # Вт/м

    conductor_temp = ambient_temp + (joule_heating + solar_heating) / (convective_cooling + 1)

    # Провисание растёт с температурой
    base_sag_m = line_parameters.get('design_sag_m', 5.0)
    thermal_expansion = 0.023e-3 * (conductor_temp - 20) * length_km * 1000
    actual_sag_m = base_sag_m + thermal_expansion * 0.5

    clearance_violation = actual_sag_m > line_parameters.get('max_sag_m', 7.0)

    return {
        'conductor_temp_c': round(conductor_temp, 1),
        'actual_sag_m': round(actual_sag_m, 2),
        'clearance_violation': clearance_violation,
        'thermal_limit_pct': min(100, conductor_temp / 90 * 100)
    }

Планирование ТО

Приоритизация объектов: Риск-матрица: вероятность отказа × последствия (кВт·ч недоотпуска × штрафы × потребители). Объекты ранжируются по ожидаемым потерям от неплановой аварии. Интеграция с SAP PM / Maximo для создания нарядов.

Сроки: DGA мониторинг + базовый алертинг + трансформаторный дашборд — 4-5 недель. Тренды DGA, PD мониторинг кабелей, тепловая модель ЛЭП, оптимизация планирования ТО, SAP PM интеграция — 3-4 месяца.