utils_preprocessing.py

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import openslide
from PIL import Image

import cv2
import torch
from torch import nn
import torchvision
#from torchvision.models import resnet50
import torchvision.transforms as transforms
# from transformers import ViTImageProcessor, ViTModel
# from timm.models.vision_transformer import VisionTransformer
# import timm
from ctrans_model import CTransPath

import utils_color_norm
color_norm = utils_color_norm.macenko_normalizer()

## check available device
device = (torch.device('cuda') if torch.cuda.is_available() else torch.device('cpu'))
print("device:", device)
##======================================================================================================
class resnet50_feature_extraction(nn.Module):
    def __init__(self, model_type="load_from_saved_file"):
        super().__init__()

        if model_type == "load_from_internet":
            self.resnet = resnet50(weights=torchvision.models.ResNet50_Weights.IMAGENET1K_V2)
        elif model_type == "load_from_saved_file":
            self.resnet = resnet50(weights=None)
        else:
            print("cannot find model_type can only be load_from_internet or load_from_saved_file")

        
    def forward(self, x):
        x = self.resnet.conv1(x)
        x = self.resnet.bn1(x)
        x = self.resnet.relu(x)
        x = self.resnet.maxpool(x)

        x = self.resnet.layer1(x)
        x = self.resnet.layer2(x)
        x = self.resnet.layer3(x)
        x = self.resnet.layer4(x)

        x = self.resnet.avgpool(x)
        x = torch.flatten(x, 1)
        return x

##======================================================================================================
def evaluate_tile_edge(img_np, edge_mag_thrsh, edge_fraction_thrsh):

    select = 1  ## initial value
    
    #img_np = np.array(img_RGB)
    tile_size = img_np.shape[0]
        
    ##---------------------------------------
    ## 0) exclude if edge_mag > 0.5
    img_gray=cv2.cvtColor(img_np, cv2.COLOR_RGB2GRAY)

    # Remove noise using a Gaussian filter
    #img_gray = cv2.GaussianBlur(img_gray, (5,5), 0)

    sobelx = cv2.Sobel(img_gray, cv2.CV_32F, 1, 0)
    sobely = cv2.Sobel(img_gray, cv2.CV_32F, 0, 1)

    sobelx1 = cv2.convertScaleAbs(sobelx)
    sobely1 = cv2.convertScaleAbs(sobely)

    mag = cv2.addWeighted(sobelx1, 0.5, sobely1, 0.5, 0)

    unique, counts = np.unique(mag, return_counts=True)

    edge_mag = counts[np.argwhere(unique < edge_mag_thrsh)].sum()/(tile_size*tile_size)

    if edge_mag > edge_fraction_thrsh:
        select = 0
    
    return select
        
##======================================================================================================
def evaluate_tile_color(img_np,black_thrsh,black_pct_thrsh,blue_level_thrsh,red_level_thrsh,
                        H_min,H_max,S_min,S_max,V_min,V_max,select):

    #img_np = np.array(img_RGB)

    L, A, B = cv2.split(cv2.cvtColor((img_np), cv2.COLOR_RGB2LAB)) 

    ##---------------------------------------
    ## 1) remove if percentage of black spot > 0.01
    black_pct = np.mean(L < black_thrsh)
    if black_pct > black_pct_thrsh:
        select = 0
        return select
    ##---------------------------------------
    ## 2) remove if too blue (heavy mark), or too red (blood)
    red,green,blue = np.mean(img_np[:,:,0]),np.mean(img_np[:,:,1]),np.mean(img_np[:,:,2])
    blue_level = blue/(red + green)
    blue_level2 = blue*blue_level

    if blue_level2 > blue_level_thrsh:
        select = 0
        return select

    ##---
    red_level = red/(green + blue)
    red_level2 = red*red_level

    if red_level2 > red_level_thrsh:
        select = 0
        return select

    ##---------------------------------------
    ## 3) remove if tile has the same color suggested (using color detection)
    H,S,V = cv2.split(cv2.cvtColor(img_np, cv2.COLOR_RGB2HSV))    
    H,S,V = np.mean(H),np.mean(S),np.mean(V)

    if (H_min <= H and H <= H_max and S_min <= S and S <= S_max and V_min <= V and V <= V_max):
        select = 0
        return select
    
    return select

