Merged recent changes from 'main' branch

tieneupin · tieneupin · commit ff10009ec619 · 2025-04-10T14:01:21.000+01:00
diff --git a/.bumpclient.toml b/.bumpclient.toml
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.17.7"
+current_version = "0.17.8"
 commit = true
 tag = false
 
diff --git a/.bumpversion.toml b/.bumpversion.toml
@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.17.7"
+current_version = "0.17.8"
 commit = true
 tag = true
 
diff --git a/Helm/Chart.yaml b/Helm/Chart.yaml
@@ -1,7 +1,7 @@
 apiVersion: v1
 name: murfey-services
 description: Umbrella Helm chart for deploying the servers and daemons needed to enable Murfey to transfer and process data
-version: 0.17.7
+version: 0.17.8
 dependencies:
   - name: murfey-instrument-server-clem
   - name: murfey-instrument-server-tem
diff --git a/Helm/charts/murfey-instrument-server-clem/Chart.yaml b/Helm/charts/murfey-instrument-server-clem/Chart.yaml
@@ -1,4 +1,4 @@
 apiVersion: v1
 name: murfey-instrument-server-clem
 description: Helm chart for deploying a Murfey instrument server, which executes orders to detect, modify, and transfer files on the instrument PC, and notifies the backend server about transferred files
-version: 0.17.7
+version: 0.17.8
diff --git a/Helm/charts/murfey-rsync/Chart.yaml b/Helm/charts/murfey-rsync/Chart.yaml
@@ -1,4 +1,4 @@
 apiVersion: v1
 name: murfey-rsync
 description: Helm chart for deploying an rsync daemon, which is responsible for executing the transfer of files from the client storage directory to the server storage system
-version: 0.17.7
+version: 0.17.8
diff --git a/Helm/charts/murfey-server/Chart.yaml b/Helm/charts/murfey-server/Chart.yaml
@@ -1,4 +1,4 @@
 apiVersion: v1
 name: murfey-server
 description: Helm chart for deploying a Murfey backend server, which is responsible for orchestrating the data transfer and processing workflow between the client PC and the storage system
