Upload 21 files

app.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import torch
+import gradio as gr
+import numpy as np
+import torchvision.transforms as T
+
+from lib.framework import create_model
+from lib.options import ParamSet, _retrieve_parameter, _dispatch_by_group
+from lib.dataloader import ImageMixin
+
+# ===========================================
+# 1) パスなど（修正があれば適宜変更）
+# ===========================================
+test_weight = './weight_epoch-011_best.pt'
+parameter = './parameters.json'
+
+# ===========================================
+# 2) クラスラベルの定義
+#    - 撮像方向(label_APorPA)と回転方向(label_round)のクラス
+# ===========================================
+LABEL_APorPA = [
+    "AP",  # class 0
+    "PA"   # class 1
+]
+
+LABEL_ROUND = [
+    "0° Rotation",    # class 0
+    "90° Rotation",   # class 1
+    "180° Rotation",  # class 2
+    "270° Rotation"   # class 3
+]
+
+# ===========================================
+# 3) 前処理用の ImageHandlerクラス
+#    - 画像が既に256×256前提。Resizeはコメントアウトで残す
+# ===========================================
+class ImageHandler(ImageMixin):
+    def __init__(self, params):
+        self.params = params
+        # ここでリサイズは省略（推論が重いので）
+        # 入力画像は既に256×256であることを想定
+        self.transform = T.Compose([
+            # T.Resize((256, 256)),  # コメントアウト: 画像を256×256にリサイズ
+            T.ToTensor(),            # Tensor化 (0~1, shape: C,H,W)
+        ])
+
+    def set_image(self, image):
+        # PIL画像 -> transform -> バッチ次元を付ける
+        image = self.transform(image)
+        image = {'image': image.unsqueeze(0)}
+        return image
+
+# ===========================================
+# 4) パラメータのロード
+# ===========================================
+def load_parameter(parameter):
+    _args = ParamSet()
+    params = _retrieve_parameter(parameter)
+    for _param, _arg in params.items():
+        setattr(_args, _param, _arg)
+
+    # 推論用に書き換え (学習関連は無効化または無視)
+    _args.augmentation = 'no'
+    _args.sampler = 'no'
+    _args.pretrained = False
+    _args.mlp = None
+    _args.net = _args.model
+    _args.device = torch.device('cpu')
+
+    args_model = _dispatch_by_group(_args, 'model')
+    args_dataloader = _dispatch_by_group(_args, 'dataloader')
+    return args_model, args_dataloader
+
+args_model, args_dataloader = load_parameter(parameter)
+
+# ===========================================
+# 5) モデルを作成し学習済み重みをロード
+# ===========================================
+model = create_model(args_model)
+print(f"Load weight: {test_weight}")
+model.load_weight(test_weight)
+model.eval()  # 推論モード
+
+# ===========================================
+# 6) 推論関数
+# ===========================================
+def classify_APorPA_and_round(image):
+    """
+    モデルが以下を出力する想定:
+      outputs["label_APorPA"]  -> shape=[1, 2] (2クラス: AP/PA)
+      outputs["label_round"]   -> shape=[1, 4] (4クラス: 0°, 90°, 180°, 270°)
+    """
+    image_handler = ImageHandler(args_dataloader)
+    image_tensor = image_handler.set_image(image)
+
+    with torch.no_grad():
+        outputs = model(image_tensor)
+
+        # --- label_APorPA ---
+        if "label_APorPA" not in outputs:
+            print(f"[ERROR] 'label_APorPA' not found in outputs. Actual keys: {list(outputs.keys())}")
+            return "ERROR: Missing 'label_APorPA'", "ERROR: Missing 'label_round'"
+
+        scores_APorPA = outputs["label_APorPA"]  # shape=[1,2]想定
+        pred_APorPA_idx = torch.argmax(scores_APorPA, dim=1).item()
+        predicted_APorPA = LABEL_APorPA[pred_APorPA_idx]
+
+        # --- label_round ---
+        if "label_round" not in outputs:
+            print(f"[ERROR] 'label_round' not found in outputs. Actual keys: {list(outputs.keys())}")
+            return predicted_APorPA, "ERROR: Missing 'label_round'"
+
+        scores_round = outputs["label_round"]  # shape=[1,4]想定
+        pred_round_idx = torch.argmax(scores_round, dim=1).item()
+        predicted_round = LABEL_ROUND[pred_round_idx]
+
+    return predicted_APorPA, predicted_round
+
+# ===========================================
+# 7) Gradio UI
+# ===========================================
+html_content = """
+<div style="padding: 15px; border: 1px solid #e0e0e0; border-radius: 5px;">
+    <h3>Chest X-ray: AP/PA & Rotation Classification</h3>
+    <p>入力画像は既に256×256（グレースケール）であることを想定し、内部でのリサイズは行いません。</p>
+    <p>胸部レントゲン画像に対して、撮像方向(AP or PA)と回転方向(0°, 90°, 180°, 270°)を同時に推定します。</p>
+</div>
+"""
+
+with gr.Blocks(title="Chest X-ray: AP/PA & Rotation Classification") as demo:
+    gr.HTML("<div style='text-align:center'><h2>Chest X-ray AP/PA & Rotation Classification</h2></div>")
+    gr.HTML(html_content)
+
+    with gr.Row():
+        input_image = gr.Image(type="pil", image_mode="L")
+        output_APorPA = gr.Label(label="Predicted AP or PA")
+        output_round = gr.Label(label="Predicted Rotation")
+
+    send_btn = gr.Button("Inference")
+    send_btn.click(
+        fn=classify_APorPA_and_round, 
+        inputs=input_image, 
+        outputs=[output_APorPA, output_round]
+    )
+
+    with gr.Row():
+        # サンプルファイルは実際のパスに置き換えてください
+        gr.Examples(
+            examples=[
+                './sample/sample_AP_inverted.png',
+                './sample/sample_PA_right.png',
+                './sample/sample_lateral_upright.png'
+            ],
+            inputs=input_image
+        )
+
+demo.launch(debug=True)

lib/__init__.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from .options import (
+    ParamSet,
+    set_options,
+    save_parameter,
+    print_parameter
+    )
+from .dataloader import create_dataloader
+from .framework import create_model
+from .metrics import set_eval
+from .logger import BaseLogger
+
+__all__ = [
+            'ParamSet',
+            'set_options',
+            'print_parameter',
+            'save_parameter',
+            'create_dataloader',
+            'create_model',
+            'set_eval',
+            'BaseLogger'
+        ]

lib/component/__init__.py

@@ -0,0 +1,16 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from .net import create_net
+from .criterion import set_criterion
+from .optimizer import set_optimizer
+from .loss import set_loss_store
+from .likelihood import set_likelihood
+
+__all__ = [
+            'create_net',
+            'set_criterion',
+            'set_optimizer',
+            'set_loss_store',
+            'set_likelihood'
+        ]

lib/component/criterion.py

@@ -0,0 +1,332 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import torch
+import torch.nn as nn
+from typing import Dict, Union
+
+# Alias of typing
+# eg. {'labels': {'label_A: torch.Tensor([0, 1, ...]), ...}}
+LabelDict = Dict[str, Dict[str, Union[torch.IntTensor, torch.FloatTensor]]]
+
+
+class RMSELoss(nn.Module):
+    """
+    Class to calculate RMSE.
+    """
+    def __init__(self, eps: float = 1e-7) -> None:
+        """
+        Args:
+            eps (float, optional): value to avoid 0. Defaults to 1e-7.
+        """
+        super().__init__()
+        self.mse = nn.MSELoss()
+        self.eps = eps
+
+    def forward(self, yhat: float, y: float) -> torch.FloatTensor:
+        """
+        Calculate RMSE.
+
+        Args:
+            yhat (float): prediction value
+            y (float): ground truth value
+
+        Returns:
+            float: RMSE
+        """
+        _loss = self.mse(yhat, y) + self.eps
+        return torch.sqrt(_loss)
+
+
+class Regularization:
+    """
+    Class to calculate regularization loss.
+
+    Args:
+        object (object): object
+    """
+    def __init__(self, order: int, weight_decay: float) -> None:
+        """
+        The initialization of Regularization class.
+
+        Args:
+            order: (int) norm order number
+            weight_decay: (float) weight decay rate
+        """
+        super().__init__()
+        self.order = order
+        self.weight_decay = weight_decay
+
+    def __call__(self, network: nn.Module) -> torch.FloatTensor:
+        """"
+        Calculates regularization(self.order) loss for network.
+
+        Args:
+            model: (torch.nn.Module object)
+
+        Returns:
+            torch.FloatTensor: the regularization(self.order) loss
+        """
+        reg_loss = 0
+        for name, w in network.named_parameters():
+            if 'weight' in name:
+                reg_loss = reg_loss + torch.norm(w, p=self.order)
+        reg_loss = self.weight_decay * reg_loss
+        return reg_loss
+
+
+class NegativeLogLikelihood(nn.Module):
+    """
+    Class to calculate RMSE.
+    """
+    def __init__(self, device: torch.device) -> None:
+        """
+        Args:
+            device (torch.device): device
+        """
+        super().__init__()
+        self.L2_reg = 0.05
+        self.reg = Regularization(order=2, weight_decay=self.L2_reg)
+        self.device = device
+
+    def forward(
+                self,
+                output: torch.FloatTensor,
+                label: torch.IntTensor,
+                periods: torch.FloatTensor,
+                network: nn.Module
+                ) -> torch.FloatTensor:
+        """
+        Calculates Negative Log Likelihood.
+
+        Args:
+            output (torch.FloatTensor): prediction value, ie risk prediction
+            label (torch.IntTensor): occurrence of event
+            periods (torch.FloatTensor): period
+            network (nn.Network): network
+
+        Returns:
+            torch.FloatTensor: Negative Log Likelihood
+        """
+        mask = torch.ones(periods.shape[0], periods.shape[0]).to(self.device)  # output and mask should be on the same device.
+        mask[(periods.T - periods) > 0] = 0
+
+        _loss = torch.exp(output) * mask
+        # Note: torch.sum(_loss, dim=0) possibly returns nan, in particular MLP.
+        _loss = torch.sum(_loss, dim=0) / torch.sum(mask, dim=0)
+        _loss = torch.log(_loss).reshape(-1, 1)
+        num_occurs = torch.sum(label)
+
+        if num_occurs.item() == 0.0:
+            loss = torch.tensor([1e-7], requires_grad=True).to(self.device)  # To avoid zero division, set small value as loss
+            return loss
+        else:
+            neg_log_loss = -torch.sum((output - _loss) * label) / num_occurs
+            l2_loss = self.reg(network)
+            loss = neg_log_loss + l2_loss
+            return loss
+
+
+class ClsCriterion:
+    """
+    Class of criterion for classification.
+    """
+    def __init__(self, device: torch.device = None) -> None:
+        """
+        Set CrossEntropyLoss.
+
+        Args:
+            device (torch.device): device
+        """
+        self.device = device
+        self.criterion = nn.CrossEntropyLoss()
+
+    def __call__(
+                self,
+                outputs: Dict[str, torch.FloatTensor],
+                labels: Dict[str, LabelDict]
+                ) -> Dict[str, torch.FloatTensor]:
+        """
+        Calculate loss.
+
+        Args:
+            outputs (Dict[str, torch.FloatTensor], optional): output
+            labels (Dict[str, LabelDict]): labels
+
+        Returns:
+            Dict[str, torch.FloatTensor]: loss for each label and their total loss
+
+        # No reshape and no cast:
+        output: [64, 2]: torch.float32
+        label:  [64]   : torch.int64
+        label.dtype should be torch.int64, otherwise nn.CrossEntropyLoss() causes error.
+
+        eg.
+        outputs = {'label_A': [[0.8, 0.2], ...] 'label_B': [[0.7, 0.3]], ...}
+        labels = { 'labels': {'label_A: 1: [1, 1, 0, ...], 'label_B': [0, 0, 1, ...], ...} }
+
+        -> losses = {total: loss_total, label_A: loss_A, label_B: loss_B, ... }
+        """
+        _labels = labels['labels']
+
+        # loss for each label and total of their losses
+        losses = dict()
+        losses['total'] = torch.tensor([0.0], requires_grad=True).to(self.device)
+        for label_name in labels['labels'].keys():
+            _output = outputs[label_name]
+            _label = _labels[label_name]
+            _label_loss = self.criterion(_output, _label)
+            losses[label_name] = _label_loss
+            losses['total'] = torch.add(losses['total'], _label_loss)
+        return losses
+
+
+class RegCriterion:
+    """
+    Class of criterion for regression.
+    """
+    def __init__(self, criterion_name: str = None, device: torch.device = None) -> None:
+        """
+        Set MSE, RMSE or MAE.
+
+        Args:
+            criterion_name (str): 'MSE', 'RMSE', or 'MAE'
+            device (torch.device): device
+        """
+        self.device = device
+
+        if criterion_name == 'MSE':
+            self.criterion = nn.MSELoss()
+        elif criterion_name == 'RMSE':
+            self.criterion = RMSELoss()
+        elif criterion_name == 'MAE':
+            self.criterion = nn.L1Loss()
+        else:
+            raise ValueError(f"Invalid criterion for regression: {criterion_name}.")
+
+    def __call__(
+                self,
+                outputs: Dict[str, torch.FloatTensor],
+                labels: Dict[str, LabelDict]
+                ) -> Dict[str, torch.FloatTensor]:
+        """
+        Calculate loss.
+
+        Args:
+            Args:
+            outputs (Dict[str, torch.FloatTensor], optional): output
+            labels (Dict[str, LabelDict]): labels
+
+        Returns:
+            Dict[str, torch.FloatTensor]: loss for each label and their total loss
+
+        # Reshape and cast
+        output: [64, 1] -> [64]: torch.float32
+        label:             [64]: torch.float64 -> torch.float32
+        # label.dtype should be torch.float32, otherwise cannot backward.
+
+        eg.
+        outputs = {'label_A': [[10.8], ...] 'label_B': [[15.7]], ...}
+        labels = {'labels': {'label_A: 1: [10, 9, ...], 'label_B': [12, 17,], ...}}
+        -> losses = {total: loss_total, label_A: loss_A, label_B: loss_B, ... }
+        """
+        _outputs = {label_name: _output.squeeze() for label_name, _output in outputs.items()}
+        _labels = {label_name: _label.to(torch.float32) for label_name, _label in labels['labels'].items()}
+
+        # loss for each label and total of their losses
+        losses = dict()
+        losses['total'] = torch.tensor([0.0], requires_grad=True).to(self.device)
+        for label_name in labels['labels'].keys():
+            _output = _outputs[label_name]
+            _label = _labels[label_name]
+            _label_loss = self.criterion(_output, _label)
+            losses[label_name] = _label_loss
+            losses['total'] = torch.add(losses['total'], _label_loss)
+        return losses
+
+
+class DeepSurvCriterion:
+    """
+    Class of criterion for deepsurv.
+    """
+    def __init__(self, device: torch.device = None) -> None:
+        """
+        Set NegativeLogLikelihood.
+
+        Args:
+            device (torch.device, optional): device
+        """
+        self.device = device
+        self.criterion = NegativeLogLikelihood(self.device).to(self.device)
+
+    def __call__(
+                self,
+                outputs: Dict[str, torch.FloatTensor],
+                labels: Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                ) -> Dict[str, torch.FloatTensor]:
+        """
+        Calculate loss.
+
+        Args:
+            outputs (Dict[str, torch.FloatTensor], optional): output
+            labels (Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]): labels, periods, and network
+
+        Returns:
+            Dict[str, torch.FloatTensor]: loss for each label and their total loss
+
+        # Reshape and no cast
+        output:         [64, 1]: torch.float32
+        label:  [64] -> [64, 1]: torch.int64
+        period: [64] -> [64, 1]: torch.float32
+
+        eg.
+        outputs = {'label_A': [[10.8], ...] 'label_B': [[15.7]], ...}
+        labels = {
+                    'labels': {'label_A: 1: [1, 0, 1, ...] },
+                    'periods': [5, 10, 7, ...],
+                    'network': network
+                }
+        -> losses = {total: loss_total, label_A: loss_A, label_B: loss_B, ... }
+        """
+        _labels = {label_name: _label.reshape(-1, 1) for label_name, _label in labels['labels'].items()}
+        _periods = labels['periods'].reshape(-1, 1)
+        _network = labels['network']
+
+        # loss for each label and total of their losses
+        losses = dict()
+        losses['total'] = torch.tensor([0.0], requires_grad=True).to(self.device)
+        for label_name in labels['labels'].keys():
+            _output = outputs[label_name]
+            _label = _labels[label_name]
+            _label_loss = self.criterion(_output, _label, _periods, _network)
+            losses[label_name] = _label_loss
+            losses['total'] = torch.add(losses['total'], _label_loss)
+        return losses
+
+
+def set_criterion(
+                criterion_name: str,
+                device: torch.device
+                ) -> Union[ClsCriterion, RegCriterion, DeepSurvCriterion]:
+    """
+    Return criterion class
+
+    Args:
+        criterion_name (str): criterion name
+        device (torch.device): device
+
+    Returns:
+        Union[ClsCriterion, RegCriterion, DeepSurvCriterion]: criterion class
+    """
+
+    if criterion_name == 'CEL':
+        return ClsCriterion(device=device)
+
+    elif criterion_name in ['MSE', 'RMSE', 'MAE']:
+        return RegCriterion(criterion_name=criterion_name, device=device)
+
+    elif criterion_name == 'NLL':
+        return DeepSurvCriterion(device=device)
+
+    else:
+        raise ValueError(f"Invalid criterion: {criterion_name}.")

