Разработка AI-системы мониторинга строительной площадки с помощью дронов

AI-мониторинг строительной площадки с помощью дронов

Дрон над стройплощадкой раз в неделю даёт то, что 20 камер дать не могут: вид сверху всего объекта, возможность облёта конструкций, 3D-модель прогресса. Но «запустить дрон и посмотреть видео» — это не мониторинг. Мониторинг — это автоматический анализ, сравнение с планом и генерация отчёта без участия человека.

Система автоматического мониторинга

import numpy as np
import cv2
from ultralytics import YOLO
from pathlib import Path
import json
from datetime import datetime

class ConstructionDroneMonitor:
    def __init__(self, project_config: dict):
        # Детектор для стройплощадки
        self.detector = YOLO(project_config['model_path'])

        # Классы: crane, excavator, concrete_mixer, scaffolding,
        #         worker, pile, rebar, formwork, brickwork, concrete_poured
        self.equipment_classes = project_config['equipment_classes']
        self.material_classes = project_config['material_classes']

        # Эталонный план + зоны
        self.site_plan = project_config['site_plan']
        self.gps_calibration = project_config['gps_calibration']

    def process_flight_mission(self, images_dir: str,
                                 flight_log: dict) -> dict:
        """
        Обрабатываем серию снимков одного облёта.
        flight_log содержит GPS-координаты для каждого снимка.
        """
        images = sorted(Path(images_dir).glob('*.jpg'))
        all_detections = []

        for img_path in images:
            frame = cv2.imread(str(img_path))
            if frame is None:
                continue

            img_name = img_path.name
            gps = flight_log.get(img_name, {})

            results = self.detector(frame, conf=0.4)
            img_detections = self._process_detections(
                results, frame, gps, str(img_path)
            )
            all_detections.extend(img_detections)

        # Агрегируем по типам и зонам
        summary = self._aggregate_detections(all_detections)
        return summary

    def _process_detections(self, results, frame: np.ndarray,
                              gps: dict, img_path: str) -> list:
        detections = []
        for box in results[0].boxes:
            cls = self.detector.model.names[int(box.cls)]
            bbox = list(map(int, box.xyxy[0]))
            conf = float(box.conf)

            # Пиксели → GPS-координаты через проективное преобразование
            cx = (bbox[0] + bbox[2]) // 2
            cy = (bbox[1] + bbox[3]) // 2
            lat, lon = self._px_to_gps(cx, cy, gps, frame.shape)

            detections.append({
                'class': cls,
                'confidence': conf,
                'bbox': bbox,
                'lat': lat,
                'lon': lon,
                'source_image': img_path
            })

        return detections

    def _px_to_gps(self, px: int, py: int, gps: dict,
                    frame_shape: tuple) -> tuple:
        """
        Преобразование пикселей в GPS.
        Упрощённо: для точной геопривязки нужен RTK-GPS + GCP.
        """
        if not gps:
            return None, None

        fov = gps.get('fov', 84)  # DJI Phantom 4: 84°
        altitude = gps.get('altitude', 50)  # метры
        drone_lat = gps.get('lat', 0)
        drone_lon = gps.get('lon', 0)

        h, w = frame_shape[:2]
        # Ground footprint
        gsd = (altitude * 2 * np.tan(np.radians(fov/2))) / w
        dx_m = (px - w/2) * gsd
        dy_m = (h/2 - py) * gsd

        lat = drone_lat + dy_m / 111320
        lon = drone_lon + dx_m / (111320 * np.cos(np.radians(drone_lat)))

        return lat, lon

    def _aggregate_detections(self, detections: list) -> dict:
        summary = {
            'equipment_count': {},
            'workers_on_site': 0,
            'zones_progress': {},
            'timestamp': datetime.now().isoformat()
        }

        for det in detections:
            cls = det['class']
            if cls in self.equipment_classes:
                summary['equipment_count'][cls] = \
                    summary['equipment_count'].get(cls, 0) + 1
            elif cls == 'worker':
                summary['workers_on_site'] += 1

        return summary

Ортофото и 3D-модель прогресса

class SiteProgressAnalyzer:
    """Сравнение ортофото текущей недели с предыдущей"""

    def compare_orthomap(self, current_path: str,
                          previous_path: str) -> dict:
        current = cv2.imread(current_path)
        previous = cv2.imread(previous_path)

        if current.shape != previous.shape:
            previous = cv2.resize(previous, (current.shape[1], current.shape[0]))

        # Semantic segmentation для классификации зон
        # Классы: bare_ground, formwork, concrete, rebar, masonry,
        #         roofing, scaffolding, finished
        current_seg = self._segment_site(current)
        previous_seg = self._segment_site(previous)

        # Прогресс по зонам
        progress_by_zone = {}
        for zone_name, zone_bbox in self.site_zones.items():
            x1, y1, x2, y2 = zone_bbox
            curr_zone = current_seg[y1:y2, x1:x2]
            prev_zone = previous_seg[y1:y2, x1:x2]

            # Изменения в классе стадии строительства
            progress_score = self._compute_progress_delta(curr_zone, prev_zone)
            progress_by_zone[zone_name] = progress_score

        return progress_by_zone

Автоматический недельный отчёт

• Тепловая карта активности: где идут работы, где простой
• Инвентаризация техники: кран №3 простаивает 3-й день
• Прогресс по секциям: секция B — 78% от плана, отставание 5 дней
• Нарушения безопасности: фото + GPS-координаты