Container setup for TEM workflow + tomography/SPA bugfixes

Helm/Chart.yaml

@@ -4,5 +4,6 @@ description: Umbrella Helm chart for deploying the servers and daemons needed to
 version: 0.16.12
 dependencies:
   - name: murfey-instrument-server-clem
+  - name: murfey-instrument-server-tem
   - name: murfey-server
   - name: murfey-rsync

Helm/charts/murfey-instrument-server-tem/Chart.yaml

@@ -0,0 +1,4 @@
+apiVersion: v1
+name: murfey-instrument-server-tem
+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.16.12

Helm/charts/murfey-instrument-server-tem/templates/deployment.yaml

@@ -0,0 +1,68 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{ .Values.appName }}
+  namespace: {{ .Values.global.namespace }}
+  labels:
+    app: {{ .Values.appName }}
+spec:
+  type: LoadBalancer
+  externalTrafficPolicy: Cluster
+  ports:
+    {{- toYaml .Values.servicePorts | nindent 2 }}
+  selector:
+    app: {{ .Values.appName }}
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ .Values.appName }}
+  namespace: {{ .Values.global.namespace }}
+  labels:
+    app: {{ .Values.appName }}
+spec:
+  replicas: {{ .Values.replicas }}
+  selector:
+    matchLabels:
+      app: {{ .Values.appName }}
+  template:
+    metadata:
+      labels:
+        app: {{ .Values.appName }}
+    spec:
+      securityContext:
+        runAsUser: {{ .Values.global.runAsUser }}
+        runAsGroup: {{ .Values.global.runAsGroup }}
+      volumes:
+        # Mount config files from secrets
+        - name: murfey-client-config
+          secret:
+            secretName: {{ .Values.global.murfeyClientConfigTEMSecretName }}
+            items:
+              - key: {{ .Values.global.murfeyClientConfigTEMFileName }}
+                path: .murfey
+        # Mount data directories
+        {{- toYaml .Values.extraVolumes | nindent 8 }}
+      containers:
+        - name: {{ .Values.appName }}
+          image: {{ .Values.image }}
+          imagePullPolicy: Always
+          securityContext:
+            privileged: false
+          volumeMounts:
+            # Mount Murfey client config
+            - name: murfey-client-config
+              mountPath: /murfey/config/.murfey
+              subPath: .murfey
+              readOnly: false
+            # Mount data directories
+            {{- toYaml .Values.extraVolumeMounts | nindent 12 }}
+          env:
+            - name: MURFEY_CLIENT_CONFIG_HOME
+              value: "/tmp"
+          ports:
+            - containerPort: {{ .Values.containerPort }}
+          command:
+            {{- toYaml .Values.command | nindent 12 }}
+          args:
+            {{- toYaml .Values.args | nindent 12 }}

pyproject.toml

@@ -64,7 +64,7 @@ server = [
     "ispyb", # Responsible for setting requirements for SQLAlchemy and mysql-connector-python; v10.0.0: sqlalchemy <2, mysql-connector-python >=8.0.32
     "jinja2",
     "mrcfile",
-    "numpy",
+    "numpy<2",
     "packaging",
     "passlib",
     "pillow",

src/murfey/client/__init__.py

@@ -322,6 +322,11 @@ def run():
         default_destination=args.destination or str(datetime.now().year),
         demo=args.demo,
         processing_only_mode=server_routing_prefix_found,
+        rsync_url=(
+            urlparse(machine_data["rsync_url"]).hostname
+            if machine_data.get("rsync_url")
+            else ""
+        ),
     )
 
     ws.environment = instance_environment

src/murfey/client/analyser.py

@@ -23,6 +23,7 @@
 from murfey.client.rsync import RSyncerUpdate, TransferResult
 from murfey.client.tui.forms import FormDependency
 from murfey.util import Observer, get_machine_config_client
+from murfey.util.mdoc import get_block
 from murfey.util.models import PreprocessingParametersTomo, ProcessingParametersSPA
 
 logger = logging.getLogger("murfey.client.analyser")
@@ -113,6 +114,13 @@
         ):
             logger.info(f"File extension re-evaluated: {file_path.suffix}")
             self._extension = file_path.suffix
+        # If we see an .mdoc file first, use that to determine the file extensions
+        elif file_path.suffix == ".mdoc":
+            with open(file_path, "r") as md:
+                md.seek(0)
+                mdoc_data_block = get_block(md)
+            if subframe_path := mdoc_data_block.get("SubFramePath"):
+                self._extension = Path(subframe_path).suffix
         # Check for LIF files separately
         elif file_path.suffix == ".lif":
             self._extension = file_path.suffix
@@ -124,6 +132,7 @@
         stages of processing. Actions to take for individual files will be determined
         in the Context classes themselves.
         """
+        logger.debug(f"Finding context using file {str(file_path)!r}")
         if "atlas" in file_path.parts:
             self._context = SPAMetadataContext("epu", self._basepath)
             return True
@@ -258,9 +267,9 @@
                     self._find_extension(transferred_file)
                     found = self._find_context(transferred_file)
                     if not found:
-                        # logger.warning(
-                        #     f"Context not understood for {transferred_file}, stopping analysis"
-                        # )
+                        logger.debug(
+                            f"Couldn't find context for {str(transferred_file)!r}"
+                        )
                         self.queue.task_done()
                         continue
                     elif self._extension:
@@ -383,6 +392,10 @@
                         SPAMetadataContext,
                     ),
                 ):
+                    context = str(self._context).split(" ")[0].split(".")[-1]
+                    logger.debug(
+                        f"Transferring file {str(transferred_file)} with context {context!r}"
+                    )
                     self.post_transfer(transferred_file)
             self.queue.task_done()
 

src/murfey/client/contexts/tomo.py

@@ -366,13 +366,19 @@ def _check_tilt_series(
         newly_completed_series: List[str] = []
         mdoc_tilt_series_size = self._tilt_series_sizes.get(tilt_series, 0)
         if not self._tilt_series or not mdoc_tilt_series_size:
+            logger.debug(f"Tilt series size not yet set for {tilt_series!r}")
             return newly_completed_series
 
         counted_tilts = len(self._tilt_series.get(tilt_series, []))
         tilt_series_size_check = counted_tilts >= mdoc_tilt_series_size
         if tilt_series_size_check and tilt_series not in self._completed_tilt_series:
             self._completed_tilt_series.append(tilt_series)
             newly_completed_series.append(tilt_series)
+        else:
+            logger.debug(
+                f"{tilt_series!r} not complete yet. Counted {counted_tilts} tilts. "
+                f"Expected number of tilts was {mdoc_tilt_series_size}"
+            )
         return newly_completed_series
 
     def _add_tomo_tilt(

src/murfey/client/instance_environment.py

@@ -52,6 +52,7 @@ class MurfeyInstanceEnvironment(BaseModel):
     superres: bool = True
     murfey_session: Optional[int] = None
     samples: Dict[Path, SampleInfo] = {}
+    rsync_url: str = ""
 
     class Config:
         validate_assignment: bool = True

src/murfey/client/multigrid_control.py

@@ -410,7 +410,7 @@
                 f"{self._environment.url.geturl()}/visits/{self._environment.visit}/{self._environment.murfey_session}/flush_tomography_processing",
                 json={"rsync_source": str(source)},
             )
-            log.info("tomography processing flushed")
+            log.info("Tomography processing flushed")
 
         elif isinstance(context, SPAModularContext):
             url = f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/{self.session_id}/register_data_collection_group"

src/murfey/client/tui/app.py

@@ -511,7 +511,7 @@
                 f"{self.app._environment.url.geturl()}/visits/{self._visit}/{self.app._environment.murfey_session}/flush_tomography_processing",
                 json={"rsync_source": str(source)},
             )
-            log.info("tomography processing flushed")
+            log.info("Tomography processing flushed")
         elif isinstance(context, SPAModularContext):
             url = f"{str(self._url.geturl())}/visits/{str(self._visit)}/{self._environment.murfey_session}/register_data_collection_group"
             dcg_data = {

src/murfey/client/tui/screens.py

@@ -886,7 +886,7 @@ def on_button_pressed(self, event):
             rsync_cmd = [
                 "rsync",
                 f"{posix_path(self._dir_tree._gain_reference)!r}",
-                f"{self.app._environment.url.hostname}::{self.app._machine_config.get('rsync_module', 'data')}/{visit_path}/processing/{secure_filename(self._dir_tree._gain_reference.name)}",
+                f"{self.app._environment.rsync_url or self.app._environment.url.hostname}::{self.app._machine_config.get('rsync_module', 'data')}/{visit_path}/processing/{secure_filename(self._dir_tree._gain_reference.name)}",
             ]
             # Encase in bash shell
             cmd = ["bash", "-c", " ".join(rsync_cmd)]

src/murfey/server/__init__.py

@@ -8,6 +8,7 @@
 import time
 from datetime import datetime
 from functools import partial, singledispatch
+from importlib.metadata import EntryPoint  # For type hinting only
 from importlib.resources import files
 from pathlib import Path
 from threading import Thread
@@ -20,7 +21,6 @@
 from backports.entry_points_selectable import entry_points
 from fastapi import Request
 from fastapi.templating import Jinja2Templates
-from importlib_metadata import EntryPoint  # For type hinting only
 from ispyb.sqlalchemy._auto_db_schema import (
     Atlas,
     AutoProcProgram,
@@ -620,8 +620,6 @@
             )
         ).one()
 
-        particle_diameter = relion_params.particle_diameter
-
         if feedback_params.picker_ispyb_id is None:
             if demo or not _transport_object:
                 feedback_params.picker_ispyb_id = 1000
@@ -649,15 +647,15 @@
                 _db.delete(s)
                 _db.commit()
 
-        if not particle_diameter:
+        # Calculate diameter if it wasn't provided
+        if not relion_params.particle_diameter:
             # If the diameter has not been calculated then find it
             picking_db = _db.exec(
                 select(db.ParticleSizes.particle_size).where(
                     db.ParticleSizes.pj_id == pj_id
                 )
             ).all()
-            particle_diameter = np.quantile(list(picking_db), 0.75)
-            relion_params.particle_diameter = particle_diameter
+            relion_params.particle_diameter = np.quantile(list(picking_db), 0.75)
             _db.add(relion_params)
             _db.commit()
 
@@ -682,7 +680,7 @@
                         "defocus_u": saved_message.defocus_u,
                         "defocus_v": saved_message.defocus_v,
                         "defocus_angle": saved_message.defocus_angle,
-                        "particle_diameter": particle_diameter,
+                        "particle_diameter": relion_params.particle_diameter,
                         "downscale": relion_options["downscale"],
                         "kv": relion_options["voltage"],
                         "node_creator_queue": machine_config.node_creator_queue,
@@ -702,9 +700,9 @@
                     _transport_object.send(
                         "processing_recipe", zocalo_message, new_connection=True
                     )
+        # Use provided diameter for next step
         else:
             # If the diameter is known then just send the new message
-            particle_diameter = relion_params.particle_diameter
             zocalo_message = {
                 "parameters": {
                     "micrographs_file": params_to_forward["micrographs_file"],
@@ -723,7 +721,7 @@
                     "defocus_u": params_to_forward["ctf_values"]["DefocusU"],
                     "defocus_v": params_to_forward["ctf_values"]["DefocusV"],
                     "defocus_angle": params_to_forward["ctf_values"]["DefocusAngle"],
-                    "particle_diameter": particle_diameter,
+                    "particle_diameter": relion_params.particle_diameter,
                     "downscale": relion_options["downscale"],
                     "kv": relion_options["voltage"],
                     "node_creator_queue": machine_config.node_creator_queue,
@@ -2944,12 +2942,19 @@
         elif (
             message["register"] in entry_points().select(group="murfey.workflows").names
         ):
-            # Run the workflow if a match is found
-            workflow: EntryPoint = list(  # Returns a list of either 1 or 0
+            # Search for corresponding workflow
+            workflows: list[EntryPoint] = list(
                 entry_points().select(
                     group="murfey.workflows", name=message["register"]
                 )
-            )[0]
+            )  # Returns either 1 item or empty list
+            if not workflows:
+                logger.error(f"No workflow found for {sanitise(message['register'])}")
+                if _transport_object:
+                    _transport_object.transport.nack(header, requeue=False)
+                return None
+            # Run the workflow if a match is found
+            workflow: EntryPoint = workflows[0]
             result = workflow.load()(
                 message=message,
                 murfey_db=murfey_db,

src/murfey/server/api/__init__.py

@@ -1056,7 +1056,7 @@ async def request_spa_preprocessing(
                 "mrc_out": str(mrc_out),
                 "pixel_size": proc_params["angpix"],
                 "image_number": proc_file.image_number,
-                "microscope": get_microscope(),
+                "microscope": instrument_name,
                 "mc_uuid": murfey_ids[0],
                 "foil_hole_id": foil_hole_id,
                 "ft_bin": proc_params["motion_corr_binning"],
@@ -1164,7 +1164,7 @@ async def request_tomography_preprocessing(
                 "pixel_size": (proc_file.pixel_size) * 10**10,
                 "image_number": proc_file.image_number,
                 "kv": int(proc_file.voltage),
-                "microscope": get_microscope(),
+                "microscope": instrument_name,
                 "mc_uuid": murfey_ids[0],
                 "ft_bin": proc_file.mc_binning,
                 "fm_dose": proc_file.dose_per_frame,
@@ -1375,7 +1375,7 @@ def start_dc(
         instrument_name
     ]
     log.info(
-        f"Starting data collection on microscope {get_microscope(machine_config=machine_config)} "
+        f"Starting data collection on microscope {instrument_name!r} "
         f"with basepath {sanitise(str(machine_config.rsync_basepath))} and directory {sanitise(dc_params.image_directory)}"
     )
     dc_parameters = {
@@ -1408,7 +1408,7 @@ def start_dc(
             {
                 "register": "data_collection",
                 **dc_parameters,
-                "microscope": get_microscope(machine_config=machine_config),
+                "microscope": instrument_name,
                 "proposal_code": ispyb_proposal_code,
                 "proposal_number": ispyb_proposal_number,
                 "visit_number": ispyb_visit_number,
@@ -1552,7 +1552,7 @@ async def write_eer_fractionation_file(
             Path(machine_config.rsync_basepath)
             / str(datetime.datetime.now().year)
             / secure_filename(visit_name)
-            / "processing"
+            / machine_config.gain_directory_name
             / secure_filename(fractionation_params.fractionation_file_name)
         )
     if file_path.is_file():

src/murfey/server/demo_api.py

@@ -1575,10 +1575,11 @@
         )
     else:
         return {"gain_ref": None}
+    gain_ref_folder = machine_config.get("gain_directory_name", "processing")
     gain_ref_out = (
-        (filepath / "processing" / f"gain_{gain_reference_params.tag}.mrc")
+        (filepath / gain_ref_folder / f"gain_{gain_reference_params.tag}.mrc")
         if gain_reference_params.tag
-        else (filepath / "processing" / "gain.mrc")
+        else (filepath / gain_ref_folder / "gain.mrc")
     )
     return {
         "gain_ref": gain_ref_out.relative_to(Path(machine_config["rsync_basepath"]))