AI-система мониторинга биоразнообразия

Разработка AI для мониторинга биоразнообразия

Мониторинг биоразнообразия традиционно опирался на ручные полевые наблюдения: 2–3 специалиста за сезон обследуют ограниченную территорию. Фотоловушки, дроны с мультиспектральными камерами, гидрофоны и акустические датчики генерируют терабайты данных в сутки — без AI обработать их невозможно. Ключевые задачи: Species identification (определение вида по фото/звуку), population estimation (подсчёт особей по аэрофотосъёмке), individual identification (re-ID конкретной особи для долгосрочного мониторинга), habitat change detection.

Детектор и классификатор диких животных

import numpy as np
import cv2
import torch
from ultralytics import YOLO
from torchvision import models, transforms
from PIL import Image
from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class WildlifeDetection:
    track_id: int
    species: str
    common_name_ru: str
    confidence: float
    bbox: list
    individual_id: Optional[str]   # re-ID если обучена модель
    behavior: Optional[str]        # resting / moving / feeding
    camera_id: str
    timestamp: float

class WildlifeMonitor:
    """
    Детекция и классификация диких животных по снимкам с фотоловушек.
    Датасеты:
    - iNaturalist (1.4M изображений, 5000+ видов) — для предобучения
    - LILA BC (camera trap images) — leopard, snow leopard, пр.
    - Snapshot Serengeti (22M изображений, 48 видов)
    - CaltechCameraTraps (20 видов, Северная Америка)
    Двухступенчатый пайплайн: YOLO detection → EfficientNet species classification.
    """
    def __init__(self, detector_path: str,
                  classifier_path: str,
                  species_vocab_path: str,
                  reid_model_path: Optional[str] = None,
                  device: str = 'cuda'):
        self.detector = YOLO(detector_path)
        self.device = device

        # Classifier: EfficientNet-B3 fine-tuned on iNaturalist
        with open(species_vocab_path) as f:
            vocab_data = json.load(f)
        self.species_list = vocab_data['species']
        self.species_ru = vocab_data.get('species_ru', {})

        self.classifier = models.efficientnet_b3(pretrained=False)
        n_classes = len(self.species_list)
        self.classifier.classifier[-1] = torch.nn.Linear(
            self.classifier.classifier[-1].in_features, n_classes
        )
        state = torch.load(classifier_path, map_location=device)
        self.classifier.load_state_dict(state)
        self.classifier = self.classifier.to(device).eval()

        self.classify_transform = transforms.Compose([
            transforms.Resize((300, 300)),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406],
                                  [0.229, 0.224, 0.225])
        ])

        # Re-ID (опционально): мегапиксельная модель для паттернов меха
        self.reid_model = None
        if reid_model_path:
            self.reid_model = torch.load(reid_model_path,
                                          map_location=device).eval()
        self._individual_gallery: dict[str, np.ndarray] = {}

    def process_camera_trap_image(self, image: np.ndarray,
                                   camera_id: str,
                                   timestamp: float) -> list[WildlifeDetection]:
        """
        Обработка снимка с фотоловушки.
        Фотоловушки: ночные (ИК) + дневные, разное разрешение.
        """
        # Улучшение ночного снимка
        enhanced = self._enhance_camera_trap(image)

        detections_raw = self.detector(
            enhanced, conf=0.25, verbose=False
        )
        results = []

        for box in detections_raw[0].boxes:
            x1, y1, x2, y2 = map(int, box.xyxy[0])
            track_id = int(box.id.item()) if box.id is not None else -1

            # Вырезать crop для классификации
            crop = enhanced[max(0,y1):y2, max(0,x1):x2]
            if crop.size == 0:
                continue

            species, conf = self._classify_species(crop)
            species_ru = self.species_ru.get(species, species)
            behavior = self._estimate_behavior(crop, box)

            # Re-ID
            individual_id = None
            if self.reid_model:
                individual_id = self._get_individual_id(crop, species)

            results.append(WildlifeDetection(
                track_id=track_id,
                species=species,
                common_name_ru=species_ru,
                confidence=round(conf, 3),
                bbox=[x1, y1, x2, y2],
                individual_id=individual_id,
                behavior=behavior,
                camera_id=camera_id,
                timestamp=timestamp
            ))

        return results

    @torch.no_grad()
    def _classify_species(self, crop: np.ndarray) -> tuple[str, float]:
        pil = Image.fromarray(cv2.cvtColor(crop, cv2.COLOR_BGR2RGB))
        tensor = self.classify_transform(pil).unsqueeze(0).to(self.device)
        logits = self.classifier(tensor)
        probs = torch.softmax(logits, dim=-1).squeeze()
        conf, idx = probs.max(dim=0)
        species = self.species_list[int(idx.item())]
        return species, float(conf.item())

    def _enhance_camera_trap(self, image: np.ndarray) -> np.ndarray:
        """Улучшение ИК-снимков фотоловушек"""
        # Проверка на ИК (низкая цветность)
        hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)
        saturation = float(np.mean(hsv[:, :, 1]))

        if saturation < 20:  # ИК снимок
            gray = cv2.cvtColor(image, cv2.COLOR_BGR2GRAY)
            clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8, 8))
            enhanced_gray = clahe.apply(gray)
            return cv2.cvtColor(enhanced_gray, cv2.COLOR_GRAY2BGR)

        # Дневной снимок с избыточной тенью
        lab = cv2.cvtColor(image, cv2.COLOR_BGR2LAB)
        clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8, 8))
        lab[:, :, 0] = clahe.apply(lab[:, :, 0])
        return cv2.cvtColor(lab, cv2.COLOR_LAB2BGR)

    def _estimate_behavior(self, crop: np.ndarray,
                            box) -> str:
        """Простая классификация поведения по bbox aspect ratio"""
        x1, y1, x2, y2 = map(int, box.xyxy[0])
        w, h = x2 - x1, y2 - y1
        aspect = w / max(h, 1)
        if aspect > 2.0:
            return 'resting'  # лежит горизонтально
        elif aspect < 0.6:
            return 'alert'    # стоит вертикально, голова поднята
        return 'moving'

    @torch.no_grad()
    def _get_individual_id(self, crop: np.ndarray,
                            species: str) -> Optional[str]:
        """Re-ID через embedding similarity (для леопардов, тигров)"""
        pil = Image.fromarray(cv2.cvtColor(crop, cv2.COLOR_BGR2RGB))
        transform = transforms.Compose([
            transforms.Resize((224, 224)),
            transforms.ToTensor(),
        ])
        tensor = transform(pil).unsqueeze(0).to(self.device)
        embedding = self.reid_model(tensor).squeeze().cpu().numpy()
        embedding /= np.linalg.norm(embedding) + 1e-8

        # Поиск в галерее
        best_match = None
        best_sim = 0.75  # порог

        gallery_key = f'{species}_'
        for ind_id, gallery_emb in self._individual_gallery.items():
            if not ind_id.startswith(gallery_key):
                continue
            sim = float(np.dot(embedding, gallery_emb))
            if sim > best_sim:
                best_sim = sim
                best_match = ind_id

        if best_match is None:
            # Новая особь
            new_id = f'{species}_{len(self._individual_gallery)+1:04d}'
            self._individual_gallery[new_id] = embedding
            return new_id
        return best_match


