Merge pull request #67 from astronomy-commons/issue/50/plot

Add file helper methods needed for hipscat-import.

Author: delucchi-cmu

src/hipscat/io/file_io/file_io.py

@@ -1,5 +1,6 @@
 import json
 import os
+import shutil
 from typing import Any
 
 import healpy as hp
@@ -19,14 +20,24 @@ def make_directory(file_pointer: FilePointer, exist_ok: bool = False):
     Args:
         file_pointer: location in file system to make directory
         exist_ok: Default False. If false will raise error if directory exists. If true existing
-        directories will be ignored and not modified
+            directories will be ignored and not modified
 
     Raises:
         OSError
     """
     os.makedirs(file_pointer, exist_ok=exist_ok)
 
 
+def remove_directory(file_pointer: FilePointer, ignore_errors=False):
+    """Remove a directory, and all contents, recursively.
+
+    Args:
+        file_pointer: directory in file system to remove
+        ignore_errors: if True errors resulting from failed removals will be ignored
+    """
+    shutil.rmtree(file_pointer, ignore_errors=ignore_errors)
+
+
 def write_string_to_file(
     file_pointer: FilePointer, string: str, encoding: str = "utf-8"
 ):

src/hipscat/io/file_io/file_pointer.py

@@ -1,5 +1,6 @@
+import glob
 import os
-from typing import NewType
+from typing import List, NewType
 
 FilePointer = NewType("FilePointer", str)
 
@@ -33,3 +34,58 @@ def does_file_or_directory_exist(pointer: FilePointer) -> bool:
         True if file or directory at `pointer` exists, False if not
     """
     return os.path.exists(pointer)
+
+
+def is_regular_file(pointer: FilePointer) -> bool:
+    """Checks if a regular file (NOT a directory) exists for a given file pointer.
+
+    Args:
+        pointer: File Pointer to check if a regular file
+
+    Returns:
+        True if regular file at `pointer` exists, False if not or is a directory
+    """
+    return os.path.isfile(pointer)
+
+
+def find_files_matching_path(*paths: str) -> List[FilePointer]:
+    """Find files or directories matching the provided path parts.
+
+    Args:
+        paths: any number of directory names optionally followed by a file name.
+            directory or file names may be replaced with `*` as a matcher.
+    Returns:
+        New file pointers to files found matching the path
+    """
+    matcher = append_paths_to_pointer(*paths)
+    return [get_file_pointer_from_path(x) for x in glob.glob(matcher)]
+
+
+def directory_has_contents(pointer: FilePointer) -> bool:
+    """Checks if a directory already has some contents (any files or subdirectories)
+
+    Args:
+        pointer: File Pointer to check for existing contents
+
+    Returns:
+        True if there are any files or subdirectories below this directory.
+    """
+    return len(find_files_matching_path(pointer, "*")) > 0
+
+
+def get_directory_contents(pointer: FilePointer) -> List[FilePointer]:
+    """Finds all files and directories in the specified directory.
+
+    NBL This is not recursive, and will return only the first level of directory contents.
+
+    Args:
+        pointer: File Pointer in which to find contents
+
+    Returns:
+        New file pointers to files or subdirectories below this directory.
+    """
+    contents = os.listdir(pointer)
+    if len(contents) == 0:
+        return []
+    contents.sort()
+    return [append_paths_to_pointer(pointer, x) for x in contents]

tests/hipscat/io/file_io/test_file_io.py

@@ -7,11 +7,12 @@
 
 from hipscat.io.file_io import (
     get_file_pointer_from_path,
-    make_directory,
-    write_string_to_file,
-    load_json_file,
     load_csv_to_pandas,
+    load_json_file,
+    make_directory,
+    remove_directory,
     write_dataframe_to_csv,
+    write_string_to_file,
 )
 
 
@@ -45,6 +46,24 @@ def test_make_existing_directory_existok(tmp_path):
     assert os.path.exists(test_inner_dir_path)
 
 
+def test_make_and_remove_directory(tmp_path):
+    test_dir_path = os.path.join(tmp_path, "test_path")
+    assert not os.path.exists(test_dir_path)
+    test_dir_pointer = get_file_pointer_from_path(test_dir_path)
+    make_directory(test_dir_pointer)
+    assert os.path.exists(test_dir_path)
+    remove_directory(test_dir_pointer)
+    assert not os.path.exists(test_dir_path)
+
+    ## Directory no longer exists to be deleted.
+    with pytest.raises(FileNotFoundError):
+        remove_directory(test_dir_pointer)
+
+    ## Directory doesn't exist, but shouldn't throw an error.
+    remove_directory(test_dir_pointer, ignore_errors=True)
+    assert not os.path.exists(test_dir_path)
+
+
 def test_write_string_to_file(tmp_path):
     test_file_path = os.path.join(tmp_path, "text_file.txt")
     test_file_pointer = get_file_pointer_from_path(test_file_path)

tests/hipscat/io/file_io/test_file_pointers.py

@@ -1,10 +1,13 @@
 import os
 
 from hipscat.io.file_io import (
-    get_file_pointer_from_path,
-    does_file_or_directory_exist,
     append_paths_to_pointer,
-    FilePointer,
+    directory_has_contents,
+    does_file_or_directory_exist,
+    find_files_matching_path,
+    get_directory_contents,
+    get_file_pointer_from_path,
+    is_regular_file,
 )
 
 
@@ -31,3 +34,49 @@ def test_append_paths_to_pointer(tmp_path):
     test_path = os.path.join(tmp_path, *test_paths)
     tmp_pointer = get_file_pointer_from_path(str(tmp_path))
     assert append_paths_to_pointer(tmp_pointer, *test_paths) == test_path
+
+
+def test_is_regular_file(small_sky_dir):
+    partition_info_file = os.path.join(small_sky_dir, "partition_info.csv")
+    assert is_regular_file(partition_info_file)
+
+    assert not is_regular_file(small_sky_dir)
+
+    partition_dir = os.path.join(small_sky_dir, "Norder=0")
+    assert not is_regular_file(partition_dir)
+
+
+def test_find_files_matching_path(small_sky_dir):
+    ## no_wildcard
+    assert len(find_files_matching_path(small_sky_dir, "partition_info.csv")) == 1
+
+    ## wilcard in the name
+    assert len(find_files_matching_path(small_sky_dir, "*.csv")) == 1
+
+
+def test_find_files_matching_path_directory(small_sky_order1_dir):
+    assert len(find_files_matching_path(small_sky_order1_dir)) == 1
+
+    ## wildcard in directory - will match all files at indicated depth
+    assert len(find_files_matching_path(small_sky_order1_dir, "*", "*", "*")) == 4
+
+
+def test_directory_has_contents(small_sky_order1_dir, tmp_path):
+    assert directory_has_contents(small_sky_order1_dir)
+    assert not directory_has_contents(tmp_path)
+
+
+def test_get_directory_contents(small_sky_order1_dir, tmp_path):
+    small_sky_contents = get_directory_contents(small_sky_order1_dir)
+    assert len(small_sky_contents) == 4
+
+    expected = [
+        os.path.join(small_sky_order1_dir, "Norder=1"),
+        os.path.join(small_sky_order1_dir, "catalog_info.json"),
+        os.path.join(small_sky_order1_dir, "partition_info.csv"),
+        os.path.join(small_sky_order1_dir, "point_map.fits"),
+    ]
+
+    assert small_sky_contents == expected
+
+    assert len(get_directory_contents(tmp_path)) == 0