Multiplex filtering

setup.py

@@ -93,6 +93,7 @@
     "xchem_collate = dlstbx.wrapper.xchem_collate:XChemCollateWrapper",
     "xia2 = dlstbx.wrapper.xia2:Xia2Wrapper",
     "xia2.multiplex = dlstbx.wrapper.xia2_multiplex:Xia2MultiplexWrapper",
+    "xia2.multiplex_filtering = dlstbx.wrapper.xia2_multiplex_filtering:Xia2MultiplexFilteringWrapper",
     "xia2.overload = dlstbx.wrapper.xia2_overload:Xia2OverloadWrapper",
     "xia2.strategy = dlstbx.wrapper.xia2_strategy:Xia2StrategyWrapper",
     "xia2.to_shelxcde = dlstbx.wrapper.xia2_to_shelxcde:Xia2toShelxcdeWrapper",

src/dlstbx/services/trigger.py

@@ -229,6 +229,10 @@ class MultiplexParameters(pydantic.BaseModel):
     diffraction_plan_info: Optional[DiffractionPlanInfo] = None
     recipe: Optional[str] = None
     use_clustering: Optional[List[str]] = None
+    use_filtering: Optional[List[str]] = None
+    filtering_group_size: Dict[str, int] = pydantic.Field(
+        default={"default": 50}, alias="filtering-group-size"
+    )
     beamline: str
     trigger_every_collection: bool
 
@@ -2168,6 +2172,7 @@ def trigger_multiplex(
                 job_parameters.append(("sample_id", str(group.sample_id)))
             else:
                 job_parameters.append(("sample_group_id", str(group.sample_group_id)))
+
             if parameters.spacegroup:
                 job_parameters.append(("spacegroup", parameters.spacegroup))
             if (
@@ -2187,6 +2192,27 @@ def trigger_multiplex(
                         ("clustering.output_clusters", "true"),
                     ]
                 )
+
+            # See if beamline is in list of allowed ones for filtering
+            # If so, add filtering parameters to job_parameters
+            # This will set the xia2.multiplex wrapper to send the job for filtering after completed
+
+            if (
+                parameters.use_filtering
+                and parameters.beamline in parameters.use_filtering
+            ):
+                group_size = parameters.filtering_group_size.get(
+                    parameters.beamline, parameters.filtering_group_size["default"]
+                )
+
+                job_parameters.extend(
+                    [
+                        ("filtering.method", "deltacchalf"),
+                        ("deltacchalf.stdcutoff", "3"),
+                        ("deltacchalf.mode", "image_group"),
+                        ("deltacchalf.group_size", str(group_size)),
+                    ]
+                )
             for k, v in job_parameters:
                 jpp = self.ispyb.mx_processing.get_job_parameter_params()
                 jpp["job_id"] = jobid
@@ -2202,7 +2228,9 @@ def trigger_multiplex(
             message = {"recipes": [], "parameters": {"ispyb_process": jobid}}
             rw.transport.send("processing_recipe", message)
 
-            self.log.info(f"xia2.multiplex trigger: Processing job {jobid} triggered")
+            self.log.info(
+                f"xia2.multiplex_filtering trigger: Processing job {jobid} triggered"
+            )
 
         return {"success": True, "return_value": jobids}
 

src/dlstbx/wrapper/xia2_multiplex.py

@@ -27,7 +27,12 @@ class Xia2MultiplexWrapper(Wrapper):
     name = "xia2.multiplex"
 
     def send_results_to_ispyb(
-        self, z, xtriage_results=None, cluster_num=None, attachments=[]
+        self,
+        z,
+        xtriage_results=None,
+        cluster_num=None,
+        attachments=[],
+        multiplex_filtering=False,
     ):
         ispyb_command_list = []
 
@@ -138,6 +143,14 @@ def send_results_to_ispyb(
         self.log.debug("Sending %s", str(ispyb_command_list))
         self.recwrap.send_to("ispyb", {"ispyb_command_list": ispyb_command_list})
 
+        # After xia2.multiplex, xia2.multiplex_filtering can be optionally run to improve data reduction
+        # Currently not supported for clusters
+        # Only triggered if filtering parameters exist -> logic handled in trigger_multiplex
+
+        if not cluster_num and multiplex_filtering:
+            self.log.info("Triggering xia2.multiplex filtering.")
+            self.recwrap.send_to("filtering", True)
+
     def construct_commandline(self, params):
         """Construct xia2.multiplex command line.
         Takes job parameter dictionary, returns array."""
@@ -150,7 +163,15 @@ def construct_commandline(self, params):
             command.append(f)
 
         if params.get("ispyb_parameters"):
-            ignore = {"sample_id", "sample_group_id"}
+            # ignore filtering parameters for xia2.multiplex_filtering
+            ignore = {
+                "sample_id",
+                "sample_group_id",
+                "filtering.method",
+                "deltacchalf.stdcutoff",
+                "deltacchalf.mode",
+                "deltacchalf.group_size",
+            }
             translation = {
                 "d_min": "resolution.d_min",
                 "spacegroup": "symmetry.space_group",
@@ -425,11 +446,27 @@ def is_final_result(final_file: pathlib.Path) -> bool:
                 self.log.info(
                     f"Triggering downstream recipe steps for dataset: '{dataset_name}'"
                 )
+
+                # Check if filtering parameters present -> if so, trigger filtering when sending results to ispyb
+                # Whether or not these are present is currently handled by the trigger service
+
+                filtering = False
+
+                if params.get("ispyb_parameters"):
+                    if ("filtering.method", ["deltacchalf"]) in params[
+                        "ispyb_parameters"
+                    ].items():
+                        self.log.info(
+                            "Additional filtering with xia2.multiplex_filtering selected."
+                        )
+                        filtering = True
+
                 self.send_results_to_ispyb(
                     ispyb_d,
                     xtriage_results=xtriage_results,
                     cluster_num=cluster_num,
                     attachments=attachments,
+                    multiplex_filtering=filtering,
                 )
 
             self._success_counter.inc()