Initial implementation of reprocessing xia2.multiplex jobs via SynchWeb

src/dlstbx/services/trigger.py

@@ -25,6 +25,8 @@
     BLSession,
     DataCollection,
     ProcessingJob,
+    ProcessingJobImageSweep,
+    ProcessingJobParameter,
     Proposal,
     Protein,
 )
@@ -237,6 +239,26 @@ class MultiplexParameters(pydantic.BaseModel):
     trigger_every_collection: bool
 
 
+class MultiplexReprocessingParameters(pydantic.BaseModel):
+    rpid: Optional[int] = None
+    d_min: Optional[float] = None
+    apply_cchalf_filtering: Optional[int] = None
+    cchalf_filtering_method: Optional[str] = None
+    sd_cutoff: Optional[float] = None
+    image_group_size: Optional[int] = None
+    scaling_id: Optional[int] = None
+    dcid: int = pydantic.Field(gt=0)
+    wavelength: Optional[float] = pydantic.Field(default=None, gt=0)
+    spacegroup: Optional[str] = None
+    automatic: Optional[bool] = False
+    comment: Optional[str] = None
+    wavelength_tolerance: float = pydantic.Field(default=1e-4, ge=0)
+    diffraction_plan_info: Optional[DiffractionPlanInfo] = None
+    recipe: Optional[str] = None
+    use_clustering: Optional[List[str]] = None
+    beamline: str
+
+
 class Xia2SsxReduceParameters(pydantic.BaseModel):
     dcid: int = pydantic.Field(gt=0)
     related_dcids: List[RelatedDCIDs]
@@ -1718,6 +1740,69 @@ class BigEPParams(pydantic.BaseModel):
 
         return {"success": True, "return_value": None}
 
