Development of AI-Based Leak Detection Systems for Pipelines

Why standard leak detection methods are slow?

On a 100 km product pipeline, a leak of 0.5% of flow remains undetected by balance methods for up to 20 minutes. At a flow rate of 500 m³/h, this means a loss of 150 m³ of product — tens of thousands of rubles just for the product. Add environmental fines, which can reach up to 50 million rubles per spill, and downtime for repairs — the incident cost reaches hundreds of millions. Standard systems based on balance or pressure drop are too slow and generate many false alarms. We develop an AI system that replaces slow balance and expensive Fiber Optic DAS. We combine ML on balance data, negative pressure wave (NPW) detection, and acoustic monitoring — this reduces detection time to seconds and localization accuracy to meters. On a 120 km gas pipeline, the system detected a 0.2% leak within 15 seconds, an order of magnitude faster than traditional methods. Contact us — we'll assess your pipeline in 2 days.

Leak detection methods

Comparison of methods:

The acoustic method detects leaks 10 times faster and more accurately than the balance method, as confirmed by research on Wikipedia.

How does the balance method with ML work?

Balance of input and output is simple but noisy. We clean the signal using exponentially weighted moving average (EWMA) and an adaptive threshold. Example implementation in Python:

import numpy as np import pandas as pd from scipy.stats import chi2 class PipelineBalanceMonitor: def __init__(self, line_id: str, balance_threshold_pct: float = 1.0): self.line_id = line_id self.threshold = balance_threshold_pct / 100 self.ewma_balance = None self.alpha = 0.1 def update(self, inlet_flow_m3h: float, outlet_flow_m3h: float, line_pack_change: float = 0.0) -> dict: """ Balance = Inlet - Outlet - Change in line pack (compressibility) For gas: line pack changes with pressure changes. """ balance = inlet_flow_m3h - outlet_flow_m3h - line_pack_change balance_pct = balance / (inlet_flow_m3h + 1e-9) # EWMA filter if self.ewma_balance is None: self.ewma_balance = balance_pct else: self.ewma_balance = self.alpha * balance_pct + (1 - self.alpha) * self.ewma_balance leak_detected = self.ewma_balance > self.threshold return { 'line_id': self.line_id, 'instantaneous_balance_pct': round(balance_pct * 100, 3), 'ewma_balance_pct': round(self.ewma_balance * 100, 3), 'leak_detected': leak_detected, 'estimated_leak_rate_m3h': max(0, balance) if leak_detected else 0 } 

How does negative pressure wave (NPW) determine the exact leak location?

On one project, we implemented NPW on a 120 km gas pipeline. For detection, two pressure sensors with a sampling rate of 1 Hz are sufficient. The wave speed for gas is about 400 m/s, for liquid — 1100 m/s. The algorithm detects a sharp pressure drop (>5 bar/s) and localizes the leak by the difference in wave arrival time. Localization accuracy ±50 m.

def detect_negative_pressure_wave(pressure_sensors: dict, pipeline_params: dict) -> dict: """ When a pipe ruptures → sharp pressure drop at the rupture point → a wave of reduced pressure propagates in both directions. Wave speed: a = sqrt(K/ρ) ≈ 1000-1300 m/s for liquids. Knowing the wave arrival time at two sensors → rupture location. """ wave_speed = pipeline_params.get('pressure_wave_speed_ms', 1100) # m/s # Determine arrival time at each sensor arrival_times = {} for sensor_id, pressure_series in pressure_sensors.items(): # Look for sharp drop: derivative < -threshold dP_dt = np.gradient(pressure_series.values, 1) # 1-second data sharp_drop_indices = np.where(dP_dt < -5)[0] # > 5 bar/s = shock wave if len(sharp_drop_indices) > 0: arrival_times[sensor_id] = sharp_drop_indices[0] if len(arrival_times) < 2: return {'leak_detected': False} # Localization using two sensors A and B sensor_ids = list(arrival_times.keys())[:2] t_A = arrival_times[sensor_ids[0]] t_B = arrival_times[sensor_ids[1]] delta_t = abs(t_A - t_B) # seconds # Distance from A to rupture point distance_AB = pipeline_params['sensor_distances'].get( (sensor_ids[0], sensor_ids[1]), 5000 # m between sensors ) dist_from_A = (distance_AB - wave_speed * delta_t) / 2 dist_from_A = max(0, min(dist_from_A, distance_AB)) return { 'leak_detected': True, 'leak_location_m_from_sensor_A': round(dist_from_A, 0), 'confidence': 'high', 'delta_t_seconds': delta_t } 

