Upload 3 files

ctranspath.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:7c998680060c8743551a412583fac689db43cec07053b72dfec6dcd810113539
+size 111292151

utils_color_norm.py

@@ -0,0 +1,270 @@
+"""
+Stain normalization based on the method of:
+
+M. Macenko et al., ‘A method for normalizing histology slides for quantitative analysis’,
+in 2009 IEEE International Symposium on Biomedical Imaging: From Nano to Macro, 2009, pp. 1107–1110.
+
+Uses the spams package:
+
+http://spams-devel.gforge.inria.fr/index.html
+(https://anaconda.org/conda-forge/python-spams
+https://pypi.org/project/spams/)
+
+"""
+
+from __future__ import division
+
+import numpy as np
+import cv2 as cv
+import spams
+import matplotlib.pyplot as plt
+
+##================================================================================================
+
+def read_image(path):
+    """
+    Read an image to RGB uint8
+    :param path:
+    :return:
+    """
+    im = cv.imread(path)
+    im = cv.cvtColor(im, cv.COLOR_BGR2RGB)
+    return im
+
+def show_colors(C):
+    """
+    Shows rows of C as colors (RGB)
+    :param C:
+    :return:
+    """
+    n = C.shape[0]
+    for i in range(n):
+        if C[i].max() > 1.0:
+            plt.plot([0, 1], [n - 1 - i, n - 1 - i], c=C[i] / 255, linewidth=20)
+        else:
+            plt.plot([0, 1], [n - 1 - i, n - 1 - i], c=C[i], linewidth=20)
+        plt.axis('off')
+        plt.axis([0, 1, -1, n])
+
+
+def show(image, now=True, fig_size=(10, 10)):
+    """
+    Show an image (np.array).
+    Caution! Rescales image to be in range [0,1].
+    :param image:
+    :param now:
+    :param fig_size:
+    :return:
+    """
+    image = image.astype(np.float32)
+    m, M = image.min(), image.max()
+    if fig_size != None:
+        plt.rcParams['figure.figsize'] = (fig_size[0], fig_size[1])
+    plt.imshow((image - m) / (M - m), cmap='gray')
+    plt.axis('off')
+    if now == True:
+        plt.show()
+
+
+def build_stack(tup):
+    """
+    Build a stack of images from a tuple of images
+    :param tup:
+    :return:
+    """
+    N = len(tup)
+    if len(tup[0].shape) == 3:
+        h, w, c = tup[0].shape
+        stack = np.zeros((N, h, w, c))
+    if len(tup[0].shape) == 2:
+        h, w = tup[0].shape
+        stack = np.zeros((N, h, w))
+    for i in range(N):
+        stack[i] = tup[i]
+    return stack
+
+
+def patch_grid(ims, width=5, sub_sample=None, rand=False, save_name=None):
+    """
+    Display a grid of patches
+    :param ims:
+    :param width:
+    :param sub_sample:
+    :param rand:
+    :return:
+    """
+    N0 = np.shape(ims)[0]
+    if sub_sample is None:
+        N = N0
+        stack = ims
+    elif sub_sample != None and rand == False:
+        N = sub_sample
+        stack = ims[:N]
+    elif sub_sample != None and rand == True:
+        N = sub_sample
+        idx = np.random.choice(range(N), sub_sample, replace=False)
+        stack = ims[idx]
+    height = np.ceil(float(N) / width).astype(np.uint16)
+    plt.rcParams['figure.figsize'] = (18, (18 / width) * height)
+    plt.figure()
+    for i in range(N):
+        plt.subplot(height, width, i + 1)
+        im = stack[i]
+        show(im, now=False, fig_size=None)
+    if save_name != None:
+        plt.savefig(save_name)
+    plt.show()
+
+######################################
+
+def standardize_brightness(I):
+    """
+    :param I:
+    :return:
+    """
+    p = np.percentile(I, 95)
+    return np.clip(I * 255.0 / p, 0, 255).astype(np.uint8)
+
+
+def remove_zeros(I):
+    """
+    Remove zeros, replace with 1's.
+    :param I: uint8 array
+    :return:
+    """
+    mask = (I == 0)
+    I[mask] = 1
+    return I
+
+
+def RGB_to_OD(I):
+    """
+    Convert from RGB to optical density
+    :param I:
+    :return:
+    """
+    I = remove_zeros(I)
+    return -1 * np.log(I / 255)
+
+
+def OD_to_RGB(OD):
+    """
+    Convert from optical density to RGB
+    :param OD:
+    :return:
+    """
+    return (255 * np.exp(-1 * OD)).astype(np.uint8)
+
+
+def normalize_rows(A):
+    """
+    Normalize rows of an array
+    :param A:
+    :return:
+    """
+    return A / np.linalg.norm(A, axis=1)[:, None]
+
+
+def notwhite_mask(I, thresh=0.8):
+    """
+    Get a binary mask where true denotes 'not white'
+    :param I:
+    :param thresh:
+    :return:
+    """
+    I_LAB = cv.cvtColor(I, cv.COLOR_RGB2LAB)
+    L = I_LAB[:, :, 0] / 255.0
+    return (L < thresh)
+
+
+def sign(x):
+    """
+    Returns the sign of x
+    :param x:
+    :return:
+    """
+    if x > 0:
+        return +1
+    elif x < 0:
+        return -1
+    elif x == 0:
+        return 0
+
+def get_concentrations(I, stain_matrix, lamda=0.01):
+    """
+    Get concentrations, a npix x 2 matrix
+    :param I:
+    :param stain_matrix: a 2x3 stain matrix
+    :return:
+    """
+    OD = RGB_to_OD(I).reshape((-1, 3))
+    return spams.lasso(OD.T, D=stain_matrix.T, mode=2, lambda1=lamda, pos=True).toarray().T
+
+##================================================================================================
+def get_stain_matrix(I, beta=0.15, alpha=1):
+    """
+    Get stain matrix (2x3)
+    :param I:
+    :param beta:
+    :param alpha:
+    :return:
+    """
+    OD = RGB_to_OD(I).reshape((-1, 3))
+    OD = OD[(OD > beta).any(axis=1), :]
+    _, V = np.linalg.eigh(np.cov(OD, rowvar=False))
+    V = V[:, [2, 1]]
+    if V[0, 0] < 0:
+        V[:, 0] *= -1
+    if V[0, 1] < 0:
+        V[:, 1] *= -1
+    That = np.dot(OD, V)
+    phi = np.arctan2(That[:, 1], That[:, 0])
+    minPhi = np.percentile(phi, alpha)
+    maxPhi = np.percentile(phi, 100 - alpha)
+    v1 = np.dot(V, np.array([np.cos(minPhi), np.sin(minPhi)]))
+    v2 = np.dot(V, np.array([np.cos(maxPhi), np.sin(maxPhi)]))
+    if v1[0] > v2[0]:
+        HE = np.array([v1, v2])
+    else:
+        HE = np.array([v2, v1])
+    return normalize_rows(HE)
+
+##================================================================================================
+
+class macenko_normalizer(object):
+    """
+    A stain normalization object
+    """
+
+    def __init__(self):
+        self.stain_matrix_target = np.array(
+            [[0.5626, 0.2159], [0.7201, 0.8012], [0.4062, 0.5581]], dtype=np.float32).T
+        self.target_concentrations = None
+
+    def fit(self, target):
+        target = standardize_brightness(target)
+        self.stain_matrix_target = get_stain_matrix(target)
+        self.target_concentrations = get_concentrations(target, self.stain_matrix_target)
+
+    def target_stains(self):
+        return OD_to_RGB(self.stain_matrix_target)
+
+    def transform(self, I):
+        I = standardize_brightness(I)
+        stain_matrix_source = get_stain_matrix(I)
+        source_concentrations = get_concentrations(I, stain_matrix_source)
+        maxC_source = np.percentile(source_concentrations, 99, axis=0).reshape((1, 2))
+        maxC_target = np.array([1.9705, 1.0308], dtype=float).reshape((1, 2))
+        source_concentrations *= maxC_target / maxC_source
+        return (255* np.exp(-1*np.dot(source_concentrations, self.stain_matrix_target).reshape(I.shape))).astype(np.uint8)
+
+    def hematoxylin(self, I):
+        I = standardize_brightness(I)
+        h, w, c = I.shape
+        stain_matrix_source = get_stain_matrix(I)
+        source_concentrations = get_concentrations(I, stain_matrix_source)
+        H = source_concentrations[:, 0].reshape(h, w)
+        H = np.exp(-1 * H)
+        return H
+
+

utils_preprocessing.py

@@ -0,0 +1,335 @@
+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