lib/component/likelihood.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import pandas as pd
+import torch
+from typing import List, Dict
+
+
+class Likelihood:
+    """
+    Class for making likelihood.
+    """
+    def __init__(self, task: str, num_outputs_for_label: Dict[str, int]) -> None:
+        """
+        Args:
+            task (str): task
+            num_outputs_for_label (Dict[str, int]): number of classes for each label
+        """
+        self.task = task
+        self.num_outputs_for_label = num_outputs_for_label
+        self.base_column_list = self._set_base_columns(self.task)
+        self.pred_column_list = self._make_pred_columns(self.task, self.num_outputs_for_label)
+
+    def _set_base_columns(self, task: str) -> List[str]:
+        """
+        Return base columns.
+
+        Args:
+            task (str): task
+
+        Returns:
+            List[str]: base columns except columns of label and prediction
+        """
+        if (task == 'classification') or (task == 'regression'):
+            base_columns = ['uniqID', 'group', 'imgpath', 'split']
+            return base_columns
+        elif task == 'deepsurv':
+            base_columns = ['uniqID', 'group', 'imgpath', 'split', 'periods']
+            return base_columns
+        else:
+            raise ValueError(f"Invalid task: {task}.")
+
+    def _make_pred_columns(self, task: str, num_outputs_for_label: Dict[str, int]) -> Dict[str, List[str]]:
+        """
+        Make column names of predictions with label name and its number of classes.
+
+        Args:
+            task (str):  task
+            num_outputs_for_label (Dict[str, int]): number of classes for each label
+
+        Returns:
+            Dict[str, List[str]]: label and list of columns of predictions with its class number
+
+        eg.
+        {label_A: 2, label_B: 2} -> {label_A: [pred_label_A_0, pred_label_A_1], label_B: [pred_label_B_0, pred_label_B_1]}
+        {label_A: 1, label_B: 1} -> {label_A: [pred_label_A], label_B: [pred_label_B]}
+        """
+        pred_columns = dict()
+        if task == 'classification':
+            for label_name, num_classes in num_outputs_for_label.items():
+                pred_columns[label_name] = ['pred_' + label_name + '_' + str(i) for i in range(num_classes)]
+            return pred_columns
+        elif (task == 'regression') or (task == 'deepsurv'):
+            for label_name, num_classes in num_outputs_for_label.items():
+                pred_columns[label_name] = ['pred_' + label_name]
+            return pred_columns
+        else:
+            raise ValueError(f"Invalid task: {task}.")
+
+    def make_format(self, data: Dict, output: Dict[str, torch.Tensor]) -> pd.DataFrame:
+            """
+            Make a new DataFrame of likelihood every batch.
+
+            Args:
+                data (Dict): batch data from dataloader
+                output (Dict[str, torch.Tensor]): output of model
+            """
+            _likelihood = {column_name: data[column_name] for column_name in self.base_column_list}
+            df_likelihood = pd.DataFrame(_likelihood)
+
+            if any(data['labels']):
+                for label_name, pred in output.items():
+                    _df_label = pd.DataFrame({label_name: data['labels'][label_name].tolist()})
+                    pred = pred.to('cpu').detach().numpy().copy()
+                    _df_pred = pd.DataFrame(pred, columns=self.pred_column_list[label_name])
+                    df_likelihood = pd.concat([df_likelihood, _df_label, _df_pred], axis=1)
+                return df_likelihood
+            else:
+                for label_name, pred in output.items():
+                    pred = pred.to('cpu').detach().numpy().copy()
+                    _df_pred = pd.DataFrame(pred, columns=self.pred_column_list[label_name])
+                    df_likelihood = pd.concat([df_likelihood, _df_pred], axis=1)
+                return df_likelihood
+
+
+def set_likelihood(task: str, num_outputs_for_label: Dict[str, int]) -> Likelihood:
+    """
+    Set likelihood.
+
+    Args:
+        task (str): task
+        num_outputs_for_label (Dict[str, int]): number of classes for each label
+
+    Returns:
+            Likelihood: instance of class Likelihood
+    """
+    return Likelihood(task, num_outputs_for_label)

lib/component/loss.py

@@ -0,0 +1,248 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from pathlib import Path
+import torch
+import pandas as pd
+from ..logger import BaseLogger
+from typing import List, Dict, Union
+
+
+logger = BaseLogger.get_logger(__name__)
+
+
+class LabelLoss:
+    """
+    Class to store loss for every bash and epoch loss of each label.
+    """
+    def __init__(self) -> None:
+        # Accumulate batch_loss(=loss * batch_size)
+        self.train_batch_loss = 0.0
+        self.val_batch_loss = 0.0
+
+        # epoch_loss = batch_loss / dataset_size
+        self.train_epoch_loss = []       # List[float]
+        self.val_epoch_loss = []         # List[float]
+
+        self.best_val_loss = None        # float
+        self.best_epoch = None           # int
+        self.is_val_loss_updated = None  # bool
+
+    def get_loss(self, phase: str, target: str) -> Union[float, List[float]]:
+        """
+        Return loss depending on phase and target
+
+        Args:
+            phase (str): 'train' or 'val'
+            target (str): 'batch' or 'epoch'
+
+        Returns:
+            Union[float, List[float]]: batch_loss or epoch_loss
+        """
+        _target = phase + '_' + target + '_loss'
+        return getattr(self, _target)
+
+    def store_batch_loss(self, phase: str, new_batch_loss: torch.FloatTensor, batch_size: int) -> None:
+        """
+        Add new batch loss to previous one for phase by multiplying by batch_size.
+
+        Args:
+            phase (str): 'train' or 'val'
+            new_batch_loss (torch.FloatTensor): batch loss calculated by criterion
+            batch_size (int): batch size
+        """
+        _new = new_batch_loss.item() * batch_size  # torch.FloatTensor -> float
+        _prev = self.get_loss(phase, 'batch')
+        _added = _prev + _new
+        _target = phase + '_' + 'batch_loss'
+        setattr(self, _target, _added)
+
+    def append_epoch_loss(self, phase: str, new_epoch_loss: float) -> None:
+        """
+        Append epoch loss depending on phase and target
+
+        Args:
+            phase (str): 'train' or 'val'
+            new_epoch_loss (float): batch loss or epoch loss
+        """
+        _target = phase + '_' + 'epoch_loss'
+        getattr(self, _target).append(new_epoch_loss)
+
+    def get_latest_epoch_loss(self, phase: str) -> float:
+        """
+        Return the latest loss of phase.
+
+        Args:
+            phase (str): train or val
+
+        Returns:
+            float: the latest loss
+        """
+        return self.get_loss(phase, 'epoch')[-1]
+
+    def update_best_val_loss(self, at_epoch: int = None) -> None:
+        """
+        Update val_epoch_loss is the best.
+
+        Args:
+            at_epoch (int): epoch when checked
+        """
+        _latest_val_loss = self.get_latest_epoch_loss('val')
+
+        if at_epoch == 1:
+            self.best_val_loss = _latest_val_loss
+            self.best_epoch = at_epoch
+            self.is_val_loss_updated = True
+        else:
+            # When at_epoch > 1
+            if _latest_val_loss < self.best_val_loss:
+                self.best_val_loss = _latest_val_loss
+                self.best_epoch = at_epoch
+                self.is_val_loss_updated = True
+            else:
+                self.is_val_loss_updated = False
+
+
+class LossStore:
+    """
+    Class for calculating loss and store it.
+    """
+    def __init__(self, label_list: List[str], num_epochs: int, dataset_info: Dict[str, int]) -> None:
+        """
+        Args:
+            label_list (List[str]): list of internal labels
+            num_epochs (int) : number of epochs
+            dataset_info (Dict[str, int]):  dataset sizes of 'train' and 'val'
+        """
+        self.label_list = label_list
+        self.num_epochs = num_epochs
+        self.dataset_info = dataset_info
+
+        # Added a special label 'total' to store total of losses of all labels.
+        self.label_losses = {label_name: LabelLoss() for label_name in self.label_list + ['total']}
+
+    def store(self, phase: str, losses: Dict[str, torch.FloatTensor], batch_size: int = None) -> None:
+        """
+        Store label-wise batch losses of phase to previous one.
+
+        Args:
+            phase (str): 'train' or 'val'
+            losses (Dict[str, torch.FloatTensor]): loss for each label calculated by criterion
+            batch_size (int): batch size
+
+        # Note:
+            self.loss_stores['total'] is already total of losses of all label, which is calculated in criterion.py,
+            therefore, it is OK just to multiply by batch_size. This is done in add_batch_loss().
+        """
+        for label_name in self.label_list + ['total']:
+            _new_batch_loss = losses[label_name]
+            self.label_losses[label_name].store_batch_loss(phase, _new_batch_loss, batch_size)
+
+    def cal_epoch_loss(self, at_epoch: int = None) -> None:
+        """
+        Calculate epoch loss for each phase all at once.
+
+        Args:
+            at_epoch (int): epoch number
+        """
+        # For each label
+        for label_name in self.label_list:
+            for phase in ['train', 'val']:
+                _batch_loss = self.label_losses[label_name].get_loss(phase, 'batch')
+                _dataset_size = self.dataset_info[phase]
+                _new_epoch_loss = _batch_loss / _dataset_size
+                self.label_losses[label_name].append_epoch_loss(phase, _new_epoch_loss)
+
+        # For total, average by dataset_size and the number of labels.
+        for phase in ['train', 'val']:
+            _batch_loss = self.label_losses['total'].get_loss(phase, 'batch')
+            _dataset_size = self.dataset_info[phase]
+            _new_epoch_loss = _batch_loss / (_dataset_size * len(self.label_list))
+            self.label_losses['total'].append_epoch_loss(phase, _new_epoch_loss)
+
+        # Update val_best_loss and best_epoch.
+        for label_name in self.label_list + ['total']:
+            self.label_losses[label_name].update_best_val_loss(at_epoch=at_epoch)
+
+        # Initialize batch_loss after calculating epoch loss.
+        for label_name in self.label_list + ['total']:
+            self.label_losses[label_name].train_batch_loss = 0.0
+            self.label_losses[label_name].val_batch_loss = 0.0
+
+    def is_val_loss_updated(self) -> bool:
+        """
+        Check if val_loss of 'total' is updated.
+
+        Returns:
+            bool: Updated or not
+        """
+        return self.label_losses['total'].is_val_loss_updated
+
+    def get_best_epoch(self) -> int:
+        """
+        Returns best epoch.
+
+        Returns:
+            int: best epoch
+        """
+        return self.label_losses['total'].best_epoch
+
+    def print_epoch_loss(self, at_epoch: int = None) -> None:
+        """
+        Print train_loss and val_loss for the ith epoch.
+
+        Args:
+            at_epoch (int): epoch number
+        """
+        train_epoch_loss = self.label_losses['total'].get_latest_epoch_loss('train')
+        val_epoch_loss = self.label_losses['total'].get_latest_epoch_loss('val')
+
+        _epoch_comm = f"epoch [{at_epoch:>3}/{self.num_epochs:<3}]"
+        _train_comm = f"train_loss: {train_epoch_loss :>8.4f}"
+        _val_comm = f"val_loss: {val_epoch_loss:>8.4f}"
+        _updated_comment = ''
+        if (at_epoch > 1) and (self.is_val_loss_updated()):
+            _updated_comment = '   Updated best val_loss!'
+        comment = _epoch_comm + ', ' + _train_comm + ', ' + _val_comm + _updated_comment
+        logger.info(comment)
+
+    def save_learning_curve(self, save_datetime_dir: str) -> None:
+        """
+        Save learning curve.
+
+        Args:
+            save_datetime_dir (str): save_datetime_dir
+        """
+        save_dir = Path(save_datetime_dir, 'learning_curve')
+        save_dir.mkdir(parents=True, exist_ok=True)
+
+        for label_name in self.label_list + ['total']:
+            _label_loss = self.label_losses[label_name]
+            _train_epoch_loss = _label_loss.get_loss('train', 'epoch')
+            _val_epoch_loss = _label_loss.get_loss('val', 'epoch')
+
+            df_label_epoch_loss = pd.DataFrame({
+                                                'train_loss': _train_epoch_loss,
+                                                'val_loss': _val_epoch_loss
+                                            })
+
+            _best_epoch = str(_label_loss.best_epoch).zfill(3)
+            _best_val_loss = f"{_label_loss.best_val_loss:.4f}"
+            save_name = 'learning_curve_' + label_name + '_val-best-epoch-' + _best_epoch + '_val-best-loss-' + _best_val_loss + '.csv'
+            save_path = Path(save_dir, save_name)
+            df_label_epoch_loss.to_csv(save_path, index=False)
+
+
+def set_loss_store(label_list: List[str], num_epochs: int, dataset_info: Dict[str, int]) -> LossStore:
+    """
+    Return class LossStore.
+
+    Args:
+        label_list (List[str]): label list
+        num_epochs (int) : number of epochs
+        dataset_info (Dict[str, int]):  dataset sizes of 'train' and 'val'
+
+    Returns:
+        LossStore: LossStore
+    """
+    return LossStore(label_list, num_epochs, dataset_info)

lib/component/net.py