##================================================================================================
def slide2tiles(path2slide, slide_name, slide_file_name, mag_assumed, mag_selected, tile_size, 
    mask_downsampling,edge_mag_thrsh,edge_fraction_thrsh,save_tile_file,
    path2mask,path2coordinates):
    
    ## open slide
    slide = openslide.OpenSlide(f"{path2slide}{slide_file_name}")

    ## magnification max
    if openslide.PROPERTY_NAME_OBJECTIVE_POWER in slide.properties:
        mag_max = slide.properties[openslide.PROPERTY_NAME_OBJECTIVE_POWER]
        print("mag_max:", mag_max)
        mag_original = mag_max
    else:
        print("[WARNING] mag not found, assuming: {mag_assumed}")
        mag_max = mag_assumed
        mag_original = 0

    ## downsample_level
    downsampling = int(int(mag_max)/mag_selected)
    print(f"downsampling: {downsampling}")


    mask_tile_size = int(np.ceil(tile_size/mask_downsampling))
    #print("mask_tile_size:", mask_tile_size)

    ##------------------------------------------------------------------
    ## slide partitioning
    ## slide size at largest level (level=0)
    px0, py0 = slide.level_dimensions[0]
    tile_size0 = int(tile_size*downsampling)
    print(f"px0: {px0}, py0: {py0}, tile_size0: {tile_size0}")

    n_rows,n_cols = int(py0/tile_size0), int(px0/tile_size0)
    print(f"n_rows: {n_rows}, n_cols: {n_cols}")

    n_tiles_total = n_rows*n_cols
    print(f"n_tiles_total: {n_tiles_total}")

    ##-----------------------
    img_mask = np.full((int((n_rows)*mask_tile_size),int((n_cols)*mask_tile_size),3),255).astype(np.uint8)
    mask = np.full((int((n_rows)*mask_tile_size),int((n_cols)*mask_tile_size),3),255).astype(np.uint8)

    i_tile = 0
    tiles_list = []

    col_list = []
    row_list = []
    i_tile_list = []
    for row in range(n_rows):
        print(f"row: {row}/{n_rows}")
        for col in range(n_cols):

            tile = slide.read_region((col*tile_size0, row*tile_size0),\
                                     level=0, size=[tile_size0, tile_size0]).convert("RGB") ## RGBA image --> RGB

            if tile.size[0] == tile_size0 and tile.size[1] == tile_size0:
                # downsample to target tile size
                tile = tile.resize((tile_size, tile_size))

                mask_tile = np.array(tile.resize((mask_tile_size, mask_tile_size)))

                img_mask[int(row*mask_tile_size):int((row+1)*mask_tile_size),\
                         int(col*mask_tile_size):int((col+1)*mask_tile_size),:] = mask_tile

                tile = np.array(tile)
                #print(tile.shape)

                ## evaluate tile
                select = evaluate_tile_edge(tile, edge_mag_thrsh, edge_fraction_thrsh)

                if select == 1:
                    ## 2022.09.08: color normalization:
                    tile_norm = Image.fromarray(color_norm.transform(tile))

                    mask_tile_norm = np.array(tile_norm.resize((mask_tile_size, mask_tile_size)))

                    mask[int(row*mask_tile_size):int((row+1)*mask_tile_size),\
                         int(col*mask_tile_size):int((col+1)*mask_tile_size),:] = mask_tile_norm    

                    #tiles_list.append(np.array(tile_norm).astype(np.uint8))
                    tiles_list.append(tile_norm)

                    if save_tile_file:
                        tile_name = "tile_" + str(row).zfill(5)+"_" + str(col).zfill(5) + "_" \
                                 + str(i_tile).zfill(5) + "_" + str(downsampling).zfill(3)

                        tile_norm.save(f"{tile_folder}/{tile_name}.png")

                    ## 2023.05.27: tile information
                    col_list.append(col)
                    row_list.append(row)
                    i_tile_list.append(i_tile)

            i_tile += 1

    ## 2023.05.27: save tile coordinates:
    downsampling_list = [downsampling]*len(row_list)
    df_coordinates = pd.DataFrame({"row": row_list, "col": col_list, "i_tile": i_tile_list, "downsampling": downsampling})
    df_coordinates.to_csv(f"{path2coordinates}{slide_name}.csv", index_label="tile_idx")
    
    ##====================================================================================================== 
    ## plot: draw color lines on the mask
    line_color = [0,255,0]

    n_tiles = len(tiles_list)

    img_mask[:,::mask_tile_size,:] = line_color
    img_mask[::mask_tile_size,:,:] = line_color
    mask[:,::mask_tile_size,:] = line_color
    mask[::mask_tile_size,:,:] = line_color

    fig, ax = plt.subplots(1,2,figsize=(30,15))
    ax[0].imshow(img_mask)
    ax[1].imshow(mask)

    ax[0].set_title(f"{slide_name}, mag_original: {mag_original}, mag_assumed: {mag_assumed}")
    ax[1].set_title(f"n_rows: {n_rows}, n_cols: {n_cols}, n_tiles_total: {n_tiles_total}, n_tiles_selected: {n_tiles}")

    plt.tight_layout(h_pad=0.4, w_pad=0.5)
    plt.savefig(f"{path2mask}{slide_name}.pdf", format="pdf", dpi=50)
    plt.close()

    img_mask = 0 ; mask = 0

    print("completed cleaning")
    
    return tiles_list