class AerialAnimalCounter:
    """
    Подсчёт животных на аэрофотоснимках (дрон/самолёт).
    Применение: мониторинг стад копытных, подсчёт пингвинов,
    инвентаризация морских котиков на лежбищах.
    SAHI обязателен для ортофото 50+ МПикс.
    """
    from sahi import AutoDetectionModel
    from sahi.predict import get_sliced_prediction

    def __init__(self, model_path: str, device: str = 'cuda'):
        from sahi import AutoDetectionModel
        self.sahi_model = AutoDetectionModel.from_pretrained(
            model_type='ultralytics',
            model_path=model_path,
            confidence_threshold=0.35,
            device=device
        )

    def count_herd(self, aerial_image: np.ndarray,
                    species_hint: str = 'ungulate') -> dict:
        from sahi.predict import get_sliced_prediction
        result = get_sliced_prediction(
            aerial_image, self.sahi_model,
            slice_height=640, slice_width=640,
            overlap_height_ratio=0.2, overlap_width_ratio=0.2
        )
        count = len(result.object_prediction_list)
        density = count / (aerial_image.shape[0] * aerial_image.shape[1] / 1e6)

        return {
            'species_hint': species_hint,
            'count': count,
            'density_per_km2': round(density * 1e6, 1),  # пикс² → км²
            'detections': [
                {'bbox': [p.bbox.minx, p.bbox.miny, p.bbox.maxx, p.bbox.maxy],
                 'conf': p.score.value}
                for p in result.object_prediction_list
            ]
        }