Duplicate from ianpan/chest-x-ray-basic

Co-authored-by: Ian Pan <ianpan@users.noreply.huggingface.co>

.gitattributes

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

README.md

@@ -0,0 +1,94 @@
+---
+library_name: transformers
+tags:
+- chest_x_ray
+- x_ray
+- medical_imaging
+- radiology
+- segmentation
+- classification
+- lungs
+- heart
+base_model:
+- timm/tf_efficientnetv2_s.in21k_ft_in1k
+pipeline_tag: image-segmentation
+---
+
+This model performs both segmentation and classification on chest radiographs (X-rays).
+The model uses a `tf_efficientnetv2_s` backbone with a U-Net decoder for segmentation and linear layer for classification.
+For frontal radiographs, the model segments the: 1) right lung, 2) left lung, and 3) heart. 
+The model also predicts the chest X-ray view (AP, PA, lateral), patient age, and patient sex.
+The [CheXpert](https://stanfordmlgroup.github.io/competitions/chexpert/) (small version) and [NIH Chest X-ray](https://nihcc.app.box.com/v/ChestXray-NIHCC) datasets were used to train the model.
+Segmentation masks were obtained from the CheXmask [dataset](https://physionet.org/content/chexmask-cxr-segmentation-data/0.4/) ([paper](https://www.nature.com/articles/s41597-024-03358-1)).
+The final dataset comprised 335,516 images from 96,385 patients and was split into 80% training/20% validation. A holdout test set was not used since minimal tuning was performed.
+The view classifier was trained only on CheXpert images (NIH images excluded from loss function), given that lateral radiographs are only present in CheXpert.
+This is to avoid unwanted bias in the model, which can occur if one class originates only from a single dataset.
+
+Validation performance as follows:
+```
+Segmentation (Dice similarity coefficient):
+  Right Lung: 0.957
+   Left Lung: 0.948
+       Heart: 0.943
+
+Age Prediction:
+  Mean Absolute Error: 5.25 years
+
+Classification:
+  View (AP, PA, lateral): 99.42% accuracy
+  Female: 0.999 AUC
+```
+
+To use the model:
+```
+import cv2
+import torch
+from transformers import AutoModel
+
+device = "cuda" if torch.cuda.is_available() else "cpu"
+
+model = AutoModel.from_pretrained("ianpan/chest-x-ray-basic", trust_remote_code=True)
+model = model.eval().to(device)
+img = cv2.imread(..., 0)
+x = model.preprocess(img) # only takes single image as input
+x = torch.from_numpy(x).unsqueeze(0).unsqueeze(0) # add channel, batch dims
+x = x.float()
+
+with torch.inference_mode():
+  out = model(x.to(device))
+```
+
+The output is a dictionary which contains 4 keys:
+* `mask` has 3 channels containing the segmentation masks. Take the argmax over the channel dimension to create a single image mask (i.e., `out["mask"].argmax(1)`): 1 = right lung, 2 = left lung, 3 = heart.
+* `age`, in years.
+* `view`, with 3 classes for each possible view. Take the argmax to select the predicted view (i.e., `out["view"].argmax(1)`): 0 = AP, 1 = PA, 2 = lateral.
+* `female`, binarize with `out["female"] >= 0.5`. 
+
+You can use the segmentation mask to crop the region containing the lungs from the rest of the X-ray.
+You can also calculate the [cardiothoracic ratio (CTR)](https://radiopaedia.org/articles/cardiothoracic-ratio?lang=us) using this function:
+```
+import numpy as np
+
+def calculate_ctr(mask): # single mask with dims (height, width)
+    lungs = np.zeros_like(mask)
+    lungs[mask == 1] = 1
+    lungs[mask == 2] = 1
+    heart = (mask == 3).astype("int")
+    y, x = np.stack(np.where(lungs == 1))
+    lung_min = x.min()
+    lung_max = x.max()
+    y, x = np.stack(np.where(heart == 1))    
+    heart_min = x.min()
+    heart_max = x.max()
+    lung_range = lung_max - lung_min
+    heart_range = heart_max - heart_min
+    return heart_range / lung_range
+```
+
+If you have `pydicom` installed, you can also load a DICOM image directly:
+```
+img = model.load_image_from_dicom(path_to_dicom)
+```
+ 
+This model is for demonstration and research purposes only and has NOT been approved by any regulatory agency for clinical use. 
+The user assumes any and all responsibility regarding their own use of this model and its outputs.

config.json

@@ -0,0 +1,41 @@
+{
+  "architectures": [
+    "CXRModel"
+  ],
+  "auto_map": {
+    "AutoConfig": "configuration.CXRConfig",
+    "AutoModel": "modeling.CXRModel"
+  },
+  "backbone": "tf_efficientnetv2_s",
+  "cls_dropout": 0.1,
+  "cls_num_classes": 5,
+  "decoder_attention_type": null,
+  "decoder_center_block": false,
+  "decoder_channels": [
+    256,
+    128,
+    64,
+    32,
+    16
+  ],
+  "decoder_n_blocks": 5,
+  "decoder_norm_layer": "bn",
+  "encoder_channels": [
+    24,
+    48,
+    64,
+    160,
+    256
+  ],
+  "feature_dim": 256,
+  "img_size": [
+    320,
+    320
+  ],
+  "in_chans": 1,
+  "model_type": "cxr_basic",
+  "seg_dropout": 0.1,
+  "seg_num_classes": 4,
+  "torch_dtype": "float32",
+  "transformers_version": "4.47.0"
+}

configuration.py

@@ -0,0 +1,40 @@
+from transformers import PretrainedConfig
+from typing import List, Optional, Tuple
+
+
+class CXRConfig(PretrainedConfig):
+    model_type = "cxr_basic"
+
+    def __init__(
+        self,
+        backbone: str = "tf_efficientnetv2_s",
+        feature_dim: int = 256,
+        seg_dropout: float = 0.1,
+        cls_dropout: float = 0.1,
+        seg_num_classes: int = 4,
+        cls_num_classes: int = 5,
+        in_chans: int = 1,
+        img_size: Tuple[int, int] = (320, 320),  # height, width
+        decoder_n_blocks: int = 5,
+        decoder_channels: List[int] = [256, 128, 64, 32, 16],
+        encoder_channels: List[int] = [24, 48, 64, 160, 256],
+        decoder_center_block: bool = False,
+        decoder_norm_layer: str = "bn",
+        decoder_attention_type: Optional[str] = None,
+        **kwargs,
+    ):
+        self.backbone = backbone
+        self.feature_dim = feature_dim
+        self.seg_dropout = seg_dropout
+        self.cls_dropout = cls_dropout
+        self.seg_num_classes = seg_num_classes
+        self.cls_num_classes = cls_num_classes
+        self.in_chans = in_chans
+        self.img_size = img_size
+        self.decoder_n_blocks = decoder_n_blocks
+        self.decoder_channels = decoder_channels
+        self.encoder_channels = encoder_channels
+        self.decoder_center_block = decoder_center_block
+        self.decoder_norm_layer = decoder_norm_layer
+        self.decoder_attention_type = decoder_attention_type
+        super().__init__(**kwargs)

model.safetensors

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:213f9a181a0ac6b49999e4f63dc5dbc6167e719ae7c72da5bc682a04365f39b1
+size 89078700

modeling.py

@@ -0,0 +1,98 @@
+import albumentations as A
+import torch
+import torch.nn as nn
+
+from numpy.typing import NDArray
+from transformers import PreTrainedModel
+from timm import create_model
+from typing import Optional
+from .configuration import CXRConfig
+from .unet import UnetDecoder, SegmentationHead
+
+_PYDICOM_AVAILABLE = False
+try:
+    from pydicom import dcmread
+    from pydicom.pixels import apply_voi_lut
+
+    _PYDICOM_AVAILABLE = True
+except ModuleNotFoundError:
+    pass
+
+
+class CXRModel(PreTrainedModel):
+    config_class = CXRConfig
+
+    def __init__(self, config):
+        super().__init__(config)
+        self.encoder = create_model(
+            model_name=config.backbone,
+            features_only=True,
+            pretrained=False,
+            in_chans=config.in_chans,
+        )
+        self.decoder = UnetDecoder(
+            decoder_n_blocks=config.decoder_n_blocks,
+            decoder_channels=config.decoder_channels,
+            encoder_channels=config.encoder_channels,
+            decoder_center_block=config.decoder_center_block,
+            decoder_norm_layer=config.decoder_norm_layer,
+            decoder_attention_type=config.decoder_attention_type,
+        )
+        self.img_size = config.img_size
+        self.segmentation_head = SegmentationHead(
+            in_channels=config.decoder_channels[-1],
+            out_channels=config.seg_num_classes,
+            size=self.img_size,
+        )
+        self.pooling = nn.AdaptiveAvgPool2d(1)
+        self.dropout = nn.Dropout(p=config.cls_dropout)
+        self.classifier = nn.Linear(config.feature_dim, config.cls_num_classes)
+
+    def normalize(self, x: torch.Tensor) -> torch.Tensor:
+        # [0, 255] -> [-1, 1]
+        mini, maxi = 0.0, 255.0
+        x = (x - mini) / (maxi - mini)
+        x = (x - 0.5) * 2.0
+        return x
+
+    @staticmethod
+    def load_image_from_dicom(path: str) -> Optional[NDArray]:
+        if not _PYDICOM_AVAILABLE:
+            print("`pydicom` is not installed, returning None ...")
+            return None
+        dicom = dcmread(path)
+        arr = apply_voi_lut(dicom.pixel_array, dicom)
+        if dicom.PhotometricInterpretation == "MONOCHROME1":
+            # invert image if needed
+            arr = arr.max() - arr
+
+        arr = arr - arr.min()
+        arr = arr / arr.max()
+        arr = (arr * 255).astype("uint8")
+        return arr
+
+    def preprocess(self, x: NDArray) -> NDArray:
+        x = A.Resize(self.img_size[0], self.img_size[1], p=1)(image=x)["image"]
+        return x
+
+    def forward(self, x: torch.Tensor, return_logits: bool = False) -> torch.Tensor:
+        x = self.normalize(x)
+        features = self.encoder(x)
+        decoder_output = self.decoder(features)
+        logits = self.segmentation_head(decoder_output[-1])
+        b, n = features[-1].shape[:2]
+        features = self.pooling(features[-1]).reshape(b, n)
+        features = self.dropout(features)
+        cls_logits = self.classifier(features)
+        out = {
+            "mask": logits,
+            "age": cls_logits[:, 0].unsqueeze(1),
+            "view": cls_logits[:, 1:4],
+            "female": cls_logits[:, 4].unsqueeze(1),
+        }
+        if return_logits:
+            return out
+        out["mask"] = out["mask"].softmax(1)
+        out["view"] = out["view"].softmax(1)
+        out["female"] = out["female"].sigmoid()
+        return out

unet.py

@@ -0,0 +1,243 @@
+import torch
+import torch.nn as nn
+import torch.nn.functional as F
+
+from functools import partial
+from typing import List, Optional
+
+
+class Conv2dAct(nn.Sequential):
+    def __init__(
+        self,
+        in_channels: int,
+        out_channels: int,
+        kernel_size: int,
+        padding: int = 0,
+        stride: int = 1,
+        norm_layer: str = "bn",
+        num_groups: int = 32,  # for GroupNorm,
+        activation: str = "ReLU",
+        inplace: bool = True,  # for activation
+    ):
+        if norm_layer == "bn":
+            NormLayer = nn.BatchNorm2d
+        elif norm_layer == "gn":
+            NormLayer = partial(nn.GroupNorm, num_groups=num_groups)
+        else:
+            raise Exception(
+                f"`norm_layer` must be one of [`bn`, `gn`], got `{norm_layer}`"
+            )
+        super().__init__()
+        self.conv = nn.Conv2d(
+            in_channels,
+            out_channels,
+            kernel_size=kernel_size,
+            stride=stride,
+            padding=padding,
+            bias=False,
+        )
+        self.norm = NormLayer(out_channels)
+        self.act = getattr(nn, activation)(inplace=inplace)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.act(self.norm(self.conv(x)))
+
+
+class SCSEModule(nn.Module):
+    def __init__(
+        self,
+        in_channels: int,
+        reduction: int = 16,
+        activation: str = "ReLU",
+        inplace: bool = False,
+    ):
+        super().__init__()
+        self.cSE = nn.Sequential(
+            nn.AdaptiveAvgPool2d(1),
+            nn.Conv2d(in_channels, in_channels // reduction, 1),
+            getattr(nn, activation)(inplace=inplace),
+            nn.Conv2d(in_channels // reduction, in_channels, 1),
+        )
+        self.sSE = nn.Conv2d(in_channels, 1, 1)
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return x * self.cSE(x).sigmoid() + x * self.sSE(x).sigmoid()
+
+
+class Attention(nn.Module):
+    def __init__(self, name: str, **params):
+        super().__init__()
+
+        if name is None:
+            self.attention = nn.Identity(**params)
+        elif name == "scse":
+            self.attention = SCSEModule(**params)
+        else:
+            raise ValueError("Attention {} is not implemented".format(name))
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.attention(x)
+
+
+class DecoderBlock(nn.Module):
+    def __init__(
+        self,
+        in_channels: int,
+        skip_channels: int,
+        out_channels: int,
+        norm_layer: str = "bn",
+        activation: str = "ReLU",
+        attention_type: Optional[str] = None,
+    ):
+        super().__init__()
+        self.conv1 = Conv2dAct(
+            in_channels + skip_channels,
+            out_channels,
+            kernel_size=3,
+            padding=1,
+            norm_layer=norm_layer,
+            activation=activation,
+        )
+        self.attention1 = Attention(
+            attention_type, in_channels=in_channels + skip_channels
+        )
+        self.conv2 = Conv2dAct(
+            out_channels,
+            out_channels,
+            kernel_size=3,
+            padding=1,
+            norm_layer=norm_layer,
+            activation=activation,
+        )
+        self.attention2 = Attention(attention_type, in_channels=out_channels)
+
+    def forward(
+        self, x: torch.Tensor, skip: Optional[torch.Tensor] = None
+    ) -> torch.Tensor:
+        if skip is not None:
+            h, w = skip.shape[2:]
+            x = F.interpolate(x, size=(h, w), mode="nearest")
+            x = torch.cat([x, skip], dim=1)
+            x = self.attention1(x)
+        else:
+            x = F.interpolate(x, scale_factor=(2, 2), mode="nearest")
+        x = self.conv1(x)
+        x = self.conv2(x)
+        x = self.attention2(x)
+        return x
+
+
+class CenterBlock(nn.Sequential):
+    def __init__(
+        self,
+        in_channels: int,
+        out_channels: int,
+        norm_layer: str = "bn",
+        activation: str = "ReLU",
+    ):
+        conv1 = Conv2dAct(
+            in_channels,
+            out_channels,
+            kernel_size=3,
+            padding=1,
+            norm_layer=norm_layer,
+            activation=activation,
+        )
+        conv2 = Conv2dAct(
+            out_channels,
+            out_channels,
+            kernel_size=3,
+            padding=1,
+            norm_layer=norm_layer,
+            activation=activation,
+        )
+        super().__init__(conv1, conv2)
+
+
+class UnetDecoder(nn.Module):
+    def __init__(
+        self,
+        decoder_n_blocks: int,
+        decoder_channels: List[int],
+        encoder_channels: List[int],
+        decoder_center_block: bool = False,
+        decoder_norm_layer: str = "bn",
+        decoder_attention_type: Optional[str] = None,
+    ):
+        super().__init__()
+
+        self.decoder_n_blocks = decoder_n_blocks
+        self.decoder_channels = decoder_channels
+        self.encoder_channels = encoder_channels
+        self.decoder_center_block = decoder_center_block
+        self.decoder_norm_layer = decoder_norm_layer
+        self.decoder_attention_type = decoder_attention_type
+
+        if self.decoder_n_blocks != len(self.decoder_channels):
+            raise ValueError(
+                "Model depth is {}, but you provide `decoder_channels` for {} blocks.".format(
+                    self.decoder_n_blocks, len(self.decoder_channels)
+                )
+            )
+        # reverse channels to start from head of encoder
+        encoder_channels = encoder_channels[::-1]
+
+        # computing blocks input and output channels
+        head_channels = encoder_channels[0]
+        in_channels = [head_channels] + list(self.decoder_channels[:-1])
+        skip_channels = list(encoder_channels[1:]) + [0]
+        out_channels = self.decoder_channels
+
+        if self.decoder_center_block:
+            self.center = CenterBlock(
+                head_channels, head_channels, norm_layer=self.decoder_norm_layer
+            )
+        else:
+            self.center = nn.Identity()
+
+        # combine decoder keyword arguments
+        kwargs = dict(
+            norm_layer=self.decoder_norm_layer,
+            attention_type=self.decoder_attention_type,
+        )
+        blocks = [
+            DecoderBlock(in_ch, skip_ch, out_ch, **kwargs)
+            for in_ch, skip_ch, out_ch in zip(in_channels, skip_channels, out_channels)
+        ]
+        self.blocks = nn.ModuleList(blocks)
+
+    def forward(self, features: List[torch.Tensor]) -> torch.Tensor:
+        features = features[::-1]  # reverse channels to start from head of encoder
+
+        head = features[0]
+        skips = features[1:]
+
+        output = [self.center(head)]
+        for i, decoder_block in enumerate(self.blocks):
+            skip = skips[i] if i < len(skips) else None
+            output.append(decoder_block(output[-1], skip))
+
+        return output
+
+
+class SegmentationHead(nn.Module):
+    def __init__(
+        self,
+        in_channels: int,
+        out_channels: int,
+        size: int,
+        kernel_size: int = 3,
+        dropout: float = 0.0,
+    ):
+        super().__init__()
+        self.drop = nn.Dropout2d(p=dropout)
+        self.conv = nn.Conv2d(
+            in_channels, out_channels, kernel_size=kernel_size, padding=kernel_size // 2
+        )
+        if isinstance(size, (tuple, list)):
+            self.up = nn.Upsample(size=size, mode="bilinear")
+        else:
+            self.up = nn.Identity()
+
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        return self.up(self.conv(self.drop(x)))