@@ -0,0 +1,624 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-r
+
+from collections import OrderedDict
+import torch
+import torch.nn as nn
+from torchvision.ops import MLP
+import torchvision.models as models
+from typing import Dict, Optional
+
+
+class BaseNet:
+    """
+    Class to construct network
+    """
+    cnn = {
+            'ResNet18': models.resnet18,
+            'ResNet': models.resnet50,
+            'DenseNet': models.densenet161,
+            'EfficientNetB0': models.efficientnet_b0,
+            'EfficientNetB2': models.efficientnet_b2,
+            'EfficientNetB4': models.efficientnet_b4,
+            'EfficientNetB6': models.efficientnet_b6,
+            'EfficientNetV2s': models.efficientnet_v2_s,
+            'EfficientNetV2m': models.efficientnet_v2_m,
+            'EfficientNetV2l': models.efficientnet_v2_l,
+            'ConvNeXtTiny': models.convnext_tiny,
+            'ConvNeXtSmall': models.convnext_small,
+            'ConvNeXtBase': models.convnext_base,
+            'ConvNeXtLarge': models.convnext_large
+            }
+
+    vit = {
+            'ViTb16': models.vit_b_16,
+            'ViTb32': models.vit_b_32,
+            'ViTl16': models.vit_l_16,
+            'ViTl32': models.vit_l_32,
+            'ViTH14': models.vit_h_14
+            }
+
+    net = {**cnn, **vit}
+
+    _classifier = {
+            'ResNet': 'fc',
+            'DenseNet': 'classifier',
+            'EfficientNet': 'classifier',
+            'ConvNext': 'classifier',
+            'ViT': 'heads'
+            }
+
+    classifier = {
+                'ResNet18': _classifier['ResNet'],
+                'ResNet': _classifier['ResNet'],
+                'DenseNet': _classifier['DenseNet'],
+                'EfficientNetB0': _classifier['EfficientNet'],
+                'EfficientNetB2': _classifier['EfficientNet'],
+                'EfficientNetB4': _classifier['EfficientNet'],
+                'EfficientNetB6': _classifier['EfficientNet'],
+                'EfficientNetV2s': _classifier['EfficientNet'],
+                'EfficientNetV2m': _classifier['EfficientNet'],
+                'EfficientNetV2l': _classifier['EfficientNet'],
+                'ConvNeXtTiny': _classifier['ConvNext'],
+                'ConvNeXtSmall': _classifier['ConvNext'],
+                'ConvNeXtBase':  _classifier['ConvNext'],
+                'ConvNeXtLarge':  _classifier['ConvNext'],
+                'ViTb16': _classifier['ViT'],
+                'ViTb32': _classifier['ViT'],
+                'ViTl16': _classifier['ViT'],
+                'ViTl32': _classifier['ViT'],
+                'ViTH14': _classifier['ViT']
+                }
+
+    mlp_config = {
+                'hidden_channels': [256, 256, 256],
+                'dropout': 0.2
+                }
+
+    DUMMY = nn.Identity()
+
+    @classmethod
+    def MLPNet(cls, mlp_num_inputs: int = None, inplace: bool = None) -> MLP:
+        """
+        Construct MLP.
+
+        Args:
+            mlp_num_inputs (int): the number of input of MLP
+            inplace (bool, optional): parameter for the activation layer, which can optionally do the operation in-place. Defaults to None.
+
+        Returns:
+            MLP: MLP
+        """
+        assert isinstance(mlp_num_inputs, int), f"Invalid number of inputs for MLP: {mlp_num_inputs}."
+        mlp = MLP(in_channels=mlp_num_inputs, hidden_channels=cls.mlp_config['hidden_channels'], inplace=inplace, dropout=cls.mlp_config['dropout'])
+        return mlp
+
+    @classmethod
+    def align_in_channels_1ch(cls, net_name: str = None, net: nn.Module = None) -> nn.Module:
+        """
+        Modify network to handle gray scale image.
+
+        Args:
+            net_name (str): network name
+            net (nn.Module): network itself
+
+        Returns:
+            nn.Module: network available for gray scale
+        """
+        if net_name.startswith('ResNet'):
+            net.conv1.in_channels = 1
+            net.conv1.weight = nn.Parameter(net.conv1.weight.sum(dim=1).unsqueeze(1))
+
+        elif net_name.startswith('DenseNet'):
+            net.features.conv0.in_channels = 1
+            net.features.conv0.weight = nn.Parameter(net.features.conv0.weight.sum(dim=1).unsqueeze(1))
+
+        elif net_name.startswith('Efficient'):
+            net.features[0][0].in_channels = 1
+            net.features[0][0].weight = nn.Parameter(net.features[0][0].weight.sum(dim=1).unsqueeze(1))
+
+        elif net_name.startswith('ConvNeXt'):
+            net.features[0][0].in_channels = 1
+            net.features[0][0].weight = nn.Parameter(net.features[0][0].weight.sum(dim=1).unsqueeze(1))
+
+        elif net_name.startswith('ViT'):
+            net.conv_proj.in_channels = 1
+            net.conv_proj.weight = nn.Parameter(net.conv_proj.weight.sum(dim=1).unsqueeze(1))
+
+        else:
+            raise ValueError(f"No specified net: {net_name}.")
+        return net
+
+    @classmethod
+    def set_net(
+                cls,
+                net_name: str = None,
+                in_channel: int = None,
+                vit_image_size: int = None,
+                pretrained: bool = None
+                ) -> nn.Module:
+        """
+        Modify network depending on in_channel and vit_image_size.
+
+        Args:
+            net_name (str): network name
+            in_channel (int, optional): image channel(any of 1ch or 3ch). Defaults to None.
+            vit_image_size (int, optional): image size which ViT handles if ViT is used. Defaults to None.
+                                            vit_image_size should be power of patch size.
+            pretrained (bool, optional): True when use pretrained CNN or ViT, otherwise False. Defaults to None.
+
+        Returns:
+            nn.Module: modified network
+        """
+        assert net_name in cls.net, f"No specified net: {net_name}."
+        if net_name in cls.cnn:
+            if pretrained:
+                net = cls.cnn[net_name](weights='DEFAULT')
+            else:
+                net = cls.cnn[net_name]()
+        else:
+            # When ViT
+            # always use pretrained
+            net = cls.set_vit(net_name=net_name, vit_image_size=vit_image_size)
+
+        if in_channel == 1:
+            net = cls.align_in_channels_1ch(net_name=net_name, net=net)
+        return net
+
+    @classmethod
+    def set_vit(cls, net_name: str = None, vit_image_size: int = None) -> nn.Module:
+        """
+        Modify ViT depending on vit_image_size.
+
+        Args:
+            net_name (str): ViT name
+            vit_image_size (int): image size which ViT handles if ViT is used.
+
+        Returns:
+            nn.Module: modified ViT
+        """
+        base_vit = cls.vit[net_name]
+        # pretrained_vit = base_vit(weights=cls.vit_weight[net_name])
+        pretrained_vit = base_vit(weights='DEFAULT')
+
+        # Align weight depending on image size
+        weight = pretrained_vit.state_dict()
+        patch_size = int(net_name[-2:])  # 'ViTb16' -> 16
+        aligned_weight = models.vision_transformer.interpolate_embeddings(
+                                                    image_size=vit_image_size,
+                                                    patch_size=patch_size,
+                                                    model_state=weight
+                                                    )
+        aligned_vit = base_vit(image_size=vit_image_size)  # Specify new image size.
+        aligned_vit.load_state_dict(aligned_weight)        # Load weight which can handle the new image size.
+        return aligned_vit
+
+    @classmethod
+    def construct_extractor(
+                            cls,
+                            net_name: str = None,
+                            mlp_num_inputs: int = None,
+                            in_channel: int = None,
+                            vit_image_size: int = None,
+                            pretrained: bool = None
+                            ) -> nn.Module:
+        """
+        Construct extractor of network depending on net_name.
+
+        Args:
+            net_name (str): network name.
+            mlp_num_inputs (int, optional): number of input of MLP. Defaults to None.
+            in_channel (int, optional): image channel(any of 1ch or 3ch). Defaults to None.
+            vit_image_size (int, optional): image size which ViT handles if ViT is used. Defaults to None.
+            pretrained (bool, optional): True when use pretrained CNN or ViT, otherwise False. Defaults to None.
+
+        Returns:
+            nn.Module: extractor of network
+        """
+        if net_name == 'MLP':
+            extractor = cls.MLPNet(mlp_num_inputs=mlp_num_inputs)
+        else:
+            extractor = cls.set_net(net_name=net_name, in_channel=in_channel, vit_image_size=vit_image_size, pretrained=pretrained)
+            setattr(extractor, cls.classifier[net_name], cls.DUMMY)  # Replace classifier with DUMMY(=nn.Identity()).
+        return extractor
+
+    @classmethod
+    def get_classifier(cls, net_name: str) -> nn.Module:
+        """
+        Get classifier of network depending on net_name.
+
+        Args:
+            net_name (str): network name
+
+        Returns:
+            nn.Module: classifier of network
+        """
+        net = cls.net[net_name]()
+        classifier = getattr(net, cls.classifier[net_name])
+        return classifier
+
+    @classmethod
+    def construct_multi_classifier(cls, net_name: str = None, num_outputs_for_label: Dict[str, int] = None) -> nn.ModuleDict:
+        """
+        Construct classifier for multi-label.
+
+        Args:
+            net_name (str): network name
+            num_outputs_for_label (Dict[str, int]): number of outputs for each label
+
+        Returns:
+            nn.ModuleDict: classifier for multi-label
+        """
+        classifiers = dict()
+        if net_name == 'MLP':
+            in_features = cls.mlp_config['hidden_channels'][-1]
+            for label_name, num_outputs in num_outputs_for_label.items():
+                classifiers[label_name] = nn.Linear(in_features, num_outputs)
+
+        elif net_name.startswith('ResNet') or net_name.startswith('DenseNet'):
+            base_classifier = cls.get_classifier(net_name)
+            in_features = base_classifier.in_features
+            for label_name, num_outputs in num_outputs_for_label.items():
+                classifiers[label_name] = nn.Linear(in_features, num_outputs)
+
+        elif net_name.startswith('EfficientNet'):
+            base_classifier = cls.get_classifier(net_name)
+            dropout = base_classifier[0].p
+            in_features = base_classifier[1].in_features
+            for label_name, num_outputs in num_outputs_for_label.items():
+                classifiers[label_name] = nn.Sequential(
+                                                        nn.Dropout(p=dropout, inplace=False),
+                                                        nn.Linear(in_features, num_outputs)
+                                                    )
+
+        elif net_name.startswith('ConvNeXt'):
+            base_classifier = cls.get_classifier(net_name)
+            layer_norm = base_classifier[0]
+            flatten = base_classifier[1]
+            in_features = base_classifier[2].in_features
+            for label_name, num_outputs in num_outputs_for_label.items():
+                # Shape is changed before nn.Linear.
+                classifiers[label_name] = nn.Sequential(
+                                                        layer_norm,
+                                                        flatten,
+                                                        nn.Linear(in_features, num_outputs)
+                                                    )
+
+        elif net_name.startswith('ViT'):
+            base_classifier = cls.get_classifier(net_name)
+            in_features = base_classifier.head.in_features
+            for label_name, num_outputs in num_outputs_for_label.items():
+                classifiers[label_name] = nn.Sequential(
+                                                OrderedDict([
+                                                        ('head', nn.Linear(in_features, num_outputs))
+                                                        ])
+                                                )
+
+        else:
+            raise ValueError(f"No specified net: {net_name}.")
+
+        multi_classifier = nn.ModuleDict(classifiers)
+        return multi_classifier
+
+    @classmethod
+    def get_classifier_in_features(cls, net_name: str) -> int:
+        """
+        Return in_feature of network indicating by net_name.
+        This class is used in class MultiNetFusion() only.
+
+        Args:
+            net_name (str): net_name
+
+        Returns:
+            int : in_feature
+
+        Required:
+        classifier.in_feature
+        classifier.[1].in_features
+        classifier.[2].in_features
+        classifier.head.in_features
+        """
+        if net_name == 'MLP':
+            in_features = cls.mlp_config['hidden_channels'][-1]
+
+        elif net_name.startswith('ResNet') or net_name.startswith('DenseNet'):
+            base_classifier = cls.get_classifier(net_name)
+            in_features = base_classifier.in_features
+
+        elif net_name.startswith('EfficientNet'):
+            base_classifier = cls.get_classifier(net_name)
+            in_features = base_classifier[1].in_features
+
+        elif net_name.startswith('ConvNeXt'):
+            base_classifier = cls.get_classifier(net_name)
+            in_features = base_classifier[2].in_features
+
+        elif net_name.startswith('ViT'):
+            base_classifier = cls.get_classifier(net_name)
+            in_features = base_classifier.head.in_features
+
+        else:
+            raise ValueError(f"No specified net: {net_name}.")
+        return in_features
+
+    @classmethod
+    def construct_aux_module(cls, net_name: str) -> nn.Sequential:
+        """
+        Construct module to align the shape of feature from extractor depending on network.
+        Actually, only when net_name == 'ConvNeXt'.
+        Because ConvNeXt has the process of aligning the dimensions in its classifier.
+
+        Needs to align shape of the feature extractor when ConvNeXt
+        (classifier):
+        Sequential(
+            (0): LayerNorm2d((768,), eps=1e-06, elementwise_affine=True)
+            (1): Flatten(start_dim=1, end_dim=-1)
+            (2): Linear(in_features=768, out_features=1000, bias=True)
+        )
+
+        Args:
+            net_name (str): net name
+
+        Returns:
+            nn.Module: layers such that they align the dimension of the output from the extractor like the original ConvNeXt.
+        """
+        aux_module = cls.DUMMY
+        if net_name.startswith('ConvNeXt'):
+            base_classifier = cls.get_classifier(net_name)
+            layer_norm = base_classifier[0]
+            flatten = base_classifier[1]
+            aux_module = nn.Sequential(
+                                layer_norm,
+                                flatten
+                                )
+        return aux_module
+
+    @classmethod
+    def get_last_extractor(cls, net: nn.Module = None, mlp: str = None, net_name: str = None) -> nn.Module:
+        """
+        Return the last extractor of network.
+        This is for Grad-CAM.
+        net should be one loaded weight.
+
+        Args:
+            net (nn.Module): network itself
+            mlp (str): 'MLP', otherwise None
+            net_name (str): network name
+
+        Returns:
+            nn.Module: last extractor of network
+        """
+        assert (net_name is not None), f"Network does not contain CNN or ViT: mlp={mlp}, net={net_name}."
+
+        _extractor = net.extractor_net
+
+        if net_name.startswith('ResNet'):
+            last_extractor = _extractor.layer4[-1]
+        elif net_name.startswith('DenseNet'):
+            last_extractor = _extractor.features.denseblock4.denselayer24
+        elif net_name.startswith('EfficientNet'):
+            last_extractor = _extractor.features[-1]
+        elif net_name.startswith('ConvNeXt'):
+            last_extractor = _extractor.features[-1][-1].block
+        elif net_name.startswith('ViT'):
+            last_extractor = _extractor.encoder.layers[-1]
+        else:
+            raise ValueError(f"Cannot get last extractor of net: {net_name}.")
+        return last_extractor
+
+
+class MultiMixin:
+    """
+    Class to define auxiliary function to handle multi-label.
+    """
+    def multi_forward(self, out_features: int) -> Dict[str, float]:
+        """
+        Forward out_features to classifier for each label.
+
+        Args:
+            out_features (int): output from extractor
+
+        Returns:
+            Dict[str, float]: output of classifier of each label
+        """
+        output = dict()
+        for label_name, classifier in self.multi_classifier.items():
+            output[label_name] = classifier(out_features)
+        return output
+
+
+class MultiWidget(nn.Module, BaseNet, MultiMixin):
+    """
+    Class for a widget to inherit multiple classes simultaneously.
+    """
+    pass
+
+
+class MultiNet(MultiWidget):
+    """
+    Model of MLP, CNN or ViT.
+    """
+    def __init__(
+                self,
+                net_name: str = None,
+                num_outputs_for_label: Dict[str, int] = None,
+                mlp_num_inputs: int = None,
+                in_channel: int = None,
+                vit_image_size: int = None,
+                pretrained: bool = None
+                ) -> None:
+        """
+        Args:
+            net_name (str): MLP, CNN or ViT name
+            num_outputs_for_label (Dict[str, int]): number of classes for each label
+            mlp_num_inputs (int): number of input of MLP.
+            in_channel (int): number of image channel, ie gray scale(=1) or color image(=3).
+            vit_image_size (int): image size to be input to ViT.
+            pretrained (bool): True when use pretrained CNN or ViT, otherwise False.
+        """
+        super().__init__()
+
+        self.net_name = net_name
+        self.num_outputs_for_label = num_outputs_for_label
+        self.mlp_num_inputs = mlp_num_inputs
+        self.in_channel = in_channel
+        self.vit_image_size = vit_image_size
+        self.pretrained = pretrained
+
+        # self.extractor_net = MLP or CVmodel
+        self.extractor_net = self.construct_extractor(
+                                                    net_name=self.net_name,
+                                                    mlp_num_inputs=self.mlp_num_inputs,
+                                                    in_channel=self.in_channel,
+                                                    vit_image_size=self.vit_image_size,
+                                                    pretrained=self.pretrained
+                                                    )
+        self.multi_classifier = self.construct_multi_classifier(net_name=self.net_name, num_outputs_for_label=self.num_outputs_for_label)
+
+    def forward(self, x: torch.Tensor) -> Dict[str, torch.Tensor]:
+        """
+        Forward.
+
+        Args:
+            x (torch.Tensor): tabular data or image
+
+        Returns:
+            Dict[str, torch.Tensor]: output
+        """
+        out_features = self.extractor_net(x)
+        output = self.multi_forward(out_features)
+        return output
+
+
+class MultiNetFusion(MultiWidget):
+    """
+    Fusion model of MLP and CNN or ViT.
+    """
+    def __init__(
+                self,
+                net_name: str = None,
+                num_outputs_for_label: Dict[str, int] = None,
+                mlp_num_inputs: int = None,
+                in_channel: int = None,
+                vit_image_size: int = None,
+                pretrained: bool = None
+                ) -> None:
+        """
+        Args:
+            net_name (str): CNN or ViT name. It is clear that MLP is used in fusion model.
+            num_outputs_for_label (Dict[str, int]): number of classes for each label
+            mlp_num_inputs (int): number of input of MLP. Defaults to None.
+            in_channel (int): number of image channel, ie gray scale(=1) or color image(=3).
+            vit_image_size (int): image size to be input to ViT.
+            pretrained (bool): True when use pretrained CNN or ViT, otherwise False.
+        """
+        assert (net_name != 'MLP'), 'net_name should not be MLP.'
+
+        super().__init__()
+
+        self.net_name = net_name
+        self.num_outputs_for_label = num_outputs_for_label
+        self.mlp_num_inputs = mlp_num_inputs
+        self.in_channel = in_channel
+        self.vit_image_size = vit_image_size
+        self.pretrained = pretrained
+
+        # Extractor of MLP and Net
+        self.extractor_mlp = self.construct_extractor(net_name='MLP', mlp_num_inputs=self.mlp_num_inputs)
+        self.extractor_net = self.construct_extractor(
+                                                    net_name=self.net_name,
+                                                    in_channel=self.in_channel,
+                                                    vit_image_size=self.vit_image_size,
+                                                    pretrained=self.pretrained
+                                                    )
+        self.aux_module = self.construct_aux_module(self.net_name)
+
+        # Intermediate MLP
+        self.in_features_from_mlp = self.get_classifier_in_features('MLP')
+        self.in_features_from_net = self.get_classifier_in_features(self.net_name)
+        self.inter_mlp_in_feature = self.in_features_from_mlp + self.in_features_from_net
+        self.inter_mlp = self.MLPNet(mlp_num_inputs=self.inter_mlp_in_feature, inplace=False)
+
+        # Multi classifier
+        self.multi_classifier = self.construct_multi_classifier(net_name='MLP', num_outputs_for_label=num_outputs_for_label)
+
+    def forward(self, x_mlp: torch.Tensor, x_net: torch.Tensor) -> Dict[str, torch.Tensor]:
+        """
+        Forward.
+
+        Args:
+            x_mlp (torch.Tensor): tabular data
+            x_net (torch.Tensor): image
+
+        Returns:
+            Dict[str, torch.Tensor]: output
+        """
+        out_mlp = self.extractor_mlp(x_mlp)
+        out_net = self.extractor_net(x_net)
+        out_net = self.aux_module(out_net)
+
+        out_features = torch.cat([out_mlp, out_net], dim=1)
+        out_features = self.inter_mlp(out_features)
+        output = self.multi_forward(out_features)
+        return output
+
+
+def create_net(
+            mlp: Optional[str] = None,
+            net: Optional[str] = None,
+            num_outputs_for_label: Dict[str, int] = None,
+            mlp_num_inputs: int = None,
+            in_channel: int = None,
+            vit_image_size: int = None,
+            pretrained: bool = None
+            ) -> nn.Module:
+    """
+    Create network.
+
+    Args:
+        mlp (Optional[str]): 'MLP' or None
+        net (Optional[str]):  CNN, ViT name or None
+        num_outputs_for_label (Dict[str, int]): number of outputs for each label
+        mlp_num_inputs (int): number of input of MLP.
+        in_channel (int): number of image channel, ie gray scale(=1) or color image(=3).
+        vit_image_size (int): image size to be input to ViT.
+        pretrained (bool): True when use pretrained CNN or ViT, otherwise False.
+
+    Returns:
+        nn.Module: network
+    """
+    _isMLPModel = (mlp is not None) and (net is None)
+    _isCVModel = (mlp is None) and (net is not None)
+    _isFusion = (mlp is not None) and (net is not None)
+
+    if _isMLPModel:
+        multi_net = MultiNet(
+                            net_name='MLP',
+                            num_outputs_for_label=num_outputs_for_label,
+                            mlp_num_inputs=mlp_num_inputs,
+                            in_channel=in_channel,
+                            vit_image_size=vit_image_size,
+                            pretrained=False   # No need of pretrained for MLP
+                            )
+    elif _isCVModel:
+        multi_net = MultiNet(
+                            net_name=net,
+                            num_outputs_for_label=num_outputs_for_label,
+                            mlp_num_inputs=mlp_num_inputs,
+                            in_channel=in_channel,
+                            vit_image_size=vit_image_size,
+                            pretrained=pretrained
+                            )
+    elif _isFusion:
+        multi_net = MultiNetFusion(
+                                net_name=net,
+                                num_outputs_for_label=num_outputs_for_label,
+                                mlp_num_inputs=mlp_num_inputs,
+                                in_channel=in_channel,
+                                vit_image_size=vit_image_size,
+                                pretrained=pretrained
+                                )
+    else:
+        raise ValueError(f"Invalid model type: mlp={mlp}, net={net}.")
+
+    return multi_net