-version: 0.17.7
+version: 0.17.8
diff --git a/pyproject.toml b/pyproject.toml
@@ -7,7 +7,7 @@ requires = [
 
 [project]
 name = "murfey"
-version = "0.17.7"
+version = "0.17.8"
 description = "Client-Server architecture hauling Cryo-EM data"
 readme = "README.md"
 keywords = [
diff --git a/src/murfey/__init__.py b/src/murfey/__init__.py
@@ -1,4 +1,4 @@
 from __future__ import annotations
 
-__version__ = "0.17.7"
-__supported_client_version__ = "0.17.7"
+__version__ = "0.17.8"
+__supported_client_version__ = "0.17.8"
diff --git a/src/murfey/instrument_server/api.py b/src/murfey/instrument_server/api.py
@@ -138,7 +138,7 @@ def check_token(session_id: MurfeySessionID):
 
 
 @router.post("/sessions/{session_id}/multigrid_watcher")
-def start_multigrid_watcher(
+def setup_multigrid_watcher(
     session_id: MurfeySessionID, watcher_spec: MultigridWatcherSpec
 ):
     if controllers.get(session_id) is not None:
@@ -182,6 +182,13 @@ def start_multigrid_watcher(
     watchers[session_id].subscribe(
         controllers[session_id]._multigrid_watcher_finalised, final=True
     )
+    return {"success": True}
+
+
+@router.post("/sessions/{session_id}/start_multigrid_watcher")
+def start_multigrid_watcher(session_id: MurfeySessionID):
+    if watchers.get(session_id) is None:
+        return {"success": False}
     watchers[session_id].start()
     return {"success": True}
 
@@ -232,7 +239,7 @@ def restart_rsyncer(session_id: MurfeySessionID, rsyncer_source: RsyncerSource):
     return {"success": True}
 
 
-class RSyncerInfo(BaseModel):
+class ObserverInfo(BaseModel):
     source: str
     num_files_transferred: int
     num_files_in_queue: int
@@ -241,11 +248,11 @@ class RSyncerInfo(BaseModel):
 
 
 @router.get("/sessions/{session_id}/rsyncer_info")
-def get_rsyncer_info(session_id: MurfeySessionID) -> list[RSyncerInfo]:
+def get_rsyncer_info(session_id: MurfeySessionID) -> list[ObserverInfo]:
     info = []
     for k, v in controllers[session_id].rsync_processes.items():
         info.append(
-            RSyncerInfo(
+            ObserverInfo(
                 source=str(k),
                 num_files_transferred=v._files_transferred,
                 num_files_in_queue=v.queue.qsize(),
@@ -256,6 +263,22 @@ def get_rsyncer_info(session_id: MurfeySessionID) -> list[RSyncerInfo]:
     return info
 
 
+@router.get("/sessions/{session_id}/analyser_info")
+def get_analyser_info(session_id: MurfeySessionID) -> list[ObserverInfo]:
+    info = []
+    for k, v in controllers[session_id].analysers.items():
+        info.append(
+            ObserverInfo(
+                source=str(k),
+                num_files_transferred=0,
+                num_files_in_queue=v.queue.qsize(),
+                alive=v.thread.is_alive(),
+                stopping=v._stopping,
+            )
+        )
+    return info
+
+
 class ProcessingParameters(BaseModel):
     gain_ref: str
     dose_per_frame: Optional[float] = None
diff --git a/src/murfey/server/__init__.py b/src/murfey/server/__init__.py
@@ -1948,7 +1948,9 @@ def _register_bfactors(message: dict, _db=murfey_db, demo: bool = False):
 
     if message["symmetry"] != relion_params.symmetry:
         # Currently don't do anything with a symmetrised re-run of the refinement
-        logger.info(f"Recieved symmetrised structure of {message['symmetry']}")
+        logger.info(
+            f"Received symmetrised structure of {sanitise(message['symmetry'])}"
+        )
         return True
 
     if not feedback_params.hold_refine:
diff --git a/src/murfey/server/api/__init__.py b/src/murfey/server/api/__init__.py
@@ -90,7 +90,6 @@
     MillingParameters,
     PostInfo,
     PreprocessingParametersTomo,
-    ProcessFile,
     ProcessingJobParameters,
     ProcessingParametersSPA,
     ProcessingParametersTomo,
@@ -104,6 +103,7 @@
     TiltInfo,
     TiltSeriesGroupInfo,
     TiltSeriesInfo,
+    TomoProcessFile,
     Visit,
 )
 from murfey.util.processing_params import default_spa_parameters
@@ -1210,7 +1210,10 @@ async def request_spa_preprocessing(
 
 @router.post("/visits/{visit_name}/{session_id}/tomography_preprocess")
 async def request_tomography_preprocessing(
-    visit_name: str, session_id: MurfeySessionID, proc_file: ProcessFile, db=murfey_db
+    visit_name: str,
+    session_id: MurfeySessionID,
+    proc_file: TomoProcessFile,
+    db=murfey_db,
 ):
     instrument_name = (
         db.exec(select(Session).where(Session.id == session_id)).one().instrument_name
@@ -1260,15 +1263,15 @@ async def request_tomography_preprocessing(
         if not mrc_out.parent.exists():
             mrc_out.parent.mkdir(parents=True, exist_ok=True)
 
-        processing_job_parameters = db.exec(
-            select(TomographyProcessingParameters).where(
-                TomographyProcessingParameters.pj_id == data_collection[0][2].id
+        session_processing_parameters = db.exec(
+            select(SessionProcessingParameters).where(
+                SessionProcessingParameters.session_id == session_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[
+        if session_processing_parameters:
+            proc_file.gain_ref = session_processing_parameters[0].gain_ref
+            proc_file.dose_per_frame = session_processing_parameters[0].dose_per_frame
+            proc_file.eer_fractionation_file = session_processing_parameters[
                 0
             ].eer_fractionation_file
 
diff --git a/src/murfey/server/api/instrument.py b/src/murfey/server/api/instrument.py
@@ -94,7 +94,7 @@ async def check_if_session_is_active(instrument_name: str, session_id: int):
 
 
 @router.post("/sessions/{session_id}/multigrid_watcher")
-async def start_multigrid_watcher(
+async def setup_multigrid_watcher(
     session_id: MurfeySessionID, watcher_spec: MultigridWatcherSetup, db=murfey_db
 ):
     data = {}
@@ -139,6 +139,27 @@ async def start_multigrid_watcher(
     return data
 
 
+@router.post("/sessions/{session_id}/start_multigrid_watcher")
+async def start_multigrid_watcher(session_id: MurfeySessionID, db=murfey_db):
+    data = {}
+    instrument_name = (
+        db.exec(select(Session).where(Session.id == session_id)).one().instrument_name
+    )
+    machine_config = get_machine_config(instrument_name=instrument_name)[
+        instrument_name
+    ]
+    if machine_config.instrument_server_url:
+        async with aiohttp.ClientSession() as clientsession:
+            async with clientsession.post(
+                f"{machine_config.instrument_server_url}/sessions/{session_id}/start_multigrid_watcher",
+                headers={
+                    "Authorization": f"Bearer {instrument_server_tokens[session_id]['access_token']}"
+                },
+            ) as resp:
+                data = await resp.json()
+    return data
+
+
 class ProvidedProcessingParameters(BaseModel):
     dose_per_frame: float
     extract_downscale: bool = True
@@ -434,8 +455,11 @@ class RSyncerInfo(BaseModel):
     source: str
     num_files_transferred: int
     num_files_in_queue: int
+    num_files_to_analyse: int
     alive: bool
     stopping: bool
+    analyser_alive: bool
+    analyser_stopping: bool
     destination: str
     tag: str
     files_transferred: int
@@ -448,7 +472,8 @@ class RSyncerInfo(BaseModel):
 async def get_rsyncer_info(
     instrument_name: str, session_id: MurfeySessionID, db=murfey_db
 ) -> List[RSyncerInfo]:
-    data = []
+    rsyncer_list = []
+    analyser_list = []
     machine_config = get_machine_config(instrument_name=instrument_name)[
         instrument_name
     ]
@@ -464,25 +489,54 @@ async def get_rsyncer_info(
                     f"{machine_config.instrument_server_url}/sessions/{session_id}/rsyncer_info",
                     headers={"Authorization": f"Bearer {token}"},
                 ) as resp:
-                    data = await resp.json()
+                    if resp.status == 200:
+                        rsyncer_list = await resp.json()
+                    else:
+                        rsyncer_list = []
         except KeyError:
-            data = []
+            rsyncer_list = []
         except Exception:
             log.warning(
                 "Exception encountered gathering rsyncer info from the instrument server",
                 exc_info=True,
             )
+
+        try:
+            async with lock:
+                token = instrument_server_tokens[session_id]["access_token"]
+            async with aiohttp.ClientSession() as clientsession:
+                async with clientsession.get(
+                    f"{machine_config.instrument_server_url}/sessions/{session_id}/analyser_info",
+                    headers={"Authorization": f"Bearer {token}"},
+                ) as resp:
+                    if resp.status == 200:
+                        analyser_list = await resp.json()
+                    else:
+                        analyser_list = []
+        except KeyError:
+            analyser_list = []
+        except Exception:
+            log.warning(
+                "Exception encountered gathering analyser info from the instrument server",
+                exc_info=True,
+            )
+
     combined_data = []
-    data_source_lookup = {d["source"]: d for d in data}
+    rsyncer_source_lookup = {d["source"]: d for d in rsyncer_list}
+    analyser_source_lookup = {d["source"]: d for d in analyser_list}
     for ri in rsync_instances:
-        d = data_source_lookup.get(ri.source, {})
+        rsync_data = rsyncer_source_lookup.get(ri.source, {})
+        analyser_data = analyser_source_lookup.get(ri.source, {})
         combined_data.append(
             RSyncerInfo(
                 source=ri.source,
-                num_files_transferred=d.get("num_files_transferred", 0),
-                num_files_in_queue=d.get("num_files_in_queue", 0),
-                alive=d.get("alive", False),
-                stopping=d.get("stopping", True),
+                num_files_transferred=rsync_data.get("num_files_transferred", 0),
+                num_files_in_queue=rsync_data.get("num_files_in_queue", 0),
+                num_files_to_analyse=analyser_data.get("num_files_in_queue", 0),
+                alive=rsync_data.get("alive", False),
+                stopping=rsync_data.get("stopping", True),
+                analyser_alive=analyser_data.get("alive", False),
+                analyser_stopping=analyser_data.get("stopping", True),
                 destination=ri.destination,
                 tag=ri.tag,
                 files_transferred=ri.files_transferred,
diff --git a/src/murfey/server/demo_api.py b/src/murfey/server/demo_api.py
@@ -73,7 +73,6 @@
     GridSquareParameters,
     PostInfo,
     PreprocessingParametersTomo,
-    ProcessFile,
     ProcessingJobParameters,
     ProcessingParametersSPA,
     ProcessingParametersTomo,
@@ -86,6 +85,7 @@
     TiltInfo,
     TiltSeriesGroupInfo,
     TiltSeriesInfo,
+    TomoProcessFile,
     Visit,
 )
 from murfey.util.processing_params import default_spa_parameters
@@ -1196,7 +1196,7 @@ def flush_tomography_processing(
 
 @router.post("/visits/{visit_name}/{client_id}/tomography_preprocess")
 async def request_tomography_preprocessing(
-    visit_name: str, client_id: int, proc_file: ProcessFile, db=murfey_db
+    visit_name: str, client_id: int, proc_file: TomoProcessFile, db=murfey_db
 ):
     if not sanitise_path(Path(proc_file.path)).exists():
         log.warning(
diff --git a/src/murfey/util/models.py b/src/murfey/util/models.py
@@ -308,13 +308,13 @@ class Token(BaseModel):
 """
 
 
-class ProcessFile(BaseModel):  # Rename to TomoProcessFile
+class TomoProcessFile(BaseModel):
     path: str
     description: str
     tag: str
     image_number: int
     pixel_size: float
-    dose_per_frame: float
+    dose_per_frame: Optional[float]
     frame_count: int
     tilt_axis: Optional[float]
     mc_uuid: Optional[int] = None
@@ -350,7 +350,7 @@ class CompletedTiltSeries(BaseModel):
 
 
 class PreprocessingParametersTomo(BaseModel):
-    dose_per_frame: float
+    dose_per_frame: Optional[float]
     frame_count: int
     tilt_axis: float
     gain_ref: Optional[str]
