36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185 | def block_predict(config):
""" Internal code block for AutoML inference.
This function is called by the Command-Line-Interface (CLI) of AUCMEDI.
Args:
config (dict): Configuration dictionary containing all required
parameters for performing an AutoML inference.
The following attributes are stored in the `config` dictionary:
Attributes:
path_imagedir (str): Path to the directory containing the images for prediction.
path_modeldir (str): Path to the model directory in which fitted model weights and metadata are stored.
path_pred (str): Path to the output file in which predicted csv file should be stored.
xai_method (str or None): Key for XAI method.
xai_directory (str or None): Path to the output directory in which predicted image xai heatmaps should be stored.
batch_size (int): Number of samples inside a single batch.
workers (int): Number of workers/threads which preprocess batches during runtime.
"""
# Peak into the dataset via the input interface
ds = input_interface("directory",
config["path_imagedir"],
path_data=None,
training=False,
ohe=False,
image_format=None)
(index_list, _, _, _, image_format) = ds
# Verify existence of input directory
if not os.path.exists(config["path_modeldir"]):
raise FileNotFoundError(config["path_modeldir"])
# Load metadata from training
path_meta = os.path.join(config["path_modeldir"], "meta.training.json")
with open(path_meta, "r") as json_file:
meta_training = json.load(json_file)
# Define neural network parameters
nn_paras = {"n_labels": 1, # placeholder
"channels": 1, # placeholder
}
# Select input shape for 3D
if meta_training["three_dim"]:
nn_paras["input_shape"] = tuple(meta_training["shape_3D"])
# Subfunctions
sf_list = []
if meta_training["three_dim"]:
sf_norm = Standardize(mode="grayscale")
sf_pad = Padding(mode="constant", shape=meta_training["shape_3D"])
sf_crop = Crop(shape=meta_training["shape_3D"], mode="random")
sf_chromer = Chromer(target="rgb")
sf_list.extend([sf_norm, sf_pad, sf_crop, sf_chromer])
# Define parameters for DataGenerator
paras_datagen = {
"path_imagedir": config["path_imagedir"],
"batch_size": config["batch_size"],
"img_aug": None,
"subfunctions": sf_list,
"prepare_images": False,
"sample_weights": None,
"seed": None,
"image_format": image_format,
"workers": config["workers"],
"shuffle": False,
"grayscale": False,
}
if not meta_training["three_dim"] : paras_datagen["loader"] = image_loader
else : paras_datagen["loader"] = sitk_loader
# Apply MIC pipelines
if meta_training["analysis"] == "minimal":
# Setup neural network
if not meta_training["three_dim"]:
arch_dim = "2D." + meta_training["architecture"]
else : arch_dim = "3D." + meta_training["architecture"]
model = NeuralNetwork(architecture=arch_dim, **nn_paras)
# Build DataGenerator
pred_gen = DataGenerator(samples=index_list,
labels=None,
resize=model.meta_input,
standardize_mode=model.meta_standardize,
**paras_datagen)
# Load model
path_model = os.path.join(config["path_modeldir"], "model.last.keras")
model.load(path_model)
# Start model inference
preds = model.predict(prediction_generator=pred_gen)
elif meta_training["analysis"] == "standard":
# Setup neural network
if not meta_training["three_dim"]:
arch_dim = "2D." + meta_training["architecture"]
else : arch_dim = "3D." + meta_training["architecture"]
model = NeuralNetwork(architecture=arch_dim, **nn_paras)
# Build DataGenerator
pred_gen = DataGenerator(samples=index_list,
labels=None,
resize=model.meta_input,
standardize_mode=model.meta_standardize,
**paras_datagen)
# Load model
path_model = os.path.join(config["path_modeldir"],
"model.best_loss.keras")
model.load(path_model)
# Start model inference via Augmenting
preds = predict_augmenting(model, pred_gen)
else:
# Build multi-model list
model_list = []
for arch in meta_training["architecture"]:
if not meta_training["three_dim"] : arch_dim = "2D." + arch
else : arch_dim = "3D." + arch
model_part = NeuralNetwork(architecture=arch_dim, **nn_paras)
model_list.append(model_part)
el = Composite(model_list, metalearner=meta_training["metalearner"],
k_fold=len(meta_training["architecture"]))
# Build DataGenerator
pred_gen = DataGenerator(samples=index_list,
labels=None,
resize=None,
standardize_mode=None,
**paras_datagen)
# Load composite model directory
el.load(config["path_modeldir"])
# Start model inference via ensemble learning
preds = el.predict(pred_gen)
# Create prediction dataset
df_index = pd.DataFrame(data={"SAMPLE": index_list})
df_pd = pd.DataFrame(data=preds, columns=meta_training["class_names"])
df_merged = pd.concat([df_index, df_pd], axis=1, sort=False)
df_merged.sort_values(by=["SAMPLE"], inplace=True)
# Store predictions to disk
df_merged.to_csv(config["path_pred"], index=False)
# Create XAI heatmaps
if config["xai_method"] is not None and config["xai_directory"] is not None:
if meta_training["analysis"] == "advanced":
raise ValueError("XAI is only supported for single model pipelines!")
# Create xai output directory
if not os.path.exists(config["xai_directory"]):
os.mkdir(config["xai_directory"])
# Run XAI decoder
xai_decoder(pred_gen, model, preds=preds, method=config["xai_method"],
layerName=None, alpha=0.4, out_path=config["xai_directory"])
|