Migrate remove_empty_directories to lib.util.fs

python/lib/dcm2bids_imaging_pipeline_lib/push_imaging_files_to_s3_pipeline.py

@@ -3,9 +3,9 @@
 import sys
 
 import lib.exitcode
-import lib.utilities
 from lib.dcm2bids_imaging_pipeline_lib.base_pipeline import BasePipeline
 from lib.logging import log_error_exit
+from lib.util.fs import remove_empty_directories
 
 __license__ = "GPLv3"
 
@@ -268,6 +268,6 @@ def _clean_up_empty_folders(self):
         # remove empty folders from file system
         print("Cleaning up empty folders")
         bids_cand_id = f"sub-{self.session.candidate.cand_id}"
-        lib.utilities.remove_empty_folders(os.path.join(self.data_dir, "assembly_bids", bids_cand_id))
-        lib.utilities.remove_empty_folders(os.path.join(self.data_dir, "pic", str(self.session.candidate.cand_id)))
-        lib.utilities.remove_empty_folders(os.path.join(self.data_dir, "trashbin"))
+        remove_empty_directories(os.path.join(self.data_dir, "assembly_bids", bids_cand_id))
+        remove_empty_directories(os.path.join(self.data_dir, "pic", str(self.session.candidate.cand_id)))
+        remove_empty_directories(os.path.join(self.data_dir, "trashbin"))

python/lib/util/fs.py

@@ -58,3 +58,22 @@ def copy_file(env: Env, old_path: str, new_path: str):
     shutil.copytree(old_path, new_path, dirs_exist_ok=True)
     if not os.path.exists(new_path):
         log_error_exit(env, f"Could not copy {old_path} to {new_path}", lib.exitcode.COPY_FAILURE)
+
+
+def is_directory_empty(dir_path: str) -> bool:
+    """
+    Check whether a directory is empty or not.
+    """
+
+    with os.scandir(dir_path) as dir_iterator:
+        return next(dir_iterator, None) is None
+
+
+def remove_empty_directories(dir_path: str):
+    """
+    Recursively remove all the empty directories in a directory, including itself if needed.
+    """
+
+    for subdir_path, _, _ in os.walk(dir_path, topdown=False):
+        if is_directory_empty(subdir_path):
+            os.rmdir(subdir_path)

python/lib/utilities.py

@@ -265,6 +265,7 @@ def create_processing_tmp_dir(template_prefix):
     return tmp_dir
 
 
+@deprecated('Use `lib.util.fs.remove_empty_subdirectories` instead')
 def remove_empty_folders(path_abs):
 
     walk = list(os.walk(path_abs))