diff --git a/src/dlstbx/services/trigger.py b/src/dlstbx/services/trigger.py index 6aa4f5514..5f129be35 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,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( diff --git a/src/dlstbx/wrapper/xia2_multiplex_filtering.py b/src/dlstbx/wrapper/xia2_multiplex_filtering.py index 66f8b238b..558ccce35 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,9 @@ def construct_commandline(self, params, multiplex_directory): "data", "clustering.method", "clustering.output_clusters", + "spacegroup", + "anomalous", + "absorption_level", } translation = { "d_min": "resolution.d_min",