Save parameters from instrument server and reuse them from the murfey database (#522)

Restarting the instrument server causes parameters to be lost, and the gain reference is not set correctly in messages from the instrument server.
This save these into a new table, and uses them in preference to values from the client/instrument

Author: stephen-riggs

src/murfey/cli/spa_ispyb_messages.py

@@ -223,12 +223,12 @@ def run():
     metadata["image_size_y"] = str(int(metadata["image_size_y"]) * binning_factor)
     metadata["motion_corr_binning"] = 1 if binning_factor_xml == 2 else 2
     metadata["gain_ref"] = (
-        f"data/{datetime.now().year}/{args.visit}/processing/gain.mrc"
+        f"{datetime.now().year}/{args.visit}/processing/gain.mrc"
         if args.gain_ref is None
         else args.gain_ref
     )
     metadata["gain_ref_superres"] = (
-        f"data/{datetime.now().year}/{args.visit}/processing/gain_superres.mrc"
+        f"{datetime.now().year}/{args.visit}/processing/gain_superres.mrc"
         if args.gain_ref_superres is None
         else args.gain_ref_superres
     )

src/murfey/client/contexts/spa.py

@@ -298,14 +298,14 @@ def gather_metadata(
                 if environment
                 and environment.data_collection_parameters.get("gain_ref")
                 not in (None, "None")
-                else f"data/{datetime.now().year}/{environment.visit}/processing/gain.mrc"
+                else f"{datetime.now().year}/{environment.visit}/processing/gain.mrc"
             )
             metadata["gain_ref_superres"] = (
                 environment.data_collection_parameters.get("gain_ref_superres")
                 if environment
                 and environment.data_collection_parameters.get("gain_ref_superres")
                 not in (None, "None")
-                else f"data/{datetime.now().year}/{environment.visit}/processing/gain_superres.mrc"
+                else f"{datetime.now().year}/{environment.visit}/processing/gain_superres.mrc"
             )
         else:
             metadata["gain_ref"] = None

src/murfey/client/contexts/tomo.py

@@ -618,7 +618,7 @@ def gather_metadata(
                     environment.data_collection_parameters.get("gain_ref")
                     if environment.data_collection_parameters.get("gain_ref")
                     not in (None, "None")
-                    else f"data/{datetime.now().year}/{environment.visit}/processing/gain.mrc"
+                    else f"{datetime.now().year}/{environment.visit}/processing/gain.mrc"
                 )
             else:
                 mdoc_metadata["gain_ref"] = None

src/murfey/instrument_server/api.py

@@ -220,6 +220,7 @@ def restart_rsyncer(session_id: MurfeySessionID, rsyncer_source: RsyncerSource):
 
 
 class ProcessingParameters(BaseModel):
+    gain_ref: str
     dose_per_frame: Optional[float] = None
     extract_downscale: bool = True
     particle_diameter: Optional[float] = None
@@ -238,7 +239,8 @@ def register_processing_parameters(
 ):
     data_collection_parameters[proc_param_block.label] = {}
     for k, v in proc_param_block.params.dict().items():
-        data_collection_parameters[proc_param_block.label][k] = v
+        if v is not None:
+            data_collection_parameters[proc_param_block.label][k] = v
     return {"success": True}
 
 

src/murfey/server/__init__.py

@@ -2490,7 +2490,7 @@ def feedback_callback(header: dict, message: dict) -> None:
                     ),
                     voltage=message["voltage"],
                     motion_corr_binning=message["motion_corr_binning"],
-                    eer_grouping=message["eer_fractionation"],
+                    eer_fractionation_file=message["eer_fractionation_file"],
                     symmetry=message["symmetry"],
                     particle_diameter=message["particle_diameter"],
                     downscale=message["downscale"],

src/murfey/server/api/__init__.py

@@ -65,6 +65,7 @@
     ProcessingJob,
     RsyncInstance,
     Session,
+    SessionProcessingParameters,
     SPAFeedbackParameters,
     SPARelionParameters,
     Tilt,
@@ -477,6 +478,19 @@ def _parse(ps, i, dcg_id):
 def register_spa_proc_params(
     session_id: MurfeySessionID, proc_params: ProcessingParametersSPA, db=murfey_db
 ):
+    session_processing_parameters = db.exec(
+        select(SessionProcessingParameters).where(
+            SessionProcessingParameters.session_id == session_id
+        )
+    ).all()
+    if session_processing_parameters:
+        proc_params.gain_ref = session_processing_parameters[0].gain_ref
+        proc_params.dose_per_frame = session_processing_parameters[0].dose_per_frame
+        proc_params.eer_fractionation_file = session_processing_parameters[
+            0
+        ].eer_fractionation_file
+        proc_params.symmetry = session_processing_parameters[0].symmetry
+
     zocalo_message = {
         "register": "spa_processing_parameters",
         **dict(proc_params),
@@ -605,6 +619,18 @@ def post_foil_hole(
 def register_tomo_preproc_params(
     session_id: MurfeySessionID, proc_params: PreprocessingParametersTomo, db=murfey_db
 ):
+    session_processing_parameters = db.exec(
+        select(SessionProcessingParameters).where(
+            SessionProcessingParameters.session_id == session_id
+        )
+    ).all()
+    if session_processing_parameters:
+        proc_params.gain_ref = session_processing_parameters[0].gain_ref
+        proc_params.dose_per_frame = session_processing_parameters[0].dose_per_frame
+        proc_params.eer_fractionation_file = session_processing_parameters[
+            0
+        ].eer_fractionation_file
+
     zocalo_message = {
         "register": "tomography_processing_parameters",
         **dict(proc_params),
@@ -1142,7 +1168,11 @@ async def request_spa_preprocessing(
                 "picker_uuid": murfey_ids[1],
                 "session_id": session_id,
                 "particle_diameter": proc_params["particle_diameter"] or 0,
-                "fm_int_file": proc_file.eer_fractionation_file,
+                "fm_int_file": (
+                    proc_params["eer_fractionation_file"]
+                    if proc_params["eer_fractionation_file"]
+                    else proc_file.eer_fractionation_file
+                ),
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "cryolo_model_weights": str(
                     _cryolo_model_path(visit_name, instrument_name)
@@ -1229,6 +1259,20 @@ async def request_tomography_preprocessing(
         murfey_ids = _murfey_id(appid, db, number=1, close=False)
         if not mrc_out.parent.exists():
             mrc_out.parent.mkdir(parents=True, exist_ok=True)
+
+        processing_job_parameters = db.exec(
+            select(TomographyProcessingParameters).where(
+                TomographyProcessingParameters.processing_job_id
+                == data_collection[2].id
+            )
+        ).all()
+        if processing_job_parameters:
+            proc_file.gain_ref = processing_job_parameters[0].gain_ref
+            proc_file.dose_per_frame = processing_job_parameters[0].dose_per_frame
+            proc_file.eer_fractionation_file = processing_job_parameters[
+                0
+            ].eer_fractionation_file
+
         zocalo_message: dict = {
             "recipes": [recipe_name],
             "parameters": {
@@ -1514,6 +1558,24 @@ def register_proc(
         },
     }
 
+    session_processing_parameters = db.exec(
+        select(SessionProcessingParameters).where(
+            SessionProcessingParameters.session_id == session_id
+        )
+    ).all()
+
+    if session_processing_parameters:
+        proc_params["job_parameters"].update(
+            {
+                "gain_ref": session_processing_parameters[0].gain_ref,
+                "dose_per_frame": session_processing_parameters[0].dose_per_frame,
+                "eer_fractionation_file": session_processing_parameters[
+                    0
+                ].eer_fractionation_file,
+                "symmetry": session_processing_parameters[0].symmetry,
+            }
+        )
+
     if _transport_object:
         _transport_object.send(
             _transport_object.feedback_queue,
@@ -1654,6 +1716,17 @@ async def write_eer_fractionation_file(
             / machine_config.gain_directory_name
             / secure_filename(fractionation_params.fractionation_file_name)
         )
+
+    session_parameters = db.exec(
+        select(SessionProcessingParameters).where(
+            SessionProcessingParameters.session_id == session_id
+        )
+    ).all()
+    if session_parameters:
+        session_parameters[0].eer_fractionation_file = str(file_path)
+        db.add(session_parameters)
+        db.commit()
+
     if file_path.is_file():
         return {"eer_fractionation_file": str(file_path)}
 

src/murfey/server/api/instrument.py

@@ -23,7 +23,7 @@
 from murfey.server.murfey_db import murfey_db
 from murfey.util import secure_path
 from murfey.util.config import get_machine_config
-from murfey.util.db import Session
+from murfey.util.db import Session, SessionProcessingParameters
 from murfey.util.models import File, MultigridWatcherSetup
 
 # Create APIRouter class object
@@ -151,6 +151,16 @@ class ProvidedProcessingParameters(BaseModel):
 async def pass_proc_params_to_instrument_server(
     session_id: MurfeySessionID, proc_params: ProvidedProcessingParameters, db=murfey_db
 ):
+    session = db.exec(select(Session).where(Session.id == session_id)).one()
+
+    session_processing_parameters = SessionProcessingParameters(
+        dose_per_frame=proc_params.dose_per_frame,
+        gain_ref=session.current_gain_ref,
+        symmetry=proc_params.symmetry,
+    )
+    db.add(session_processing_parameters)
+    db.commit()
+
     data = {}
     instrument_name = (
         db.exec(select(Session).where(Session.id == session_id)).one().instrument_name
@@ -171,6 +181,7 @@ async def pass_proc_params_to_instrument_server(
                         "particle_diameter": proc_params.particle_diameter,
                         "symmetry": proc_params.symmetry,
                         "eer_fractionation": proc_params.eer_fractionation,
+                        "gain_ref": session.current_gain_ref,
                     },
                 },
                 headers={

src/murfey/util/db.py

@@ -326,6 +326,14 @@ class CLEMImageStack(SQLModel, table=True):  # type: ignore
 """
 
 
+class SessionProcessingParameters(SQLModel, table=True):  # type: ignore
+    session_id: int = Field(foreign_key="session.id", primary_key=True)
+    gain_ref: str
+    dose_per_frame: float
+    eer_fractionation_file: str = ""
+    symmetry: str = "C1"
+
+
 class TiltSeries(SQLModel, table=True):  # type: ignore
     id: int = Field(primary_key=True)
     tag: str
@@ -641,7 +649,7 @@ class SPARelionParameters(SQLModel, table=True):  # type: ignore
     gain_ref: Optional[str]
     voltage: int
     motion_corr_binning: int
-    eer_grouping: int
+    eer_fractionation_file: str = ""
     symmetry: str
     particle_diameter: Optional[float]
     downscale: bool

src/murfey/util/models.py

@@ -226,7 +226,7 @@ class ProcessingParametersSPA(BaseModel):
     boxsize: Optional[int]
     downscale: bool
     small_boxsize: Optional[int]
-    eer_fractionation: int
+    eer_fractionation_file: str = ""
     particle_diameter: Optional[float]
     magnification: Optional[int] = None
     total_exposed_dose: Optional[float] = None

src/murfey/workflows/spa/flush_spa_preprocess.py

@@ -417,7 +417,11 @@ def flush_spa_preprocess(message: dict, murfey_db: Session, demo: bool = False)
                 "picker_uuid": murfey_ids[2 * i + 1],
                 "session_id": session_id,
                 "particle_diameter": proc_params.particle_diameter or 0,
-                "fm_int_file": f.eer_fractionation_file,
+                "fm_int_file": (
+                    proc_params.eer_fractionation_file
+                    if proc_params.eer_fractionation_file
+                    else f.eer_fractionation_file
+                ),
                 "do_icebreaker_jobs": default_spa_parameters.do_icebreaker_jobs,
                 "foil_hole_id": foil_hole_id,
             },