lib/component/optimizer.py

@@ -0,0 +1,34 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import torch.optim as optim
+import torch.nn as nn
+
+
+def set_optimizer(optimizer_name: str, network: nn.Module, lr: float) -> optim:
+    """
+    Set optimizer.
+    Args:
+        optimizer_name (str): criterion name
+        network (torch.nn.Module): network
+        lr (float): learning rate
+    Returns:
+        torch.optim: optimizer
+    """
+    optimizers = {
+        'SGD': optim.SGD,
+        'Adadelta': optim.Adadelta,
+        'Adam': optim.Adam,
+        'RMSprop': optim.RMSprop,
+        'RAdam': optim.RAdam
+        }
+
+    assert (optimizer_name in optimizers), f"No specified optimizer: {optimizer_name}."
+
+    _optim = optimizers[optimizer_name]
+
+    if lr is None:
+        optimizer = _optim(network.parameters())
+    else:
+        optimizer = _optim(network.parameters(), lr=lr)
+    return optimizer

lib/dataloader.py

@@ -0,0 +1,400 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import numpy as np
+import torch
+import torchvision.transforms as transforms
+from torch.utils.data.dataset import Dataset
+from torch.utils.data.dataloader import DataLoader
+from torch.utils.data.sampler import WeightedRandomSampler
+from PIL import Image
+from sklearn.preprocessing import MinMaxScaler
+import pickle
+from .logger import BaseLogger
+from typing import List, Dict, Union
+import pandas as pd
+
+
+logger = BaseLogger.get_logger(__name__)
+
+
+class PrivateAugment(torch.nn.Module):
+    """
+    Augmentation defined privately.
+    Variety of augmentation can be written in this class if necessary.
+    """
+    # For X-ray photo.
+    xray_augs_list = [
+                    transforms.RandomAffine(degrees=(-3, 3), translate=(0.02, 0.02)),
+                    transforms.RandomAdjustSharpness(sharpness_factor=2),
+                    transforms.RandomAutocontrast()
+                    ]
+
+
+class InputDataMixin:
+    """
+    Class to normalizes input data.
+    """
+    def _make_scaler(self) -> MinMaxScaler:
+        """
+        Make scaler to normalize input data by min-max normalization with train data.
+
+        Returns:
+            MinMaxScaler: scaler
+        """
+        scaler = MinMaxScaler()
+        _df_train = self.df_source[self.df_source['split'] == 'train']  # should be normalized with min and max of training data
+        _ = scaler.fit(_df_train[self.input_list])                      # fit only
+        return scaler
+
+    def save_scaler(self, save_path :str) -> None:
+        """
+        Save scaler
+
+        Args:
+            save_path (str): path for saving scaler.
+        """
+        #save_scaler_path = Path(save_datetime_dir, 'scaler.pkl')
+        with open(save_path, 'wb') as f:
+            pickle.dump(self.scaler, f)
+
+    def load_scaler(self, scaler_path :str) -> None:
+        """
+        Load scaler.
+
+        Args:
+            scaler_path (str): path to scaler
+        """
+        with open(scaler_path, 'rb') as f:
+            scaler = pickle.load(f)
+        return scaler
+
+    def _normalize_inputs(self, df_inputs: pd.DataFrame) -> torch.FloatTensor:
+        """
+        Normalize inputs.
+
+        Args:
+            df_inputs (pd.DataFrame): DataFrame of inputs
+
+        Returns:
+            torch.FloatTensor: normalized inputs
+
+        Note:
+        After iloc[[idx], index_input_list], pd.DataFrame is obtained.
+        DataFrame fits the input type of self.scaler.transform.
+        However, after normalizing, the shape of inputs_value is (1, N), where N is the number of input values.
+        Since the shape (1, N) is not acceptable when forwarding, convert (1, N) -> (N,) is needed.
+        """
+        inputs_value = self.scaler.transform(df_inputs).reshape(-1)  #    np.float64
+        inputs_value = np.array(inputs_value, dtype=np.float32)      # -> np.float32
+        inputs_value = torch.from_numpy(inputs_value).clone()        # -> torch.float32
+        return inputs_value
+
+    def _load_input_value_if_mlp(self, idx: int) -> Union[torch.FloatTensor, str]:
+        """
+        Load input values after converting them into tensor if MLP is used.
+
+        Args:
+            idx (int): index
+
+        Returns:
+            Union[torch.Tensor[float], str]: tensor of input values, or empty string
+        """
+        inputs_value = ''
+
+        if self.params.mlp is None:
+            return inputs_value
+
+        index_input_list = [self.col_index_dict[input] for input in self.input_list]
+        _df_inputs = self.df_split.iloc[[idx], index_input_list]
+        inputs_value = self._normalize_inputs( _df_inputs)
+        return inputs_value
+
+
+class ImageMixin:
+    """
+    Class to normalize and transform image.
+    """
+    def _make_augmentations(self) -> List:
+        """
+        Define which augmentation is applied.
+
+        When training, augmentation is needed for train data only.
+        When test, no need of augmentation.
+        """
+        _augmentation = []
+        if (self.params.isTrain) and (self.split == 'train'):
+            if self.params.augmentation == 'xrayaug':
+                _augmentation = PrivateAugment.xray_augs_list
+            elif self.params.augmentation == 'trivialaugwide':
+                _augmentation.append(transforms.TrivialAugmentWide())
+            elif self.params.augmentation == 'randaug':
+                _augmentation.append(transforms.RandAugment())
+            else:
+                # ie. self.params.augmentation == 'no':
+                pass
+
+        _augmentation = transforms.Compose(_augmentation)
+        return _augmentation
+
+    def _make_transforms(self) -> List:
+        """
+        Make list of transforms.
+
+        Returns:
+            list of transforms: image normalization
+        """
+        _transforms = []
+        _transforms.append(transforms.ToTensor())
+
+        if self.params.normalize_image == 'yes':
+            # transforms.Normalize accepts only Tensor.
+            if self.params.in_channel == 1:
+                _transforms.append(transforms.Normalize(mean=(0.5, ), std=(0.5, )))
+            else:
+                # ie. self.params.in_channel == 3
+                _transforms.append(transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]))
+
+        _transforms = transforms.Compose(_transforms)
+        return _transforms
+
+    def _open_image_in_channel(self, imgpath: str, in_channel: int) -> Image:
+        """
+        Open image in channel.
+
+        Args:
+            imgpath (str): path to image
+            in_channel (int): channel, or 1 or 3
+
+        Returns:
+            Image: PIL image
+        """
+        if in_channel == 1:
+            image = Image.open(imgpath).convert('L')    # eg. np.array(image).shape = (64, 64)
+            return image
+        else:
+            # ie. self.params.in_channel == 3
+            image = Image.open(imgpath).convert('RGB')  # eg. np.array(image).shape = (64, 64, 3)
+            return image
+
+    def _load_image_if_cnn(self, idx: int) -> Union[torch.Tensor, str]:
+        """
+        Load image and convert it to tensor if any of CNN or ViT is used.
+
+        Args:
+            idx (int): index
+
+        Returns:
+            Union[torch.Tensor[float], str]: tensor converted from image, or empty string
+        """
+        image = ''
+
+        if self.params.net is None:
+            return image
+
+        imgpath = self.df_split.iat[idx, self.col_index_dict['imgpath']]
+        image = self._open_image_in_channel(imgpath, self.params.in_channel)
+        image = self.augmentation(image)
+        image = self.transform(image)
+        return image
+
+
+class DeepSurvMixin:
+    """
+    Class to handle required data for deepsurv.
+    """
+    def _load_periods_if_deepsurv(self, idx: int) -> Union[torch.FloatTensor, str]:
+        """
+        Return period if deepsurv.
+
+        Args:
+            idx (int): index
+
+        Returns:
+            Union[torch.FloatTensor, str]: period, or empty string
+        """
+        periods = ''
+
+        if self.params.task != 'deepsurv':
+            return periods
+
+        assert (self.params.task == 'deepsurv') and (len(self.label_list) == 1), 'Deepsurv cannot work in multi-label.'
+        periods = self.df_split.iat[idx, self.col_index_dict[self.period_name]]  #    int64
+        periods = np.array(periods, dtype=np.float32)                            # -> np.float32
+        periods = torch.from_numpy(periods).clone()                              # -> torch.float32
+        return periods
+
+
+class DataSetWidget(InputDataMixin, ImageMixin, DeepSurvMixin):
+    """
+    Class for a widget to inherit multiple classes simultaneously.
+    """
+    pass
+
+
+class LoadDataSet(Dataset, DataSetWidget):
+    """
+    Dataset for split.
+    """
+    def __init__(
+                self,
+                params,
+                split: str
+                ) -> None:
+        """
+        Args:
+            params (ParamSet): parameter for model
+            split (str): split
+        """
+        self.params = params
+        self.df_source = self.params.df_source
+        self.split = split
+
+        self.input_list = self.params.input_list
+        self.label_list = self.params.label_list
+
+        if self.params.task == 'deepsurv':
+            self.period_name = self.params.period_name
+
+        self.df_split = self.df_source[self.df_source['split'] == self.split]
+        self.col_index_dict = {col_name: self.df_split.columns.get_loc(col_name) for col_name in self.df_split.columns}
+
+        # For input data
+        if self.params.mlp is not None:
+            assert (self.input_list != []), f"input list is empty."
+            if params.isTrain:
+                self.scaler = self._make_scaler()
+            else:
+                # load scaler used at training.
+                self.scaler = self.load_scaler(self.params.scaler_path)
+
+        # For image
+        if self.params.net is not None:
+            self.augmentation = self._make_augmentations()
+            self.transform = self._make_transforms()
+
+    def __len__(self) -> int:
+        """
+        Return length of DataFrame.
+
+        Returns:
+            int: length of DataFrame
+        """
+        return len(self.df_split)
+
+    def _load_label(self, idx: int) -> Dict[str, Union[int, float]]:
+        """
+        Return labels.
+        If no column of label when csv of external dataset is used,
+        empty dictionary is returned.
+
+        Args:
+            idx (int): index
+
+        Returns:
+            Dict[str, Union[int, float]]: dictionary of label name and its value
+        """
+        # For checking if columns of labels exist when used csv for external dataset.
+        label_list_in_split = list(self.df_split.columns[self.df_split.columns.str.startswith('label')])
+        label_dict = dict()
+        if label_list_in_split != []:
+            for label_name in self.label_list:
+                label_dict[label_name] = self.df_split.iat[idx, self.col_index_dict[label_name]]
+        else:
+            # no label
+            pass
+        return label_dict
+
+    def __getitem__(self, idx: int) -> Dict:
+        """
+        Return data row specified by index.
+
+        Args:
+            idx (int): index
+
+        Returns:
+            Dict: dictionary of data to be passed model
+        """
+        uniqID = self.df_split.iat[idx, self.col_index_dict['uniqID']]
+        group = self.df_split.iat[idx, self.col_index_dict['group']]
+        imgpath = self.df_split.iat[idx, self.col_index_dict['imgpath']]
+        split = self.df_split.iat[idx, self.col_index_dict['split']]
+        inputs_value = self._load_input_value_if_mlp(idx)
+        image = self._load_image_if_cnn(idx)
+        label_dict = self._load_label(idx)
+        periods = self._load_periods_if_deepsurv(idx)
+
+        _data = {
+                'uniqID': uniqID,
+                'group': group,
+                'imgpath': imgpath,
+                'split': split,
+                'inputs': inputs_value,
+                'image': image,
+                'labels': label_dict,
+                'periods': periods
+                }
+        return _data
+
+
+def _make_sampler(split_data: LoadDataSet) -> WeightedRandomSampler:
+    """
+    Make sampler.
+
+    Args:
+        split_data (LoadDataSet): dataset
+
+    Returns:
+        WeightedRandomSampler: sampler
+    """
+    _target = []
+    for _, data in enumerate(split_data):
+        _target.append(list(data['labels'].values())[0])
+
+    class_sample_count = np.array([len(np.where(_target == t)[0]) for t in np.unique(_target)])
+    weight = 1. / class_sample_count
+    samples_weight = np.array([weight[t] for t in _target])
+    sampler = WeightedRandomSampler(samples_weight, len(samples_weight))
+    return sampler
+
+
+def create_dataloader(
+                    params,
+                    split: str = None
+                    ) -> DataLoader:
+    """
+    Create data loader ofr split.
+
+    Args:
+        params (ParamSet): parameter for dataloader
+        split (str): split. Defaults to None.
+
+    Returns:
+        DataLoader: data loader
+    """
+    split_data = LoadDataSet(params, split)
+
+    if params.isTrain:
+        batch_size = params.batch_size
+        shuffle = True
+    else:
+        batch_size = params.test_batch_size
+        shuffle = False
+
+    if params.sampler == 'yes':
+        assert ((params.task == 'classification') or (params.task == 'deepsurv')), 'Cannot make sampler in regression.'
+        assert (len(params.label_list) == 1), 'Cannot make sampler for multi-label.'
+        shuffle = False
+        sampler = _make_sampler(split_data)
+    else:
+        # When params.sampler == 'no'
+        sampler = None
+
+    split_loader = DataLoader(
+                            dataset=split_data,
+                            batch_size=batch_size,
+                            shuffle=shuffle,
+                            num_workers=0,
+                            sampler=sampler
+                            )
+    return split_loader

