Fixing beegfs path and ensuring recursive transfers

config.yml

@@ -140,15 +140,15 @@ globus:
       name: nersc832
 
     bl832-beegfs-raw:
-      root_path: /global/beegfs/beamlines/bl832/raw/
+      root_path: /raw/
       uri: beegfs.als.lbl.gov
-      uuid: d33b5d6e-1603-414e-93cb-bcb732b7914a
+      uuid: ad1d70a4-649e-4e1a-8cd4-3074b825bd82
       name: bl832-beegfs-raw
 
     bl832-beegfs-scratch:
-      root_path: /global/beegfs/beamlines/bl832/scratch/
+      root_path: /processed/
       uri: beegfs.als.lbl.gov
-      uuid: d33b5d6e-1603-414e-93cb-bcb732b7914a
+      uuid: ad1d70a4-649e-4e1a-8cd4-3074b825bd82
       name: bl832-beegfs-scratch
 
   globus_apps:

orchestration/flows/bl832/nersc.py

@@ -1896,7 +1896,8 @@ def nersc_recon_flow(
     nersc_to_beegfs_zarr_future = globus_transfer_task.submit(
         file_path=zarr_file_path,
         source=config.nersc832_alsdev_pscratch_scratch,
-        destination=config.beegfs_scratch
+        destination=config.beegfs_scratch,
+        config=config
     )
 
     # Resolve before pruning (which needs to know what landed where)
@@ -1909,17 +1910,22 @@ def nersc_recon_flow(
     logger.info("All transfers complete.")
 
     # Register the reconstructed TIFFs in tiled
-    register_file_to_tiled(
-        path=Path(config.beegfs_scratch.root_path+tiff_file_path),
-        prefix="beamlines/bl832/scratch",
-        overwrite=False,
-        tags=["scratch", "bl832"],
-    )
+    # register_file_to_tiled(
+    #     path=Path(config.beegfs_scratch.root_path+tiff_file_path),
+    #     prefix="beamlines/bl832/scratch",
+    #     overwrite=False,
+    #     tags=["scratch", "bl832"],
+    # )
+
+    #TODO: get project ID from the raw.h5 metadata and add it to the tags for the zarr registration
+    # Probably will only work if the data is still on Spot832
 
     # Register the reconstructed ZARRs in tiled
+    p = Path("/global/beegfs/beamlines/bl832/" + config.beegfs_scratch.root_path + zarr_file_path)
+    logger.info(f"Registering Zarr file in Tiled: {p}")
     register_file_to_tiled(
-        path=Path(config.beegfs_scratch.root_path+zarr_file_path),
-        prefix="beamlines/bl832/scratch",
+        path=p,
+        prefix=f"beamlines/bl832/processed/{folder_name}",
         overwrite=False,
         tags=["8.3.2", folder_name],
     )

orchestration/globus/transfer.py

@@ -127,7 +127,7 @@ def start_transfer(
             relative_path = item.relative_to(source_path.parent)
             tdata.add_item(str(item), os.path.join(dest_path, str(relative_path)))
     else:
-        tdata.add_item(str(source_path), dest_path)
+        tdata.add_item(str(source_path), dest_path, recursive=True)
     logger.info(
         f"starting transfer {source_endpoint.uri}:{source_path} to {dest_endpoint.uri}:{dest_path}"
     )

orchestration/tiled.py

@@ -23,6 +23,19 @@ def register_file_to_tiled(
     overwrite: bool = False,
     tags: list[str] | None = None,
 ) -> None:
+    """
+    Register a file or Zarr/HDF5 store to the Tiled catalog.
+
+    Args:
+        path: The path to the file or Zarr/HDF5 store to register.
+        prefix: The sub-path within the Tiled catalog where the entry should be registered.
+        overwrite: Whether to overwrite an existing entry with the same key.
+        tags: A list of tags to apply to the registered entry.
+    Raises:
+        RuntimeError: If registration fails for any reason.
+    Returns:
+        None
+    """
     logger = get_run_logger()
     path = Path(path)
     tiled_uri = os.environ["TILED_URI"]
@@ -76,6 +89,17 @@ def register_file_to_tiled(
 
 @task(name="apply-tags", task_run_name="apply-tags-{tags}")
 def _apply_tags(entry_node, tags: list[str]) -> None:
+    """
+    Apply tags to a Tiled entry node. If the entry already has tags, merge them with the new tags.
+
+    Args:
+        entry_node: The Tiled entry node to which tags should be applied.
+        tags: A list of tags to apply to the entry node.
+    Raises:
+        Exception: If the tagging operation fails for any reason.
+    Returns:
+        None
+    """
     logger = get_run_logger()
     existing_blob = entry_node.access_blob
     existing_tags = (existing_blob or {}).get("tags", [])

pyproject.toml

@@ -28,8 +28,8 @@ dependencies = [
     "scicat_beamline @ git+https://github.com/als-computing/scicat_beamline.git@4828273f5f49ba4eba5442728729e0545b3f5b79",
     "sfapi_client",
     "starlette==1.0.0",
-    "tiled[client]",
-    "watchfiles"
+    "tiled[all]==0.2.8",
+    "watchfiles",
 ]
 
 [dependency-groups]