#!/usr/bin/env python
# Copyright 2016-2023 Biomedical Imaging Group Rotterdam, Departments of
# Medical Informatics and Radiology, Erasmus MC, Rotterdam, The Netherlands
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from .simpleworc import SimpleWORC, _for_all_methods, _error_bulldozer
import os
import fastr
from .helpers.processing import convert_radiomix_features
[docs]@_for_all_methods(_error_bulldozer)
class BasicWORC(SimpleWORC):
"""Facade around the main WORC object for simple interaction.
Based upon the SimpleWORC object, but with additional functionality:
- Sources of WORC (e.g. images_train, segmentations_train, ...)
can be directly assessed.
Please also see the `WORCTutorial Github <https://github.com/MStarmans91/WORCTutorial/>`_.
"""
[docs] def __init__(self, name='WORC'):
super().__init__(name)
# All hidden objects are now linked to unhidden objects
self.images_train = []
self.images_test = []
self.features_train = []
self.features_test = []
self.segmentations_train = []
self.segmentations_test = []
self.masks_train = []
self.masks_test = []
self.metadata_train = []
self.metadata_test = []
self.semantics_file_train = []
self.elastix_parameter_file = []
self.semantics_file_test = []
self.radiomix_feature_file = None
self.labels_file_train = None
self.labels_file_test = None
self.label_names = []
self.trained_model = None
self.config_files = []
self.fixed_splits = None
[docs] def execute(self):
"""Execute the experiment.
Before executing the actual experiment, this function will first run several validators
and check the provided setup to make sure some of the most common
made error are caught before running the experiment.
"""
# Run some sanity checks before starting the execution
self._validate()
if self._fixed_splits:
self._worc.fixedsplits = self._fixed_splits
elif self.fixed_splits:
self._worc.fixedsplits = self.fixed_splits
if self._radiomix_feature_file:
# Convert radiomix features and use those as inputs
output_folder = os.path.join(fastr.config.mounts['tmp'],
'Radiomix_features')
# Check if output folder exists: otherwise create
if not os.path.exists(output_folder):
os.mkdir(output_folder)
# convert the features
convert_radiomix_features(self._radiomix_feature_file, output_folder)
# Set the newly created feature files as the WORC input
self.features_from_this_directory(output_folder)
# Training sources
if self.images_train:
self._worc.images_train = self.images_train
elif self._images_train:
self._worc.images_train = self._images_train
if self.features_train:
self._worc.features_train = self.features_train
elif self._features_train:
self._worc.features_train = self._features_train
if self.segmentations_train:
self._worc.segmentations_train = self.segmentations_train
elif self._segmentations_train:
self._worc.segmentations_train = self._segmentations_train
if self.masks_train:
self._worc.masks_train = self.masks_train
elif self._masks_train:
self._worc.masks_train = self._masks_train
if self.labels_file_train:
self._worc.labels_train = self.labels_file_train
elif self._labels_file_train:
self._worc.labels_train = self._labels_file_train
if self.semantics_file_train:
self._worc.semantics_train = self.semantics_file_train
elif self._semantics_file_train:
self._worc.semantics_train = self._semantics_file_train
# Testing sources
if self.images_test:
self._worc.images_test = self.images_test
elif self._images_test:
self._worc.images_test = self._images_test
if self.features_test:
self._worc.features_test = self.features_test
elif self._features_test:
self._worc.features_test = self._features_test
if self.segmentations_test:
self._worc.segmentations_test = self.segmentations_test
elif self._segmentations_test:
self._worc.segmentations_test = self._segmentations_test
if self.masks_test:
self._worc.masks_test = self.masks_test
elif self._masks_test:
self._worc.masks_test = self._masks_test
if self.labels_file_test:
self._worc.labels_test = self.labels_file_test
elif self._labels_file_test:
self._worc.labels_test = self._labels_file_test
if self.semantics_file_test:
self._worc.semantics_test = self.semantics_file_test
elif self._semantics_file_test:
self._worc.semantics_test = self._semantics_file_test
if self.elastix_parameter_file:
self._worc.Elastix_Para = self.elastix_parameter_file
elif self._elastix_parameter_file:
self._worc.Elastix_Para = self._elastix_parameter_file
if self.trained_model:
self._worc.trained_model = self.trained_model
elif self._trained_model:
self._worc.trained_model = self._trained_model
# Set the labels to predict
self._worc.label_names = ', '.join(self._label_names)
if 'Labels' not in self._config_builder._custom_overrides.keys():
self._config_builder._custom_overrides['Labels'] = dict()
self._config_builder._custom_overrides['Labels']['label_names'] = self._worc.label_names
# Find out how many configs we need to make
if self._worc.images_train:
nmod = len(self._worc.images_train)
else:
nmod = len(self._worc.features_train)
# Check whether user has provided a separate train and test set
self._check_traintest()
# Create configuration files if needed
if self.config_files:
self._worc.configs = self.config_files
elif self._config_files:
self._worc.configs = self._config_files
else:
self._worc.configs = [self._config_builder.build_config(self._worc.defaultconfig())] * nmod
for cnum, _ in enumerate(self._worc.configs):
self._worc.configs[cnum]['ImageFeatures']['image_type'] = self._image_types[cnum]
# Build the fastr network
self._worc.build(buildtype=self._buildtype)
if self._add_evaluation:
self._worc.add_evaluation(label_type=self._label_names[self._selected_label],
modus=self._method)
# Set the sources and sinks and execute the experiment.
self._worc.set()
self._worc.execute()