diff --git a/config.yml b/config.yml index fb0b9989..a59f09a4 100644 --- a/config.yml +++ b/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: diff --git a/orchestration/flows/bl832/nersc.py b/orchestration/flows/bl832/nersc.py index f5850a61..bae6f63d 100644 --- a/orchestration/flows/bl832/nersc.py +++ b/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], ) diff --git a/orchestration/globus/transfer.py b/orchestration/globus/transfer.py index 7a9c010a..f8be2c62 100644 --- a/orchestration/globus/transfer.py +++ b/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}" ) diff --git a/orchestration/tiled.py b/orchestration/tiled.py index 90687c7b..1a25b2f4 100644 --- a/orchestration/tiled.py +++ b/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", []) diff --git a/pyproject.toml b/pyproject.toml index 8c007895..a9ef7724 100644 --- a/pyproject.toml +++ b/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]