AI Automatic Road Pavement Inspection System Development

AI road surface inspection system

Potholes, cracks, and ruts—the annual cost of fixing them runs into the billions. The problem is that by the time road workers notice them, the defect has already reached a critical stage. AI inspections using vehicle cameras or specialized machines can detect early signs of pavement degradation and prioritize repairs.

Classification of road surface defects

ASTM D6433 identifies 20 types of distress. In practice, we work with 7–8 key ones:

Detection and segmentation model

import torch
import numpy as np
import segmentation_models_pytorch as smp
from ultralytics import YOLO
import cv2

class PavementInspector:
    def __init__(self, seg_model_path: str, det_model_path: str):
        # Сегментация трещин: UNet++ с ResNet50 энкодером
        # Дообучен на RDD2022 (Road Damage Dataset, 47k изображений)
        self.seg_model = smp.UnetPlusPlus(
            encoder_name='resnet50',
            encoder_weights=None,
            in_channels=3,
            classes=4,  # background, longitudinal, transverse, alligator
        )
        seg_ckpt = torch.load(seg_model_path)
        self.seg_model.load_state_dict(seg_ckpt)
        self.seg_model.eval()

        # YOLOv8m для ям и выбоин (bbox достаточно)
        self.det_model = YOLO(det_model_path)

        # Маппинг классов сегментации
        self.seg_classes = {
            0: 'background',
            1: 'longitudinal_crack',
            2: 'transverse_crack',
            3: 'alligator_crack'
        }

        # Маппинг для оценки тяжести (PCI-based)
        self.severity_thresholds = {
            'pothole': {'low': 0.01, 'medium': 0.05},    # % площади кадра
            'crack': {'low': 0.02, 'medium': 0.08}
        }

    @torch.no_grad()
    def inspect(self, frame: np.ndarray) -> dict:
        h, w = frame.shape[:2]

        # 1. Сегментация трещин
        input_tensor = self._preprocess(frame)
        seg_output = self.seg_model(input_tensor)
        seg_mask = seg_output.argmax(dim=1)[0].numpy()

        crack_analysis = self._analyze_cracks(seg_mask, w * h)

        # 2. Детекция ям
        det_results = self.det_model(frame, conf=0.45)
        potholes = self._analyze_potholes(det_results, w * h)

        # 3. Индекс состояния покрытия (упрощённый PCI)
        pci = self._compute_pci(crack_analysis, potholes)

        return {
            'crack_analysis': crack_analysis,
            'potholes': potholes,
            'pci_score': pci,
            'condition': self._pci_to_condition(pci),
            'seg_mask': seg_mask
        }

    def _analyze_cracks(self, mask: np.ndarray,
                          total_pixels: int) -> dict:
        analysis = {}
        for cls_id, cls_name in self.seg_classes.items():
            if cls_id == 0:
                continue
            crack_pixels = int((mask == cls_id).sum())
            ratio = crack_pixels / total_pixels
            analysis[cls_name] = {
                'pixel_count': crack_pixels,
                'area_ratio': ratio,
                'severity': 'high' if ratio > 0.08 else
                             'medium' if ratio > 0.02 else 'low'
            }
        return analysis

    def _compute_pci(self, cracks: dict, potholes: list) -> float:
        """
        PCI 0–100: 100 = идеальное покрытие, 0 = полная деградация.
        Упрощённая формула на основе ASTM D6433.
        """
        deduct = 0.0
        for crack_type, data in cracks.items():
            ratio = data['area_ratio']
            if ratio > 0.08:
                deduct += 25
            elif ratio > 0.02:
                deduct += 12
            elif ratio > 0.005:
                deduct += 5

        for pothole in potholes:
            area = pothole['area_ratio']
            if area > 0.03:
                deduct += 30
            elif area > 0.01:
                deduct += 15

        return max(0, 100 - deduct)

    def _pci_to_condition(self, pci: float) -> str:
        if pci >= 85:   return 'excellent'
        elif pci >= 70: return 'good'
        elif pci >= 55: return 'fair'
        elif pci >= 40: return 'poor'
        elif pci >= 25: return 'very_poor'
        else:           return 'failed'

Mobile Inspection: Vehicle-Mounted Camera

For public roads, a camera under the front bumper or in the radiator grille records at 25 fps and is linked to GPS. Additionally, an accelerometer automatically detects potholes based on vibration.

class MobileRoadSurvey:
    def __init__(self, gps_logger, inspector: PavementInspector):
        self.gps = gps_logger
        self.inspector = inspector
        self.survey_log = []

    def process_frame_with_geotagging(self, frame: np.ndarray,
                                       timestamp: float) -> dict:
        gps_coords = self.gps.get_coords(timestamp)
        results = self.inspector.inspect(frame)

        record = {
            'timestamp': timestamp,
            'lat': gps_coords['lat'],
            'lon': gps_coords['lon'],
            'pci': results['pci_score'],
            'condition': results['condition'],
            'defects': results
        }
        self.survey_log.append(record)
        return record

Case: Inspection of 120 km of city roads

Task: Prioritizing road repairs. Equipment: Ford Transit with 4 cameras (front + 2 sides + rear), RTK GPS. 120 km covered over 3 days of filming.

• Processed: 1.2 million frames
• Found: 3400 pits (P> 0.5), 47 km of cracks (segmentation)
• Of these, critical (PCI < 25): 8.2 km - priority repair
• Savings compared to manual inspection: 12 working days → 6 hours of processing + 3 hours of verification