Merge branch 'main' into 73-add-hexel-channel-location-figure-alongsi…

…de-expression-plots

invokers/invoker_create_trialwise_data.py

@@ -1,9 +1,9 @@
 from pathlib import Path
-from kymata.io.yaml import load_config
+
+from kymata.io.config import load_config
 from kymata.preproc.data_cleansing import create_trialwise_data
 
 
-# noinspection DuplicatedCode
 def main():
     config = load_config(str(Path(Path(__file__).parent.parent, "kymata", "config", "dataset4.yaml")))
 

invokers/invoker_estimate_noise_cov_matrix.py

@@ -1,31 +1,21 @@
 from pathlib import Path
 
-from kymata.io.yaml import load_config
+from kymata.io.config import load_config, get_root_dir
 from kymata.preproc.data_cleansing import estimate_noise_cov
 
 
-# noinspection DuplicatedCode
 def main():
     config = load_config(str(Path(Path(__file__).parent.parent, "kymata", "config", "dataset4.yaml")))
 
-    if config['data_location'] == "local":
-        data_root_dir = str(Path(Path(__file__).parent.parent, "kymata-toolbox-data", "emeg_study_data")) + "/"
-    elif config['data_location'] == "cbu":
-        data_root_dir = '/imaging/projects/cbu/kymata/data/'
-    elif config['data_location'] == "cbu-local":
-        data_root_dir = '//cbsu/data/imaging/projects/cbu/kymata/data/'
-    else:
-        raise Exception("The 'data_location' parameter in the config file must be either 'cbu' or 'local' or 'cbu-local'.")
-
-    estimate_noise_cov( data_root_dir = data_root_dir,
+    estimate_noise_cov(data_root_dir = get_root_dir(config),
                        emeg_machine_used_to_record_data = config['emeg_machine_used_to_record_data'],
                        list_of_participants = config['list_of_participants'],
                        dataset_directory_name  = config['dataset_directory_name'],
                        n_runs = config['number_of_runs'],
                        cov_method = config['cov_method'],
                        duration_emp = config['duration'],
-                       reg_method = config['reg_method']
-                        )
+                       reg_method = config['reg_method'])
+
 
 if __name__ == '__main__':
     main()

invokers/invoker_run_data_cleansing.py

@@ -1,23 +1,14 @@
 from pathlib import Path
-from colorama import Fore
 
-from kymata.io.yaml import load_config
-from kymata.io.cli import print_with_color
-from kymata.preproc.data_cleansing import run_first_pass_cleansing_and_maxwell_filtering, run_second_pass_cleansing_and_EOG_removal
+from kymata.io.config import load_config, get_root_dir
+from kymata.preproc.data_cleansing import run_first_pass_cleansing_and_maxwell_filtering, \
+    run_second_pass_cleansing_and_EOG_removal
 
 
-# noinspection DuplicatedCode
 def main():
     config = load_config(str(Path(Path(__file__).parent.parent, "kymata", "config", "dataset4.yaml")))
 
-    if config['data_location'] == "local":
-        data_root_dir = str(Path(Path(__file__).parent.parent, "kymata-toolbox-data", "emeg_study_data")) + "/"
-    elif config['data_location'] == "cbu":
-        data_root_dir = '/imaging/projects/cbu/kymata/data/'
-    elif config['data_location'] == "cbu-local":
-        data_root_dir = '//cbsu/data/imaging/projects/cbu/kymata/data/'
-    else:
-        raise Exception("The 'data_location' parameter in the config file must be either 'cbu' or 'local' or 'cbu-local'.")
+    data_root_dir = get_root_dir(config)
 
     run_first_pass_cleansing_and_maxwell_filtering(
         data_root_dir = data_root_dir,
@@ -41,16 +32,6 @@ def main():
         supress_excessive_plots_and_prompts=config['supress_excessive_plots_and_prompts'],
     )
 
-def _display_welcome_message_to_terminal():
-    """Runs welcome message"""
-    print_with_color("-----------------------------------------------", Fore.BLUE)
-    print_with_color(" Kymata Preprocessing and Analysis Pipeline    ", Fore.BLUE)
-    print_with_color("-----------------------------------------------", Fore.BLUE)
-    print_with_color("", Fore.BLUE)
-
-def _run_cleanup():
-    """Runs clean up"""
-    print_with_color("Exited successfully.", Fore.GREEN)
 
 if __name__ == '__main__':
     main()

invokers/invoker_run_hexel_current_estimation.py

