Add 'ispyb_detectorname' for use in fast feedback service (#328)

Author: jbeilstenedmands

src/dlstbx/ispybtbx/__init__.py

@@ -194,6 +194,22 @@ def dc_info_to_detectorclass(
         elif template.endswith(".cbf"):
             return "pilatus"
 
+    def dc_info_to_detectorname(
+        self, dc_info, session: sqlalchemy.orm.session.Session
+    ):
+        ## Get a detector name if it is one of a set of allowed values for fast feedback service.
+        det_id = dc_info.get("detectorId")
+        if det_id is not None and (det := crud.get_detector(det_id, session)):
+            if det.detectorModel == "Eiger2 XE 16M":
+                return "Eiger16M"
+            elif det.detectorModel == "Eiger2 X 4M":
+                return "Eiger4M"
+            elif det.detectorModel == "Eiger2 X CdTe 9M":
+                return "Eiger9MCdTe"
+
+        # Not one of the set of detectors allowed for fast feedback service.
+        return None
+
     def get_sample_group_dcids(
         self,
         ispyb_info,
@@ -847,6 +863,7 @@ def ispyb_filter(
     parameters["ispyb_beamline"] = i.get_beamline_from_dcid(dc_id, session)
 
     parameters["ispyb_detectorclass"] = i.dc_info_to_detectorclass(dc_info, session)
+    parameters["ispyb_detectorname"] = i.dc_info_to_detectorname(dc_info, session)
     parameters["ispyb_dc_info"] = dc_info
     parameters["ispyb_dc_info"]["gridinfo"] = i.get_gridscan_info(
         dc_info.get("dataCollectionId"), dc_info.get("dataCollectionGroupId"), session

tests/ispybtbx/test_ispyb.py

@@ -362,6 +362,16 @@ def test_filter_function(db_session):
     param = {"ispyb_dcid": ds["i19_screening"]}
     msg, param = ispyb_filter(msg, param, db_session)
 
+def test_get_detectorname(db_session):
+    test_data = {
+        "i03" : {"dcid" : 19714078, "expected" : "Eiger16M"},
+        "i24" : {"dcid" : 16030696, "expected" : "Eiger9MCdTe"}
+    }
+    for d in test_data.values():
+        message = {}
+        parameters = {"ispyb_dcid": d["dcid"]}
+        message, parameters = ispyb_filter(message, parameters, db_session)
+        assert parameters["ispyb_detectorname"] == d["expected"]
 
 def test_filter_function_with_load_config_file_timeout(monkeypatch, db_session):
     def mock_load_config_file(*args, **kwargs):