From 4685f80774c25a59b773a1f18c2efe11abaf9c6a Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Thu, 24 Sep 2026 18:49:10 +0100 Subject: [PATCH 1/2] Let 'feedback_callback' handle raised exceptions and database connection closure --- src/murfey/workflows/fib/register_atlas.py | 139 +++++++-------------- 1 file changed, 46 insertions(+), 93 deletions(-) diff --git a/src/murfey/workflows/fib/register_atlas.py b/src/murfey/workflows/fib/register_atlas.py index 8b7ca0315..9089b60f3 100644 --- a/src/murfey/workflows/fib/register_atlas.py +++ b/src/murfey/workflows/fib/register_atlas.py @@ -1,5 +1,4 @@ import logging -import traceback from importlib.metadata import entry_points from pathlib import Path from typing import Any, cast @@ -185,101 +184,55 @@ def run( message: dict[str, Any], murfey_db: Session, ): - # Outer try-finally block to ensure database connection closes - try: - try: - # Validate incoming message - fib_info = FIBAtlasRegistrationInfo(**message) - except Exception: - logger.error("Could not validate incoming message", exc_info=True) - return {"success": False, "requeue": False} + # Validate incoming message + fib_info = FIBAtlasRegistrationInfo(**message) - try: - # Load visit information - murfey_session = murfey_db.exec( - select(MurfeyDB.Session).where( - MurfeyDB.Session.id == fib_info.session_id - ) - ).one() - visit_name = murfey_session.visit - instrument_name = murfey_session.instrument_name - except Exception: - logger.error( - "Exception encountered while querying Murfey database", exc_info=True - ) - return {"success": False, "requeue": False} - - try: - # Load the machine config - machine_config = get_machine_config(instrument_name)[instrument_name] - rotation_offset: float = cast( - float, machine_config.calibrations.get("rotation_offset", 0) - ) + # Load visit information + murfey_session = murfey_db.exec( + select(MurfeyDB.Session).where(MurfeyDB.Session.id == fib_info.session_id) + ).one() + visit_name = murfey_session.visit + instrument_name = murfey_session.instrument_name - # Extract metadata from Electron Snapshot image - metadata = FIBImageMetadata( - visit_name=visit_name, - file=fib_info.atlas_file, - **parse_image_metadata( - fib_info.atlas_file, - rotation_offset=rotation_offset, - ), - ) - except Exception: - logger.error( - f"Error extracting metadata from file {fib_info.atlas_file}", - exc_info=True, - ) - return {"success": False, "requeue": False} + # Load the machine config + machine_config = get_machine_config(instrument_name)[instrument_name] + rotation_offset: float = cast( + float, machine_config.calibrations.get("rotation_offset", 0) + ) - try: - # Make a thumbnail of the image and update metadata accordingly - metadata.thumbnail_path = _make_thumbnail( - file=metadata.file, - metadata=metadata, - visit_name=visit_name, - ) - except Exception: - logger.warning( - f"Error creating thumbnail of file {fib_info.atlas_file}", exc_info=True - ) + # Extract metadata from Electron Snapshot image + metadata = FIBImageMetadata( + visit_name=visit_name, + file=fib_info.atlas_file, + **parse_image_metadata( + fib_info.atlas_file, + rotation_offset=rotation_offset, + ), + ) - try: - # Register imaging site in Murfey, or update existing one - fib_imaging_site = _register_fib_imaging_site( - fib_info.session_id, metadata, murfey_db - ) - logger.info( - f"Registered FIB atlas image {fib_info.atlas_file} " - f"for slot {metadata.slot_number} in Murfey database" - ) - except Exception: - logger.error( - "Error registering FIB atlas image " - f"{fib_info.atlas_file} in Murfey database", - exc_info=True, - ) - return {"success": False, "requeue": False} + # Make a thumbnail of the image and update metadata accordingly + metadata.thumbnail_path = _make_thumbnail( + file=metadata.file, + metadata=metadata, + visit_name=visit_name, + ) - try: - # Register data collection group and atlas in ISPyB - _register_dcg_and_atlas( - session_id=fib_info.session_id, - instrument_name=murfey_session.instrument_name, - visit_name=murfey_session.visit, - imaging_site=fib_imaging_site, - metadata=metadata, - murfey_db=murfey_db, - ) - except Exception: - # Log error but allow workflow to proceed - logger.error( - "Error registering data collection group for FIB workflow " - f"for {metadata.site_name!r}: \n" - f"{traceback.format_exc()}" - ) - return {"success": False, "requeue": True} - return {"success": True, "requeue": False} + # Register imaging site in Murfey, or update existing one + fib_imaging_site = _register_fib_imaging_site( + fib_info.session_id, metadata, murfey_db + ) + logger.info( + f"Registered FIB atlas image {fib_info.atlas_file} " + f"for slot {metadata.slot_number} in Murfey database" + ) - finally: - murfey_db.close() + # Register data collection group and atlas in ISPyB + _register_dcg_and_atlas( + session_id=fib_info.session_id, + instrument_name=murfey_session.instrument_name, + visit_name=murfey_session.visit, + imaging_site=fib_imaging_site, + metadata=metadata, + murfey_db=murfey_db, + ) + return {"success": True, "requeue": False} From f84c22a0107a6e5a5ab52751990ec91e4dabecbf Mon Sep 17 00:00:00 2001 From: Eu Pin Tien Date: Thu, 24 Sep 2026 18:53:42 +0100 Subject: [PATCH 2/2] Use new 'register_dcg' function instead of loading the entry point --- src/murfey/workflows/fib/register_atlas.py | 24 ++++++++-------------- 1 file changed, 8 insertions(+), 16 deletions(-) diff --git a/src/murfey/workflows/fib/register_atlas.py b/src/murfey/workflows/fib/register_atlas.py index 9089b60f3..a38de6466 100644 --- a/src/murfey/workflows/fib/register_atlas.py +++ b/src/murfey/workflows/fib/register_atlas.py @@ -15,6 +15,7 @@ parse_image_metadata, populate_fib_imaging_site_entry, ) +from murfey.workflows.register_data_collection_group import register_dcg logger = logging.getLogger("murfey.workflows.fib.register_atlas") @@ -152,23 +153,14 @@ def _register_dcg_and_atlas( "atlas_pixel_size": atlas_pixel_size, "sample": metadata.slot_number, } - if entry_point_result := entry_points( - group="murfey.workflows", name="data_collection_group" - ): - (workflow,) = entry_point_result - # Register grid square - _ = workflow.load()( - message=dcg_message, - murfey_db=murfey_db, + dcg_entry = register_dcg( + message=dcg_message, + murfey_db=murfey_db, + ) + if not dcg_entry: + raise RuntimeError( + f"Could not register DataCollectionGroup entry for {imaging_site.image_path}" ) - else: - logger.warning("No workflow found for 'data_collection_group'") - dcg_entry = murfey_db.exec( - select(MurfeyDB.DataCollectionGroup) - .where(MurfeyDB.DataCollectionGroup.session_id == session_id) - .where(MurfeyDB.DataCollectionGroup.tag == imaging_site.site_name) - ).one() - imaging_site.dcg_id = dcg_entry.id imaging_site.dcg_name = dcg_entry.tag murfey_db.add(imaging_site)