Spaces:
Sleeping
Sleeping
File size: 4,535 Bytes
3ca8ae1 9d26f07 3ca8ae1 9d26f07 3ca8ae1 9d26f07 3ca8ae1 9d26f07 3ca8ae1 9d26f07 3ca8ae1 9d26f07 3ca8ae1 9d26f07 3ca8ae1 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 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 |
import configparser
import logging
import os
import shutil
import json
import fnmatch
def run_model(
input_path: str,
model_path: str,
verbose: str = "info",
task: str = "MRI_TumorCore",
name: str = "Tumor",
stripped_inputs_status: bool = False,
):
logging.basicConfig()
logging.getLogger().setLevel(logging.WARNING)
if verbose == "debug":
logging.getLogger().setLevel(logging.DEBUG)
elif verbose == "info":
logging.getLogger().setLevel(logging.INFO)
elif verbose == "error":
logging.getLogger().setLevel(logging.ERROR)
else:
raise ValueError("Unsupported verbose value provided:", verbose)
# delete patient/result folder if they exist
if os.path.exists("./patient/"):
shutil.rmtree("./patient/")
if os.path.exists("./result/"):
shutil.rmtree("./result/")
try:
# setup temporary patient directory
filename = input_path.split("/")[-1]
splits = filename.split(".")
extension = ".".join(splits[1:])
patient_directory = "./patient/"
os.makedirs(patient_directory + "T0/", exist_ok=True)
shutil.copy(
input_path,
patient_directory + "T0/" + splits[0] + "-t1gd." + extension,
)
# define output directory to save results
output_path = "./result/prediction-" + splits[0] + "/"
os.makedirs(output_path, exist_ok=True)
# Setting up the configuration file
rads_config = configparser.ConfigParser()
rads_config.add_section("Default")
rads_config.set("Default", "task", "neuro_diagnosis")
rads_config.set("Default", "caller", "")
rads_config.add_section("System")
rads_config.set("System", "gpu_id", "-1")
rads_config.set("System", "input_folder", patient_directory)
rads_config.set("System", "output_folder", output_path)
rads_config.set("System", "model_folder", model_path)
rads_config.set('System', 'pipeline_filename', os.path.join(output_path,
'test_pipeline.json'))
rads_config.add_section("Runtime")
rads_config.set(
"Runtime", "reconstruction_method", "thresholding"
) # thresholding, probabilities
rads_config.set("Runtime", "reconstruction_order", "resample_first")
rads_config.set("Runtime", "use_preprocessed_data", "False")
rads_config.set('Runtime', 'use_stripped_data', 'True' if stripped_inputs_status else 'False')
with open("rads_config.ini", "w") as f:
rads_config.write(f)
pip = {}
step_index = 1
pip_num = str(step_index)
pip[pip_num] = {}
pip[pip_num]["task"] = "Classification"
pip[pip_num]["inputs"] = {} # Empty input means running it on all existing data for the patient
pip[pip_num]["target"] = ["MRSequence"]
pip[pip_num]["model"] = "MRI_SequenceClassifier"
pip[pip_num]["description"] = "Classification of the MRI sequence type for all input scans."
step_index = step_index + 1
pip_num = str(step_index)
pip[pip_num] = {}
pip[pip_num]["task"] = 'Model selection'
pip[pip_num]["model"] = task
pip[pip_num]["timestamp"] = 0
pip[pip_num]["format"] = "thresholding"
pip[pip_num]["description"] = f"Identifying the best {task} segmentation model for existing inputs"
with open(os.path.join(output_path, 'test_pipeline.json'), 'w', newline='\n') as outfile:
json.dump(pip, outfile, indent=4, sort_keys=True)
# finally, run inference
from raidionicsrads.compute import run_rads
run_rads(config_filename="rads_config.ini")
logging.info(f"Looking for the following pattern: {task}")
patterns = ["*_" + task + '.*', "*_" + name + '.*']
existing_files = os.listdir(os.path.join(output_path, "T0"))
logging.info(f"Existing files: {existing_files}")
fileName = str(os.path.join(output_path, "T0",
[x for x in existing_files if
any(fnmatch.fnmatch(x, pattern) for pattern in patterns)][0]))
os.rename(src=fileName, dst="./prediction.nii.gz")
except Exception as e:
print(e)
# Clean-up
if os.path.exists(patient_directory):
shutil.rmtree(patient_directory)
if os.path.exists(output_path):
shutil.rmtree(output_path)
|