##======================================================================================================
def tile_transform(tiles_list, data_mean, data_std):
    data_transform = transforms.Compose([transforms.Resize(224),
                                     transforms.ToTensor(),
                                     transforms.Normalize(mean=data_mean, std=data_std)])

    ## data transform:
    n_tiles = len(tiles_list)
    print("n_tiles:", n_tiles)

    tiles = []
    for i in range(n_tiles):
        tiles.append(data_transform(tiles_list[i]).unsqueeze(0))
    tiles = torch.cat(tiles, dim=0)
    print("tiles.shape:", tiles.shape)
    tiles_list = 0
    
    return tiles  ## [n_tiles,3,224,224]

##================================================================================================
def tiles2features(tiles_list, model_name, batch_size):

    ##----------------------------------------
    ## model config
    if model_name == "vit":
        path2model = "../vit-base-patch16-224-in21k"
        model = ViTModel.from_pretrained(path2model)
        model.to(device)
        data_mean=[0.5, 0.5, 0.5] ; data_std = [0.5, 0.5, 0.5]

    if model_name == "dino":
        path2model = "../dino_vit_small_patch16_ep200.pt"
        model = VisionTransformer(img_size=224, patch_size=16, 
                                  embed_dim=384, num_heads=6, num_classes=0)
        model.to(device)
        model.load_state_dict(torch.load(path2model,map_location=device))
        data_mean=[0.485, 0.456, 0.406] ; data_std = [0.229, 0.224, 0.225]

    if model_name == "ctrans":
        path2model = "../ctranspath.pth"
        model = CTransPath(num_classes=0)
        model.to(device)
        model.load_state_dict(torch.load(path2model)['model'])
        model = model.cpu()
        data_mean=[0.485, 0.456, 0.406] ; data_std = [0.229, 0.224, 0.225]

    model.eval()
    
    ## tile transform
    tiles = tile_transform(tiles_list, data_mean, data_std)

    ## extract features from tiles
    n_tiles = tiles.shape[0]
    features = []
    for idx_start in range(0, n_tiles, batch_size):
        idx_end = idx_start + min(batch_size, n_tiles - idx_start)

        with torch.no_grad():
            y = model(tiles[idx_start:idx_end])

        if model_name == "vit":
            y = y.last_hidden_state[:, 0]

        features.append(y.detach().cpu().numpy())

    features = np.concatenate(features)
    print("features.shape:", features.shape)
    
    return features
##================================================================================================
def init_random_seed(random_seed=42):
    # Python RNG
    np.random.seed(random_seed)

    # Torch RNG
    torch.manual_seed(random_seed)
    torch.cuda.manual_seed(random_seed)
    torch.cuda.manual_seed_all(random_seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False