|
|
|
|
|
|
|
|
|
|
|
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) |
|
|
|
