Fixes for the processing parameters on instrument server

src/murfey/client/analyser.py

@@ -218,7 +218,9 @@ def post_transfer(self, transferred_file: Path):
                     transferred_file, environment=self._environment
                 )
         except Exception as e:
-            logger.error(f"An exception was encountered post transfer: {e}")
+            logger.error(
+                f"An exception was encountered post transfer: {e}", exc_info=True
+            )
 
     def _analyse(self):
         logger.info("Analyser thread started")

src/murfey/instrument_server/__init__.py

@@ -41,7 +41,9 @@ def run():
     )
 
     handler = CustomHandler(ws.send)
-    logging.getLogger().addHandler(handler)
+    logging.getLogger("murfey").addHandler(handler)
+    logging.getLogger("fastapi").addHandler(handler)
+    logging.getLogger("uvicorn").addHandler(handler)
 
     logger.info(
         f"Starting Murfey server version {murfey.__version__}, listening on {args.host}:{args.port}"

src/murfey/server/api/__init__.py

@@ -1262,8 +1262,7 @@ async def request_tomography_preprocessing(
 
         processing_job_parameters = db.exec(
             select(TomographyProcessingParameters).where(
-                TomographyProcessingParameters.processing_job_id
-                == data_collection[2].id
+                TomographyProcessingParameters.pj_id == data_collection[2].id
             )
         ).all()
         if processing_job_parameters:
@@ -1547,7 +1546,7 @@ def register_proc(
     proc_params: ProcessingJobParameters,
     db=murfey_db,
 ):
-    proc_parameters = {
+    proc_parameters: dict = {
         "session_id": session_id,
         "experiment_type": proc_params.experiment_type,
         "recipe": proc_params.recipe,
@@ -1565,7 +1564,8 @@ def register_proc(
     ).all()
 
     if session_processing_parameters:
-        proc_params["job_parameters"].update(
+        job_parameters: dict = proc_parameters["job_parameters"]
+        job_parameters.update(
             {
                 "gain_ref": session_processing_parameters[0].gain_ref,
                 "dose_per_frame": session_processing_parameters[0].dose_per_frame,
@@ -1575,6 +1575,7 @@ def register_proc(
                 "symmetry": session_processing_parameters[0].symmetry,
             }
         )
+        proc_parameters["job_parameters"] = job_parameters
 
     if _transport_object:
         _transport_object.send(

src/murfey/server/api/instrument.py

@@ -60,8 +60,8 @@
         machine_config = get_machine_config(instrument_name=instrument_name)[
             instrument_name
         ]
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.post(
                 f"{machine_config.instrument_server_url}/sessions/{int(sanitise(str(session_id)))}/token",
                 json={"access_token": token, "token_type": "bearer"},
             ) as response:
@@ -80,11 +80,11 @@
     if instrument_server_tokens.get(session_id) is None:
         return {"active": False}
     async with lock:
-        async with aiohttp.ClientSession() as session:
+        async with aiohttp.ClientSession() as clientsession:
             machine_config = get_machine_config(instrument_name=instrument_name)[
                 instrument_name
             ]
-            async with session.get(
+            async with clientsession.get(
                 f"{machine_config.instrument_server_url}/sessions/{int(sanitise(str(session_id)))}/check_token",
                 headers={
                     "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
@@ -116,25 +116,25 @@
             "visit": visit,
             "default_model": str(machine_config.default_model),
         }
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.post(
                 f"{machine_config.instrument_server_url}/sessions/{session_id}/multigrid_watcher",
                 json={
                     "source": str(secure_path(watcher_spec.source / visit)),
                     "visit": visit,
                     "configuration": _config,
                     "label": visit,
                     "instrument_name": instrument_name,
                     "skip_existing_processing": watcher_spec.skip_existing_processing,
                     "destination_overrides": {
                         str(k): v for k, v in watcher_spec.destination_overrides.items()
                     },
                     "rsync_restarts": watcher_spec.rsync_restarts,
                 },
                 headers={
                     "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                 },
             ) as resp:
                 data = await resp.json()
     return data
 
@@ -154,6 +154,7 @@
     session = db.exec(select(Session).where(Session.id == session_id)).one()
 
     session_processing_parameters = SessionProcessingParameters(
+        session_id=session_id,
         dose_per_frame=proc_params.dose_per_frame,
         gain_ref=session.current_gain_ref,
         symmetry=proc_params.symmetry,
@@ -170,24 +171,24 @@
     ]
     if machine_config.instrument_server_url:
         label = db.exec(select(Session).where(Session.id == session_id)).one().name
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.post(
                 f"{machine_config.instrument_server_url}/sessions/{session_id}/processing_parameters",
                 json={
                     "label": label,
                     "params": {
                         "dose_per_frame": proc_params.dose_per_frame,
                         "extract_downscale": proc_params.extract_downscale,
                         "particle_diameter": proc_params.particle_diameter,
                         "symmetry": proc_params.symmetry,
                         "eer_fractionation": proc_params.eer_fractionation,
                         "gain_ref": session.current_gain_ref,
                     },
                 },
                 headers={
                     "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                 },
             ) as resp:
                 data = await resp.json()
     return data
 
@@ -199,8 +200,8 @@
         instrument_name
     ]
     if machine_config.instrument_server_url:
-        async with aiohttp.ClientSession() as session:
-            async with session.get(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.get(
                 f"{machine_config.instrument_server_url}/health",
             ) as resp:
                 data = await resp.json()
@@ -220,11 +221,11 @@
     if machine_config.instrument_server_url:
         async with lock:
             token = instrument_server_tokens[session_id]["access_token"]
-        async with aiohttp.ClientSession() as session:
-            async with session.get(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.get(
                 f"{machine_config.instrument_server_url}/instruments/{sanitise(instrument_name)}/sessions/{sanitise(str(session_id))}/possible_gain_references",
                 headers={"Authorization": f"Bearer {token}"},
             ) as resp:
                 data = await resp.json()
     return data
 
@@ -249,18 +250,18 @@
     visit_path = f"{datetime.datetime.now().year}/{visit}"
     data = {}
     if machine_config.instrument_server_url:
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.post(
                 f"{machine_config.instrument_server_url}/instruments/{instrument_name}/sessions/{session_id}/upload_gain_reference",
                 json={
                     "gain_path": str(gain_reference_request.gain_path),
                     "visit_path": visit_path,
                     "gain_destination_dir": machine_config.gain_directory_name,
                 },
                 headers={
                     "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                 },
             ) as resp:
                 data = await resp.json()
     return data
 
@@ -282,14 +283,14 @@
             / secure_filename(visit_name)
         )
         if machine_config.instrument_server_url:
-            async with aiohttp.ClientSession() as session:
-                async with session.post(
+            async with aiohttp.ClientSession() as clientsession:
+                async with clientsession.post(
                     f"{machine_config.instrument_server_url}/visits/{secure_filename(visit_name)}/sessions/{sanitise(str(session_id))}/upstream_tiff_data_request",
                     json={"download_dir": download_dir},
                     headers={
                         "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                     },
                 ) as resp:
                     data = await resp.json()
     return data
 
@@ -310,17 +311,17 @@
         instrument_name
     ]
     if machine_config.instrument_server_url:
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.post(
                 f"{machine_config.instrument_server_url}/sessions/{session_id}/stop_rsyncer",
                 json={
                     "label": session_id,
                     "source": str(secure_path(Path(rsyncer_source.source))),
                 },
                 headers={
                     "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                 },
             ) as resp:
                 data = await resp.json()
     return data
 
@@ -337,17 +338,17 @@
         instrument_name
     ]
     if machine_config.instrument_server_url:
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.post(
                 f"{machine_config.instrument_server_url}/sessions/{session_id}/finalise_rsyncer",
                 json={
                     "label": session_id,
                     "source": str(secure_path(Path(rsyncer_source.source))),
                 },
                 headers={
                     "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                 },
             ) as resp:
                 data = await resp.json()
     return data
 
@@ -365,17 +366,17 @@
     ]
     if isinstance(session_id, int):
         if machine_config.instrument_server_url:
-            async with aiohttp.ClientSession() as session:
-                async with session.post(
+            async with aiohttp.ClientSession() as clientsession:
+                async with clientsession.post(
                     f"{machine_config.instrument_server_url}/sessions/{session_id}/remove_rsyncer",
                     json={
                         "label": session_id,
                         "source": str(secure_path(Path(rsyncer_source.source))),
                     },
                     headers={
                         "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                     },
                 ) as resp:
                     data = await resp.json()
     return data
 
@@ -393,16 +394,16 @@
     ]
     if isinstance(session_id, int):
         if machine_config.instrument_server_url:
-            async with aiohttp.ClientSession() as session:
-                async with session.post(
+            async with aiohttp.ClientSession() as clientsession:
+                async with clientsession.post(
                     f"{machine_config.instrument_server_url}/sessions/{session_id}/restart_rsyncer",
                     json={
                         "label": session_id,
                         "source": str(secure_path(Path(rsyncer_source.source))),
                     },
                     headers={
                         "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
                     },
                 ) as resp:
                     data = await resp.json()
     return data

src/murfey/server/gain.py

@@ -18,8 +18,11 @@ class Camera(Enum):
     FALCON = 3
 
 
-def _sanitise(gain_path: Path) -> Path:
-    dest = gain_path.parent / "gain" / gain_path.name.replace(" ", "_")
+def _sanitise(gain_path: Path, tag: str) -> Path:
+    if tag:
+        dest = gain_path.parent / f"gain_{tag}" / gain_path.name.replace(" ", "_")
+    else:
+        dest = gain_path.parent / "gain" / gain_path.name.replace(" ", "_")
     dest.write_bytes(gain_path.read_bytes())
     return dest
 
@@ -57,7 +60,7 @@ async def prepare_gain(
             secure_path(gain_path.parent / f"gain_{tag}").mkdir(exist_ok=True)
         else:
             secure_path(gain_path.parent / "gain").mkdir(exist_ok=True)
-        gain_path = _sanitise(gain_path)
+        gain_path = _sanitise(gain_path, tag)
         flip = "flipx" if camera == Camera.K3_FLIPX else "flipy"
         gain_path_mrc = gain_path.with_suffix(".mrc")
         gain_path_superres = gain_path.parent / (gain_path.name + "_superres.mrc")