Create cubids mv command to rename individual files or directories

cubids/cli.py

@@ -55,6 +55,36 @@ def _path_exists(path, parser):
     return path.absolute()
 
 
+def _path_does_not_exist(path, parser):
+    """Ensure a given path does not exist.
+
+    Parameters
+    ----------
+    path : str or Path
+        The path to check.
+    parser : argparse.ArgumentParser
+        The argument parser instance to use for error reporting.
+
+    Returns
+    -------
+    pathlib.Path
+        The absolute path if it exists.
+
+    Raises
+    ------
+    argparse.ArgumentError
+        If the path exists.
+    """
+    if path is not None:
+        path = Path(path)
+
+    if path.exists():
+        raise parser.error(f"Path already exists: <{path.absolute()}>.")
+    elif path is None:
+        raise parser.error("Path is None.")
+    return path.absolute()
+
+
 def _is_file(path, parser):
     """Ensure a given path exists and it is a file.
 
@@ -1394,6 +1424,80 @@ def _enter_print_metadata_fields(argv=None):
     workflows.print_metadata_fields(**args)
 
 
+def _parse_mv():
+    """Create the parser for the `cubids mv` command.
+
+    This function sets up an argument parser for the `cubids mv` command, which is used
+    to move files between directories in a BIDS dataset. It defines the required
+    arguments and their types, as well as optional arguments.
+
+    Parameters
+    ----------
+    None
+
+    Returns
+    -------
+    argparse.ArgumentParser
+        The argument parser for the `cubids mv` command.
+
+    Notes
+    -----
+    The parser includes the following arguments:
+
+    - bids_dir: The root of a BIDS dataset, which should contain sub-X directories
+        and dataset_description.json.
+    - source: The path to the file or directory to move.
+    - destination: The path to the new location for the file or directory.
+    """
+    parser = argparse.ArgumentParser(
+        description="Move or rename a file, a directory, or a symlink within a BIDS dataset",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+        allow_abbrev=False,
+    )
+    PathExists = partial(_path_exists, parser=parser)
+
+    parser.add_argument(
+        "source",
+        type=PathExists,
+        action="store",
+        help="The path to the file or directory to move.",
+    )
+    parser.add_argument(
+        "destination",
+        type=Path,
+        action="store",
+        help="The path to the new location for the file or directory.",
+    )
+    parser.add_argument(
+        "--use-datalad",
+        action="store_true",
+        default=False,
+        help="Use Datalad to track the move or rename operation.",
+    )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        default=False,
+        help="Force the move or rename operation.",
+    )
+    parser.add_argument(
+        "-n",
+        "--dry-run",
+        action="store_true",
+        default=False,
+        help="Do nothing; only show what would happen.",
+    )
+    parser.add_argument(
+        "-v",
+        "--verbose",
+        action="store_true",
+        default=False,
+        help="Report the details of the move or rename operation.",
+    )
+
+    return parser
+
+
 COMMANDS = [
     ("validate", _parse_validate, workflows.validate),
     ("bids-version", _parse_bids_version, workflows.bids_version),
@@ -1408,6 +1512,7 @@ def _enter_print_metadata_fields(argv=None):
     ("datalad-save", _parse_datalad_save, workflows.datalad_save),
     ("print-metadata-fields", _parse_print_metadata_fields, workflows.print_metadata_fields),
     ("remove-metadata-fields", _parse_remove_metadata_fields, workflows.remove_metadata_fields),
+    ("mv", _parse_mv, workflows.mv),
 ]
 
 

cubids/cubids.py

@@ -439,6 +439,48 @@ def add_file_collections(self):
 
         self.reset_bids_layout()
 
+    def rename_subject(self, old_subject, new_subject):
+        """Rename a subject in the dataset.
+
+        This method renames a subject in the dataset and updates all associated files.
+
+        Parameters
+        ----------
+        old_subject : :obj:`str`
+            The old subject ID.
+        new_subject : :obj:`str`
+            The new subject ID.
+        """
+        # check if force_unlock is set
+        if self.force_unlock:
+            # CHANGE TO SUBPROCESS.CALL IF NOT BLOCKING
+            subprocess.run(["datalad", "unlock"], cwd=self.path)
+
+        # get all files in the dataset
+        files = self.layout.get(subject=old_subject, extension=[".nii", ".nii.gz"])
+        for file in files:
+            # get the filename
+            filename = file.filename
+            # get the entities
+            entities = file.entities
+            # rename the subject
+            entities["subject"] = new_subject
+            # build the new path
+            new_path = utils.build_path(
+                filepath=filename,
+                out_entities=entities,
+                out_dir=self.path,
+                schema=self.schema,
+            )
+            # rename the file
+            os.rename(file.path, new_path)
+
+        if self.use_datalad:
+            self.datalad_save(message="Renamed subject")
+
+        self.reset_bids_layout()
+
+
     def apply_tsv_changes(self, summary_tsv, files_tsv, new_prefix, raise_on_error=True):
         """Apply changes documented in the edited summary tsv and generate the new tsv files.
 