+    def register_multiplex_job(
+        self,
+        parameters,
+        dcids,
+        session,
+        job_parameters,
+        rw,
+    ):
+        jp = self.ispyb.mx_processing.get_job_params()
+        jp["automatic"] = parameters.automatic
+        jp["comments"] = parameters.comment
+        jp["datacollectionid"] = parameters.dcid
+        jp["display_name"] = "xia2.multiplex"
+        jp["recipe"] = parameters.recipe
+        self.log.info(jp)
+        jobid = self.ispyb.mx_processing.upsert_job(list(jp.values()))
+        self.log.debug(f"xia2.multiplex trigger: generated JobID {jobid}")
+
+        query = (
+            session.query(DataCollection)
+            .filter(DataCollection.dataCollectionId.in_(dcids))
+            .options(
+                Load(DataCollection).load_only(
+                    DataCollection.dataCollectionId,
+                    DataCollection.wavelength,
+                    DataCollection.startImageNumber,
+                    DataCollection.numberOfImages,
+                    raiseload=True,
+                )
+            )
+        )
+
+        for dc in query.all():
+            jisp = self.ispyb.mx_processing.get_job_image_sweep_params()
+            jisp["datacollectionid"] = dc.dataCollectionId
+            jisp["start_image"] = dc.startImageNumber
+            jisp["end_image"] = dc.startImageNumber + dc.numberOfImages - 1
+
+            jisp["job_id"] = jobid
+            jispid = self.ispyb.mx_processing.upsert_job_image_sweep(
+                list(jisp.values())
+            )
+            self.log.debug(
+                f"xia2.multiplex trigger: generated JobImageSweepID {jispid}"
+            )
+
+        for k, v in job_parameters:
+            jpp = self.ispyb.mx_processing.get_job_parameter_params()
+            jpp["job_id"] = jobid
+            jpp["parameter_key"] = k
+            jpp["parameter_value"] = v
+            jppid = self.ispyb.mx_processing.upsert_job_parameter(list(jpp.values()))
+            self.log.debug(
+                f"xia2.multiplex trigger generated JobParameterID {jppid} with {k}={v}",
+            )
+
+        message = {"recipes": [], "parameters": {"ispyb_process": jobid}}
+        rw.transport.send("processing_recipe", message)
+
+        self.log.info(f"xia2.multiplex trigger: Processing job {jobid} triggered")
+
+        return jobid
+
     @pydantic.validate_call(config={"arbitrary_types_allowed": True})
     def trigger_multiplex(
         self,
@@ -2127,44 +2212,6 @@ def trigger_multiplex(
                 continue
             multiplex_job_dcids.append(set_dcids)
 
-            jp = self.ispyb.mx_processing.get_job_params()
-            jp["automatic"] = parameters.automatic
-            jp["comments"] = parameters.comment
-            jp["datacollectionid"] = dcid
-            jp["display_name"] = "xia2.multiplex"
-            jp["recipe"] = parameters.recipe
-            self.log.info(jp)
-            jobid = self.ispyb.mx_processing.upsert_job(list(jp.values()))
-            jobids.append(jobid)
-            self.log.debug(f"xia2.multiplex trigger: generated JobID {jobid}")
-
-            query = (
-                session.query(DataCollection)
-                .filter(DataCollection.dataCollectionId.in_(dcids))
-                .options(
-                    Load(DataCollection).load_only(
-                        DataCollection.dataCollectionId,
-                        DataCollection.wavelength,
-                        DataCollection.startImageNumber,
-                        DataCollection.numberOfImages,
-                        raiseload=True,
-                    )
-                )
-            )
-            for dc in query.all():
-                jisp = self.ispyb.mx_processing.get_job_image_sweep_params()
-                jisp["datacollectionid"] = dc.dataCollectionId
-                jisp["start_image"] = dc.startImageNumber
-                jisp["end_image"] = dc.startImageNumber + dc.numberOfImages - 1
-
-                jisp["job_id"] = jobid
-                jispid = self.ispyb.mx_processing.upsert_job_image_sweep(
-                    list(jisp.values())
-                )
-                self.log.debug(
-                    f"xia2.multiplex trigger: generated JobImageSweepID {jispid}"
-                )
-
             job_parameters: list[tuple[str, str]] = [
                 ("data", ";".join(files)) for files in data_files
             ]
@@ -2210,26 +2257,204 @@ def trigger_multiplex(
                         ("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
-                jpp["parameter_key"] = k
-                jpp["parameter_value"] = v
-                jppid = self.ispyb.mx_processing.upsert_job_parameter(
-                    list(jpp.values())
-                )
-                self.log.debug(
-                    f"xia2.multiplex trigger generated JobParameterID {jppid} with {k}={v}",
+
+            jobids.append(
+                self.register_multiplex_job(
+                    parameters,
+                    dcids,
+                    session,
+                    job_parameters,
+                    rw,
                 )
+            )
 
-            message = {"recipes": [], "parameters": {"ispyb_process": jobid}}
-            rw.transport.send("processing_recipe", message)
+        return {"success": True, "return_value": jobids}
 
-            self.log.info(
-                f"xia2.multiplex trigger: Processing job {jobid} triggered"
+    @pydantic.validate_call(config={"arbitrary_types_allowed": True})
+    def trigger_multiplex_reprocessing(
+        self,
+        rw: RecipeWrapper,
+        message: Dict,
+        parameters: MultiplexReprocessingParameters,
+        session: sqlalchemy.orm.session.Session,
+        transaction: int,
+        **kwargs,
+    ):
+        """
+        Trigger reprocessing of an existing xia2.multiplex job for a given data collection and sample group.
+        Intended use case: reprocess an existing job with different (manually specified) phil parameters.
+        - add/change deltacchalf filtering
+        - specify resolution limit
+        - change space group
+
+        As the intended use case is to "edit" an existing job, much of the logic of trigger_multiplex is not required.
+        Trigger called via registering a new processingJobId with associated processigJobParameterIds.
+        One of these is the autoProcScalingId of the original multiplex job.
+        From the autoProcScalingId, query the database for the corresponding DCIDs and sample_id (or sample_group_id).
+        Use this to know exactly which datasets were selected out of the sample group, and which sample group it corresponds to.
+        This allows for a new job to be registered with the new processing parameters.
+        The data files input to xia2.multiplex via commandline are also found in this same query.
+        Currently not supporting any use cases where further xia2-dials jobs are added in order to maintain
+        consistency within SynchWeb (ie if the parent and reprocessed multiplex jobs have different numbers of datasets
+        then comparison tables and plots are not valid).
+
+        Recipe parameters:
+        - rpid: processingJobId input to trigger recipe (links to user defined parameters in SynchWeb)
+        - d_min: custom resolution cutoff
+        - apply_cchalf_filtering: perform deltacchalf filtering (xia2.multiplex_filtering)
+        - cchalf_filtering_method: filter by image_group or datasets
+        - sd_cutoff: STD cutoff for deltacchalf filtering
+        - image_group_size: image group size for optional deltacchalf filtering (only when cchalf_filtering_method=image_group)
+        - scaling_id: autoProcScalingId of the original multiplex job
+        - target: set this to "multiplex_reprocessing"
+        - beamline: the beamline as a string
+        - dcid: the dataCollectionId for the given data collection
+        - wavelength: wavelength of data collection
+        - spacegroup: space group for processing
+        - wavelength_tolerance: tolerance for selecting datasets with same wavelength
+        - comment: a comment to be stored in the ProcessingJob.comment field
+        - automatic: boolean value passed to ProcessingJob.automatic field
+        - ispyb_parameters: a dictionary of ispyb_reprocessing_parameters set by user interaction in SynchWeb
+        - recipe: this will be set in this function to the name of the recipe this function triggers
+        - use_clustering: list of beamlines which allow output of clusters
+
+        Example recipe parameters:
+        {
+            "target": "multiplex_reprocessing",
+            "dcid": 1234,
+            "rpid": 4006800,
+            "wavelength": "1.03936",
+            "comment": "xia2.multiplex triggered manually",
+            "automatic": false,
+            "use_clustering": ["i02-2", "i02-1", "i24"],
+            "beamline": "i02-1",
+            "ispyb_parameters": {"spacegroup": "P1", "apply_cchalf_filtering":"1", "cchalf_filtering_method": "image_group"},
+        }
+        """
+
+        parameters.recipe = "postprocessing-xia2-multiplex"
+
+        if (
+            parameters.use_clustering
+            and parameters.beamline in parameters.use_clustering
+        ):
+            output_clusters = True
+        else:
+            output_clusters = False
+
+        # First, use the input scaling ID to find the DCIDs of all multiplex jobs as well as the sample (group) id value
+
+        query = (
+            session.query(ProcessingJobImageSweep, ProcessingJobParameter)
+            .join(
+                AutoProcProgram,
+                AutoProcProgram.processingJobId
+                == ProcessingJobImageSweep.processingJobId,
+            )
+            .join(
+                ProcessingJobParameter,
+                ProcessingJobParameter.processingJobId
+                == AutoProcProgram.processingJobId,
+            )
+            .join(
+                AutoProc,
+                AutoProc.autoProcProgramId == AutoProcProgram.autoProcProgramId,
+            )
+            .join(AutoProcScaling, AutoProcScaling.autoProcId == AutoProc.autoProcId)
+            .where(AutoProcScaling.autoProcScalingId == parameters.scaling_id)
+            .options(
+                Load(ProcessingJobImageSweep).load_only(
+                    ProcessingJobImageSweep.dataCollectionId,
+                    raiseload=True,
+                ),
+                Load(ProcessingJobParameter).load_only(
+                    ProcessingJobParameter.parameterKey,
+                    ProcessingJobParameter.parameterValue,
+                    raiseload=True,
+                ),
             )
+        )
 
-        return {"success": True, "return_value": jobids}
+        found_dcids = set()
+        group = None
+        found_data_files = set()
+
+        for dc, dc_params in query.all():
+            found_dcids.add(dc.dataCollectionId)
+            if "sample_id" == dc_params.parameterKey:
+                group = ("sample_id", dc_params.parameterValue)
+            elif "sample_group_id" == dc_params.parameterKey:
+                group = ("sample_group_id", dc_params.parameterValue)
+            elif "data" == dc_params.parameterKey:
+                found_data_files.add(dc_params.parameterValue)
+
+        # Set parameters
+
+        dcids = list(found_dcids)
+        data_files = list(found_data_files)
+
+        self.log.info(f"Found {dcids} corresponding to {group}")
+
+        self.log.debug(f"Found data files {data_files}")
+
+        job_parameters: list[tuple[str, str]] = [
+            ("data", files) for files in data_files
+        ]
+        if group:
+            job_parameters.append(group)
+
+        if parameters.spacegroup:
+            job_parameters.append(("spacegroup", parameters.spacegroup))
+
+        if parameters.d_min:
+            job_parameters.append(("d_min", str(parameters.d_min)))
+
+        if (
+            parameters.diffraction_plan_info
+            and parameters.diffraction_plan_info.anomalousScatterer
+        ):
+            job_parameters.extend(
+                [
+                    ("anomalous", "true"),
+                    ("absorption_level", "high"),
+                ]
+            )
+        if output_clusters:
+            job_parameters.extend(
+                [
+                    ("clustering.method", "coordinate"),
+                    ("clustering.output_clusters", "true"),
+                ]
+            )
+
+        # Unlike regular multiplex, filtering allowed regardless of beamline
+
+        if parameters.apply_cchalf_filtering:
+            job_parameters.append(("filtering.method", "deltacchalf"))
+            if parameters.cchalf_filtering_method:
+                job_parameters.append(
+                    ("deltacchalf.mode", parameters.cchalf_filtering_method)
+                )
+            if parameters.image_group_size:
+                job_parameters.append(
+                    ("deltacchalf.group_size", str(parameters.image_group_size))
+                )
+            if parameters.sd_cutoff:
+                job_parameters.append(
+                    ("deltacchalf.stdcutoff", str(parameters.sd_cutoff))
+                )
+
+        # Register and trigger jobs (same as automatic multiplex)
+
+        jobid = self.register_multiplex_job(
+            parameters,
+            dcids,
+            session,
+            job_parameters,
+            rw,
+        )
+
+        return {"success": True, "return_value": jobid}
 
     @pydantic.validate_call(config={"arbitrary_types_allowed": True})
     def trigger_xia2_ssx_reduce(

src/dlstbx/wrapper/xia2_multiplex_filtering.py

@@ -129,6 +129,7 @@ def construct_commandline(self, params, multiplex_directory):
         # xia2.multiplex_filtering uses previous multiplex job files
         #   so no specific data files to be input, just add ispyb_parameters to cmdline
         #   ignore sample id params and also clustering params from parent multiplex
+        #   ignore spacegroup from parent multiplex as well -> filtering pipeline does not accept
 
         if params.get("ispyb_parameters"):
             ignore = {
@@ -137,6 +138,9 @@ def construct_commandline(self, params, multiplex_directory):
                 "data",
                 "clustering.method",
                 "clustering.output_clusters",
+                "spacegroup",
+                "anomalous",
+                "absorption_level",
             }
             translation = {
                 "d_min": "resolution.d_min",