Loading datahipy/bids/version.py +13 −0 Original line number Diff line number Diff line Loading @@ -6,6 +6,7 @@ import os from datetime import date from packaging import version import datalad.api from datahipy.bids.const import BIDS_VERSION Loading Loading @@ -84,3 +85,15 @@ def update_bids_changes(bids_dir, changes_tag_entry): # Create a new CHANGES file with the new release text block at the top with open(os.path.join(bids_dir, "CHANGES"), "w") as f: f.writelines(changes_tag_entry + content) def manage_bids_dataset_with_datalad(bids_dir): """Create a Datalad dataset out of an existing BIDS dataset not Datalad-managed yet.""" print(f"Initializing Datalad dataset at {bids_dir}...") # Initialize the BIDS dataset as a Datalad-managed dataset create_params = { "dataset": bids_dir, "cfg_proc": ["text2git", "bids"], "force": True, # Enforce dataset creation in a non-empty directory } datalad.api.create(**create_params) datahipy/handlers/participants.py +25 −0 Original line number Diff line number Diff line Loading @@ -27,6 +27,7 @@ except ImportError: from datahipy.handlers.dataset import DatasetHandler from datahipy.bids.participant import get_subject_bidsfile_info from datahipy.bids.bids_manager import post_import_bids_refinement from datahipy.bids.version import manage_bids_dataset_with_datalad class ParticipantHandler: Loading @@ -38,6 +39,12 @@ class ParticipantHandler: def sub_import(self, input_data=None): """Import subject(s) data into a BIDS dataset.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager and check converters Loading Loading @@ -77,6 +84,12 @@ class ParticipantHandler: def sub_delete(self, input_data=None): """Delete a subject from an already existing BIDS dataset.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager Loading @@ -100,6 +113,12 @@ class ParticipantHandler: def sub_delete_file(self, input_data=None): """Delete data files from /raw and /source.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager Loading Loading @@ -138,6 +157,12 @@ class ParticipantHandler: def sub_edit_clinical(self, input_data=None): """Update subject clinical info in BIDS dataset.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager Loading datahipy/handlers/project.py +37 −0 Original line number Diff line number Diff line Loading @@ -123,6 +123,32 @@ def create_project(input_data: str, output_file: str): print(SUCCESS) def manage_project_with_datalad(project_dir: str): """Create a Datalad dataset out of an existing project not Datalad-managed yet. Parameters ---------- project_dir : str Path to the project directory. """ print(f"Initializing Datalad dataset and subdataset at {project_dir}...") # Initialize the project dataset as a Datalad-managed dataset create_project_params = { "dataset": project_dir, "cfg_proc": ["text2git"], "force": True, # Enforce dataset creation in a non-empty directory } datalad.api.create(**create_project_params) # Initialize the BIDS dataset as a Datalad-managed sub-dataset create_bids_params = { "dataset": project_dir, "path": os.path.join(project_dir, "inputs", "bids-dataset"), "cfg_proc": ["text2git", "bids"], "force": True, # Enforce dataset creation in a non-empty directory } datalad.api.create(**create_bids_params) def import_subject(input_data: str, output_file: str): """Import a new subject from a BIDS dataset of the HIP Center space to the BIDS dataset of the HIP Collaborative Project. Loading @@ -149,6 +175,12 @@ def import_subject(input_data: str, output_file: str): f"from {input_data['sourceDatasetPath']} " f"to {input_data['targetDatasetPath']}..." ) # Check if project root directory is already Datalad-managed. If not, # initialize it. Otherwise, this would fail when trying to save the # dataset state with Datalad. project_dir = os.path.join(input_data["targetDatasetPath"], "..", '..', ".datalad") if not os.path.exists(project_dir): manage_project_with_datalad(project_dir) # Copy subject directory from source to target shutil.copytree( ( Loading Loading @@ -244,6 +276,11 @@ def import_document(input_data: str): f"Importing document {source_document_path} from HIP Center space " f"to HIP Collaborative Project at {target_document_path}... " ) # Check if project root directory is already Datalad-managed. If not, # initialize it. Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.exists(input_data["targetProjectAbsPath"]): manage_project_with_datalad(input_data["targetProjectAbsPath"]) # Create intermediate directories if they don't exist os.makedirs(target_document_path.parent, exist_ok=True) # Remove target document if it already exists Loading Loading
datahipy/bids/version.py +13 −0 Original line number Diff line number Diff line Loading @@ -6,6 +6,7 @@ import os from datetime import date from packaging import version import datalad.api from datahipy.bids.const import BIDS_VERSION Loading Loading @@ -84,3 +85,15 @@ def update_bids_changes(bids_dir, changes_tag_entry): # Create a new CHANGES file with the new release text block at the top with open(os.path.join(bids_dir, "CHANGES"), "w") as f: f.writelines(changes_tag_entry + content) def manage_bids_dataset_with_datalad(bids_dir): """Create a Datalad dataset out of an existing BIDS dataset not Datalad-managed yet.""" print(f"Initializing Datalad dataset at {bids_dir}...") # Initialize the BIDS dataset as a Datalad-managed dataset create_params = { "dataset": bids_dir, "cfg_proc": ["text2git", "bids"], "force": True, # Enforce dataset creation in a non-empty directory } datalad.api.create(**create_params)
datahipy/handlers/participants.py +25 −0 Original line number Diff line number Diff line Loading @@ -27,6 +27,7 @@ except ImportError: from datahipy.handlers.dataset import DatasetHandler from datahipy.bids.participant import get_subject_bidsfile_info from datahipy.bids.bids_manager import post_import_bids_refinement from datahipy.bids.version import manage_bids_dataset_with_datalad class ParticipantHandler: Loading @@ -38,6 +39,12 @@ class ParticipantHandler: def sub_import(self, input_data=None): """Import subject(s) data into a BIDS dataset.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager and check converters Loading Loading @@ -77,6 +84,12 @@ class ParticipantHandler: def sub_delete(self, input_data=None): """Delete a subject from an already existing BIDS dataset.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager Loading @@ -100,6 +113,12 @@ class ParticipantHandler: def sub_delete_file(self, input_data=None): """Delete data files from /raw and /source.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager Loading Loading @@ -138,6 +157,12 @@ class ParticipantHandler: def sub_edit_clinical(self, input_data=None): """Update subject clinical info in BIDS dataset.""" # Check if the dataset is managed by Datalad based on # the presence of the .datalad directory. If not, create it. # Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.isdir(os.path.join(self.dataset_path, ".datalad")): manage_bids_dataset_with_datalad(self.dataset_path) # Load the input_data json in a dict input_data = self.load_input_data(input_data) # Load the targeted BIDS dataset in BIDS Manager Loading
datahipy/handlers/project.py +37 −0 Original line number Diff line number Diff line Loading @@ -123,6 +123,32 @@ def create_project(input_data: str, output_file: str): print(SUCCESS) def manage_project_with_datalad(project_dir: str): """Create a Datalad dataset out of an existing project not Datalad-managed yet. Parameters ---------- project_dir : str Path to the project directory. """ print(f"Initializing Datalad dataset and subdataset at {project_dir}...") # Initialize the project dataset as a Datalad-managed dataset create_project_params = { "dataset": project_dir, "cfg_proc": ["text2git"], "force": True, # Enforce dataset creation in a non-empty directory } datalad.api.create(**create_project_params) # Initialize the BIDS dataset as a Datalad-managed sub-dataset create_bids_params = { "dataset": project_dir, "path": os.path.join(project_dir, "inputs", "bids-dataset"), "cfg_proc": ["text2git", "bids"], "force": True, # Enforce dataset creation in a non-empty directory } datalad.api.create(**create_bids_params) def import_subject(input_data: str, output_file: str): """Import a new subject from a BIDS dataset of the HIP Center space to the BIDS dataset of the HIP Collaborative Project. Loading @@ -149,6 +175,12 @@ def import_subject(input_data: str, output_file: str): f"from {input_data['sourceDatasetPath']} " f"to {input_data['targetDatasetPath']}..." ) # Check if project root directory is already Datalad-managed. If not, # initialize it. Otherwise, this would fail when trying to save the # dataset state with Datalad. project_dir = os.path.join(input_data["targetDatasetPath"], "..", '..', ".datalad") if not os.path.exists(project_dir): manage_project_with_datalad(project_dir) # Copy subject directory from source to target shutil.copytree( ( Loading Loading @@ -244,6 +276,11 @@ def import_document(input_data: str): f"Importing document {source_document_path} from HIP Center space " f"to HIP Collaborative Project at {target_document_path}... " ) # Check if project root directory is already Datalad-managed. If not, # initialize it. Otherwise, this would fail when trying to save the # dataset state with Datalad. if not os.path.exists(input_data["targetProjectAbsPath"]): manage_project_with_datalad(input_data["targetProjectAbsPath"]) # Create intermediate directories if they don't exist os.makedirs(target_document_path.parent, exist_ok=True) # Remove target document if it already exists Loading
