From 8e9d10d63caa4215e30596f39299e554c07a5c5e Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Fri, 11 Sep 2026 16:44:44 +0100 Subject: [PATCH 1/9] test fixes --- tests/client/contexts/test_atlas.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/client/contexts/test_atlas.py b/tests/client/contexts/test_atlas.py index a3ca5122f..cf37e3c44 100644 --- a/tests/client/contexts/test_atlas.py +++ b/tests/client/contexts/test_atlas.py @@ -100,6 +100,7 @@ def test_atlas_context_dm(mock_capture_post, tmp_path): ) # Write sample dm file + (tmp_path / "cm12345-6/Supervisor_atlas/Sample2/Atlas/Atlas_01.mrc").touch() atlas_dm = tmp_path / "cm12345-6/Supervisor_atlas/Sample2/Atlas/Atlas.dm" atlas_dm.parent.mkdir(parents=True) (tmp_path / "cm12345-6/Supervisor_atlas/Sample2/Atlas/Atlas_01.mrc").touch() From 7d07478c54b8fc7b12ad277614f8ea26588ef8ae Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Fri, 11 Sep 2026 16:50:27 +0100 Subject: [PATCH 2/9] need to make the directories first --- tests/client/contexts/test_atlas.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/client/contexts/test_atlas.py b/tests/client/contexts/test_atlas.py index cf37e3c44..a3ca5122f 100644 --- a/tests/client/contexts/test_atlas.py +++ b/tests/client/contexts/test_atlas.py @@ -100,7 +100,6 @@ def test_atlas_context_dm(mock_capture_post, tmp_path): ) # Write sample dm file - (tmp_path / "cm12345-6/Supervisor_atlas/Sample2/Atlas/Atlas_01.mrc").touch() atlas_dm = tmp_path / "cm12345-6/Supervisor_atlas/Sample2/Atlas/Atlas.dm" atlas_dm.parent.mkdir(parents=True) (tmp_path / "cm12345-6/Supervisor_atlas/Sample2/Atlas/Atlas_01.mrc").touch() From e431a20343150126d8b0cdb2b1e5536a4ee4ed13 Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Wed, 16 Sep 2026 10:33:36 +0100 Subject: [PATCH 3/9] add utility function to parse rabbitmq credentials and construct url --- src/murfey/util/config.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/src/murfey/util/config.py b/src/murfey/util/config.py index 30d456f89..39b8c2593 100644 --- a/src/murfey/util/config.py +++ b/src/murfey/util/config.py @@ -1,5 +1,6 @@ from __future__ import annotations +import configparser import copy import os import socket @@ -7,6 +8,7 @@ from importlib.metadata import entry_points from pathlib import Path from typing import Any, Literal, Optional +from urllib.parser import quote import yaml from pydantic import BaseModel, ConfigDict, RootModel, ValidationInfo, field_validator @@ -373,3 +375,26 @@ def get_smartem_keycloak_client(): load_keycloak_config(Path(keycloak_config)) ) return keycloak_client + + +def get_rabbitmq_url() -> str: + rabbitmq_defaults = { + "host": "localhost", + "port": "5672", + "username": "guest", + "password": "guest", + "vhost": "/", + } + rabbitmq_credentials_file = get_security_config().rabbitmq_credentials + cfgparser = configparser.ConfigParser(allow_no_value=True) + if rabbitmq_credentials_file: + cfgparser.read(rabbitmq_credentials_file) + rabbitmq_creds = ( + {**rabbitmq_defaults, **cfgparser["rabbit"]} + if cfgparser.has_section("rabbit") + else rabbitmq_defaults + ) + user = quote(rabbitmq_creds["username"], safe="") + password = quote(rabbitmq_creds["password"], safe="") + vhost = quote(rabbitmq_creds["vhost"], safe="") + return f"amqp://{user}:{password}@{rabbitmq_creds['host']}:{rabbitmq_creds['port']}/{vhost}" From 1bdd60aee435e5a2be7a4165e02462ee04fb5cff Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Wed, 16 Sep 2026 10:44:01 +0100 Subject: [PATCH 4/9] add publication of particle picking complete smartem message to trigger model weight updates --- src/murfey/util/config.py | 1 + src/murfey/workflows/spa/picking.py | 47 ++++++++++++++++++++++++++++- 2 files changed, 47 insertions(+), 1 deletion(-) diff --git a/src/murfey/util/config.py b/src/murfey/util/config.py index 39b8c2593..ad80cfd0c 100644 --- a/src/murfey/util/config.py +++ b/src/murfey/util/config.py @@ -377,6 +377,7 @@ def get_smartem_keycloak_client(): return keycloak_client +@lru_cache(maxsize=1) def get_rabbitmq_url() -> str: rabbitmq_defaults = { "host": "localhost", diff --git a/src/murfey/workflows/spa/picking.py b/src/murfey/workflows/spa/picking.py index 7de4bcbce..1f5610282 100644 --- a/src/murfey/workflows/spa/picking.py +++ b/src/murfey/workflows/spa/picking.py @@ -1,3 +1,4 @@ +import asyncio from logging import getLogger from typing import List @@ -11,7 +12,7 @@ _app_id, _pj_id, ) -from murfey.util.config import get_machine_config +from murfey.util.config import get_machine_config, get_rabbitmq_url from murfey.util.db import ( AutoProcProgram, ClassificationFeedbackParameters, @@ -33,6 +34,11 @@ from smartem_backend.api_client import SmartEMAPIClient from smartem_backend.model.http_request import MicrographUpdateRequest from smartem_backend.model.http_response import MicrographResponse + from smartem_backend.model.mq_event import ( + MessageQueueEventType, + ParticlePickingCompleteBody, + ) + from smartem_backend.rmq.publisher import AioPikaPublisher from smartem_common.entity_status import MicrographStatus from murfey.util.config import get_smartem_keycloak_client @@ -411,6 +417,45 @@ def particles_picked(message: dict, murfey_db: Session) -> dict[str, bool]: update, MicrographResponse, ) + + async def _publish_particle_picking_completed( + micrograph_uuid: str, number_of_particles_picked: int + ) -> None: + publisher = AioPikaPublisher( + url=get_rabbitmq_url(), + exchange_name="smartem", + routing_key="smartem", + exchange_type="fanout", + ) + await publisher.connect() + try: + await publisher.publish_event( + MessageQueueEventType.PARTICLE_PICKING_COMPLETE, + ParticlePickingCompleteBody( + event_type=MessageQueueEventType.PARTICLE_PICKING_COMPLETE, + micrograph_uuid=micrograph_uuid, + number_of_particles_picked=number_of_particles_picked, + ), + ) + except Exception: + logger.warning( + f"smartem failed to publish picking completion {micrograph_uuid}", + exc_info=True, + ) + finally: + await publisher.close() + + number_of_particles_picked = message.get("particle_count") + if number_of_particles_picked is None: + number_of_particles_picked = len( + message.get("particle_diameters") or [] + ) + asyncio.run( + _publish_particle_picking_completed( + movie.smartem_uuid, number_of_particles_picked + ) + ) + except Exception: logger.warning( "Failed to emit particle picking complete event to smartem", From 633dcf06bccd72fea478d0a0aa2c208cea71a227 Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Wed, 16 Sep 2026 10:48:49 +0100 Subject: [PATCH 5/9] add publication of motion correction complete smartem message --- src/murfey/workflows/spa/motion_correction.py | 45 ++++++++++++++++++- 1 file changed, 44 insertions(+), 1 deletion(-) diff --git a/src/murfey/workflows/spa/motion_correction.py b/src/murfey/workflows/spa/motion_correction.py index d1b0c5043..cd6b89a3e 100644 --- a/src/murfey/workflows/spa/motion_correction.py +++ b/src/murfey/workflows/spa/motion_correction.py @@ -1,8 +1,9 @@ +import asyncio from logging import getLogger from sqlmodel import Session, select -from murfey.util.config import get_machine_config +from murfey.util.config import get_machine_config, get_rabbitmq_url from murfey.util.db import ( Movie, Session as MurfeySession, @@ -20,6 +21,11 @@ MicrographResponse, ProcessingFeedbackPublishResponse, ) + from smartem_backend.model.mq_event import ( + MessageQueueEventType, + MotionCorrectionCompleteBody, + ) + from smartem_backend.rmq.publisher import AioPikaPublisher from smartem_common.entity_status import MicrographStatus from murfey.util.config import get_smartem_keycloak_client @@ -71,6 +77,43 @@ def motion_corrected(message: dict, murfey_db: Session) -> dict[str, bool]: registered_request, ProcessingFeedbackPublishResponse, ) + + async def _publish_motion_correction_completed( + micrograph_uuid: str, total_motion: float, average_motion: float + ) -> None: + publisher = AioPikaPublisher( + url=get_rabbitmq_url(), + exchange_name="smartem", + routing_key="smartem", + exchange_type="fanout", + ) + await publisher.connect() + try: + await publisher.publish_event( + MessageQueueEventType.MOTION_CORRECTION_COMPLETE, + MotionCorrectionCompleteBody( + event_type=MessageQueueEventType.MOTION_CORRECTION_COMPLETE, + micrograph_uuid=micrograph_uuid, + total_motion=total_motion, + average_motion=average_motion, + ), + ) + except Exception: + logger.warning( + f"smartem failed to motion correction completion {micrograph_uuid}", + exc_info=True, + ) + finally: + await publisher.close() + + asyncio.run( + _publish_motion_correction_completed( + movie.smartem_uuid, + message.get("total_motion", 0), + message.get("average_motion", 0), + ) + ) + except Exception: logger.warning( "Failed to emit motion correction complete event to smartem", From cf7fb5bcf7fa5e9dad19047909e4236314a63ea0 Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Wed, 16 Sep 2026 10:53:17 +0100 Subject: [PATCH 6/9] add publication of ctf estimation complete smartem message --- src/murfey/workflows/spa/ctf_estimation.py | 42 ++++++++++++++++++- src/murfey/workflows/spa/motion_correction.py | 4 +- 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/src/murfey/workflows/spa/ctf_estimation.py b/src/murfey/workflows/spa/ctf_estimation.py index 7e78374ce..c837b45fe 100644 --- a/src/murfey/workflows/spa/ctf_estimation.py +++ b/src/murfey/workflows/spa/ctf_estimation.py @@ -1,8 +1,9 @@ +import asyncio from logging import getLogger from sqlmodel import Session, select -from murfey.util.config import get_machine_config +from murfey.util.config import get_machine_config, get_rabbitmq_url from murfey.util.db import ( Movie, Session as MurfeySession, @@ -20,6 +21,11 @@ MicrographResponse, ProcessingFeedbackPublishResponse, ) + from smartem_backend.model.mq_event import ( + CtfCompleteBody, + MessageQueueEventType, + ) + from smartem_backend.rmq.publisher import AioPikaPublisher from smartem_common.entity_status import MicrographStatus from murfey.util.config import get_smartem_keycloak_client @@ -69,6 +75,40 @@ def ctf_estimated(message: dict, murfey_db: Session) -> dict[str, bool]: registered_request, ProcessingFeedbackPublishResponse, ) + + async def _publish_ctf_completed( + micrograph_uuid: str, ctf_max_resolution: float + ) -> None: + publisher = AioPikaPublisher( + url=get_rabbitmq_url(), + exchange_name="smartem", + routing_key="smartem", + exchange_type="fanout", + ) + await publisher.connect() + try: + await publisher.publish_event( + MessageQueueEventType.CTF_COMPLETE, + CtfCompleteBody( + event_type=MessageQueueEventType.CTF_COMPLETE, + micrograph_uuid=micrograph_uuid, + ctf_max_resolution_estimate=ctf_max_resolution, + ), + ) + except Exception: + logger.warning( + f"smartem failed to ctf estimation completion {micrograph_uuid}", + exc_info=True, + ) + finally: + await publisher.close() + + asyncio.run( + _publish_ctf_completed( + movie.smartem_uuid, message.get("ctf_max_resolution", 1000) + ) + ) + except Exception: logger.warning( "Failed to emit CTF estimation complete event to smartem", diff --git a/src/murfey/workflows/spa/motion_correction.py b/src/murfey/workflows/spa/motion_correction.py index cd6b89a3e..07a05a6a9 100644 --- a/src/murfey/workflows/spa/motion_correction.py +++ b/src/murfey/workflows/spa/motion_correction.py @@ -109,8 +109,8 @@ async def _publish_motion_correction_completed( asyncio.run( _publish_motion_correction_completed( movie.smartem_uuid, - message.get("total_motion", 0), - message.get("average_motion", 0), + message.get("total_motion", 1000), + message.get("average_motion", 1000), ) ) From e3cb4143506020eb490201fcd95bbc64f1a0a94c Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Wed, 16 Sep 2026 10:54:52 +0100 Subject: [PATCH 7/9] add publication of ctf estimation complete smartem message --- src/murfey/workflows/spa/ctf_estimation.py | 2 +- src/murfey/workflows/spa/motion_correction.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/murfey/workflows/spa/ctf_estimation.py b/src/murfey/workflows/spa/ctf_estimation.py index c837b45fe..2d8afff2f 100644 --- a/src/murfey/workflows/spa/ctf_estimation.py +++ b/src/murfey/workflows/spa/ctf_estimation.py @@ -43,7 +43,7 @@ def ctf_estimated(message: dict, murfey_db: Session) -> dict[str, bool]: if not SMARTEM_ACTIVE: return {"success": True} movie = murfey_db.exec( - select(Movie).where(Movie.murfey_id == message["motion_correction_id"]) + select(Movie).where(Movie.murfey_id == message["mc_id"]) ).one() if movie.smartem_uuid: try: diff --git a/src/murfey/workflows/spa/motion_correction.py b/src/murfey/workflows/spa/motion_correction.py index 07a05a6a9..2d151c914 100644 --- a/src/murfey/workflows/spa/motion_correction.py +++ b/src/murfey/workflows/spa/motion_correction.py @@ -43,7 +43,7 @@ def motion_corrected(message: dict, murfey_db: Session) -> dict[str, bool]: if not SMARTEM_ACTIVE: return {"success": True} movie = murfey_db.exec( - select(Movie).where(Movie.murfey_id == message["motion_correction_id"]) + select(Movie).where(Movie.murfey_id == message["mc_id"]) ).one() if movie.smartem_uuid: try: From 464824c5d9876f6c2a77b5751d13b459f673aee5 Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Wed, 16 Sep 2026 11:00:17 +0100 Subject: [PATCH 8/9] typo on import --- src/murfey/util/config.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/murfey/util/config.py b/src/murfey/util/config.py index ad80cfd0c..ad664d64f 100644 --- a/src/murfey/util/config.py +++ b/src/murfey/util/config.py @@ -8,7 +8,7 @@ from importlib.metadata import entry_points from pathlib import Path from typing import Any, Literal, Optional -from urllib.parser import quote +from urllib.parse import quote import yaml from pydantic import BaseModel, ConfigDict, RootModel, ValidationInfo, field_validator From abc3552e3757d55ae6a64d352d0165888c63c114 Mon Sep 17 00:00:00 2001 From: Daniel Hatton Date: Wed, 16 Sep 2026 11:31:09 +0100 Subject: [PATCH 9/9] wrong name for movie id --- src/murfey/workflows/spa/ctf_estimation.py | 2 +- src/murfey/workflows/spa/motion_correction.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/murfey/workflows/spa/ctf_estimation.py b/src/murfey/workflows/spa/ctf_estimation.py index 2d8afff2f..605442859 100644 --- a/src/murfey/workflows/spa/ctf_estimation.py +++ b/src/murfey/workflows/spa/ctf_estimation.py @@ -43,7 +43,7 @@ def ctf_estimated(message: dict, murfey_db: Session) -> dict[str, bool]: if not SMARTEM_ACTIVE: return {"success": True} movie = murfey_db.exec( - select(Movie).where(Movie.murfey_id == message["mc_id"]) + select(Movie).where(Movie.murfey_id == message["mc_uuid"]) ).one() if movie.smartem_uuid: try: diff --git a/src/murfey/workflows/spa/motion_correction.py b/src/murfey/workflows/spa/motion_correction.py index 2d151c914..fa77be2be 100644 --- a/src/murfey/workflows/spa/motion_correction.py +++ b/src/murfey/workflows/spa/motion_correction.py @@ -43,7 +43,7 @@ def motion_corrected(message: dict, murfey_db: Session) -> dict[str, bool]: if not SMARTEM_ACTIVE: return {"success": True} movie = murfey_db.exec( - select(Movie).where(Movie.murfey_id == message["mc_id"]) + select(Movie).where(Movie.murfey_id == message["mc_uuid"]) ).one() if movie.smartem_uuid: try: