Upload model

config.json

@@ -2,6 +2,10 @@
   "architectures": [
     "PneumoniaModel"
   ],
+  "auto_map": {
+    "AutoConfig": "configuration.PneumoniaConfig",
+    "AutoModel": "modeling.PneumoniaModel"
+  },
   "backbone": "tf_efficientnetv2_s",
   "cls_dropout": 0.1,
   "cls_num_classes": 1,

configuration.py

@@ -0,0 +1,40 @@
+from transformers import PretrainedConfig
+from typing import List, Optional, Tuple
+
+
+class PneumoniaConfig(PretrainedConfig):
+    model_type = "pneumonia"
+
+    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 = 1,
+        cls_num_classes: int = 1,
+        in_chans: int = 1,
+        img_size: Tuple[int, int] = (512, 512),  # 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)

modeling.py

@@ -0,0 +1,95 @@
+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 PneumoniaConfig
+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 PneumoniaModel(PreTrainedModel):
+    config_class = PneumoniaConfig
+
+    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,
+            "cls": cls_logits
+        }
+        if return_logits:
+            return out
+        out["mask"] = out["mask"].sigmoid()
+        out["cls"] = out["cls"].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)))