Add soft x-ray tomography

.github/workflows/ci.yml

@@ -77,7 +77,7 @@ jobs:
       - name: Install Murfey
         run: |
           set -eux
-          pip install --disable-pip-version-check -e "."[cicd,server,developer]
+          pip install --disable-pip-version-check -e "."[cicd,server,developer,sxt]
 
       - uses: shogo82148/actions-setup-mysql@v1
         with:

pyproject.toml

@@ -71,6 +71,9 @@ server = [
 smartem = [
     "smartem-decisions[backend]",
 ]
+sxt = [
+    "txrm2tiff",
+]
 [project.urls]
 Bug-Tracker = "https://github.com/DiamondLightSource/python-murfey/issues"
 Documentation = "https://github.com/DiamondLightSource/python-murfey"

src/murfey/client/analyser.py

@@ -19,6 +19,7 @@
 from murfey.client.contexts.clem import CLEMContext
 from murfey.client.contexts.spa import SPAModularContext
 from murfey.client.contexts.spa_metadata import SPAMetadataContext
+from murfey.client.contexts.sxt import SXTContext
 from murfey.client.contexts.tomo import TomographyContext
 from murfey.client.contexts.tomo_metadata import TomographyMetadataContext
 from murfey.client.destinations import find_longest_data_directory
@@ -110,8 +111,12 @@ def _find_extension(self, file_path: Path) -> bool:
             if subframe_path := mdoc_data_block.get("SubFramePath"):
                 self._extension = Path(subframe_path).suffix
                 return True
-        # Check for LIF files separately
-        elif file_path.suffix == ".lif":
+        # Check for LIF files and TXRM files separately
+        elif (
+            file_path.suffix == ".lif"
+            or file_path.suffix == ".txrm"
+            or file_path.suffix == ".xrm"
+        ):
             self._extension = file_path.suffix
             return True
         return False
@@ -138,6 +143,11 @@ def _find_context(self, file_path: Path) -> bool:
             self._context = CLEMContext("leica", self._basepath, self._token)
             return True
 
+        # SXT workflow checks
+        if file_path.suffix in (".txrm", ".xrm"):
+            self._context = SXTContext("zeiss", self._basepath, self._token)
+            return True
+
         # Tomography and SPA workflow checks
         if "atlas" in file_path.parts:
             self._context = AtlasContext(
@@ -321,6 +331,10 @@ def _analyse(self):
                     )
                     self.post_transfer(transferred_file)
 
+                elif isinstance(self._context, SXTContext):
+                    logger.debug(f"File {transferred_file.name!r} is an SXT file")
+                    self.post_transfer(transferred_file)
+
                 elif isinstance(self._context, AtlasContext):
                     logger.debug(f"File {transferred_file.name!r} is part of the atlas")
                     self.post_transfer(transferred_file)

src/murfey/client/context.py

@@ -13,6 +13,44 @@
 logger = logging.getLogger("murfey.client.context")
 
 
+def _file_transferred_to(
+    environment: MurfeyInstanceEnvironment, source: Path, file_path: Path, token: str
+):
+    machine_config = get_machine_config_client(
+        str(environment.url.geturl()),
+        token,
+        instrument_name=environment.instrument_name,
+    )
+    if environment.visit in environment.default_destinations[source]:
+        return (
+            Path(machine_config.get("rsync_basepath", ""))
+            / Path(environment.default_destinations[source])
+            / file_path.relative_to(source)  # need to strip out the rsync_module name
+        )
+    return (
+        Path(machine_config.get("rsync_basepath", ""))
+        / Path(environment.default_destinations[source])
+        / environment.visit
+        / file_path.relative_to(source)
+    )
+
+
+def _get_source(file_path: Path, environment: MurfeyInstanceEnvironment) -> Path | None:
+    possible_sources = []
+    for s in environment.sources:
+        if file_path.is_relative_to(s):
+            possible_sources.append(s)
+    if not possible_sources:
+        return None
+    elif len(possible_sources) == 1:
+        return possible_sources[0]
+    source = possible_sources[0]
+    for extra_source in possible_sources[1:]:
+        if extra_source.is_relative_to(source):
+            source = extra_source
+    return source
+
+
 def _atlas_destination(
     environment: MurfeyInstanceEnvironment, source: Path, token: str
 ) -> Path:
@@ -61,6 +99,10 @@ def ensure_dcg_exists(
                     )
             except Exception as e:
                 logger.warning(f"Get EPU session hook failed: {e}")
+    elif collection_type == "sxt":
+        experiment_type_id = 47
+        session_file = metadata_source / "Session.dm"
+        source_visit_dir = metadata_source.parent
     else:
         logger.error(f"Unknown collection type {collection_type}")
         return None

src/murfey/client/contexts/atlas.py

@@ -4,8 +4,7 @@
 
 import xmltodict
 
-from murfey.client.context import Context, _atlas_destination
-from murfey.client.contexts.spa import _get_source
+from murfey.client.context import Context, _atlas_destination, _get_source
 from murfey.client.instance_environment import MurfeyInstanceEnvironment
 from murfey.util.client import capture_post
 

src/murfey/client/contexts/spa.py

@@ -7,7 +7,12 @@
 
 import xmltodict
 
-from murfey.client.context import Context, ProcessingParameter
+from murfey.client.context import (
+    Context,
+    ProcessingParameter,
+    _file_transferred_to,
+    _get_source,
+)
 from murfey.client.destinations import find_longest_data_directory
 from murfey.client.instance_environment import (
     MovieTracker,
@@ -26,28 +31,6 @@
 logger = logging.getLogger("murfey.client.contexts.spa")
 
 
-def _file_transferred_to(
-    environment: MurfeyInstanceEnvironment, source: Path, file_path: Path, token: str
-):
-    machine_config = get_machine_config_client(
-        str(environment.url.geturl()),
-        token,
-        instrument_name=environment.instrument_name,
-    )
-    if environment.visit in environment.default_destinations[source]:
-        return (
-            Path(machine_config.get("rsync_basepath", ""))
-            / Path(environment.default_destinations[source])
-            / file_path.relative_to(source)  # need to strip out the rsync_module name
-        )
-    return (
-        Path(machine_config.get("rsync_basepath", ""))
-        / Path(environment.default_destinations[source])
-        / environment.visit
-        / file_path.relative_to(source)
-    )
-
-
 def _grid_square_metadata_file(
     f: Path, data_directories: list[Path], visit: str, grid_square: int
 ) -> Path:
@@ -66,22 +49,6 @@ def _grid_square_metadata_file(
     return metadata_file
 
 
-def _get_source(file_path: Path, environment: MurfeyInstanceEnvironment) -> Path | None:
-    possible_sources = []
-    for s in environment.sources:
-        if file_path.is_relative_to(s):
-            possible_sources.append(s)
-    if not possible_sources:
-        return None
-    elif len(possible_sources) == 1:
-        return possible_sources[0]
-    source = possible_sources[0]
-    for extra_source in possible_sources[1:]:
-        if extra_source.is_relative_to(source):
-            source = extra_source
-    return source
-
-
 def _get_xml_list_index(key: str, xml_list: list) -> int:
     for i, elem in enumerate(xml_list):
         if elem["a:Key"] == key:

src/murfey/client/contexts/spa_metadata.py

@@ -4,8 +4,12 @@
 
 import xmltodict
 
-from murfey.client.context import Context, ensure_dcg_exists
-from murfey.client.contexts.spa import _file_transferred_to, _get_source
+from murfey.client.context import (
+    Context,
+    _file_transferred_to,
+    _get_source,
+    ensure_dcg_exists,
+)
 from murfey.client.instance_environment import MurfeyInstanceEnvironment
 from murfey.util.client import capture_post
 from murfey.util.spa_metadata import (