Concrete example: on a 120 km gas pipeline, we implemented NPW + acoustics. Within the first month, the system detected two small leaks (0.2% of flow) that the balance method missed. Localization accuracy was ±15 meters, confirmed by digging.

Why is fusion detection more reliable than a single method?

We combine three methods with weights: balance (0.4), NPW (0.4), acoustics (0.2). If at least one confirms a leak with high confidence, the system raises an alarm. This reduces false alarms by 70% compared to threshold methods.

Comparison of methods reliability:

Example code:

def fuse_leak_detections(balance_result: dict, npw_result: dict, acoustic_result: dict) -> dict: """ Each method has its own false positives. Voting with weights based on method reliability. """ votes = 0 total_weight = 0 location_estimates = [] if balance_result.get('leak_detected'): votes += 0.4 location_estimates.append(balance_result.get('estimated_location')) total_weight += 0.4 if npw_result.get('leak_detected'): votes += 0.4 location_estimates.append(npw_result.get('leak_location_m_from_sensor_A')) total_weight += 0.4 if acoustic_result.get('leak_detected'): votes += 0.2 location_estimates.append(acoustic_result.get('acoustic_location_m')) total_weight += 0.2 confidence = votes / total_weight best_location = min(location_estimates, key=lambda x: 0 if x is None else 1) if location_estimates else None return { 'leak_confirmed': confidence >= 0.4, # at least one reliable method 'confidence': round(confidence, 2), 'estimated_location_m': best_location, 'alert_level': 'critical' if confidence > 0.8 else 'warning' } 

Acoustic monitoring: from signal to decision

Acoustic sensors capture high-frequency leak noise (500-5000 Hz). We apply bandpass filtering and extract features: RMS, kurtosis, energy ratio in the leak band to total. A threshold of leak_score > 0.6 indicates a leak. Processing is real-time with a delay of no more than 1 second.

from scipy import signal import librosa def analyze_acoustic_signal(audio_signal: np.ndarray, sampling_rate: int = 44100) -> dict: """ A leak produces a characteristic acoustic signal: - Broadband noise 100-10000 Hz - Characteristics depend on pressure and hole size """ # Filtering: bandpass filter 500-5000 Hz (leak range) b, a = signal.butter(4, [500, 5000], btype='band', fs=sampling_rate) filtered = signal.filtfilt(b, a, audio_signal) # Features rms = np.sqrt(np.mean(filtered**2)) kurtosis = np.mean((filtered - filtered.mean())**4) / (np.std(filtered)**4 + 1e-9) # Spectral features freqs, psd = signal.welch(filtered, fs=sampling_rate, nperseg=1024) band_energy = np.trapz(psd[(freqs >= 1000) & (freqs <= 3000)], freqs[(freqs >= 1000) & (freqs <= 3000)]) total_energy = np.trapz(psd, freqs) leak_band_ratio = band_energy / (total_energy + 1e-9) # Classification: compare with pipeline normal noise templates leak_score = rms * 0.4 + leak_band_ratio * 0.4 + min(1, kurtosis / 10) * 0.2 return { 'rms_amplitude': float(rms), 'kurtosis': float(kurtosis), 'leak_band_ratio': float(leak_band_ratio), 'leak_score': float(leak_score), 'leak_detected': leak_score > 0.6 } 

What does the system deliver in practice?

Over the past years, we have implemented more than 30 systems on oil and gas pipelines. On one facility, fusion detection reduced false alarms by 70% and detected a 0.2% leak within 15 seconds, while the old balance method showed instability. Savings from prevented fines and product loss — millions of rubles.

What is included in the work

• Technical audit and collection of historical pipeline data.
• Algorithm development (balance, NPW, acoustic, fusion).
• Integration with SCADA via OPC-UA (Emerson DeltaV, Honeywell Experion, ABB 800xA).
• Testing on historical accidents and retrospective data.
• Dispatcher training on system operation.
• 12-month warranty support.

Estimated timelines

• Basic version (balance + alerts + SCADA connector) — from 3 to 4 weeks.
• Full functionality (NPW, acoustics, fusion, automatic valve closure) — from 3 to 4 months.

Request a preliminary assessment — we will prepare a commercial proposal with exact timelines for your facility. Over 5 years on the market, 30+ implemented projects in industrial automation.