Record 'last active' time for SPA particle picking and tomo motion correction jobs

src/murfey/util/db.py

@@ -62,6 +62,12 @@ class Session(SQLModel, table=True):  # type: ignore
     visit_end_time: Optional[datetime] = Field(
         default=None, sa_type=DateTime(timezone=False)
     )
+    last_active: Optional[datetime] = Field(
+        default=None, sa_type=DateTime(timezone=False)
+    )
+
+    # Relationships
+    # -------------
     smartem_acquisition_uuid: str | None = Field(default=None)
 
     # Image sites associated with this session

src/murfey/workflows/clem/register_align_and_merge_results.py

@@ -72,11 +72,7 @@ def run(message: dict, murfey_db: Session) -> dict[str, bool]:
         ...
     """
 
-    session_id: int = (
-        int(message["session_id"])
-        if not isinstance(message["session_id"], int)
-        else message["session_id"]
-    )
+    session_id = int(message["session_id"])
 
     # Validate message and try and load results
     try:

src/murfey/workflows/register_atlas_update.py

@@ -20,7 +20,7 @@ def run(
         return {"success": False, "requeue": False}
 
     logger.info(f"Registering updated atlas: \n{message}")
-
+    murfey_session_id = int(message["session_id"])
     murfey.server._transport_object.do_update_atlas(
         atlas_id=message["atlas_id"],
         atlas_image=message["atlas"],
@@ -39,7 +39,7 @@ def run(
     ):
         dcgs_atlas = murfey_db.exec(
             select(DataCollectionGroup)
-            .where(DataCollectionGroup.session_id == message["session_id"])
+            .where(DataCollectionGroup.session_id == murfey_session_id)
             .where(DataCollectionGroup.atlas == message["atlas"])
             .where(DataCollectionGroup.sample == message["sample"])
         ).all()
@@ -54,7 +54,7 @@ def run(
     ):
         try:
             for hook in dcg_hooks:
-                hook.load()(message["dcgid"], session_id=message["session_id"])
+                hook.load()(message["dcgid"], session_id=murfey_session_id)
         except Exception:
             logger.error("Call to data collection group hook failed", exc_info=True)
     return {"success": True}

src/murfey/workflows/register_data_collection.py

@@ -5,8 +5,8 @@
 from sqlmodel import select
 from sqlmodel.orm.session import Session as SQLModelSession
 
-import murfey.util.db as MurfeyDB
 import murfey.server
+import murfey.util.db as MurfeyDB
 from murfey.server.ispyb import ISPyBSession, get_session_id
 from murfey.util import sanitise
 
@@ -21,7 +21,7 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
 
     logger.info(f"Registering the following data collection: \n{message}")
 
-    murfey_session_id = message["session_id"]
+    murfey_session_id = int(message["session_id"])
     ispyb_session_id = get_session_id(
         microscope=message["microscope"],
         proposal_code=message["proposal_code"],

src/murfey/workflows/register_data_collection_group.py

@@ -22,6 +22,7 @@ def register_dcg(
         logger.error("Unable to find transport manager")
         return None
 
+    murfey_session_id = int(message["session_id"])
     ispyb_session_id = get_session_id(
         microscope=message["microscope"],
         proposal_code=message["proposal_code"],
@@ -31,7 +32,7 @@ def register_dcg(
     )
     if ispyb_session_id is None:
         murfey_dcg = DataCollectionGroup(
-            session_id=message["session_id"],
+            session_id=murfey_session_id,
             tag=message.get("tag"),
             smartem_grid_uuid=message.get("smartem_grid_uuid"),
         )
@@ -70,15 +71,15 @@ def register_dcg(
             atlas=message.get("atlas", ""),
             atlas_pixel_size=message.get("atlas_pixel_size"),
             sample=message.get("sample"),
-            session_id=message["session_id"],
+            session_id=murfey_session_id,
             tag=message.get("tag"),
             smartem_grid_uuid=message.get("smartem_grid_uuid"),
         )
     murfey_db.add(murfey_dcg)
     if dcgid is not None and message.get("atlas_x_stage_position"):
         atlas_site = ImagingSite(
             dcg_id=dcgid,
-            session_id=message["session_id"],
+            session_id=murfey_session_id,
             site_name=message.get("tag"),
             data_type="atlas",
             pos_x=message.get("atlas_x_stage_position"),
@@ -99,10 +100,10 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
         return {"success": False, "requeue": False}
 
     logger.info(f"Registering the following data collection group: \n{message}")
-
+    murfey_session_id = int(message["session_id"])
     if dcg_murfey := murfey_db.exec(
         select(DataCollectionGroup)
-        .where(DataCollectionGroup.session_id == message["session_id"])
+        .where(DataCollectionGroup.session_id == murfey_session_id)
         .where(DataCollectionGroup.tag == message.get("tag"))
     ).all():
         dcgid = dcg_murfey[0].id
@@ -128,7 +129,7 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     ):
         dcgs_atlas = murfey_db.exec(
             select(DataCollectionGroup)
-            .where(DataCollectionGroup.session_id == message["session_id"])
+            .where(DataCollectionGroup.session_id == murfey_session_id)
             .where(DataCollectionGroup.atlas == message["atlas"])
             .where(DataCollectionGroup.sample == message["sample"])
         ).all()
@@ -140,7 +141,7 @@ def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     if dcg_hooks := entry_points(group="murfey.hooks", name="data_collection_group"):
         try:
             for hook in dcg_hooks:
-                hook.load()(dcgid, session_id=message["session_id"])
+                hook.load()(dcgid, session_id=murfey_session_id)
         except Exception:
             logger.error("Call to data collection group hook failed", exc_info=True)
 

src/murfey/workflows/register_processing_job.py

@@ -5,9 +5,9 @@
 from sqlmodel import select
 from sqlmodel.orm.session import Session as SQLModelSession
 
+import murfey.server
 import murfey.server.prometheus as prom
 import murfey.util.db as MurfeyDB
-import murfey.server
 from murfey.server.ispyb import ISPyBSession
 from murfey.util import sanitise
 
@@ -22,7 +22,7 @@ def run(message: dict, murfey_db: SQLModelSession):
 
     logger.info(f"Registering the following processing job: \n{message}")
 
-    murfey_session_id = message["session_id"]
+    murfey_session_id = int(message["session_id"])
     dc = murfey_db.exec(
         select(MurfeyDB.DataCollection, MurfeyDB.DataCollectionGroup)
         .where(MurfeyDB.DataCollection.dcg_id == MurfeyDB.DataCollectionGroup.id)
@@ -86,9 +86,9 @@ def run(message: dict, murfey_db: SQLModelSession):
             record = ISPyBDB.AutoProcProgram(
                 processingJobId=pid, processingStartTime=datetime.now()
             )
-            appid = murfey.server._transport_object.do_update_processing_status(record).get(
-                "return_value", None
-            )
+            appid = murfey.server._transport_object.do_update_processing_status(
+                record
+            ).get("return_value", None)
             if appid is None:
                 return {"success": False, "requeue": True}
             murfey_app = MurfeyDB.AutoProcProgram(id=appid, pj_id=pid)

src/murfey/workflows/spa/ctf_estimation.py

@@ -42,13 +42,14 @@
 def ctf_estimated(message: dict, murfey_db: Session) -> dict[str, bool]:
     if not SMARTEM_ACTIVE:
         return {"success": True}
+    session_id = int(message["session_id"])
     movie = murfey_db.exec(
         select(Movie).where(Movie.murfey_id == message["mc_uuid"])
     ).one()
     if movie.smartem_uuid:
         try:
             session = murfey_db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
+                select(MurfeySession).where(MurfeySession.id == session_id)
             ).one()
             machine_config = get_machine_config(
                 instrument_name=session.instrument_name

src/murfey/workflows/spa/flush_spa_preprocess.py

@@ -302,7 +302,7 @@ def _flush_position_analysis(
 
 
 def flush_spa_preprocess(message: dict, murfey_db: Session) -> dict[str, bool]:
-    session_id = message["session_id"]
+    session_id = int(message["session_id"])
     stashed_files = murfey_db.exec(
         select(PreprocessStash)
         .where(PreprocessStash.session_id == session_id)
@@ -312,7 +312,7 @@ def flush_spa_preprocess(message: dict, murfey_db: Session) -> dict[str, bool]:
         return {"success": True}
 
     murfey_session = murfey_db.exec(
-        select(MurfeySession).where(MurfeySession.id == message["session_id"])
+        select(MurfeySession).where(MurfeySession.id == session_id)
     ).one()
     machine_config = get_machine_config(instrument_name=murfey_session.instrument_name)[
         murfey_session.instrument_name

src/murfey/workflows/spa/motion_correction.py

@@ -42,13 +42,14 @@
 def motion_corrected(message: dict, murfey_db: Session) -> dict[str, bool]:
     if not SMARTEM_ACTIVE:
         return {"success": True}
+    session_id = int(message["session_id"])
     movie = murfey_db.exec(
         select(Movie).where(Movie.murfey_id == message["mc_uuid"])
     ).one()
     if movie.smartem_uuid:
         try:
             session = murfey_db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
+                select(MurfeySession).where(MurfeySession.id == session_id)
             ).one()
             machine_config = get_machine_config(
                 instrument_name=session.instrument_name

src/murfey/workflows/spa/picking.py

@@ -1,4 +1,5 @@
 import asyncio
+from datetime import datetime
 from logging import getLogger
 from typing import List
 
@@ -57,7 +58,10 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
     # Add this message to the table of seen messages
     params_to_forward = message.get("extraction_parameters")
     assert isinstance(params_to_forward, dict)
-    pj_id = _pj_id(message["program_id"], _db)
+
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
+    pj_id = _pj_id(program_id, _db)
     ctf_params = CtfParameters(
         pj_id=pj_id,
         micrographs_file=params_to_forward["micrographs_file"],
@@ -80,9 +84,7 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
     if picking_db_len > default_spa_parameters.nr_picks_before_diameter:
         # If there are enough particles to get a diameter
         instrument_name = (
-            _db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
-            )
+            _db.exec(select(MurfeySession).where(MurfeySession.id == session_id))
             .one()
             .instrument_name
         )
@@ -130,9 +132,9 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
                         "downscale": relion_options["downscale"],
                         "kv": relion_options["voltage"],
                         "node_creator_queue": machine_config.node_creator_queue,
-                        "session_id": message["session_id"],
+                        "session_id": session_id,
                         "autoproc_program_id": _app_id(
-                            _pj_id(message["program_id"], _db, recipe="em-spa-extract"),
+                            _pj_id(program_id, _db, recipe="em-spa-extract"),
                             _db,
                         ),
                         "batch_size": default_spa_parameters.batch_size_2d,
@@ -171,9 +173,9 @@ def _register_picked_particles_use_diameter(message: dict, _db: Session):
                     "downscale": relion_options["downscale"],
                     "kv": relion_options["voltage"],
                     "node_creator_queue": machine_config.node_creator_queue,
-                    "session_id": message["session_id"],
+                    "session_id": session_id,
                     "autoproc_program_id": _app_id(
-                        _pj_id(message["program_id"], _db, recipe="em-spa-extract"), _db
+                        _pj_id(program_id, _db, recipe="em-spa-extract"), _db
                     ),
                     "batch_size": default_spa_parameters.batch_size_2d,
                 },
@@ -204,15 +206,17 @@ def _register_picked_particles_use_boxsize(message: dict, _db: Session):
     params_to_forward = message.get("extraction_parameters")
     assert isinstance(params_to_forward, dict)
 
+    session_id = int(message["session_id"])
     instrument_name = (
-        _db.exec(select(MurfeySession).where(MurfeySession.id == message["session_id"]))
+        _db.exec(select(MurfeySession).where(MurfeySession.id == session_id))
         .one()
         .instrument_name
     )
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
-    pj_id = _pj_id(message["program_id"], _db)
+    program_id = int(message["program_id"])
+    pj_id = _pj_id(program_id, _db)
     ctf_params = CtfParameters(
         pj_id=pj_id,
         micrographs_file=params_to_forward["micrographs_file"],
@@ -253,9 +257,9 @@ def _register_picked_particles_use_boxsize(message: dict, _db: Session):
             "downscale": relion_params.downscale,
             "kv": relion_params.voltage,
             "node_creator_queue": machine_config.node_creator_queue,
-            "session_id": message["session_id"],
+            "session_id": session_id,
             "autoproc_program_id": _app_id(
-                _pj_id(message["program_id"], _db, recipe="em-spa-extract"), _db
+                _pj_id(program_id, _db, recipe="em-spa-extract"), _db
             ),
             "batch_size": default_spa_parameters.batch_size_2d,
         },
@@ -298,11 +302,13 @@ def _request_email(
 
 
 def _check_notifications(message: dict, murfey_db: Session) -> None:
+    session_id = int(message["session_id"])
+    program_id = int(message["program_id"])
     data_collection_hierarchy = murfey_db.exec(
         select(DataCollection, ProcessingJob, AutoProcProgram)
         .where(ProcessingJob.dc_id == DataCollection.id)
         .where(AutoProcProgram.pj_id == ProcessingJob.id)
-        .where(AutoProcProgram.id == message["program_id"])
+        .where(AutoProcProgram.id == program_id)
     ).all()
     dcgid = data_collection_hierarchy[0][0].dcg_id
     notification_parameters = murfey_db.exec(
@@ -383,11 +389,19 @@ def _check_notifications(message: dict, murfey_db: Session) -> None:
             "Requested email notification for the following abnormal parameters: \n"
             f"{', '.join([f'{p}' for p in failures])}"
         )
-        _request_email(failures, dcgid, message["session_id"], murfey_db)
+        _request_email(failures, dcgid, session_id, murfey_db)
     return None
 
 
 def particles_picked(message: dict, murfey_db: Session) -> dict[str, bool]:
+    # Update the last_active' column for the session
+    session_id = int(message["session_id"])
+    session = murfey_db.exec(
+        select(MurfeySession).where(MurfeySession.id == session_id)
+    ).one()
+    session.last_active = datetime.now()
+    murfey_db.add(session)
+
     movie = murfey_db.exec(
         select(Movie).where(Movie.murfey_id == message["motion_correction_id"])
     ).one()
@@ -396,9 +410,6 @@ def particles_picked(message: dict, murfey_db: Session) -> dict[str, bool]:
     murfey_db.commit()
     if SMARTEM_ACTIVE and movie.smartem_uuid:
         try:
-            session = murfey_db.exec(
-                select(MurfeySession).where(MurfeySession.id == message["session_id"])
-            ).one()
             machine_config = get_machine_config(
                 instrument_name=session.instrument_name
             )[session.instrument_name]
@@ -461,18 +472,16 @@ async def _publish_particle_picking_completed(
                 "Failed to emit particle picking complete event to smartem",
                 exc_info=True,
             )
+    program_id = int(message["program_id"])
     feedback_params = murfey_db.exec(
         select(ClassificationFeedbackParameters).where(
-            ClassificationFeedbackParameters.pj_id
-            == _pj_id(message["program_id"], murfey_db)
+            ClassificationFeedbackParameters.pj_id == _pj_id(program_id, murfey_db)
         )
     ).one()
     if feedback_params.estimate_particle_diameter:
         _register_picked_particles_use_diameter(message, murfey_db)
     else:
         _register_picked_particles_use_boxsize(message, murfey_db)
-    prom.preprocessed_movies.labels(
-        processing_job=_pj_id(message["program_id"], murfey_db)
-    ).inc()
+    prom.preprocessed_movies.labels(processing_job=_pj_id(program_id, murfey_db)).inc()
     _check_notifications(message, murfey_db)
     return {"success": True}

src/murfey/workflows/spa/register_grid_square.py

@@ -218,7 +218,7 @@ def register_grid_square(
 
 def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     return register_grid_square(
-        message["session_id"],
+        int(message["session_id"]),
         message["gsid"],
         GridSquareParameters(**message["grid_square_params"]),
         murfey_db,

src/murfey/workflows/sxt/process_sxt_tilt_series.py

@@ -205,7 +205,7 @@ def process_sxt_tilt_series(
 def run(message: dict, murfey_db: SQLModelSession) -> dict[str, bool]:
     return process_sxt_tilt_series(
         message["visit_name"],
-        message["session_id"],
+        int(message["session_id"]),
         SXTTiltSeriesInfo(**message["tilt_series_info"]),
         murfey_db,
     )