Integrating atlas validation in wrapup.py

brainglobe_atlasapi/atlas_generation/validate_atlases.py

@@ -16,13 +16,25 @@
 from brainglobe_atlasapi.update_atlases import update_atlas
 
 
-def validate_atlas_files(atlas: BrainGlobeAtlas):
-    """Checks if basic files exist in the atlas folder"""
+def validate_atlas_files(
+    atlas: BrainGlobeAtlas, custom_atlas_path: Path = None
+):
+    """
+    Checks if basic files exist in the atlas folder
+
+    custom_atlas_path is used when the function is called as part of
+    the wrapup function in the atlas packaging script. The expected
+    input is working_dir
+    """
+
+    if custom_atlas_path is None:
+        atlas_path = (
+            Path(get_brainglobe_dir()) / f"{atlas.atlas_name}_v"
+            f"{get_local_atlas_version(atlas.atlas_name)}"
+        )
+    else:
+        atlas_path = custom_atlas_path
 
-    atlas_path = (
-        Path(get_brainglobe_dir())
-        / f"{atlas.atlas_name}_v{get_local_atlas_version(atlas.atlas_name)}"
-    )
     assert atlas_path.is_dir(), f"Atlas path {atlas_path} not found"
     expected_files = [
         "annotation.tiff",
@@ -143,17 +155,27 @@
     return True
 
 
-def catch_missing_mesh_files(atlas: BrainGlobeAtlas):
+def catch_missing_mesh_files(
+    atlas: BrainGlobeAtlas, custom_atlas_path: Path = None
+):
     """
     Checks if all the structures in the atlas have a corresponding mesh file
+
+    custom_atlas_path is used when the function is called as part of
+    the wrapup function in the atlas packaging script. The expected
+    input is working_dir
     """
 
     ids_from_bg_atlas_api = list(atlas.structures.keys())
 
-    atlas_path = (
-        Path(get_brainglobe_dir())
-        / f"{atlas.atlas_name}_v{get_local_atlas_version(atlas.atlas_name)}"
-    )
+    if custom_atlas_path is None:
+        atlas_path = (
+            Path(get_brainglobe_dir()) / f"{atlas.atlas_name}_v"
+            f"{get_local_atlas_version(atlas.atlas_name)}"
+        )
+    else:
+        atlas_path = custom_atlas_path
+
     obj_path = Path(atlas_path / "meshes")
 
     ids_from_mesh_files = [
@@ -174,18 +196,28 @@
         )
 
 
-def catch_missing_structures(atlas: BrainGlobeAtlas):
+def catch_missing_structures(
+    atlas: BrainGlobeAtlas, custom_atlas_path: Path = None
+):
     """
     Checks if all the mesh files in the atlas folder
     are listed as a structure in the atlas.
+
+    custom_atlas_path is used when the function is called as part of
+    the wrapup function in the atlas packaging script. The expected
+    input is working_dir
     """
 
     ids_from_bg_atlas_api = list(atlas.structures.keys())
 
-    atlas_path = (
-        Path(get_brainglobe_dir())
-        / f"{atlas.atlas_name}_v{get_local_atlas_version(atlas.atlas_name)}"
-    )
+    if custom_atlas_path is None:
+        atlas_path = (
+            Path(get_brainglobe_dir()) / f"{atlas.atlas_name}_v"
+            f"{get_local_atlas_version(atlas.atlas_name)}"
+        )
+    else:
+        atlas_path = custom_atlas_path
+
     obj_path = Path(atlas_path / "meshes")
 
     ids_from_mesh_files = [

brainglobe_atlasapi/atlas_generation/wrapup.py

@@ -8,7 +8,7 @@
 import tifffile
 
 import brainglobe_atlasapi.atlas_generation
-from brainglobe_atlasapi import descriptors
+from brainglobe_atlasapi import BrainGlobeAtlas, descriptors
 from brainglobe_atlasapi.atlas_generation.metadata_utils import (
     create_metadata_files,
     generate_metadata_dict,
@@ -22,6 +22,16 @@
 from brainglobe_atlasapi.atlas_generation.structures import (
     check_struct_consistency,
 )
+from brainglobe_atlasapi.atlas_generation.validate_atlases import (
+    catch_missing_mesh_files,
+    catch_missing_structures,
+    open_for_visual_check,
+    validate_additional_references,
+    validate_atlas_files,
+    validate_checksum,
+    validate_image_dimensions,
+    validate_mesh_matches_image_extents,
+)
 from brainglobe_atlasapi.utils import atlas_name_from_repr
 
 # This should be changed every time we make changes in the atlas
@@ -129,6 +139,7 @@
     atlas_dir_name = atlas_name_from_repr(
         atlas_name, resolution[0], ATLAS_VERSION, atlas_minor_version
     )
+
     dest_dir = Path(working_dir) / atlas_dir_name
 
     # exist_ok would be more permissive but error-prone here as there might
@@ -231,6 +242,72 @@
         additional_metadata=additional_metadata,
     )
 
+    atlas_name_for_validation = atlas_name_from_repr(atlas_name, resolution[0])
+
+    # creating BrainGlobe object from local folder (working_dir)
+    atlas_to_validate = BrainGlobeAtlas(
+        atlas_name=atlas_name_for_validation,
+        brainglobe_dir=working_dir,
+        check_latest=False,
+    )
+
+    # Run validation functions
+    print(f"Running atlas validation on {atlas_dir_name}")
+
+    validation_results = {}
+
+    functions = [
+        (validate_atlas_files, atlas_to_validate, atlas_to_validate.root_dir),
+        (
+            catch_missing_structures,
+            atlas_to_validate,
+            atlas_to_validate.root_dir,
+        ),
+        (
+            catch_missing_mesh_files,
+            atlas_to_validate,
+            atlas_to_validate.root_dir,
+        ),
+        (validate_mesh_matches_image_extents, atlas_to_validate),
+        (open_for_visual_check, atlas_to_validate),
+        (validate_checksum, atlas_to_validate),
+        (validate_image_dimensions, atlas_to_validate),
+        (validate_additional_references, atlas_to_validate),
+    ]
+
+    for func, *args in functions:
+        try:
+            func(*args)
+            validation_results[func.__name__] = "Pass"
+        except AssertionError as e:
+            validation_results[func.__name__] = f"Fail: {str(e)}"
+
+    def _check_validations(validation_results):
+        # Helper function to check if all validations passed
+        all_passed = all(
+            result == "Pass" for result in validation_results.values()
+        )
+
+        if all_passed:
+            print("This atlas is valid")
+        else:
+            failed_functions = [
+                func
+                for func, result in validation_results.items()
+                if result != "Pass"
+            ]
+            error_messages = [
+                result.split(": ")[1]
+                for result in validation_results.values()
+                if result != "Pass"
+            ]
+
+            print("These validation functions have failed:")
+            for func, error in zip(failed_functions, error_messages):
+                print(f"- {func}: {error}")
+
+    _check_validations(validation_results)
+
     # Compress if required:
     if compress:
         output_filename = dest_dir.parent / f"{dest_dir.name}.tar.gz"
@@ -240,6 +317,7 @@
 
     # Cleanup if required:
     if cleanup_files:
+        print(f"Cleaning up atlas data at: {dest_dir}")
         # Clean temporary directory and remove it:
         shutil.rmtree(dest_dir)