lib/framework.py

@@ -0,0 +1,373 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from pathlib import Path
+import copy
+from abc import ABC, abstractmethod
+import torch
+import torch.nn as nn
+from .component import create_net
+from .logger import BaseLogger
+from lib import ParamSet
+from typing import List, Dict, Tuple, Union
+
+# Alias of typing
+# eg. {'labels': {'label_A: torch.Tensor([0, 1, ...]), ...}}
+LabelDict = Dict[str, Dict[str, Union[torch.IntTensor, torch.FloatTensor]]]
+
+
+logger = BaseLogger.get_logger(__name__)
+
+
+class BaseModel(ABC):
+    """
+    Class to construct model. This class is the base class to construct model.
+    """
+    def __init__(self, params: ParamSet) -> None:
+        """
+        Class to define Model
+
+        Args:
+            param (ParamSet): parameters
+        """
+        self.params = params
+        self.device = self.params.device
+
+        self.network = create_net(
+                                mlp=self.params.mlp,
+                                net=self.params.net,
+                                num_outputs_for_label=self.params.num_outputs_for_label,
+                                mlp_num_inputs=self.params.mlp_num_inputs,
+                                in_channel=self.params.in_channel,
+                                vit_image_size=self.params.vit_image_size,
+                                pretrained=self.params.pretrained
+                                )
+        self.network.to(self.device)
+
+        # variables to keep temporary best_weight and best_epoch
+        self.acting_best_weight = None
+        self.acting_best_epoch = None
+
+    def train(self) -> None:
+        """
+        Make network training mode.
+        """
+        self.network.train()
+
+    def eval(self) -> None:
+        """
+        Make network evaluation mode.
+        """
+        self.network.eval()
+
+    @abstractmethod
+    def set_data(
+                self,
+                data: Dict
+                ) -> Tuple[
+                        Dict[str, torch.FloatTensor],
+                        Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                        ]:
+        raise NotImplementedError
+
+    def store_weight(self, at_epoch: int = None) -> None:
+        """
+        Store weight and epoch number when it is saved.
+
+        Args:
+            at_epoch (int): epoch number when save weight
+        """
+        self.acting_best_epoch = at_epoch
+
+        _network = copy.deepcopy(self.network)
+        if hasattr(_network, 'module'):
+            # When DataParallel used, move weight to CPU.
+            self.acting_best_weight = copy.deepcopy(_network.module.to(torch.device('cpu')).state_dict())
+        else:
+            self.acting_best_weight = copy.deepcopy(_network.state_dict())
+
+    def save_weight(self, save_datetime_dir: str, as_best: bool = None) -> None:
+        """
+        Save weight.
+
+        Args:
+            save_datetime_dir (str): save_datetime_dir
+            as_best (bool): True if weight is saved as best, otherwise False. Defaults to None.
+        """
+
+        save_dir = Path(save_datetime_dir, 'weights')
+        save_dir.mkdir(parents=True, exist_ok=True)
+        save_name = 'weight_epoch-' + str(self.acting_best_epoch).zfill(3) + '.pt'
+        save_path = Path(save_dir, save_name)
+
+        if as_best:
+            save_name_as_best = 'weight_epoch-' + str(self.acting_best_epoch).zfill(3) + '_best' + '.pt'
+            save_path_as_best = Path(save_dir, save_name_as_best)
+            if save_path.exists():
+                # Check if best weight already saved. If exists, rename with '_best'
+                save_path.rename(save_path_as_best)
+            else:
+                torch.save(self.acting_best_weight, save_path_as_best)
+        else:
+            save_name = 'weight_epoch-' + str(self.acting_best_epoch).zfill(3) + '.pt'
+            torch.save(self.acting_best_weight, save_path)
+
+    def load_weight(self, weight_path: Path) -> None:
+        """
+        Load wight from weight_path.
+
+        Args:
+            weight_path (Path): path to weight
+        """
+        logger.info(f"Load weight: {weight_path}.\n")
+        weight = torch.load(weight_path)
+        self.network.load_state_dict(weight)
+
+
+class ModelMixin:
+    def to_gpu(self, gpu_ids: List[int]) -> None:
+        """
+        Make model compute on the GPU.
+
+        Args:
+            gpu_ids (List[int]): GPU ids
+        """
+        if gpu_ids != []:
+            assert torch.cuda.is_available(), 'No available GPU on this machine.'
+            self.network = nn.DataParallel(self.network, device_ids=gpu_ids)
+
+    def init_network(self) -> None:
+        """
+        Initialize network.
+        This method is used at test to reset the current weight by redefining network.
+        """
+        self.network = create_net(
+                                mlp=self.params.mlp,
+                                net=self.params.net,
+                                num_outputs_for_label=self.params.num_outputs_for_label,
+                                mlp_num_inputs=self.params.mlp_num_inputs,
+                                in_channel=self.params.in_channel,
+                                vit_image_size=self.params.vit_image_size,
+                                pretrained=self.params.pretrained
+                                )
+        self.network.to(self.device)
+
+class ModelWidget(BaseModel, ModelMixin):
+    """
+    Class for a widget to inherit multiple classes simultaneously
+    """
+    pass
+
+
+class MLPModel(ModelWidget):
+    """
+    Class for MLP model
+    """
+
+    def __init__(self, params: ParamSet) -> None:
+        """
+        Args:
+            params: (ParamSet): parameters
+        """
+        super().__init__(params)
+
+    def set_data(
+                self,
+                data: Dict
+                ) -> Tuple[
+                        Dict[str, torch.FloatTensor],
+                        Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                        ]:
+        """
+        Unpack data for forwarding of MLP and calculating loss
+        by passing them to device.
+        When deepsurv, period and network are also returned.
+
+        Args:
+            data (Dict): dictionary of data
+
+        Returns:
+            Tuple[
+                Dict[str, torch.FloatTensor],
+                Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                ]: input of model and data for calculating loss.
+        eg.
+        ([inputs], [labels]), or ([inputs], [labels, periods, network]) when deepsurv
+        """
+        in_data = {'inputs': data['inputs'].to(self.device)}
+        labels = {'labels': {label_name: label.to(self.device) for label_name, label in data['labels'].items()}}
+
+        if not any(data['periods']):
+            return in_data, labels
+
+        # When deepsurv
+        labels = {
+                  **labels,
+                  **{'periods': data['periods'].to(self.device), 'network': self.network.to(self.device)}
+                }
+        return in_data, labels
+
+    def __call__(self, in_data: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
+        """
+        Forward
+
+        Args:
+            in_data (Dict[str, torch.Tensor]): data to be input into model
+
+        Returns:
+            Dict[str, torch.Tensor]: output
+        """
+        inputs = in_data['inputs']
+        output = self.network(inputs)
+        return output
+
+
+class CVModel(ModelWidget):
+    """
+    Class for CNN or ViT model
+    """
+    def __init__(self, params: ParamSet) -> None:
+        """
+        Args:
+            params: (ParamSet): parameters
+        """
+        super().__init__(params)
+
+    def set_data(
+                self,
+                data: Dict
+                ) -> Tuple[
+                        Dict[str, torch.FloatTensor],
+                        Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                    ]:
+        """
+        Unpack data for forwarding of CNN or ViT and calculating loss by passing them to device.
+        When deepsurv, period and network are also returned.
+
+        Args:
+            data (Dict): dictionary of data
+
+        Returns:
+            Tuple[
+                Dict[str, torch.FloatTensor],
+                Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                ]: input of model and data for calculating loss.
+        eg.
+        ([image], [labels]), or ([image], [labels, periods, network]) when deepsurv
+        """
+        in_data = {'image': data['image'].to(self.device)}
+        labels = {'labels': {label_name: label.to(self.device) for label_name, label in data['labels'].items()}}
+
+        if not any(data['periods']):
+            return in_data, labels
+
+        # When deepsurv
+        labels = {
+                  **labels,
+                  **{'periods': data['periods'].to(self.device), 'network': self.network.to(self.device)}
+                }
+        return in_data, labels
+
+    def __call__(self, in_data: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
+        """
+        Forward
+
+        Args:
+            in_data (Dict[str, torch.Tensor]): data to be input into model
+
+        Returns:
+            Dict[str, torch.Tensor]: output
+        """
+        image = in_data['image']
+        output = self.network(image)
+        return output
+
+
+class FusionModel(ModelWidget):
+    """
+    Class for MLP+CNN or MLP+ViT model.
+    """
+    def __init__(self, params: ParamSet) -> None:
+        """
+        Args:
+            params: (ParamSet): parameters
+        """
+        super().__init__(params)
+
+    def set_data(
+                self,
+                data: Dict
+                ) -> Tuple[
+                        Dict[str, torch.FloatTensor],
+                        Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                    ]:
+        """
+        Unpack data for forwarding of MLP+CNN or MLP+ViT and calculating loss
+        by passing them to device.
+        When deepsurv, period and network are also returned.
+
+        Args:
+            data (Dict): dictionary of data
+
+        Returns:
+            Tuple[
+                Dict[str, torch.FloatTensor],
+                Dict[str, Union[LabelDict, torch.IntTensor, nn.Module]]
+                ]: input of model and data for calculating loss.
+        eg.
+        ([inputs, image], [labels]), or ([inputs, image], [labels, periods, network]) when deepsurv
+        """
+        in_data = {
+                'inputs': data['inputs'].to(self.device),
+                'image': data['image'].to(self.device)
+                }
+        labels = {'labels': {label_name: label.to(self.device) for label_name, label in data['labels'].items()}}
+
+        if not any(data['periods']):
+            return in_data, labels
+
+        # When deepsurv
+        labels = {
+                  **labels,
+                  **{'periods': data['periods'].to(self.device), 'network': self.network.to(self.device)}
+                }
+        return in_data, labels
+
+    def __call__(self, in_data: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
+        """
+        Forward
+
+        Args:
+            in_data (Dict[str, torch.Tensor]): data to be input into model
+
+        Returns:
+            Dict[str, torch.Tensor]: output
+        """
+        inputs = in_data['inputs']
+        image = in_data['image']
+        output = self.network(inputs, image)
+        return output
+
+
+def create_model(params: ParamSet) -> nn.Module:
+    """
+    Construct model.
+
+    Args:
+        params (ParamSet): parameters
+
+    Returns:
+        nn.Module: model
+    """
+    _isMLPModel = (params.mlp is not None) and (params.net is None)
+    _isCVModel = (params.mlp is None) and (params.net is not None)
+    _isFusionModel = (params.mlp is not None) and (params.net is not None)
+
+    if _isMLPModel:
+        return MLPModel(params)
+    elif _isCVModel:
+        return CVModel(params)
+    elif _isFusionModel:
+        return FusionModel(params)
+    else:
+        raise ValueError(f"Invalid model type: mlp={params.mlp}, net={params.net}.")

lib/logger.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from pathlib import Path
+import logging
+
+
+class BaseLogger:
+    """
+    Class for defining logger.
+    """
+    _unexecuted_configure = True
+
+    @classmethod
+    def get_logger(cls, name: str) -> logging.Logger:
+        """
+        Set logger.
+        Args:
+            name (str): If needed, potentially hierarchical name is desired, eg. lib.net, lib.dataloader, etc.
+                        For the details, see https://docs.python.org/3/library/logging.html?highlight=logging#module-logging.
+        Returns:
+            logging.Logger: logger
+        """
+        if cls._unexecuted_configure:
+            cls._init_logger()
+
+        return logging.getLogger('nervus.{}'.format(name))
+
+    @classmethod
+    def _init_logger(cls) -> None:
+        """
+        Configure logger.
+        """
+        _root_logger = logging.getLogger('nervus')
+        _root_logger.setLevel(logging.DEBUG)
+        formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
+
+        log_dir = Path('logs')
+        log_dir.mkdir(parents=True, exist_ok=True)
+        log_path = Path(log_dir, 'log.log')
+
+        # file handler
+        ## upper warning
+        fh_err = logging.FileHandler(log_path)
+        fh_err.setLevel(logging.WARNING)
+        fh_err.setFormatter(formatter)
+        fh_err.addFilter(lambda log_record: not ('BdbQuit' in str(log_record.exc_info)) and (log_record.levelno >= logging.WARNING))
+        _root_logger.addHandler(fh_err)
+
+        ## lower warning
+        fh = logging.FileHandler(log_path)
+        fh.setLevel(logging.DEBUG)
+        fh.addFilter(lambda log_record: log_record.levelno < logging.WARNING)
+        _root_logger.addHandler(fh)
+
+        # stream handler
+        ## upper warning
+        ch_err = logging.StreamHandler()
+        ch_err.setLevel(logging.WARNING)
+        ch_err.setFormatter(formatter)
+        ch_err.addFilter(lambda log_record: log_record.levelno >= logging.WARNING)
+        _root_logger.addHandler(ch_err)
+
+        ## lower warning
+        ch = logging.StreamHandler()
+        ch.setLevel(logging.DEBUG)
+        ch.addFilter(lambda log_record: log_record.levelno < logging.WARNING)
+        _root_logger.addHandler(ch)
+
+        cls._unexecuted_configure = False
+

lib/metrics.py

@@ -0,0 +1,623 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+from pathlib import Path
+import numpy as np
+import pandas as pd
+from sklearn import metrics
+from sklearn.preprocessing import label_binarize
+import matplotlib.pyplot as plt
+from matplotlib import colors as mcolors
+from .logger import BaseLogger
+from typing import Dict, Union
+
+
+logger = BaseLogger.get_logger(__name__)
+
+
+class MetricsData:
+    """
+    Class to store metrics as class variable.
+    Metrics are defined depending on task.
+
+    For ROC
+        self.fpr: np.ndarray
+        self.tpr: np.ndarray
+        self.auc: float
+
+    For Regression
+        self.y_obs: np.ndarray
+        self.y_pred: np.ndarray
+        self.r2: float
+
+    For DeepSurv
+        self.c_index: float
+    """
+    def __init__(self) -> None:
+        pass
+
+
+class LabelMetrics:
+    """
+    Class to store metrics of each split for each label.
+    """
+    def __init__(self) -> None:
+        """
+        Metrics of split, ie 'val' and 'test'
+        """
+        self.val = MetricsData()
+        self.test = MetricsData()
+
+    def set_label_metrics(self, split: str, attr: str, value: Union[np.ndarray, float]) -> None:
+        """
+        Set value as appropriate metrics of split.
+
+        Args:
+            split (str): split
+            attr (str): attribute name as follows:
+                        classification: 'fpr', 'tpr', or 'auc',
+                        regression:     'y_obs'(ground truth), 'y_pred'(prediction) or 'r2', or
+                        deepsurv:       'c_index'
+            value (Union[np.ndarray,float]): value of attr
+        """
+        setattr(getattr(self, split), attr, value)
+
+    def get_label_metrics(self, split: str, attr: str) -> Union[np.ndarray, float]:
+        """
+        Return value of metrics of split.
+
+        Args:
+            split (str): split
+            attr (str): metrics name
+
+        Returns:
+            Union[np.ndarray,float]: value of attr
+        """
+        return getattr(getattr(self, split), attr)
+
+
+class ROCMixin:
+    """
+    Class for calculating ROC and AUC.
+    """
+    def _set_roc(self, label_metrics: LabelMetrics, split: str, fpr: np.ndarray, tpr: np.ndarray) -> None:
+        """
+        Set fpr, tpr, and auc.
+
+        Args:
+            label_metrics (LabelMetrics): metrics of 'val' and 'test'
+            split (str): 'val' or 'test'
+            fpr (np.ndarray): FPR
+            tpr (np.ndarray): TPR
+
+        self.metrics_kind = 'auc' is defined in class ClsEval below.
+        """
+        label_metrics.set_label_metrics(split, 'fpr', fpr)
+        label_metrics.set_label_metrics(split, 'tpr', tpr)
+        label_metrics.set_label_metrics(split, self.metrics_kind, metrics.auc(fpr, tpr))
+
+    def _cal_label_roc_binary(self, label_name: str, df_group: pd.DataFrame) -> LabelMetrics:
+        """
+        Calculate ROC for binary class.
+
+        Args:
+            label_name (str): label name
+            df_group (pd.DataFrame): likelihood for group
+
+        Returns:
+            LabelMetrics: metrics of 'val' and 'test'
+        """
+        required_columns = [column_name for column_name in df_group.columns if label_name in column_name] + ['split']
+        df_label = df_group[required_columns]
+        POSITIVE = 1
+        positive_pred_name = 'pred_' + label_name + '_' + str(POSITIVE)
+
+        # ! When splits is 'test' only, ie when external dataset, error occurs.
+        label_metrics = LabelMetrics()
+        for split in ['val', 'test']:
+            df_split = df_label.query('split == @split')
+            y_true = df_split[label_name]
+            y_score = df_split[positive_pred_name]
+            _fpr, _tpr, _ = metrics.roc_curve(y_true, y_score)
+            self._set_roc(label_metrics, split, _fpr, _tpr)
+        return label_metrics
+
+    def _cal_label_roc_multi(self, label_name: str, df_group: pd.DataFrame) -> LabelMetrics:
+        """
+        Calculate ROC for multi-class by macro average.
+
+        Args:
+            label_name (str): label name
+            df_group (pd.DataFrame): likelihood for group
+
+        Returns:
+            LabelMetrics: metrics of 'val' and 'test'
+        """
+        required_columns = [column_name for column_name in df_group.columns if label_name in column_name] + ['split']
+        df_label = df_group[required_columns]
+
+        pred_name_list = list(df_label.columns[df_label.columns.str.startswith('pred')])
+        class_list = [int(pred_name.rsplit('_', 1)[-1]) for pred_name in pred_name_list]  # [pred_label_0, pred_label_1, pred_label_2] -> [0, 1, 2]
+        num_classes = len(class_list)
+
+        label_metrics = LabelMetrics()
+        for split in ['val', 'test']:
+            df_split = df_label.query('split == @split')
+            y_true = df_split[label_name]
+            y_true_bin = label_binarize(y_true, classes=class_list)  # Since y_true: List[int], should be class_list: List[int]
+
+            # Compute ROC for each class by OneVsRest
+            _fpr = dict()
+            _tpr = dict()
+            for i, class_name in enumerate(class_list):
+                pred_name = 'pred_' + label_name + '_' + str(class_name)
+                _fpr[class_name], _tpr[class_name], _ = metrics.roc_curve(y_true_bin[:, i], df_split[pred_name])
+
+            # First aggregate all false positive rates
+            all_fpr = np.unique(np.concatenate([_fpr[class_name] for class_name in class_list]))
+
+            # Then interpolate all ROC at this points
+            mean_tpr = np.zeros_like(all_fpr)
+            for class_name in class_list:
+                mean_tpr += np.interp(all_fpr, _fpr[class_name], _tpr[class_name])
+
+            # Finally average it and compute AUC
+            mean_tpr /= num_classes
+
+            _fpr['macro'] = all_fpr
+            _tpr['macro'] = mean_tpr
+            self._set_roc(label_metrics, split, _fpr['macro'], _tpr['macro'])
+        return label_metrics
+
+    def cal_label_metrics(self, label_name: str, df_group: pd.DataFrame) -> LabelMetrics:
+        """
+        Calculate ROC and AUC for label depending on binary or multi-class.
+
+        Args:
+            label_name (str):label name
+            df_group (pd.DataFrame): likelihood for group
+
+        Returns:
+            LabelMetrics: metrics of 'val' and 'test'
+        """
+        pred_name_list = df_group.columns[df_group.columns.str.startswith('pred_' + label_name)]
+        isMultiClass = (len(pred_name_list) > 2)
+        if isMultiClass:
+            label_metrics = self._cal_label_roc_multi(label_name, df_group)
+        else:
+            label_metrics = self._cal_label_roc_binary(label_name, df_group)
+        return label_metrics
+
+
+class YYMixin:
+    """
+    Class for calculating YY and R2.
+    """
+    def _set_yy(self, label_metrics: LabelMetrics, split: str, y_obs: np.ndarray, y_pred: np.ndarray) -> None:
+        """
+        Set ground truth, prediction, and R2.
+
+        Args:
+            label_metrics (LabelMetrics): metrics of 'val' and 'test'
+            split (str): 'val' or 'test'
+            y_obs (np.ndarray): ground truth
+            y_pred (np.ndarray): prediction
+
+        self.metrics_kind = 'r2' is defined in class RegEval below.
+        """
+        label_metrics.set_label_metrics(split, 'y_obs', y_obs.values)
+        label_metrics.set_label_metrics(split, 'y_pred', y_pred.values)
+        label_metrics.set_label_metrics(split, self.metrics_kind, metrics.r2_score(y_obs, y_pred))
+
+    def cal_label_metrics(self, label_name: str, df_group: pd.DataFrame) -> LabelMetrics:
+        """
+        Calculate YY and R2 for label.
+
+        Args:
+            label_name (str): label name
+            df_group (pd.DataFrame): likelihood for group
+
+        Returns:
+            LabelMetrics: metrics of 'val' and 'test'
+        """
+        required_columns = [column_name for column_name in df_group.columns if label_name in column_name] + ['split']
+        df_label = df_group[required_columns]
+        label_metrics = LabelMetrics()
+        for split in ['val', 'test']:
+            df_split = df_label.query('split == @split')
+            y_obs = df_split[label_name]
+            y_pred = df_split['pred_' + label_name]
+            self._set_yy(label_metrics, split, y_obs, y_pred)
+        return label_metrics
+
+
+class C_IndexMixin:
+    """
+    Class for calculating C-Index.
+    """
+    def _set_c_index(
+                    self,
+                    label_metrics: LabelMetrics,
+                    split: str,
+                    periods: pd.Series,
+                    preds: pd.Series,
+                    labels: pd.Series
+                    ) -> None:
+        """
+        Set C-Index.
+
+        Args:
+            label_metrics (LabelMetrics): metrics of 'val' and 'test'
+            split (str): 'val' or 'test'
+            periods (pd.Series): periods
+            preds (pd.Series): prediction
+            labels (pd.Series): label
+
+        self.metrics_kind = 'c_index' is defined in class DeepSurvEval below.
+        """
+        from lifelines.utils import concordance_index
+        value_c_index = concordance_index(periods, (-1)*preds, labels)
+        label_metrics.set_label_metrics(split, self.metrics_kind, value_c_index)
+
+    def cal_label_metrics(self, label_name: str, df_group: pd.DataFrame) -> LabelMetrics:
+        """
+        Calculate C-Index for label.
+
+        Args:
+            label_name (str): label name
+            df_group (pd.DataFrame): likelihood for group
+
+        Returns:
+            LabelMetrics: metrics of 'val' and 'test'
+        """
+        required_columns = [column_name for column_name in df_group.columns if label_name in column_name] + ['periods', 'split']
+        df_label = df_group[required_columns]
+        label_metrics = LabelMetrics()
+        for split in ['val', 'test']:
+            df_split = df_label.query('split == @split')
+            periods = df_split['periods']
+            preds = df_split['pred_' + label_name]
+            labels = df_split[label_name]
+            self._set_c_index(label_metrics, split, periods, preds, labels)
+        return label_metrics
+
+
+class MetricsMixin:
+    """
+    Class to calculate metrics and make summary.
+    """
+    def _cal_group_metrics(self, df_group: pd.DataFrame) -> Dict[str, LabelMetrics]:
+        """
+        Calculate metrics for each group.
+
+        Args:
+            df_group (pd.DataFrame): likelihood for group
+
+        Returns:
+            Dict[str, LabelMetrics]: dictionary of label and its LabelMetrics
+            eg. {{label_1: LabelMetrics(), label_2: LabelMetrics(), ...}
+        """
+        label_list = list(df_group.columns[df_group.columns.str.startswith('label')])
+        group_metrics = dict()
+        for label_name in label_list:
+            label_metrics = self.cal_label_metrics(label_name, df_group)
+            group_metrics[label_name] = label_metrics
+        return group_metrics
+
+    def cal_whole_metrics(self, df_likelihood: pd.DataFrame) -> Dict[str, Dict[str, LabelMetrics]]:
+        """
+        Calculate metrics for all groups.
+
+        Args:
+            df_likelihood (pd.DataFrame) : DataFrame of likelihood
+
+        Returns:
+            Dict[str, Dict[str, LabelMetrics]]: dictionary of group and dictionary of label and its LabelMetrics
+            eg. {
+                groupA: {label_1: LabelMetrics(), label_2: LabelMetrics(), ...},
+                groupB: {label_1: LabelMetrics(), label_2: LabelMetrics()}, ...},
+                ...}
+        """
+        whole_metrics = dict()
+        for group in df_likelihood['group'].unique():
+            df_group = df_likelihood.query('group == @group')
+            whole_metrics[group] = self._cal_group_metrics(df_group)
+        return whole_metrics
+
+    def make_summary(
+                    self,
+                    whole_metrics: Dict[str, Dict[str, LabelMetrics]],
+                    likelihood_path: Path,
+                    metrics_kind: str
+                    ) -> pd.DataFrame:
+        """
+        Make summary.
+
+        Args:
+            whole_metrics (Dict[str, Dict[str, LabelMetrics]]): metrics for all groups
+            likelihood_path (Path): path to likelihood
+            metrics_kind (str): kind of metrics, ie, 'auc', 'r2', or 'c_index'
+
+        Returns:
+            pd.DataFrame: summary
+        """
+        _datetime = likelihood_path.parents[1].name
+        _weight = likelihood_path.stem.replace('likelihood_', '') + '.pt'
+        df_summary = pd.DataFrame()
+        for group, group_metrics in whole_metrics.items():
+            _new = dict()
+            _new['datetime'] = [_datetime]
+            _new['weight'] = [ _weight]
+            _new['group'] = [group]
+            for label_name, label_metrics in group_metrics.items():
+                _val_metrics = label_metrics.get_label_metrics('val', metrics_kind)
+                _test_metrics = label_metrics.get_label_metrics('test', metrics_kind)
+                _new[label_name + '_val_' + metrics_kind] = [f"{_val_metrics:.2f}"]
+                _new[label_name + '_test_' + metrics_kind] = [f"{_test_metrics:.2f}"]
+            df_summary = pd.concat([df_summary, pd.DataFrame(_new)], ignore_index=True)
+
+        df_summary = df_summary.sort_values('group')
+        return df_summary
+
+    def print_metrics(self, df_summary: pd.DataFrame, metrics_kind: str) -> None:
+        """
+        Print metrics.
+
+        Args:
+            df_summary (pd.DataFrame): summary
+            metrics_kind (str): kind of metrics, ie. 'auc', 'r2', or 'c_index'
+        """
+        label_list = list(df_summary.columns[df_summary.columns.str.startswith('label')])  # [label_1_val, label_1_test, label_2_val, label_2_test, ...]
+        num_splits = len(['val', 'test'])
+        _column_val_test_list = [label_list[i:i+num_splits] for i in range(0, len(label_list), num_splits)]  # [[label_1_val, label_1_test], [label_2_val, label_2_test], ...]
+        for _, row in df_summary.iterrows():
+            logger.info(row['group'])
+            for _column_val_test in _column_val_test_list:
+                _label_name = _column_val_test[0].replace('_val', '')
+                _label_name_val = _column_val_test[0]
+                _label_name_test = _column_val_test[1]
+                logger.info(f"{_label_name:<25} val_{metrics_kind}: {row[_label_name_val]:>7}, test_{metrics_kind}: {row[_label_name_test]:>7}")
+
+    def update_summary(self, df_summary: pd.DataFrame, likelihood_path: Path) -> None:
+        """
+        Update summary.
+
+        Args:
+            df_summary (pd.DataFrame): summary to be added to the previous summary
+            likelihood_path (Path): path to likelihood
+        """
+        _project_dir = likelihood_path.parents[3]
+        summary_dir = Path(_project_dir, 'summary')
+        summary_path = Path(summary_dir, 'summary.csv')
+        if summary_path.exists():
+            df_prev = pd.read_csv(summary_path)
+            df_updated = pd.concat([df_prev, df_summary], axis=0)
+        else:
+            summary_dir.mkdir(parents=True, exist_ok=True)
+            df_updated = df_summary
+        df_updated.to_csv(summary_path, index=False)
+
+    def make_metrics(self, likelihood_path: Path) -> None:
+        """
+        Make metrics.
+
+        Args:
+            likelihood_path (Path): path to likelihood
+        """
+        df_likelihood = pd.read_csv(likelihood_path)
+        whole_metrics = self.cal_whole_metrics(df_likelihood)
+        self.make_save_fig(whole_metrics, likelihood_path, self.fig_kind)
+        df_summary = self.make_summary(whole_metrics, likelihood_path, self.metrics_kind)
+        self.print_metrics(df_summary, self.metrics_kind)
+        self.update_summary(df_summary, likelihood_path)
+
+
+class FigROCMixin:
+    """
+    Class to plot ROC.
+    """
+    def _plot_fig_group_metrics(self, group: str, group_metrics: Dict[str, LabelMetrics]) -> plt:
+        """
+        Plot ROC.
+
+        Args:
+            group (str): group
+            group_metrics (Dict[str, LabelMetrics]): dictionary of label and its LabelMetrics
+
+        Returns:
+            plt: ROC
+        """
+        label_list = group_metrics.keys()
+        num_rows = 1
+        num_cols = len(label_list)
+        base_size = 7
+        height = num_rows * base_size
+        width = num_cols * height
+        fig = plt.figure(figsize=(width, height))
+
+        for i, label_name in enumerate(label_list):
+            label_metrics = group_metrics[label_name]
+            offset = i + 1
+            ax_i = fig.add_subplot(
+                                    num_rows,
+                                    num_cols,
+                                    offset,
+                                    title=group + ': ' + label_name,
+                                    xlabel='1 - Specificity',
+                                    ylabel='Sensitivity',
+                                    xmargin=0,
+                                    ymargin=0
+                                    )
+            ax_i.plot(label_metrics.val.fpr, label_metrics.val.tpr, label=f"AUC_val = {label_metrics.val.auc:.2f}", marker='x')
+            ax_i.plot(label_metrics.test.fpr, label_metrics.test.tpr, label=f"AUC_test = {label_metrics.test.auc:.2f}", marker='o')
+            ax_i.grid()
+            ax_i.legend()
+            fig.tight_layout()
+        return fig
+
+
+class FigYYMixin:
+    """
+    Class to plot YY-graph.
+    """
+    def _plot_fig_group_metrics(self, group: str, group_metrics: Dict[str, LabelMetrics]) -> plt:
+        """
+        Plot yy.
+
+        Args:
+            group (str): group
+            group_metrics (Dict[str, LabelMetrics]): dictionary of label and its LabelMetrics
+
+        Returns:
+            plt: YY-graph
+        """
+        label_list = group_metrics.keys()
+        num_splits = len(['val', 'test'])
+        num_rows = 1
+        num_cols = len(label_list) * num_splits
+        base_size = 7
+        height = num_rows * base_size
+        width = num_cols * height
+        fig = plt.figure(figsize=(width, height))
+
+        for i, label_name in enumerate(label_list):
+            label_metrics = group_metrics[label_name]
+            val_offset = (i * num_splits) + 1
+            test_offset = val_offset + 1
+
+            val_ax = fig.add_subplot(
+                                    num_rows,
+                                    num_cols,
+                                    val_offset,
+                                    title=group + ': ' + label_name + '\n' + 'val: Observed-Predicted Plot',
+                                    xlabel='Observed',
+                                    ylabel='Predicted',
+                                    xmargin=0,
+                                    ymargin=0
+                                    )
+
+            test_ax = fig.add_subplot(
+                                    num_rows,
+                                    num_cols,
+                                    test_offset,
+                                    title=group + ': ' + label_name + '\n' + 'test: Observed-Predicted Plot',
+                                    xlabel='Observed',
+                                    ylabel='Predicted',
+                                    xmargin=0,
+                                    ymargin=0
+                                    )
+
+            y_obs_val = label_metrics.val.y_obs
+            y_pred_val = label_metrics.val.y_pred
+
+            y_obs_test = label_metrics.test.y_obs
+            y_pred_test = label_metrics.test.y_pred
+
+            # Plot
+            color = mcolors.TABLEAU_COLORS
+            val_ax.scatter(y_obs_val, y_pred_val, color=color['tab:blue'], label='val')
+            test_ax.scatter(y_obs_test, y_pred_test, color=color['tab:orange'], label='test')
+
+            # Draw diagonal line
+            y_values_val = np.concatenate([y_obs_val.flatten(), y_pred_val.flatten()])
+            y_values_test = np.concatenate([y_obs_test.flatten(), y_pred_test.flatten()])
+
+            y_values_val_min, y_values_val_max, y_values_val_range = np.amin(y_values_val), np.amax(y_values_val), np.ptp(y_values_val)
+            y_values_test_min, y_values_test_max, y_values_test_range = np.amin(y_values_test), np.amax(y_values_test), np.ptp(y_values_test)
+
+            val_ax.plot([y_values_val_min - (y_values_val_range * 0.01), y_values_val_max + (y_values_val_range * 0.01)],
+                        [y_values_val_min - (y_values_val_range * 0.01), y_values_val_max + (y_values_val_range * 0.01)], color='red')
+
+            test_ax.plot([y_values_test_min - (y_values_test_range * 0.01), y_values_test_max + (y_values_test_range * 0.01)],
+                         [y_values_test_min - (y_values_test_range * 0.01), y_values_test_max + (y_values_test_range * 0.01)], color='red')
+
+        fig.tight_layout()
+        return fig
+
+
+class FigMixin:
+    """
+    Class for make and save figure
+    This class is for ROC and YY-graph.
+    """
+    def make_save_fig(self, whole_metrics: Dict[str, Dict[str, LabelMetrics]], likelihood_path: Path, fig_kind: str) -> None:
+        """
+        Make and save figure.
+
+        Args:
+            whole_metrics (Dict[str, Dict[str, LabelMetrics]]): metrics for all groups
+            likelihood_path (Path): path to likelihood
+            fig_kind (str): kind of figure, ie. 'roc' or 'yy'
+        """
+        _datetime_dir = likelihood_path.parents[1]
+        save_dir = Path(_datetime_dir, fig_kind)
+        save_dir.mkdir(parents=True, exist_ok=True)
+        _fig_name = fig_kind + '_' + likelihood_path.stem.replace('likelihood_', '')
+        for group, group_metrics in whole_metrics.items():
+            fig = self._plot_fig_group_metrics(group, group_metrics)
+            save_path = Path(save_dir, group + '_' + _fig_name + '.png')
+            fig.savefig(save_path)
+            plt.close()
+
+
+class ClsEval(MetricsMixin, ROCMixin, FigMixin, FigROCMixin):
+    """
+    Class for calculation metrics for classification.
+    """
+    def __init__(self) -> None:
+        self.fig_kind = 'roc'
+        self.metrics_kind = 'auc'
+
+
+class RegEval(MetricsMixin, YYMixin, FigMixin, FigYYMixin):
+    """
+    Class for calculation metrics for regression.
+    """
+    def __init__(self) -> None:
+        self.fig_kind = 'yy'
+        self.metrics_kind = 'r2'
+
+
+class DeepSurvEval(MetricsMixin, C_IndexMixin):
+    """
+    Class for calculation metrics for DeepSurv.
+    """
+    def __init__(self) -> None:
+        self.fig_kind = None
+        self.metrics_kind = 'c_index'
+
+    def make_metrics(self, likelihood_path: Path) -> None:
+        """
+        Make metrics, substantially this method handles everything all.
+
+        Args:
+            likelihood_path (Path): path to likelihood
+
+        Overwrite def make_metrics() in class MetricsMixin by deleting self.make_save_fig(),
+        because of no need to plot and save figure.
+        """
+        df_likelihood = pd.read_csv(likelihood_path)
+        whole_metrics = self.cal_whole_metrics(df_likelihood)
+        df_summary = self.make_summary(whole_metrics, likelihood_path, self.metrics_kind)
+        self.print_metrics(df_summary, self.metrics_kind)
+        self.update_summary(df_summary, likelihood_path)
+
+
+def set_eval(task: str) -> Union[ClsEval, RegEval, DeepSurvEval]:
+    """
+    Set class for evaluation depending on task depending on task.
+
+    Args:
+        task (str): task
+
+    Returns:
+        Union[ClsEval, RegEval, DeepSurvEval]: class for evaluation
+    """
+    if task == 'classification':
+        return ClsEval()
+    elif task == 'regression':
+        return RegEval()
+    elif task == 'deepsurv':
+        return DeepSurvEval()
+    else:
+        raise ValueError(f"Invalid task: {task}.")

lib/options.py

@@ -0,0 +1,655 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import argparse
+from distutils.util import strtobool
+from pathlib import Path
+import pandas as pd
+import json
+import torch
+from .logger import BaseLogger
+from typing import List, Dict, Tuple, Union
+
+
+logger = BaseLogger.get_logger(__name__)
+
+
+class Options:
+    """
+    Class for options.
+    """
+    def __init__(self,  datetime: str = None, isTrain: bool = None) -> None:
+        """
+            Args:
+            datetime (str, optional): date time    Args:
+            isTrain (bool, optional): Variable indicating whether training or not. Defaults to None.
+        """
+        self.parser = argparse.ArgumentParser(description='Options for training or test')
+
+        # CSV
+        self.parser.add_argument('--csvpath', type=str, required=True, help='path to csv for training or test')
+
+        # GPU Ids
+        self.parser.add_argument('--gpu_ids', type=str, default='cpu', help='gpu ids: e.g. 0, 0-1-2, 0-2. Use cpu for CPU (Default: cpu)')
+
+        if isTrain:
+            # Task
+            self.parser.add_argument('--task', type=str, required=True, choices=['classification', 'regression', 'deepsurv'], help='Task')
+
+            # Model
+            self.parser.add_argument('--model',      type=str, required=True, help='model: MLP, CNN, ViT, or MLP+(CNN or ViT)')
+            self.parser.add_argument('--pretrained', type=strtobool, default=False, help='For use of pretrained model(CNN or ViT)')
+
+            # Training and Internal validation
+            self.parser.add_argument('--criterion', type=str,   required=True, choices=['CEL', 'MSE', 'RMSE', 'MAE', 'NLL'], help='criterion')
+            self.parser.add_argument('--optimizer', type=str,   default='Adam', choices=['SGD', 'Adadelta', 'RMSprop', 'Adam', 'RAdam'], help='optimizer')
+            self.parser.add_argument('--lr',        type=float,                metavar='N', help='learning rate')
+            self.parser.add_argument('--epochs',    type=int,   default=10,    metavar='N', help='number of epochs (Default: 10)')
+
+            # Batch size
+            self.parser.add_argument('--batch_size', type=int,  required=True, metavar='N', help='batch size in training')
+
+            # Preprocess for image
+            self.parser.add_argument('--augmentation',       type=str,  default='no', choices=['xrayaug', 'trivialaugwide', 'randaug', 'no'], help='kind of augmentation')
+            self.parser.add_argument('--normalize_image',    type=str,                choices=['yes', 'no'], default='yes', help='image normalization: yes, no (Default: yes)')
+
+            # Sampler
+            self.parser.add_argument('--sampler',            type=str,  default='no',  choices=['yes', 'no'], help='sample data in training or not, yes or no')
+
+            # Input channel
+            self.parser.add_argument('--in_channel',         type=int,  required=True, choices=[1, 3], help='channel of input image')
+            self.parser.add_argument('--vit_image_size',     type=int,  default=0,                     help='input image size for ViT. Set 0 if not used ViT (Default: 0)')
+
+            # Weight saving strategy
+            self.parser.add_argument('--save_weight_policy', type=str,  choices=['best', 'each'], default='best', help='Save weight policy: best, or each(ie. save each time loss decreases when multi-label output) (Default: best)')
+
+        else:
+            # Directory of weight at training
+            self.parser.add_argument('--weight_dir',         type=str,  default=None, help='directory of weight to be used when test. If None, the latest one is selected')
+
+            # Test bash size
+            self.parser.add_argument('--test_batch_size',    type=int,  default=1, metavar='N', help='batch size for test (Default: 1)')
+
+            # Splits for test
+            self.parser.add_argument('--test_splits',        type=str, default='train-val-test', help='splits for test: e.g. test, val-test, train-val-test. (Default: train-val-test)')
+
+        self.args = self.parser.parse_args()
+
+        if datetime is not None:
+            self.args.datetime = datetime
+
+        assert isinstance(isTrain, bool), 'isTrain should be bool.'
+        self.args.isTrain = isTrain
+
+    def get_args(self) -> argparse.Namespace:
+        """
+        Return arguments.
+
+        Returns:
+            argparse.Namespace: arguments
+        """
+        return self.args
+
+
+class CSVParser:
+    """
+    Class to get information of csv and cast csv.
+    """
+    def __init__(self, csvpath: str, task: str, isTrain: bool = None) -> None:
+        """
+        Args:
+            csvpath (str): path to csv
+            task (str): task
+            isTrain (bool): if training or not
+        """
+        self.csvpath = csvpath
+        self.task = task
+
+        _df_source = pd.read_csv(self.csvpath)
+        _df_source = _df_source[_df_source['split'] != 'exclude']
+
+        self.input_list = list(_df_source.columns[_df_source.columns.str.startswith('input')])
+        self.label_list = list(_df_source.columns[_df_source.columns.str.startswith('label')])
+        if self.task == 'deepsurv':
+            _period_name_list = list(_df_source.columns[_df_source.columns.str.startswith('period')])
+            assert (len(_period_name_list) == 1), f"One column of period should be contained in {self.csvpath} when deepsurv."
+            self.period_name = _period_name_list[0]
+
+        _df_source = self._cast(_df_source, self.task)
+
+        # If no column of group, add it.
+        if 'group' not in _df_source.columns:
+            _df_source = _df_source.assign(group='all')
+
+        self.df_source = _df_source
+
+        if isTrain:
+            self.mlp_num_inputs = len(self.input_list)
+            self.num_outputs_for_label = self._define_num_outputs_for_label(self.df_source, self.label_list, self.task)
+
+    def _cast(self, df_source: pd.DataFrame, task: str) -> pd.DataFrame:
+        """
+        Make dictionary of cast depending on task.
+
+        Args:
+            df_source (pd.DataFrame): excluded DataFrame
+            task: (str): task
+
+        Returns:
+            DataFrame: csv excluded and cast depending on task
+        """
+        _cast_input = {input_name: float for input_name in self.input_list}
+
+        if task == 'classification':
+            _cast_label = {label_name: int for label_name in self.label_list}
+            _casts = {**_cast_input, **_cast_label}
+            df_source = df_source.astype(_casts)
+            return df_source
+
+        elif task == 'regression':
+            _cast_label = {label_name: float for label_name in self.label_list}
+            _casts = {**_cast_input, **_cast_label}
+            df_source = df_source.astype(_casts)
+            return df_source
+
+        elif task == 'deepsurv':
+            _cast_label = {label_name: int for label_name in self.label_list}
+            _cast_period = {self.period_name: int}
+            _casts = {**_cast_input, **_cast_label, **_cast_period}
+            df_source = df_source.astype(_casts)
+            return df_source
+
+        else:
+            raise ValueError(f"Invalid task: {self.task}.")
+
+    def _define_num_outputs_for_label(self, df_source: pd.DataFrame, label_list: List[str], task :str) -> Dict[str, int]:
+        """
+        Define the number of outputs for each label.
+
+        Args:
+            df_source (pd.DataFrame): DataFrame of csv
+            label_list (List[str]): list of labels
+                task: str
+
+        Returns:
+            Dict[str, int]: dictionary of the number of outputs for each label
+            eg.
+                classification:       _num_outputs_for_label = {label_A: 2, label_B: 3, ...}
+                regression, deepsurv: _num_outputs_for_label = {label_A: 1, label_B: 1, ...}
+                deepsurv:             _num_outputs_for_label = {label_A: 1}
+        """
+        if task == 'classification':
+            _num_outputs_for_label = {label_name: df_source[label_name].nunique() for label_name in label_list}
+            return _num_outputs_for_label
+
+        elif (task == 'regression') or (task == 'deepsurv'):
+            _num_outputs_for_label = {label_name: 1 for label_name in label_list}
+            return _num_outputs_for_label
+
+        else:
+            raise ValueError(f"Invalid task: {task}.")
+
+
+def _parse_model(model_name: str) -> Tuple[Union[str, None], Union[str, None]]:
+    """
+    Parse model name.
+
+    Args:
+        model_name (str): model name (eg. MLP, ResNey18, or MLP+ResNet18)
+
+    Returns:
+        Tuple[str, str]: MLP, CNN or Vision Transformer name
+        eg. 'MLP', 'ResNet18', 'MLP+ResNet18' ->
+            ['MLP'], ['ResNet18'], ['MLP', 'ResNet18']
+    """
+    _model = model_name.split('+')
+    mlp = 'MLP' if 'MLP' in _model else None
+    _net = [_n for _n in _model if _n != 'MLP']
+    net = _net[0] if _net != [] else None
+    return mlp, net
+
+
+def _parse_gpu_ids(gpu_ids: str) -> List[int]:
+    """
+    Parse GPU ids concatenated with '-' to list of integers of GPU ids.
+    eg. '0-1-2' -> [0, 1, 2], '-1' -> []
+
+    Args:
+        gpu_ids (str): GPU Ids
+
+    Returns:
+        List[int]: list of GPU ids
+    """
+    if (gpu_ids == 'cpu') or (gpu_ids == 'cpu\r'):
+        str_ids = []
+    else:
+        str_ids = gpu_ids.split('-')
+    _gpu_ids = []
+    for str_id in str_ids:
+        id = int(str_id)
+        if id >= 0:
+            _gpu_ids.append(id)
+    return _gpu_ids
+
+
+def _get_latest_weight_dir() -> str:
+    """
+    Return the latest path to directory of weight made at training.
+
+    Returns:
+        str: path to directory of the latest weight
+        eg. 'results/<project>/trials/2022-09-30-15-56-60/weights'
+    """
+    _weight_dirs = list(Path('results').glob('*/trials/*/weights'))
+    assert (_weight_dirs != []), 'No directory of weight.'
+    weight_dir = max(_weight_dirs, key=lambda weight_dir: weight_dir.stat().st_mtime)
+    return str(weight_dir)
+
+
+def _collect_weight_paths(weight_dir: str) -> List[str]:
+    """
+    Return list of weight paths.
+
+    Args:
+        weight_dir (str): path to directory of weights
+
+    Returns:
+        List[str]: list of weight paths
+    """
+    _weight_paths = list(Path(weight_dir).glob('*.pt'))
+    assert _weight_paths != [], f"No weight in {weight_dir}."
+    _weight_paths.sort(key=lambda path: path.stat().st_mtime)
+    _weight_paths = [str(weight_path) for weight_path in _weight_paths]
+    return _weight_paths
+
+
+class ParamTable:
+    """
+    Class to make table to dispatch parameters by group.
+    """
+    def __init__(self) -> None:
+        # groups
+        # key is abbreviation, value is group name
+        self.groups = {
+                        'mo': 'model',
+                        'dl': 'dataloader',
+                        'trc': 'train_conf',
+                        'tsc': 'test_conf',
+                        'sa': 'save',
+                        'lo': 'load',
+                        'trp': 'train_print',
+                        'tsp': 'test_print'
+                        }
+
+        mo = self.groups['mo']
+        dl = self.groups['dl']
+        trc = self.groups['trc']
+        tsc = self.groups['tsc']
+        sa = self.groups['sa']
+        lo = self.groups['lo']
+        trp = self.groups['trp']
+        tsp = self.groups['tsp']
+
+        # The below shows that which group each parameter dispatches to.
+        self.dispatch = {
+                'datetime': [sa],
+                'project': [sa, trp, tsp],
+                'csvpath': [sa, trp, tsp],
+                'task': [dl, tsc, sa, lo, trp, tsp],
+                'isTrain': [dl, trp, tsp],
+
+                'model': [sa, lo, trp, tsp],
+                'vit_image_size': [mo, sa, lo, trp, tsp],
+                'pretrained': [mo, sa, trp],
+                'mlp': [mo, dl],
+                'net': [mo, dl],
+
+                'weight_dir': [tsc, tsp],
+                'weight_paths': [tsc],
+
+                'criterion': [trc, sa, trp],
+                'optimizer': [trc, sa, trp],
+                'lr': [trc, sa, trp],
+                'epochs': [trc, sa, trp],
+
+                'batch_size': [dl, sa, trp],
+                'test_batch_size': [dl, tsp],
+                'test_splits': [tsc, tsp],
+
+                'in_channel': [mo, dl, sa, lo, trp, tsp],
+                'normalize_image': [dl, sa, lo, trp, tsp],
+                'augmentation': [dl, sa, trp],
+                'sampler': [dl, sa, trp],
+
+                'df_source': [dl],
+                'label_list': [dl, trc, sa, lo],
+                'input_list': [dl, sa, lo],
+                'period_name': [dl, sa, lo],
+                'mlp_num_inputs': [mo, sa, lo],
+                'num_outputs_for_label': [mo, sa, lo, tsc],
+
+                'save_weight_policy': [sa, trp, trc],
+                'scaler_path': [dl, tsp],
+                'save_datetime_dir': [trc, tsc, trp, tsp],
+
+                'gpu_ids': [trc, tsc, sa, trp, tsp],
+                'device': [mo, trc, tsc],
+                'dataset_info': [trc, sa, trp, tsp]
+                }
+
+        self.table = self._make_table()
+
+    def _make_table(self) -> pd.DataFrame:
+        """
+        Make table to dispatch parameters by group.
+
+        Returns:
+            pd.DataFrame: table which shows that which group each parameter belongs to.
+        """
+        df_table = pd.DataFrame([], index=self.dispatch.keys(), columns=self.groups.values()).fillna('no')
+        for param, grps in self.dispatch.items():
+            for grp in grps:
+                df_table.loc[param, grp] = 'yes'
+
+        df_table = df_table.reset_index()
+        df_table = df_table.rename(columns={'index': 'parameter'})
+        return df_table
+
+    def get_by_group(self, group_name: str) -> List[str]:
+        """
+        Return list of parameters which belong to group
+
+        Args:
+            group_name (str): group name
+
+        Returns:
+            List[str]: list of parameters
+        """
+        _df_table = self.table
+        _param_names = _df_table[_df_table[group_name] == 'yes']['parameter'].tolist()
+        return _param_names
+
+
+Param_Table = ParamTable()
+
+
+class ParamSet:
+    """
+    Class to store required parameters for each group.
+    """
+    pass
+
+
+def _dispatch_by_group(args: argparse.Namespace, group_name: str) -> ParamSet:
+    """
+    Dispatch parameters depending on group.
+
+    Args:
+        args (argparse.Namespace): arguments
+        group_name (str): group
+
+    Returns:
+        ParamSet: class containing parameters for group
+    """
+    _param_names = Param_Table.get_by_group(group_name)
+    param_set = ParamSet()
+    for param_name in _param_names:
+        if hasattr(args, param_name):
+            _arg = getattr(args, param_name)
+            setattr(param_set, param_name, _arg)
+    return param_set
+
+
+def save_parameter(params: ParamSet, save_path: str) -> None:
+    """
+    Save parameters.
+
+    Args:
+        params (ParamSet): parameters
+
+        save_path (str): save path for parameters
+    """
+    _saved = {_param: _arg for _param, _arg in vars(params).items()}
+    save_dir = Path(save_path).parents[0]
+    save_dir.mkdir(parents=True, exist_ok=True)
+    with open(save_path, 'w') as f:
+        json.dump(_saved, f, indent=4)
+
+
+def _retrieve_parameter(parameter_path: str) -> Dict[str, Union[str, int, float]]:
+    """
+    Retrieve only parameters required at test from parameters at training.
+
+    Args:
+        parameter_path (str): path to parameter_path
+
+    Returns:
+        Dict[str, Union[str, int, float]]: parameters at training
+    """
+    with open(parameter_path) as f:
+        params = json.load(f)
+
+    _required = Param_Table.get_by_group('load')
+    params = {p: v for p, v in params.items() if p in _required}
+    return params
+
+
+def print_parameter(params: ParamSet) -> None:
+    """
+    Print parameters.
+
+    Args:
+        params (ParamSet): parameters
+    """
+
+    LINE_LENGTH = 82
+
+    if params.isTrain:
+        phase = 'Training'
+    else:
+        phase = 'Test'
+
+    _header = f" Configuration of {phase} "
+    _padding = (LINE_LENGTH - len(_header) + 1) // 2  # round up
+    _header = ('-' * _padding) + _header + ('-' * _padding) + '\n'
+
+    _footer = ' End '
+    _padding = (LINE_LENGTH - len(_footer) + 1) // 2
+    _footer = ('-' * _padding) + _footer + ('-' * _padding) + '\n'
+
+    message = ''
+    message += _header
+
+    _params_dict = vars(params)
+    del _params_dict['isTrain']
+    for _param, _arg in _params_dict.items():
+        _str_arg = _arg2str(_param, _arg)
+        message += f"{_param:>30}: {_str_arg:<40}\n"
+
+    message += _footer
+    logger.info(message)
+
+
+def _arg2str(param: str, arg: Union[str, int, float]) -> str:
+        """
+        Convert argument to string.
+
+        Args:
+            param (str): parameter
+            arg (Union[str, int, float]): argument
+
+        Returns:
+            str: strings of argument
+        """
+        if param == 'lr':
+            if arg is None:
+                str_arg = 'Default'
+            else:
+                str_arg = str(param)
+            return str_arg
+        elif param == 'gpu_ids':
+            if arg == []:
+                str_arg = 'CPU selected'
+            else:
+                str_arg = f"{arg}  (Primary GPU:{arg[0]})"
+            return str_arg
+        elif param == 'test_splits':
+            str_arg = ', '.join(arg)
+            return str_arg
+        elif param == 'dataset_info':
+            str_arg = ''
+            for i, (split, total) in enumerate(arg.items()):
+                if i < len(arg) - 1:
+                    str_arg += (f"{split}_data={total}, ")
+                else:
+                    str_arg += (f"{split}_data={total}")
+            return str_arg
+        else:
+            if arg is None:
+                str_arg = 'No need'
+            else:
+                str_arg = str(arg)
+            return str_arg
+
+
+def _check_if_valid_criterion(task: str = None, criterion: str = None) -> None:
+    """
+    Check if criterion is valid.
+
+    Args:
+        task (str): task
+        criterion (str): criterion
+    """
+    valid_criterion = {
+        'classification': ['CEL'],
+        'regression': ['MSE', 'RMSE', 'MAE'],
+        'deepsurv': ['NLL']
+    }
+    if criterion in valid_criterion[task]:
+        pass
+    else:
+        raise ValueError(f"Invalid criterion for task: task={task}, criterion={criterion}.")
+
+
+def _train_parse(args: argparse.Namespace) -> Dict[str, ParamSet]:
+    """
+    Parse parameters required at training.
+
+    Args:
+        args (argparse.Namespace): arguments
+
+    Returns:
+        Dict[str, ParamSet]: parameters dispatched by group
+    """
+    # Check if criterion is valid.
+    _check_if_valid_criterion(task=args.task, criterion=args.criterion)
+
+    args.project = Path(args.csvpath).stem
+    args.gpu_ids = _parse_gpu_ids(args.gpu_ids)
+    args.device = torch.device(f"cuda:{args.gpu_ids[0]}") if args.gpu_ids != [] else torch.device('cpu')
+    args.mlp, args.net = _parse_model(args.model)
+    args.pretrained = bool(args.pretrained)  # strtobool('False') = 0 (== False)
+    args.save_datetime_dir = str(Path('results', args.project, 'trials', args.datetime))
+
+    # Parse csv
+    _csvparser = CSVParser(args.csvpath, args.task, args.isTrain)
+    args.df_source = _csvparser.df_source
+    args.dataset_info = {split: len(args.df_source[args.df_source['split'] == split]) for split in ['train', 'val']}
+    args.input_list = _csvparser.input_list
+    args.label_list = _csvparser.label_list
+    args.mlp_num_inputs = _csvparser.mlp_num_inputs
+    args.num_outputs_for_label = _csvparser.num_outputs_for_label
+    if args.task == 'deepsurv':
+        args.period_name = _csvparser.period_name
+
+    # Dispatch parameters
+    return {
+            'args_model': _dispatch_by_group(args, 'model'),
+            'args_dataloader': _dispatch_by_group(args, 'dataloader'),
+            'args_conf': _dispatch_by_group(args, 'train_conf'),
+            'args_print': _dispatch_by_group(args, 'train_print'),
+            'args_save': _dispatch_by_group(args, 'save')
+            }
+
+
+def _test_parse(args: argparse.Namespace) -> Dict[str, ParamSet]:
+    """
+    Parse parameters required at test.
+
+    Args:
+        args (argparse.Namespace): arguments
+
+    Returns:
+        Dict[str, ParamSet]: parameters dispatched by group
+    """
+    args.project = Path(args.csvpath).stem
+    args.gpu_ids = _parse_gpu_ids(args.gpu_ids)
+    args.device = torch.device(f"cuda:{args.gpu_ids[0]}") if args.gpu_ids != [] else torch.device('cpu')
+
+    # Collect weight paths
+    if args.weight_dir is None:
+        args.weight_dir = _get_latest_weight_dir()
+    args.weight_paths = _collect_weight_paths(args.weight_dir)
+
+    # Get datetime at training
+    _train_datetime_dir = Path(args.weight_dir).parents[0]
+    _train_datetime = _train_datetime_dir.name
+
+    args.save_datetime_dir = str(Path('results', args.project, 'trials', _train_datetime))
+
+    # Retrieve only parameters required at test
+    _parameter_path = str(Path(_train_datetime_dir, 'parameters.json'))
+    params = _retrieve_parameter(_parameter_path)
+    for _param, _arg in params.items():
+        setattr(args, _param, _arg)
+
+    # When test, the followings are always fixed.
+    args.augmentation = 'no'
+    args.sampler = 'no'
+    args.pretrained = False
+
+    args.mlp, args.net = _parse_model(args.model)
+    if args.mlp is not None:
+        args.scaler_path = str(Path(_train_datetime_dir, 'scaler.pkl'))
+
+    # Parse csv
+    _csvparser = CSVParser(args.csvpath, args.task)
+    args.df_source = _csvparser.df_source
+
+    # Align test_splits
+    args.test_splits = args.test_splits.split('-')
+    _splits = args.df_source['split'].unique().tolist()
+    if set(_splits) < set(args.test_splits):
+        args.test_splits = _splits
+
+    args.dataset_info = {split: len(args.df_source[args.df_source['split'] == split]) for split in args.test_splits}
+
+    # Dispatch parameters
+    return {
+            'args_model': _dispatch_by_group(args, 'model'),
+            'args_dataloader': _dispatch_by_group(args, 'dataloader'),
+            'args_conf': _dispatch_by_group(args, 'test_conf'),
+            'args_print': _dispatch_by_group(args, 'test_print')
+            }
+
+def set_options(datetime_name: str = None, phase: str = None) -> argparse.Namespace:
+    """
+    Parse options for training or test.
+
+    Args:
+        datetime_name (str, optional): datetime name. Defaults to None.
+        phase (str, optional): train or test. Defaults to None.
+
+    Returns:
+        argparse.Namespace: arguments
+    """
+    if phase == 'train':
+        opt = Options(datetime=datetime_name, isTrain=True)
+        _args = opt.get_args()
+        args = _train_parse(_args)
+        return args
+    else:
+        opt = Options(isTrain=False)
+        _args = opt.get_args()
+        args = _test_parse(_args)
+        return args

parameters.json

@@ -0,0 +1,36 @@
+{
+    "datetime": "2025-01-25-04-52-01",
+    "project": "CXp_checker_patientbase",
+    "csvpath": "materials/docs/CXp_checker_patientbase.csv",
+    "task": "classification",
+    "model": "EfficientNetB4",
+    "vit_image_size": 0,
+    "pretrained": true,
+    "criterion": "CEL",
+    "optimizer": "Adam",
+    "lr": null,
+    "epochs": 100,
+    "batch_size": 32,
+    "bit_depth": 8,
+    "in_channel": 1,
+    "augmentation": "trivialaugwide",
+    "normalize_image": "yes",
+    "sampler": "distributed",
+    "label_list": [
+        "label_round",
+        "label_APorPA"
+    ],
+    "input_list": [],
+    "mlp_num_inputs": 0,
+    "num_outputs_for_label": {
+        "label_round": 4,
+        "label_APorPA": 3
+    },
+    "save_weight_policy": "best",
+    "gpu_ids": [
+        0,
+        1,
+        2,
+        3
+    ]
+}

requirements.txt

@@ -0,0 +1,5 @@
+torch
+torchvision
+gradio
+matplotlib
+scikit-learn

weight_epoch-011_best.pt

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