Register colours and collection mode in CLEMImageSeries table upon arrival of message, then translate it across to Atlas and GridSquare tables in ISPyB

Author: tieneupin

src/murfey/workflows/clem/register_preprocessing_results.py

@@ -67,6 +67,27 @@ def _is_clem_atlas(result: CLEMPreprocessingResult):
     )
 
 
+COLOR_FLAGS_MURFEY = {
+    "gray": "has_grey",
+    "red": "has_red",
+    "green": "has_green",
+    "blue": "has_blue",
+    "cyan": "has_cyan",
+    "magenta": "has_magenta",
+    "yellow": "has_yellow",
+}
+
+
+def _get_color_flags(
+    colors: Collection[str] | None = None,
+):
+    colors = colors or []
+    color_flags = dict.fromkeys(COLOR_FLAGS_MURFEY.values(), False)
+    for color in colors:
+        color_flags[COLOR_FLAGS_MURFEY[color]] = True
+    return color_flags
+
+
 def _register_clem_image_series(
     session_id: int,
     result: CLEMPreprocessingResult,
@@ -160,6 +181,11 @@ def _register_clem_image_series(
     clem_img_series.image_search_string = str(output_file.parent / "*tiff")
     clem_img_series.data_type = "atlas" if _is_clem_atlas(result) else "grid_square"
     clem_img_series.number_of_members = result.number_of_members
+    for col_name, value in _get_color_flags(result.output_files.keys()):
+        setattr(clem_img_series, col_name, value)
+    clem_img_series.collection_mode = _determine_collection_mode(
+        result.output_files.keys()
+    )
     clem_img_series.image_pixels_x = result.pixels_x
     clem_img_series.image_pixels_y = result.pixels_y
     clem_img_series.image_pixel_size = result.pixel_size
@@ -187,27 +213,6 @@ def _register_clem_image_series(
     logger.info(f"CLEM preprocessing results registered for {result.series_name!r} ")
 
 
-color_columns = {
-    "gray": "hasGrey",
-    "red": "hasRed",
-    "green": "hasGreen",
-    "blue": "hasBlue",
-    "cyan": "hasCyan",
-    "magenta": "hasMagenta",
-    "yellow": "hasYellow",
-}
-
-
-def _get_color_flags(
-    colors: Collection[str] | None = None,
-):
-    colors = colors or []
-    color_flags = dict.fromkeys(color_columns.values(), 0)
-    for color in colors:
-        color_flags[color_columns[color]] = 1
-    return color_flags
-
-
 def _determine_collection_mode(
     colors: Collection[str] | None = None,
 ):
@@ -223,6 +228,16 @@ def _determine_collection_mode(
         return "Fluorescent"
 
 
+def _snake_to_camel_case(string: str):
+    parts = string.split("_")
+    return parts[0] + "".join(part.capitalize() for part in parts[1:])
+
+
+COLOR_FLAGS_MURFEY_TO_ISPYB = {
+    value: _snake_to_camel_case(value) for value in COLOR_FLAGS_MURFEY.values()
+}
+
+
 def _register_dcg_and_atlas(
     session_id: int,
     instrument_name: str,
@@ -262,7 +277,11 @@ def _register_dcg_and_atlas(
         else:
             atlas_name = str(output_file.parent / "*.tiff")
             atlas_pixel_size = result.pixel_size
-        color_flags = _get_color_flags(result.output_files.keys())
+        # Translate colour flags into ISPyB convention
+        color_flags = {
+            COLOR_FLAGS_MURFEY_TO_ISPYB[key]: int(value)
+            for key, value in _get_color_flags(result.output_files.keys()).items()
+        }
         collection_mode = _determine_collection_mode(result.output_files.keys())
     else:
         atlas_name = ""
@@ -455,8 +474,13 @@ def _register_grid_square(
                 y_stage_position=0.5 * (clem_img_series.y0 + clem_img_series.y1),
                 pixel_size=clem_img_series.image_pixel_size,
                 image=clem_img_series.thumbnail_search_string,
-                collection_mode=_determine_collection_mode(result.output_files.keys()),
+                collection_mode=clem_img_series.collection_mode,
             )
+            # Construct colour flags for ISPyB
+            color_flags = {
+                ispyb_color_flags: int(getattr(clem_img_series, murfey_color_flags, 0))
+                for murfey_color_flags, ispyb_color_flags in COLOR_FLAGS_MURFEY_TO_ISPYB.items()
+            }
             # Register or update the grid square entry as required
             if grid_square_result := murfey_db.exec(
                 select(MurfeyDB.GridSquare)
@@ -481,7 +505,7 @@ def _register_grid_square(
                 _transport_object.do_update_grid_square(
                     grid_square_id=grid_square_entry.id,
                     grid_square_parameters=grid_square_params,
-                    color_flags=_get_color_flags(result.output_files.keys()),
+                    color_flags=color_flags,
                 )
             else:
                 # Look up data collection group for current series
@@ -495,7 +519,7 @@ def _register_grid_square(
                     atlas_id=dcg_entry.atlas_id,
                     grid_square_id=clem_img_series.id,
                     grid_square_parameters=grid_square_params,
-                    color_flags=_get_color_flags(result.output_files.keys()),
+                    color_flags=color_flags,
                 )
                 # Register to Murfey
                 grid_square_entry = MurfeyDB.GridSquare(

tests/workflows/clem/test_register_preprocessing_results.py

@@ -12,11 +12,13 @@
 
 import murfey.util.db as MurfeyDB
 from murfey.workflows.clem.register_preprocessing_results import (
+    COLOR_FLAGS_MURFEY_TO_ISPYB,
     _determine_collection_mode,
     _get_color_flags,
     _register_clem_image_series,
     _register_dcg_and_atlas,
     _register_grid_square,
+    _snake_to_camel_case,
     run,
 )
 from tests.conftest import ExampleVisit, get_or_create_db_entry
@@ -118,75 +120,68 @@ def generate_preprocessing_messages(
     return messages
 
 
-def test_register_clem_image_series():
-    _register_clem_image_series
-
-
 @pytest.mark.parametrize(
     "test_params",
     (
         (
             ["gray"],
             {
-                "hasGrey": 1,
+                "has_grey": True,
             },
         ),
         (
             ["gray", "red"],
             {
-                "hasGrey": 1,
-                "hasRed": 1,
+                "has_grey": True,
+                "has_red": True,
             },
         ),
         (
             ["red", "green", "blue"],
             {
-                "hasRed": 1,
-                "hasGreen": 1,
-                "hasBlue": 1,
+                "has_red": True,
+                "has_green": True,
+                "has_blue": True,
             },
         ),
         (
             ["cyan", "magenta", "yellow"],
             {
-                "hasCyan": 1,
-                "hasMagenta": 1,
-                "hasYellow": 1,
+                "has_cyan": True,
+                "has_magenta": True,
+                "has_yellow": True,
             },
         ),
     ),
 )
-def test_get_color_flags(test_params: tuple[list[str], dict[str, int]]):
-    # Unpack test params
+def test_get_color_flags(test_params: tuple[list[str], dict[str, bool]]):
     colors, positive_flags = test_params
     expected_result = dict.fromkeys(
         (
-            "hasGrey",
-            "hasRed",
-            "hasGreen",
-            "hasBlue",
-            "hasCyan",
-            "hasMagenta",
-            "hasYellow",
+            "has_grey",
+            "has_red",
+            "has_green",
+            "has_blue",
+            "has_cyan",
+            "has_magenta",
+            "has_yellow",
         ),
-        0,
+        False,
     )
     for flag, value in positive_flags.items():
         expected_result[flag] = value
     assert _get_color_flags(colors) == expected_result
 
 
+def test_register_clem_image_series():
+    _register_clem_image_series
+
+
 @pytest.mark.parametrize(
     "test_params",
     (
-        (
-            ["gray"],
-            "Bright Field",
-        ),
-        (
-            ["gray", "blue"],
-            "Bright Field and Fluorescent",
-        ),
+        (["gray"], "Bright Field"),
+        (["gray", "blue"], "Bright Field and Fluorescent"),
         (["red", "green", "blue"], "Fluorescent"),
     ),
 )
@@ -195,6 +190,25 @@ def test_determine_collection_mode(test_params: tuple[list[str], str]):
     assert _determine_collection_mode(colors) == expected_result
 
 
+@pytest.mark.parametrize(
+    "test_params",
+    (
+        ("has_grey", "hasGrey"),
+        ("has_red", "hasRed"),
+        ("has_green", "hasGreen"),
+        ("has_blue", "hasBlue"),
+        ("has_cyan", "hasCyan"),
+        ("has_magenta", "hasMagenta"),
+        ("has_yellow", "hasYellow"),
+    ),
+)
+def test_snake_to_camel_case(
+    test_params: tuple[str, str],
+):
+    string, expected_result = test_params
+    assert _snake_to_camel_case(string) == expected_result
+
+
 def test_register_dcg_and_atlas():
     _register_dcg_and_atlas
 
@@ -267,18 +281,8 @@ def test_run(
     "test_params",
     (
         # Reverse list order? | Colors
-        (
-            False,
-            [
-                "gray",
-            ],
-        ),
-        (
-            True,
-            [
-                "gray",
-            ],
-        ),
+        (False, ["gray"]),
+        (True, ["gray"]),
         (False, ["red", "green", "blue"]),
         (True, ["cyan", "magenta", "yellow"]),
         (False, ["gray", "red", "green", "blue"]),
@@ -417,7 +421,10 @@ def test_run_with_db(
     assert len(ispyb_atlas_search) == 1
 
     # Determine the color flags and collection mode
-    color_flags = _get_color_flags(colors)
+    color_flags = {
+        COLOR_FLAGS_MURFEY_TO_ISPYB[flag]: int(value)
+        for flag, value in _get_color_flags(colors).items()
+    }
     collection_mode = _determine_collection_mode(colors)
 
     ispyb_atlas = ispyb_atlas_search[0]