@@ -1,23 +1,13 @@
 from pathlib import Path
-from colorama import Fore
 
-from kymata.io.yaml import load_config
-from kymata.io.cli import print_with_color
+from kymata.io.config import load_config, get_root_dir
 from kymata.preproc.hexel_current_estimation import create_forward_model_and_inverse_solution, create_hexel_morph_maps
 
 
-# noinspection DuplicatedCode
 def main():
     config = load_config(str(Path(Path(__file__).parent.parent, "kymata", "config", "dataset4.yaml")))
 
-    if config['data_location'] == "local":
-        data_root_dir = str(Path(Path(__file__).parent.parent, "kymata-toolbox-data", "emeg_study_data")) + "/"
-    elif config['data_location'] == "cbu":
-        data_root_dir = '/imaging/projects/cbu/kymata/data/'
-    elif config['data_location'] == "cbu-local":
-        data_root_dir = '//cbsu/data/imaging/projects/cbu/kymata/data/'
-    else:
-        raise Exception("The 'data_location' parameter in the config file must be either 'cbu' or 'local' or 'cbu-local'.")
+    data_root_dir = get_root_dir(config)
 
     # create_current_estimation_prerequisites(data_root_dir, config=config)
 
@@ -26,17 +16,5 @@ def main():
     create_hexel_morph_maps(data_root_dir, config=config)
 
 
-def _display_welcome_message_to_terminal():
-    """Runs welcome message"""
-    print_with_color("-----------------------------------------------", Fore.BLUE)
-    print_with_color(" Kymata Preprocessing and Analysis Pipeline    ", Fore.BLUE)
-    print_with_color("-----------------------------------------------", Fore.BLUE)
-    print_with_color("", Fore.BLUE)
-
-def _run_cleanup():
-    """Runs clean up"""
-    print_with_color("Exited successfully.", Fore.GREEN)
-
-
 if __name__ == '__main__':
     main()

invokers/invoker_run_nkg_plotting.py

@@ -1,21 +1,15 @@
 from pathlib import Path
 from os import path
+
+from kymata.io.config import load_config, get_root_dir
 from kymata.io.nkg import load_expression_set
 from kymata.plot.plot import expression_plot
-from kymata.io.yaml import load_config
 
 # template invoker for printing out expression set .nkgs
 def main():
     config = load_config(str(Path(Path(__file__).parent.parent, "kymata", "config", "dataset4.yaml")))
 
-    if config['data_location'] == "local":
-        data_root_dir = str(Path(Path(__file__).parent.parent, "kymata-toolbox-data", "emeg_study_data")) + "/"
-    elif config['data_location'] == "cbu":
-        data_root_dir = '/imaging/projects/cbu/kymata/data/'
-    elif config['data_location'] == "cbu-local":
-        data_root_dir = '//cbsu/data/imaging/projects/cbu/kymata/data/'
-    else:
-        raise Exception("The 'data_location' parameter in the config file must be either 'cbu' or 'local' or 'cbu-local'.")
+    data_root_dir = get_root_dir(config)
 
     mri_structurals_directory = config['mri_structurals_directory']
 

kymata/io/config.py

@@ -0,0 +1,35 @@
+from pathlib import Path
+
+import yaml
+
+from file import path_type, file_type, open_or_use
+
+
+def load_config(config_location: path_type | file_type):
+    """Load config parameters"""
+    with open_or_use(config_location) as stream:
+        return yaml.safe_load(stream)
+
+
+def modify_param_config(config_location: str, key: str, value):
+    with open(config_location, 'r') as stream:
+        """Load config parameter"""
+        data = yaml.safe_load(stream)
+        data[key] = value
+    yaml.emitter.Emitter.process_tag = lambda self, *args, **kw: None
+    with open(config_location, 'w') as file:
+        yaml.dump(data,
+                  file,
+                  sort_keys = False)
+
+
+def get_root_dir(config: dict) -> str:
+    if config['data_location'] == "local":
+        return str(Path(Path(__file__).parent.parent, "kymata-toolbox-data", "emeg_study_data")) + "/"
+    elif config['data_location'] == "cbu":
+        return '/imaging/projects/cbu/kymata/data/'
+    elif config['data_location'] == "cbu-local":
+        return '//cbsu/data/imaging/projects/cbu/kymata/data/'
+    else:
+        raise ValueError(
+            "The `data_location` parameter in the config file must be either 'cbu' or 'local' or 'cbu-local'.")