AI Digital Pathology Analysis System

AI-система цифровой патологии: анализ WSI

Whole Slide Image (WSI) — гигапиксельный скан гистологического препарата: типичный размер 100 000 × 80 000 пикселей, 8 GB TIFF-файл. Это делает невозможным прямой инференс обычной сеткой — нужна специальная архитектура. Патоморфолог изучает препарат на разных увеличениях: x4 для общей структуры, x20 для клеточных деталей, x40 для ядер.

Тайлинг WSI и многомасштабный анализ

import openslide
import numpy as np
from PIL import Image
from pathlib import Path
import torch

class WSIProcessor:
    """
    Обработка Whole Slide Image через тайлинг.
    openslide поддерживает SVS, TIFF, NDPI, SCN форматы.
    """
    def __init__(self, wsi_path: str):
        self.slide = openslide.OpenSlide(wsi_path)
        self.dimensions = self.slide.dimensions     # (W, H) на level 0
        self.level_count = self.slide.level_count
        # Обычно: level 0 = x40, level 1 = x20, level 2 = x10, level 3 = x4

        mpp = float(self.slide.properties.get(
            openslide.PROPERTY_NAME_MPP_X, 0.25
        ))  # микрон/пиксель
        self.magnifications = {
            lvl: 0.25 / (mpp * self.slide.level_downsamples[lvl])
            for lvl in range(self.level_count)
        }

    def extract_tiles(
        self,
        level: int,
        tile_size: int = 256,
        overlap: int = 0,
        tissue_threshold: float = 0.5   # минимум % ткани в тайле
    ) -> list[dict]:
        """
        Нарезка WSI на тайлы заданного уровня.
        Пропускаем тайлы с преобладанием фона (стекло/воздух).
        """
        level_w, level_h = self.slide.level_dimensions[level]
        downsample = self.slide.level_downsamples[level]
        stride = tile_size - overlap
        tiles = []

        for y in range(0, level_h - tile_size + 1, stride):
            for x in range(0, level_w - tile_size + 1, stride):
                # Координаты в level 0 для openslide.read_region
                x0 = int(x * downsample)
                y0 = int(y * downsample)

                tile = self.slide.read_region(
                    (x0, y0), level, (tile_size, tile_size)
                ).convert('RGB')

                # Фильтр по содержанию ткани
                tile_arr = np.array(tile)
                if self._tissue_ratio(tile_arr) >= tissue_threshold:
                    tiles.append({
                        'image': tile,
                        'level': level,
                        'x': x, 'y': y,
                        'x0': x0, 'y0': y0
                    })

        return tiles

    def _tissue_ratio(self, tile: np.ndarray) -> float:
        """Отношение пикселей ткани к фону через HSV-маску"""
        hsv = np.array(Image.fromarray(tile).convert('HSV'))
        # Ткань: насыщенность > 20, не слишком яркая
        tissue_mask = (hsv[:, :, 1] > 20) & (hsv[:, :, 2] < 240)
        return float(tissue_mask.mean())

Multiple Instance Learning (MIL)

Стандартная постановка: есть slide-level метка (рак / норма), но нет аннотаций на уровне тайлов. MIL решает это: bag (WSI) = набор instances (тайлов), предсказание bag из instances.

import torch
import torch.nn as nn
import timm

class AttentionMIL(nn.Module):
    """
    Attention-based Multiple Instance Learning (Ilse et al., 2018).
    Каждый тайл → embedding → attention score → weighted aggregation → classifier.
    """
    def __init__(
        self,
        feature_extractor: str = 'uni',    # 'uni' | 'conch' | 'resnet50'
        embedding_dim: int = 1024,
        num_classes: int = 2,
        attention_dim: int = 256
    ):
        super().__init__()

        # Feature extractor — лучше использовать pathology-pretrained
        # UNI (MahmoodLab) или CONCH — обученные на миллионах патопатч
        if feature_extractor in ('uni', 'conch'):
            # Загружается через Hugging Face (требует accepted license)
            self.feature_extractor = self._load_pathology_foundation(
                feature_extractor
            )
        else:
            backbone = timm.create_model(
                feature_extractor, pretrained=True, num_classes=0
            )
            self.feature_extractor = backbone

        # Attention mechanism
        self.attention = nn.Sequential(
            nn.Linear(embedding_dim, attention_dim),
            nn.Tanh(),
            nn.Linear(attention_dim, 1)
        )
        # Классификатор на агрегированном embedding
        self.classifier = nn.Sequential(
            nn.Linear(embedding_dim, 256),
            nn.ReLU(),
            nn.Dropout(0.25),
            nn.Linear(256, num_classes)
        )

    def forward(
        self,
        tile_features: torch.Tensor    # (N, embedding_dim) — предвычисленные
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """
        Возвращает (logits, attention_scores).
        attention_scores — для визуализации внимания на WSI.
        """
        # Attention weights
        A = self.attention(tile_features)   # (N, 1)
        A = torch.softmax(A, dim=0)         # нормализация по тайлам

        # Взвешенная агрегация
        aggregated = (A * tile_features).sum(dim=0, keepdim=True)  # (1, dim)

        logits = self.classifier(aggregated)
        return logits, A.squeeze()

    def _load_pathology_foundation(self, name: str) -> nn.Module:
        # Placeholder — реальная загрузка через Hugging Face Hub
        raise NotImplementedError(
            f'Load {name} from Hugging Face: '
            f'MahmoodLab/{name}'
        )

Визуализация внимания на WSI

def create_attention_heatmap(
    slide: openslide.OpenSlide,
    tile_coords: list[tuple],    # [(x, y), ...] в пикселях level-0
    attention_scores: np.ndarray, # нормализованные attention weights
    tile_size: int,
    downsample: int = 32          # уменьшение для отображения
) -> np.ndarray:
    """
    Проецируем attention scores обратно на WSI → heatmap.
    """
    W, H = slide.dimensions
    heatmap = np.zeros((H // downsample, W // downsample), dtype=np.float32)

    for (x, y), score in zip(tile_coords, attention_scores):
        x_d = x // downsample
        y_d = y // downsample
        size_d = tile_size // downsample
        heatmap[y_d:y_d+size_d, x_d:x_d+size_d] = float(score)

    # Наложение на превью WSI
    thumbnail = np.array(
        slide.get_thumbnail((W // downsample, H // downsample))
    )
    heatmap_colored = cv2.applyColorMap(
        (heatmap * 255).astype(np.uint8), cv2.COLORMAP_JET
    )
    overlay = cv2.addWeighted(thumbnail, 0.6, heatmap_colored, 0.4, 0)
    return overlay

Сроки