merge in main

Author: stephen-riggs

.bumpclient.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.14.0"
+current_version = "0.15.3"
 commit = true
 tag = false
 

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.14.0"
+current_version = "0.15.3"
 commit = true
 tag = true
 

README.md

@@ -2,7 +2,7 @@
 
 A transporter for data from Diamond eBIC microscope and detector machines onto the Diamond network.
 
-### Who is Murfey?
+## Who is Murfey?
 
 Murfey, the package, is named after [Eliza Murfey, the inventor](https://nationalrrmuseum.org/blog/mother-of-invention-women-railroad-innovators/):
 
@@ -11,38 +11,38 @@ Murfey, the package, is named after [Eliza Murfey, the inventor](https://nationa
 > it was Murfey who designed the packings that would lubricate the axles with oil, aiding
 > in the reduction of derailments caused by seized axles and bearings.
 
-### How do I set up a development environment?
+## How do I set up a development environment?
 
 We suggest you start with your favourite virtual environment (mamba/conda/python virtualenv/...),
 then install using the following command.
 
-#### From Git
-
-```bash
+```text
 $ git clone [email protected]:DiamondLightSource/python-murfey.git
 $ cd python-murfey
 $ pip install -e .[client,server,developer]
 ```
 
 The packages included under the `[developer]` installation key contain some helpful tools to aid you with developing Murfey further:
 
+- `bump-my-version` - Simplifies version control.
 - `ipykernel` - Enables interactive code development via Jupyter Notebooks.
 - `pre-commit` - Allows for the installation and running of hooks to help with linting, formatting, and type checking your code.
 - `pytest` - Used in conjunction with test functions to evaluate the reliability of your code.
-- `bump2version` - A nice little script to simplify version control.
+
+Instructions for setting up the database for Murfey to register files to can be found [here](src/murfey/server/MURFEY_DB.md).
 
 Finally, you may want to set up an ISPyB mock database server and a Zocalo
 development environment. The instructions for this are out of scope here.
 
 You can then start the Murfey server with
 
-```bash
+```text
 $ murfey.server
 ```
 
 and connect the client with
 
-```bash
+```text
 $ murfey --server http://127.0.0.1:8000
 ```
 

pyproject.toml

@@ -7,7 +7,7 @@ requires = [
 
 [project]
 name = "murfey"
-version = "0.14.0"
+version = "0.15.3"
 description = "Client-Server architecture hauling Cryo-EM data"
 readme = "README.md"
 keywords = [
@@ -82,6 +82,7 @@ murfey = "murfey.client:run"
 "murfey.create_db" = "murfey.cli.create_db:run"
 "murfey.db_sql" = "murfey.cli.murfey_db_sql:run"
 "murfey.decrypt_password" = "murfey.cli.decrypt_db_password:run"
+"murfey.dlq_murfey" = "murfey.cli.dlq_resubmit:run"
 "murfey.generate_key" = "murfey.cli.generate_crypto_key:run"
 "murfey.generate_password" = "murfey.cli.generate_db_password:run"
 "murfey.instrument_server" = "murfey.instrument_server:run"
@@ -93,6 +94,8 @@ murfey = "murfey.client:run"
 "murfey.transfer" = "murfey.cli.transfer:run"
 [project.entry-points."murfey.auth.token_validation"]
 "password" = "murfey.server.api.auth:password_token_validation"
+[project.entry-points."murfey.config.extraction"]
+"murfey_machine" = "murfey.util.config:get_extended_machine_config"
 [project.entry-points."murfey.workflows"]
 "lif_to_stack" = "murfey.workflows.lif_to_stack:zocalo_cluster_request"
 "tiff_to_stack" = "murfey.workflows.tiff_to_stack:zocalo_cluster_request"

src/murfey/__init__.py

@@ -1,4 +1,4 @@
 from __future__ import annotations
 
-__version__ = "0.14.0"
-__supported_client_version__ = "0.14.0"
+__version__ = "0.15.3"
+__supported_client_version__ = "0.15.3"

src/murfey/cli/dlq_resubmit.py

@@ -0,0 +1,95 @@
+import argparse
+import json
+import subprocess
+from pathlib import Path
+
+import requests
+
+
+def handle_failed_posts(json_folder: Path, token: str):
+    """Deal with any messages that have been sent as failed client posts"""
+    for json_file in json_folder.glob("*"):
+        with open(json_file, "r") as json_data:
+            message = json.load(json_data)
+
+        if not message.get("message") or not message["message"].get("url"):
+            print(f"{json_file} is not a failed client post")
+            continue
+        dest = message["message"]["url"]
+        message_json = message["message"]["json"]
+
+        response = requests.post(
+            dest, json=message_json, headers={"Authorization": f"Bearer {token}"}
+        )
+        if response.status_code != 200:
+            print(f"Failed to repost {json_file}")
+        else:
+            print(f"Reposted {json_file}")
+            json_file.unlink()
+
+
+def handle_dlq_messages(json_folder: Path):
+    """Reinjected to the queue"""
+    for json_file in json_folder.glob("*"):
+        reinject_result = subprocess.run(
+            ["zocalo.dlq_reinject", "-e", "devrmq", str(json_file)],
+            capture_output=True,
+        )
+        if reinject_result.returncode == 0:
+            print(f"Reinjected {json_file}")
+            json_file.unlink()
+        else:
+            print(f"Failed to reinject {json_file}")
+
+
+def run():
+    """
+    Method of checking and purging murfey queues on rabbitmq
+    Two types of messages are possible:
+    - failed client posts which need reposting to the murfey server API
+    - feedback messages that can be sent back to rabbitmq
+    """
+    parser = argparse.ArgumentParser(
+        description="Purge and reinject failed murfey messages"
+    )
+    parser.add_argument(
+        "--queue",
+        help="Queue to check and purge",
+        required=True,
+    )
+    parser.add_argument(
+        "--token",
+        help="Murfey token",
+        required=True,
+    )
+    args = parser.parse_args()
+
+    purge_result = subprocess.run(
+        ["zocalo.dlq_purge", "-e", "devrmq", args.queue],
+        capture_output=True,
+    )
+    if purge_result.returncode != 0:
+        print(f"Failed to purge {args.queue}")
+        return
+    purge_stdout = purge_result.stdout.decode("utf8")
+    export_directories = []
+    if "exported" in purge_stdout:
+        for line in purge_stdout.split("\n"):
+            if line.strip().startswith("DLQ/"):
+                dlq_dir = "DLQ/" + line.split("/")[1]
+                if dlq_dir not in export_directories:
+                    print(f"Found messages in {dlq_dir}")
+                    export_directories.append(dlq_dir)
+
+    if not export_directories:
+        print("No exported messages found")
+        return
+
+    for json_dir in export_directories:
+        handle_failed_posts(Path(json_dir), args.token)
+        handle_dlq_messages(Path(json_dir))
+    print("Done")
+
+
+if __name__ == "__main__":
+    run()

src/murfey/cli/inject_spa_processing.py

@@ -10,7 +10,7 @@
 
 from murfey.server.ispyb import TransportManager
 from murfey.server.murfey_db import url
-from murfey.util.config import get_machine_config, get_microscope
+from murfey.util.config import get_machine_config, get_microscope, get_security_config
 from murfey.util.db import (
     AutoProcProgram,
     ClientEnvironment,
@@ -97,12 +97,13 @@ def run():
         os.environ["BEAMLINE"] = args.microscope
 
     machine_config = get_machine_config()
+    security_config = get_security_config()
     _url = url(machine_config)
     engine = create_engine(_url)
     murfey_db = Session(engine)
 
     _transport_object = TransportManager(args.transport)
-    _transport_object.feedback_queue = machine_config.feedback_queue
+    _transport_object.feedback_queue = security_config.feedback_queue
 
     query = (
         select(Movie)
@@ -182,7 +183,7 @@ def run():
             zocalo_message = {
                 "recipes": ["em-spa-preprocess"],
                 "parameters": {
-                    "feedback_queue": machine_config.feedback_queue,
+                    "feedback_queue": _transport_object.feedback_queue,
                     "node_creator_queue": machine_config.node_creator_queue,
                     "dcid": detached_ids[1],
                     "kv": proc_params["voltage"],

src/murfey/cli/spa_ispyb_messages.py

@@ -22,7 +22,7 @@
 from murfey.server.ispyb import Session, TransportManager, get_session_id
 from murfey.server.murfey_db import url
 from murfey.util import db
-from murfey.util.config import get_machine_config, get_microscope
+from murfey.util.config import get_machine_config, get_microscope, get_security_config
 
 
 def run():
@@ -341,6 +341,7 @@ def run():
         .where(db.ProcessingJob.recipe == "em-spa-preprocess")
     ).one()
     machine_config = get_machine_config()
+    security_config = get_security_config()
     params = db.SPARelionParameters(
         pj_id=collected_ids[2].id,
         angpix=float(metadata["pixel_size_on_image"]) * 1e10,
@@ -377,7 +378,7 @@ def run():
 
     if args.flush_preprocess:
         _transport_object = TransportManager(args.transport)
-        _transport_object.feedback_queue = machine_config.feedback_queue
+        _transport_object.feedback_queue = security_config.feedback_queue
         stashed_files = murfey_db.exec(
             select(db.PreprocessStash)
             .where(db.PreprocessStash.session_id == args.session_id)
@@ -407,7 +408,7 @@ def run():
             zocalo_message = {
                 "recipes": ["em-spa-preprocess"],
                 "parameters": {
-                    "feedback_queue": machine_config.feedback_queue,
+                    "feedback_queue": _transport_object.feedback_queue,
                     "dcid": collected_ids[1].id,
                     "kv": metadata["voltage"],
                     "autoproc_program_id": collected_ids[3].id,

src/murfey/client/__init__.py

@@ -288,7 +288,9 @@ def run():
 
     status_bar = StatusBar()
 
-    machine_data = requests.get(f"{murfey_url.geturl()}/machine").json()
+    machine_data = requests.get(
+        f"{murfey_url.geturl()}/instruments/{instrument_name}/machine"
+    ).json()
     gain_ref: Path | None = None
 
     instance_environment = MurfeyInstanceEnvironment(

src/murfey/client/analyser.py

@@ -22,7 +22,7 @@
 from murfey.client.instance_environment import MurfeyInstanceEnvironment
 from murfey.client.rsync import RSyncerUpdate, TransferResult
 from murfey.client.tui.forms import FormDependency
-from murfey.util import Observer, get_machine_config
+from murfey.util import Observer, get_machine_config_client
 from murfey.util.models import PreprocessingParametersTomo, ProcessingParametersSPA
 
 logger = logging.getLogger("murfey.client.analyser")
@@ -73,7 +73,7 @@ def __init__(
         self._stopping = False
         self._halt_thread = False
         self._murfey_config = (
-            get_machine_config(
+            get_machine_config_client(
                 str(environment.url.geturl()),
                 instrument_name=environment.instrument_name,
                 demo=environment.demo,
@@ -86,6 +86,10 @@ def _find_extension(self, file_path: Path):
         """
         Identifies the file extension and stores that information in the class.
         """
+        if "atlas" in file_path.parts:
+            self._extension = file_path.suffix
+            return True
+
         if (
             required_substrings := self._murfey_config.get(
                 "data_required_substrings", {}
@@ -121,6 +125,10 @@ def _find_context(self, file_path: Path) -> bool:
         stages of processing. Actions to take for individual files will be determined
         in the Context classes themselves.
         """
+        if "atlas" in file_path.parts:
+            self._role = "detector"
+            self._context = SPAMetadataContext("epu", self._basepath)
+            return True
 
         # CLEM workflow checks
         # Look for LIF and XLIF files
@@ -137,7 +145,7 @@ def _find_context(self, file_path: Path) -> bool:
             and self._environment
         ):
             created_directories = set(
-                get_machine_config(
+                get_machine_config_client(
                     str(self._environment.url.geturl()),
                     instrument_name=self._environment.instrument_name,
                     demo=self._environment.demo,
@@ -157,7 +165,7 @@ def _find_context(self, file_path: Path) -> bool:
                     logger.info("Acquisition software: EPU")
                     if self._environment:
                         try:
-                            cfg = get_machine_config(
+                            cfg = get_machine_config_client(
                                 str(self._environment.url.geturl()),
                                 instrument_name=self._environment.instrument_name,
                                 demo=self._environment.demo,
@@ -310,7 +318,10 @@ def _analyse(self):
                             )
                         except Exception as e:
                             logger.error(f"Exception encountered: {e}")
-                        if self._role == "detector":
+                        if (
+                            self._role == "detector"
+                            and "atlas" not in transferred_file.parts
+                        ):
                             if not dc_metadata:
                                 try:
                                     dc_metadata = self._context.gather_metadata(
@@ -376,7 +387,10 @@ def _analyse(self):
                             )
                         except Exception as e:
                             logger.error(f"Exception encountered: {e}")
-                        if self._role == "detector":
+                        if (
+                            self._role == "detector"
+                            and "atlas" not in transferred_file.parts
+                        ):
                             if not dc_metadata:
                                 try:
                                     dc_metadata = self._context.gather_metadata(