AI for Satellite Imagery Analysis (Remote Sensing)

Разработка AI для анализа спутниковых снимков (Remote Sensing)

Спутниковые снимки дают глобальный охват с периодичностью от нескольких часов (Planet Labs, SkySat) до нескольких дней (Sentinel-2, Landsat). Задачи AI в дистанционном зондировании: семантическая сегментация земного покрова, детекция объектов (суда, самолёты, здания, транспорт), мониторинг изменений, оценка ущерба после катастроф. Ключевые сложности — геопространственная привязка, многоканальные данные (до 13 каналов Sentinel-2), разномасштабность объектов.

Работа с геопространственными данными

import rasterio
import numpy as np
from rasterio.warp import calculate_default_transform, reproject, Resampling

class SatelliteImageLoader:
    def load_sentinel2(self, safe_dir: str, resolution: int = 10) -> dict:
        """
        Загрузка Sentinel-2 L2A данных.
        Каналы: B02(Blue), B03(Green), B04(Red), B08(NIR) — 10м
                B05,B06,B07,B8A,B11,B12 — 20м
                B01,B09,B10 — 60м
        """
        import glob
        bands = {}
        band_files = glob.glob(f"{safe_dir}/**/*B0*.jp2", recursive=True)

        with rasterio.open(band_files[0]) as ref:
            self.transform = ref.transform
            self.crs = ref.crs
            self.profile = ref.profile

        for band_file in band_files:
            band_name = self._extract_band_name(band_file)
            with rasterio.open(band_file) as src:
                # Репроецирование всех каналов к единому разрешению
                data, _ = reproject(
                    source=rasterio.band(src, 1),
                    destination=np.zeros(
                        (self.profile['height'], self.profile['width']),
                        dtype=np.float32
                    ),
                    src_transform=src.transform,
                    src_crs=src.crs,
                    dst_transform=self.transform,
                    dst_crs=self.crs,
                    resampling=Resampling.bilinear
                )
                bands[band_name] = data / 10000.0  # масштабирование DN→reflectance

        return bands

    def compute_indices(self, bands: dict) -> dict:
        red, nir = bands.get('B04'), bands.get('B08')
        swir1 = bands.get('B11')
        return {
            'ndvi': (nir - red) / (nir + red + 1e-8),
            'ndwi': (bands['B03'] - nir) / (bands['B03'] + nir + 1e-8),  # вода
            'ndbi': (swir1 - nir) / (swir1 + nir + 1e-8),                # застройка
            'evi': 2.5 * (nir - red) / (nir + 6*red - 7.5*bands['B02'] + 1 + 1e-8)
        }

Семантическая сегментация земного покрова

import segmentation_models_pytorch as smp
import torch

class LandCoverSegmenter:
    CLASSES = [
        'urban', 'agriculture', 'forest', 'grassland',
        'wetland', 'water', 'barren', 'snow_ice'
    ]

    def __init__(self, checkpoint_path: str):
        # DeepLabV3+ с ResNet-101 — strong baseline для land cover
        self.model = smp.DeepLabV3Plus(
            encoder_name='resnet101',
            encoder_weights=None,
            in_channels=12,   # 12 каналов Sentinel-2
            classes=len(self.CLASSES)
        )
        self.model.load_state_dict(torch.load(checkpoint_path))
        self.model.eval()

    def segment_patch(self, patch: np.ndarray) -> np.ndarray:
        """patch: [12, 256, 256] нормализованный"""
        tensor = torch.from_numpy(patch).float().unsqueeze(0)
        with torch.no_grad():
            logits = self.model(tensor)
        return logits.argmax(dim=1).squeeze().numpy()

    def segment_large_image(self, bands: dict,
                              patch_size: int = 256,
                              overlap: int = 32) -> np.ndarray:
        """Сегментация большого изображения через скользящее окно"""
        H, W = next(iter(bands.values())).shape
        result = np.zeros((H, W), dtype=np.uint8)
        count = np.zeros((H, W), dtype=np.float32)

        stride = patch_size - overlap
        for y in range(0, H - patch_size + 1, stride):
            for x in range(0, W - patch_size + 1, stride):
                patch = np.stack([
                    bands[b][y:y+patch_size, x:x+patch_size]
                    for b in sorted(bands.keys())
                ])
                pred = self.segment_patch(patch)
                result[y:y+patch_size, x:x+patch_size] += pred
                count[y:y+patch_size, x:x+patch_size] += 1

        return (result / np.maximum(count, 1)).astype(np.uint8)

Детекция объектов на спутниковых снимках

Для детекции мелких объектов (суда 10–50 пикселей, автомобили 3–5 пикселей) стандартные детекторы теряют производительность. Специализированные подходы:

from ultralytics import YOLO

class SatelliteObjectDetector:
    def __init__(self, model_path: str):
        # YOLOv8x + Large Image Inference для HRO снимков
        self.model = YOLO(model_path)

    def detect_ships(self, image: np.ndarray,
                      tile_size: int = 640,
                      overlap: float = 0.2) -> list:
        """
        SAHI (Slicing Aided Hyper Inference) подход:
        нарезка на overlapping тайлы + NMS через всё изображение
        """
        from sahi import AutoDetectionModel
        from sahi.predict import get_sliced_prediction

        detection_model = AutoDetectionModel.from_pretrained(
            model_type='yolov8',
            model_path=self.model.ckpt_path,
            confidence_threshold=0.35,
            device='cuda'
        )

        result = get_sliced_prediction(
            image,
            detection_model,
            slice_height=tile_size,
            slice_width=tile_size,
            overlap_height_ratio=overlap,
            overlap_width_ratio=overlap
        )

        return [
            {
                'bbox': pred.bbox.to_xyxy(),
                'score': pred.score.value,
                'class': pred.category.name
            }
            for pred in result.object_prediction_list
        ]

Мониторинг изменений (Change Detection)

import torch.nn as nn

class SiameseChangeDetector(nn.Module):
    """
    Bi-temporal change detection: два снимка одной территории
    с разными датами → маска изменений
    """
    def __init__(self):
        super().__init__()
        import timm
        encoder = timm.create_model('resnet50', features_only=True, pretrained=True)
        self.encoder = encoder  # shared weights для обоих снимков

        self.change_head = nn.Sequential(
            nn.Conv2d(512 * 2, 256, 3, padding=1),
            nn.BatchNorm2d(256),
            nn.ReLU(),
            nn.Conv2d(256, 1, 1),
            nn.Sigmoid()
        )

    def forward(self, img_t1: torch.Tensor,
                img_t2: torch.Tensor) -> torch.Tensor:
        feat_t1 = self.encoder(img_t1)[-1]
        feat_t2 = self.encoder(img_t2)[-1]
        # Конкатенация + разность features
        diff = torch.cat([feat_t1 - feat_t2, feat_t1 * feat_t2], dim=1)
        return self.change_head(diff)

Публичные датасеты и бенчмарки