cubids/utils.py

@@ -1132,3 +1132,11 @@ def get_bidsuri(filename, dataset_root):
     import os
 
     return f"bids::{os.path.relpath(filename, dataset_root)}"
+
+
+class BIDSError(Exception):
+    """Exception raised for errors in BIDS operations."""
+
+    def __init__(self, message):
+        self.message = message
+        super().__init__(self.message)

cubids/workflows.py

@@ -15,6 +15,7 @@
 
 from cubids.cubids import CuBIDS
 from cubids.metadata_merge import merge_json_into_json
+from cubids.utils import BIDSError
 from cubids.validator import (
     bids_validator_version,
     build_first_subject_path,
@@ -351,6 +352,61 @@ def apply(
     )
 
 
+def mv(source, destination, use_datalad=False, force=False, dry_run=False, verbose=False):
+    """Move or rename a file, a directory, or a symlink within a BIDS dataset.
+
+    Parameters
+    ----------
+    source : :obj:`pathlib.Path`
+        The path to the file or directory to move.
+    destination : :obj:`pathlib.Path`
+        The path to the new location for the file or directory.
+    use_datalad : :obj:`bool`
+        Use Datalad to track the move or rename operation.
+    force : :obj:`bool`
+        Force the move or rename operation.
+    dry_run : :obj:`bool`
+        Do nothing; only show what would happen.
+    verbose : :obj:`bool`
+        Report the details of the move or rename operation.
+
+    Notes
+    -----
+    This function is similar to git mv. It determines if a path is a BIDS dataset by looking for the
+    dataset_description.json file in the current directory and all parent directories, analogous to
+    how git mv looks for the .git directory.
+
+    If the source is a file, it will be moved to the destination and associated files will be moved
+    with it.
+
+    If the source is a directory, it will be moved to the destination and all files within it will be
+    moved with it. Additionally, if the directory involves changing a subject label, the subject label
+    will be updated in the file names of all associated files. If the directory involves changing a
+    session label, the session label will be updated in the file names of all associated files.
+
+    Raises
+    ------
+    BIDSError
+        If the dataset_description.json file is not found in the current directory or any parent
+        directory.
+    """
+    # Find the BIDS dataset root. We can do this by looking for the dataset_description.json file.
+    bids_dir = None
+    cwd = Path.cwd()
+    to_check = [cwd] + list(cwd.parents)
+    for path in to_check:
+        if (path / "dataset_description.json").exists():
+            bids_dir = path
+            break
+
+    if bids_dir is None:
+        raise BIDSError("fatal: not a BIDS dataset (or any of the parent directories)")
+
+    # Run directly from python using
+    bod = CuBIDS(data_root=str(bids_dir), use_datalad=use_datalad)
+    ...
+
+
 def datalad_save(bids_dir, m):
     """Perform datalad save.