diff --git a/src/dlstbx/services/trigger.py b/src/dlstbx/services/trigger.py index 6aa4f5514..6788640fa 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,24 @@ 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: int + 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 + + class Xia2SsxReduceParameters(pydantic.BaseModel): dcid: int = pydantic.Field(gt=0) related_dcids: List[RelatedDCIDs] @@ -1718,6 +1738,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 +2210,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 +2255,167 @@ 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 + + CURRENT CLUSTERING BEHAVIOUR: clustering will be output if the original parent job had clustering. + Will think how to expand this in the future for more tailorability. + + 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" + - 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 + + Example recipe parameters: + { + "target": "multiplex_reprocessing", + "dcid": 1234, + "rpid": 4006800, + "wavelength": "1.03936", + "comment": "xia2.multiplex triggered manually", + "automatic": false, + "ispyb_parameters": {"spacegroup": "P1", "apply_cchalf_filtering":"1", "cchalf_filtering_method": "image_group"}, + } + """ + + parameters.recipe = "postprocessing-xia2-multiplex" + + # First, get the job parameters from the existing multiplex job + + job_parameters: list[tuple[str, str]] = ( + session.query( + ProcessingJobParameter.parameterKey, + ProcessingJobParameter.parameterValue, + ) + .select_from(AutoProcProgram) + .join( + ProcessingJobParameter, + ProcessingJobParameter.processingJobId + == AutoProcProgram.processingJobId, + ) + .join( + AutoProc, + AutoProc.autoProcProgramId == AutoProcProgram.autoProcProgramId, ) + .join(AutoProcScaling, AutoProcScaling.autoProcId == AutoProc.autoProcId) + .where(AutoProcScaling.autoProcScalingId == parameters.scaling_id) + ).all() - return {"success": True, "return_value": jobids} + # Second, get the DCIDs used in the existing multiplex job + + parent_job_dcids: list[tuple[int]] = ( + session.query(ProcessingJobImageSweep.dataCollectionId) + .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) + ).all() + + dcids: list[int] = [row[0] for row in parent_job_dcids] + + self.log.info(f"Found DCIDS: {dcids}") + self.log.info("Found parent job parameters:") + for param in job_parameters: + self.log.info(param) + + extra_params = ( + "spacegroup", + "d_min", + "apply_cchalf_filtering", + "cchalf_filtering_method", + "image_group_size", + "sd_cutoff", + ) + + for i in parameters: + if i[0] in extra_params and i[1]: + if i[0] == "apply_cchalf_filtering": + job_parameters.append((i[0], "deltacchalf")) + else: + job_parameters.append((i[0], str(i[1]))) + elif ( + i[0] == "diffraction_plan_info" + and parameters.diffraction_plan_info + and parameters.diffraction_plan_info.anomalousScatterer + ): + job_parameters.extend( + [("anomalous", "true"), ("absorption_level", "high")] + ) + + # 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( diff --git a/src/dlstbx/wrapper/xia2_multiplex.py b/src/dlstbx/wrapper/xia2_multiplex.py index 6c65a3ee5..36a446563 100644 --- a/src/dlstbx/wrapper/xia2_multiplex.py +++ b/src/dlstbx/wrapper/xia2_multiplex.py @@ -167,10 +167,10 @@ def construct_commandline(self, params): ignore = { "sample_id", "sample_group_id", - "filtering.method", - "deltacchalf.stdcutoff", - "deltacchalf.mode", - "deltacchalf.group_size", + "apply_cchalf_filtering", + "cchalf_filtering_method", + "image_group_size", + "sd_cutoff", } translation = { "d_min": "resolution.d_min", diff --git a/src/dlstbx/wrapper/xia2_multiplex_filtering.py b/src/dlstbx/wrapper/xia2_multiplex_filtering.py index 66f8b238b..78d026154 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,9 +138,16 @@ def construct_commandline(self, params, multiplex_directory): "data", "clustering.method", "clustering.output_clusters", + "spacegroup", + "anomalous", + "absorption_level", } translation = { "d_min": "resolution.d_min", + "apply_cchalf_filtering": "filtering.method", + "cchalf_filtering_method": "deltacchalf.mode", + "image_group_size": "deltacchalf.group_size", + "sd_cutoff": "deltacchalf.stdcutoff", } for param, value in params["ispyb_parameters"].items(): if param not in ignore: