Merged recent changes from 'main' branch and resolved conflicts

Author: tieneupin

src/murfey/client/contexts/spa.py

@@ -570,7 +570,7 @@ def _position_analysis(
         grid_square = _grid_square_from_file(transferred_file)
         grid_square_metadata_file = _grid_square_metadata_file(
             transferred_file,
-            machine_config["data_directories"],
+            [Path(p) for p in machine_config["data_directories"]],
             environment.visit,
             grid_square,
         )
@@ -921,6 +921,7 @@ def _register_processing_job(
             )
         msg: Dict[str, Any] = {
             "tag": tag,
+            "source": tag,
             "recipe": "ispyb-relion",
             "parameters": {
                 "acquisition_software": parameters["acquisition_software"],

src/murfey/client/contexts/spa_metadata.py

@@ -4,7 +4,6 @@
 
 import requests
 import xmltodict
-from PIL import Image
 
 from murfey.client.context import Context
 from murfey.client.contexts.spa import _get_grid_square_atlas_positions, _get_source
@@ -85,17 +84,9 @@ def post_transfer(
                 atlas_original_pixel_size = atlas_xml_data["MicroscopeImage"][
                     "SpatialScale"
                 ]["pixelSize"]["x"]["numericValue"]
-                readout_width = float(
-                    atlas_xml_data["MicroscopeImage"]["SpatialScale"]["pixelSize"]["x"][
-                        "numericValue"
-                    ]
-                )
 
             # need to calculate the pixel size of the downscaled image
-            atlas_im = Image.open(atlas_xml_path.with_suffix(".jpg"))
-            atlas_pixel_size = atlas_original_pixel_size * (
-                readout_width / atlas_im.width
-            )
+            atlas_pixel_size = atlas_original_pixel_size * 7.8
 
             source = _get_source(
                 visitless_path.parent / "Images-Disc1" / visitless_path.name,

src/murfey/client/contexts/tomo.py

@@ -375,6 +375,7 @@ def _add_tilt(
                                 proc_url,
                                 {
                                     "tag": tilt_series,
+                                    "source": str(self._basepath),
                                     "recipe": "em-tomo-preprocess",
                                     "experiment_type": "tomography",
                                 },
@@ -385,6 +386,7 @@ def _add_tilt(
                                 proc_url,
                                 {
                                     "tag": tilt_series,
+                                    "source": str(self._basepath),
                                     "recipe": "em-tomo-align",
                                     "experiment_type": "tomography",
                                 },
@@ -396,6 +398,7 @@ def _add_tilt(
                             proc_url,
                             json={
                                 "tag": tilt_series,
+                                "source": str(self._basepath),
                                 "recipe": "em-tomo-preprocess",
                                 "experiment_type": "tomography",
                             },
@@ -404,6 +407,7 @@ def _add_tilt(
                             proc_url,
                             json={
                                 "tag": tilt_series,
+                                "source": str(self._basepath),
                                 "recipe": "em-tomo-align",
                                 "experiment_type": "tomography",
                             },

src/murfey/client/multigrid_control.py

@@ -366,16 +366,9 @@ def _start_dc(self, json, from_form: bool = False):
                     f"{self._environment.url.geturl()}/clients/{self._environment.client_id}/tomography_processing_parameters",
                     json=json,
                 )
+
             source = Path(json["source"])
-            self._environment.listeners["data_collection_group_ids"] = {
-                context._flush_data_collections
-            }
-            self._environment.listeners["data_collection_ids"] = {
-                context._flush_processing_job
-            }
-            self._environment.listeners["autoproc_program_ids"] = {
-                context._flush_preprocess
-            }
+
             self._environment.id_tag_registry["data_collection_group"].append(
                 str(source)
             )
@@ -386,12 +379,85 @@ def _start_dc(self, json, from_form: bool = False):
                 "tag": str(source),
             }
             requests.post(url, json=dcg_data)
+
+            data = {
+                "voltage": json["voltage"],
+                "pixel_size_on_image": json["pixel_size_on_image"],
+                "experiment_type": json["experiment_type"],
+                "image_size_x": json["image_size_x"],
+                "image_size_y": json["image_size_y"],
+                "file_extension": json["file_extension"],
+                "acquisition_software": json["acquisition_software"],
+                "image_directory": str(self._environment.default_destinations[source]),
+                "tag": json["tilt_series_tag"],
+                "source": str(source),
+                "magnification": json["magnification"],
+                "total_exposed_dose": json.get("total_exposed_dose"),
+                "c2aperture": json.get("c2aperture"),
+                "exposure_time": json.get("exposure_time"),
+                "slit_width": json.get("slit_width"),
+                "phase_plate": json.get("phase_plate", False),
+            }
+            capture_post(
+                f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/{self._environment.murfey_session}/start_data_collection",
+                json=data,
+            )
+            for recipe in ("em-tomo-preprocess", "em-tomo-align"):
+                capture_post(
+                    f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/{self._environment.murfey_session}/register_processing_job",
+                    json={
+                        "tag": json["tilt_series_tag"],
+                        "source": str(source),
+                        "recipe": recipe,
+                    },
+                )
+            log.info("Registering tomography processing parameters")
+            if self._environment.data_collection_parameters.get("num_eer_frames"):
+                eer_response = requests.post(
+                    f"{str(self._environment.url.geturl())}/visits/{self._environment.visit}/{self._environment.murfey_session}/eer_fractionation_file",
+                    json={
+                        "num_frames": self._environment.data_collection_parameters[
+                            "num_eer_frames"
+                        ],
+                        "fractionation": self._environment.data_collection_parameters[
+                            "eer_fractionation"
+                        ],
+                        "dose_per_frame": self._environment.data_collection_parameters[
+                            "dose_per_frame"
+                        ],
+                        "fractionation_file_name": "eer_fractionation_tomo.txt",
+                    },
+                )
+                eer_fractionation_file = eer_response.json()["eer_fractionation_file"]
+                json.update({"eer_fractionation_file": eer_fractionation_file})
+            requests.post(
+                f"{self._environment.url.geturl()}/sessions/{self._environment.murfey_session}/tomography_preprocessing_parameters",
+                json=json,
+            )
+            context._flush_data_collections()
+            context._flush_processing_jobs()
+            capture_post(
+                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")
+
         elif isinstance(context, SPAContext) or isinstance(context, SPAModularContext):
             url = f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/{self.session_id}/register_data_collection_group"
             dcg_data = {
                 "experiment_type": "single particle",
                 "experiment_type_id": 37,
                 "tag": str(source),
+                "atlas": (
+                    str(self._environment.samples[source].atlas)
+                    if self._environment.samples.get(source)
+                    else ""
+                ),
+                "sample": (
+                    self._environment.samples[source].sample
+                    if self._environment.samples.get(source)
+                    else None
+                ),
             }
             capture_post(url, json=dcg_data)
             if from_form:
@@ -428,7 +494,11 @@ def _start_dc(self, json, from_form: bool = False):
                 ):
                     capture_post(
                         f"{str(self._environment.url.geturl())}/visits/{str(self._environment.visit)}/{self.session_id}/register_processing_job",
-                        json={"tag": str(source), "recipe": recipe},
+                        json={
+                            "tag": str(source),
+                            "source": str(source),
+                            "recipe": recipe,
+                        },
                     )
                 log.info(f"Posting SPA processing parameters: {json}")
                 response = capture_post(

src/murfey/client/tui/app.py

@@ -516,7 +516,11 @@ def _start_dc(self, json, from_form: bool = False):
             for recipe in ("em-tomo-preprocess", "em-tomo-align"):
                 capture_post(
                     f"{str(self._url.geturl())}/visits/{str(self._visit)}/{self._environment.murfey_session}/register_processing_job",
-                    json={"tag": json["tilt_series_tag"], "recipe": recipe},
+                    json={
+                        "tag": json["tilt_series_tag"],
+                        "source": str(source),
+                        "recipe": recipe,
+                    },
                 )
             log.info("Registering tomography processing parameters")
             if self.app._environment.data_collection_parameters.get("num_eer_frames"):
@@ -600,7 +604,11 @@ def _start_dc(self, json, from_form: bool = False):
                 ):
                     capture_post(
                         f"{str(self._url.geturl())}/visits/{str(self._visit)}/{self._environment.murfey_session}/register_processing_job",
-                        json={"tag": str(source), "recipe": recipe},
+                        json={
+                            "tag": str(source),
+                            "source": str(source),
+                            "recipe": recipe,
+                        },
                     )
                 log.info(f"Posting SPA processing parameters: {json}")
                 response = capture_post(

src/murfey/client/tui/screens.py

@@ -717,7 +717,11 @@ def on_button_pressed(self, event: Button.Pressed):
         if self._switch_status:
             self.app.install_screen(
                 DirectorySelection(
-                    [p for p in machine_data.get("data_directories", []) if p.exists()]
+                    [
+                        p
+                        for p in machine_data.get("data_directories", [])
+                        if Path(p).exists()
+                    ]
                 ),
                 "directory-select",
             )

src/murfey/server/__init__.py

@@ -11,13 +11,12 @@
 from importlib.resources import files
 from pathlib import Path
 from threading import Thread
-from typing import Any, Dict, List, NamedTuple, Tuple
+from typing import Any, Dict, List, Literal, NamedTuple, Tuple
 
 import graypy
 import mrcfile
 import numpy as np
 import uvicorn
-import workflows
 from backports.entry_points_selectable import entry_points
 from fastapi import Request
 from fastapi.templating import Jinja2Templates
@@ -42,6 +41,7 @@
 from sqlalchemy.orm.exc import ObjectDeletedError
 from sqlmodel import Session, create_engine, select
 from werkzeug.utils import secure_filename
+from workflows.transport.pika_transport import PikaTransport
 
 import murfey
 import murfey.server.ispyb
@@ -225,6 +225,7 @@ def respond_with_template(
 
 
 def run():
+    # Set up argument parser
     parser = argparse.ArgumentParser(description="Start the Murfey server")
     parser.add_argument(
         "--host",
@@ -273,28 +274,29 @@ def run():
         help="Increase logging output verbosity",
         default=0,
     )
+    # Parse and separate known and unknown args
+    args, unknown = parser.parse_known_args()
 
+    # Load the security configuration
     security_config = get_security_config()
-    # setup logging
+
+    # Set up GrayLog handler if provided in the configuration
     if security_config.graylog_host:
         handler = graypy.GELFUDPHandler(
             security_config.graylog_host, security_config.graylog_port, level_names=True
         )
         root_logger = logging.getLogger()
         root_logger.addHandler(handler)
-
     # Install a log filter to all existing handlers.
     LogFilter.install()
 
-    workflows.transport.load_configuration_file(security_config.rabbitmq_credentials)
-
-    args = parser.parse_args()
-
-    # Set up Zocalo connection
     if args.demo:
+        # Run in demo mode with no connections set up
         os.environ["MURFEY_DEMO"] = "1"
     else:
-        _set_up_transport(args.transport)
+        # Load RabbitMQ configuration and set up the connection
+        PikaTransport().load_configuration_file(security_config.rabbitmq_credentials)
+        _set_up_transport("PikaTransport")
 
     # Set up logging now that the desired verbosity is known
     _set_up_logging(quiet=args.quiet, verbosity=args.verbose)
@@ -393,7 +395,7 @@ def _set_up_logging(quiet: bool, verbosity: int):
         logging.getLogger(logger_name).setLevel(log_level)
 
 
-def _set_up_transport(transport_type):
+def _set_up_transport(transport_type: Literal["PikaTransport"]):
     global _transport_object
     _transport_object = TransportManager(transport_type)
 
@@ -2471,19 +2473,16 @@ def _save_bfactor(message: dict, _db=murfey_db, demo: bool = False):
             _transport_object.send(
                 "ispyb_connector",
                 {
-                    "parameters": {
-                        "ispyb_command": "buffer",
-                        "buffer_lookup": {
-                            "particle_classification_id": refined_class_uuid,
-                        },
-                        "buffer_command": {
-                            "ispyb_command": "insert_particle_classification"
-                        },
-                        "program_id": message["program_id"],
-                        "bfactor_fit_intercept": str(bfactor_fitting[1]),
-                        "bfactor_fit_linear": str(bfactor_fitting[0]),
+                    "ispyb_command": "buffer",
+                    "buffer_lookup": {
+                        "particle_classification_id": refined_class_uuid,
                     },
-                    "content": {"dummy": "dummy"},
+                    "buffer_command": {
+                        "ispyb_command": "insert_particle_classification"
+                    },
+                    "program_id": message["program_id"],
+                    "bfactor_fit_intercept": str(bfactor_fitting[1]),
+                    "bfactor_fit_linear": str(bfactor_fitting[0]),
                 },
                 new_connection=True,
             )
@@ -2639,7 +2638,9 @@ def feedback_callback(header: dict, message: dict) -> None:
                         cassetteSlot=message.get("sample"),
                     )
                     if _transport_object:
-                        atlas_id = _transport_object.do_insert_atlas(atlas_record)
+                        atlas_id = _transport_object.do_insert_atlas(atlas_record)[
+                            "return_value"
+                        ]
                     murfey_dcg = db.DataCollectionGroup(
                         id=dcgid,
                         atlas_id=atlas_id,
@@ -2756,7 +2757,6 @@ def feedback_callback(header: dict, message: dict) -> None:
         elif message["register"] == "processing_job":
             murfey_session_id = message["session_id"]
             logger.info("registering processing job")
-            assert isinstance(global_state["data_collection_ids"], dict)
             dc = murfey_db.exec(
                 select(db.DataCollection, db.DataCollectionGroup)
                 .where(db.DataCollection.dcg_id == db.DataCollectionGroup.id)