Розробка 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-координати