From 25e981d4badd5199cacc7aded85cccfb332a4f65 Mon Sep 17 00:00:00 2001 From: amyjaynethompson <52806925+amyjaynethompson@users.noreply.github.com> Date: Mon, 22 Jun 2026 16:01:45 +0100 Subject: [PATCH 1/4] inital test of separate multiplex reprocessing trigger function --- src/dlstbx/services/trigger.py | 196 +++++++++++++++++- .../wrapper/xia2_multiplex_filtering.py | 2 + 2 files changed, 196 insertions(+), 2 deletions(-) diff --git a/src/dlstbx/services/trigger.py b/src/dlstbx/services/trigger.py index 6aa4f5514..c907efe96 100644 --- a/src/dlstbx/services/trigger.py +++ b/src/dlstbx/services/trigger.py @@ -25,6 +25,8 @@ BLSession, DataCollection, ProcessingJob, + ProcessingJobImageSweep, + ProcessingJobParameter, Proposal, Protein, ) @@ -237,6 +239,16 @@ class MultiplexParameters(pydantic.BaseModel): trigger_every_collection: bool +class MultiplexReprocessingParameters(MultiplexParameters): + 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 + + class Xia2SsxReduceParameters(pydantic.BaseModel): dcid: int = pydantic.Field(gt=0) related_dcids: List[RelatedDCIDs] @@ -2225,9 +2237,189 @@ 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 trigger: Processing job {jobid} triggered") + + return {"success": True, "return_value": jobids} + + @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, + ): + """ + Testing a separate trigger function for multiplex reprocessing. + Because it is a retriggered job, DO NOT need to see if preceeding jobs have finished. + CHECK IF THIS IS CORRECT - CHECK SKIPPING LOGIC!!!! + ALSO CHECK IF NEED TO MATCH PROCESSING ID OR IF THIS MAKES NEW ONE AND THAT'S CORRECT BEHAVIOUR + Also, we can link back the DCIDs from the scaling ID. + """ + + parameters.recipe = "postprocessing-xia2-multiplex" + + if ( + parameters.use_clustering + and parameters.beamline in parameters.use_clustering + ): + output_clusters = True + + # 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, + ), ) + ) + + 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) + + 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}") + + jobids = [] + 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())) + 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", 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)) + ) + + 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 {"success": True, "return_value": jobids} diff --git a/src/dlstbx/wrapper/xia2_multiplex_filtering.py b/src/dlstbx/wrapper/xia2_multiplex_filtering.py index 66f8b238b..d26b6ea8a 100644 --- a/src/dlstbx/wrapper/xia2_multiplex_filtering.py +++ b/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,7 @@ def construct_commandline(self, params, multiplex_directory): "data", "clustering.method", "clustering.output_clusters", + "spacegroup", } translation = { "d_min": "resolution.d_min", From 317823d54e735b99d420cb04fefbecaefae1eddd Mon Sep 17 00:00:00 2001 From: amyjaynethompson <52806925+amyjaynethompson@users.noreply.github.com> Date: Mon, 6 Jul 2026 13:42:27 +0100 Subject: [PATCH 2/4] ignore anomalous flags in multiplex filtering as will be imported from parent job --- src/dlstbx/wrapper/xia2_multiplex_filtering.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/dlstbx/wrapper/xia2_multiplex_filtering.py b/src/dlstbx/wrapper/xia2_multiplex_filtering.py index d26b6ea8a..558ccce35 100644 --- a/src/dlstbx/wrapper/xia2_multiplex_filtering.py +++ b/src/dlstbx/wrapper/xia2_multiplex_filtering.py @@ -139,6 +139,8 @@ def construct_commandline(self, params, multiplex_directory): "clustering.method", "clustering.output_clusters", "spacegroup", + "anomalous", + "absorption_level", } translation = { "d_min": "resolution.d_min", From 45424b45f469f8d3ffb1662e07b2a6b8e813769d Mon Sep 17 00:00:00 2001 From: amyjaynethompson <52806925+amyjaynethompson@users.noreply.github.com> Date: Mon, 6 Jul 2026 15:18:56 +0100 Subject: [PATCH 3/4] separate multiplex reprocessing trigger to escape unnecessary logic --- src/dlstbx/services/trigger.py | 257 +++++++++++++++++++-------------- 1 file changed, 146 insertions(+), 111 deletions(-) diff --git a/src/dlstbx/services/trigger.py b/src/dlstbx/services/trigger.py index c907efe96..d4c1141bb 100644 --- a/src/dlstbx/services/trigger.py +++ b/src/dlstbx/services/trigger.py @@ -239,7 +239,7 @@ class MultiplexParameters(pydantic.BaseModel): trigger_every_collection: bool -class MultiplexReprocessingParameters(MultiplexParameters): +class MultiplexReprocessingParameters(pydantic.BaseModel): rpid: Optional[int] = None d_min: Optional[float] = None apply_cchalf_filtering: Optional[int] = None @@ -247,6 +247,16 @@ class MultiplexReprocessingParameters(MultiplexParameters): 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): @@ -1730,6 +1740,71 @@ class BigEPParams(pydantic.BaseModel): return {"success": True, "return_value": None} + def register_multiplex_job( + self, + parameters, + dcids, + session, + job_parameters, + rw, + ): + jobids = [] + 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())) + 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}" + ) + + 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 jobids + @pydantic.validate_call(config={"arbitrary_types_allowed": True}) def trigger_multiplex( self, @@ -2139,44 +2214,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 ] @@ -2222,22 +2259,16 @@ 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}", - ) - - message = {"recipes": [], "parameters": {"ispyb_process": jobid}} - rw.transport.send("processing_recipe", message) - self.log.info(f"xia2.multiplex trigger: Processing job {jobid} triggered") + jobids.append( + self.register_multiplex_job( + parameters, + dcids, + session, + job_parameters, + rw, + ) + ) return {"success": True, "return_value": jobids} @@ -2252,11 +2283,55 @@ def trigger_multiplex_reprocessing( **kwargs, ): """ - Testing a separate trigger function for multiplex reprocessing. - Because it is a retriggered job, DO NOT need to see if preceeding jobs have finished. - CHECK IF THIS IS CORRECT - CHECK SKIPPING LOGIC!!!! - ALSO CHECK IF NEED TO MATCH PROCESSING ID OR IF THIS MAKES NEW ONE AND THAT'S CORRECT BEHAVIOUR - Also, we can link back the DCIDs from the scaling ID. + 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" @@ -2266,6 +2341,8 @@ def trigger_multiplex_reprocessing( 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 @@ -2313,6 +2390,8 @@ def trigger_multiplex_reprocessing( elif "data" == dc_params.parameterKey: found_data_files.add(dc_params.parameterValue) + # Set parameters + dcids = list(found_dcids) data_files = list(found_data_files) @@ -2320,45 +2399,6 @@ def trigger_multiplex_reprocessing( self.log.debug(f"Found data files {data_files}") - jobids = [] - 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())) - 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", files) for files in data_files ] @@ -2406,20 +2446,15 @@ def trigger_multiplex_reprocessing( ("deltacchalf.stdcutoff", str(parameters.sd_cutoff)) ) - 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}", - ) + # Register and trigger jobs (same as automatic multiplex) - message = {"recipes": [], "parameters": {"ispyb_process": jobid}} - rw.transport.send("processing_recipe", message) - - self.log.info(f"xia2.multiplex trigger: Processing job {jobid} triggered") + jobids = self.register_multiplex_job( + parameters, + dcids, + session, + job_parameters, + rw, + ) return {"success": True, "return_value": jobids} From 3ec92e8c8b685d8ada9a526e51e22b89a615de26 Mon Sep 17 00:00:00 2001 From: amyjaynethompson <52806925+amyjaynethompson@users.noreply.github.com> Date: Mon, 6 Jul 2026 15:48:29 +0100 Subject: [PATCH 4/4] fix jobid return --- src/dlstbx/services/trigger.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/dlstbx/services/trigger.py b/src/dlstbx/services/trigger.py index d4c1141bb..5f129be35 100644 --- a/src/dlstbx/services/trigger.py +++ b/src/dlstbx/services/trigger.py @@ -1748,7 +1748,6 @@ def register_multiplex_job( job_parameters, rw, ): - jobids = [] jp = self.ispyb.mx_processing.get_job_params() jp["automatic"] = parameters.automatic jp["comments"] = parameters.comment @@ -1757,7 +1756,6 @@ def register_multiplex_job( 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 = ( @@ -1803,7 +1801,7 @@ def register_multiplex_job( self.log.info(f"xia2.multiplex trigger: Processing job {jobid} triggered") - return jobids + return jobid @pydantic.validate_call(config={"arbitrary_types_allowed": True}) def trigger_multiplex( @@ -2448,7 +2446,7 @@ def trigger_multiplex_reprocessing( # Register and trigger jobs (same as automatic multiplex) - jobids = self.register_multiplex_job( + jobid = self.register_multiplex_job( parameters, dcids, session, @@ -2456,7 +2454,7 @@ def trigger_multiplex_reprocessing( rw, ) - return {"success": True, "return_value": jobids} + return {"success": True, "return_value": jobid} @pydantic.validate_call(config={"arbitrary_types_allowed": True}) def trigger_xia2_ssx_reduce(