Making Estimate_Transmission Wrapper a standalone pipeline

src/dlstbx/mimas/core.py

@@ -189,6 +189,13 @@ def handle_characterization(
             source="automatic",
             displayname="align_crystal",
         ),
+        mimas.MimasISPyBJobInvocation(
+            DCID=scenario.DCID,
+            autostart=True,
+            recipe="strategy-estimate-transmission",
+            source="automatic",
+            displayname="estimate_transmission",
+        ),
         mimas.MimasRecipeInvocation(
             DCID=scenario.DCID, recipe="generate-diffraction-preview"
         ),

src/dlstbx/services/strategy.py

@@ -200,6 +200,13 @@ def generate_strategy(
         txn = self._transport.transaction_begin(subscription_id=header["subscription"])
         self._transport.ack(header, transaction=txn)
 
+        transmission_estimate = parameters.get("transmission_estimate", 100)
+        transmission_estimate = float(
+            transmission_estimate[0]
+            if isinstance(transmission_estimate, list)
+            else transmission_estimate
+        )
+
         beamline = (
             parameters["beamline"][0]
             if isinstance(parameters["beamline"], list)
@@ -215,6 +222,7 @@ def generate_strategy(
             if isinstance(parameters["resolution"], list)
             else float(parameters["resolution"])
         )
+        recommended_max_transmission = transmission_estimate / 100
         dc_transmission = float(parameters.get("transmission", 100)) / 100
         resolution_offset = 0.5
         min_resolution = 0.9
@@ -232,7 +240,6 @@ def generate_strategy(
             )
         beamline_config = parse_config_file(beamline_config_file)
 
-        recommended_max_transmission = parameters.get("scaled_transmission", 1.0)
         base_recipe_home = Path(f"/dls_sw/{beamline}/etc/agamemnon-recipes")
         agamemnon_recipe_config = base_recipe_home / "recipe_config.yaml"
         agamemnon_limits: dict[str, AgamemnonLimits] = parse_agamemnon_config(
@@ -346,8 +353,12 @@ def generate_strategy(
                     ),
                 )
                 if beamline_wants_dose_displayed:
+                    # Convert transmission to percentage for ISPyB
+                    relative_transmission_pct = (
+                        transmission_limits[1] / dc_transmission * 100
+                    )
                     screening_sub_wedge_command.update(
-                        dosetotal=dose, transmission=(transmission_limits[1] * 100)
+                        dosetotal=dose, transmission=(relative_transmission_pct)
                     )
                     ispyb_command_list.append(screening_sub_wedge_command)
                     continue

src/dlstbx/services/trigger.py

@@ -292,6 +292,10 @@ class StrategyParameters(pydantic.BaseModel):
     experiment_type: str
     program_id: int = pydantic.Field(gt=0)
     wavelength: float = pydantic.Field(gt=0)
+    max_transmission: Optional[float] = None
+    backoff_delay: float = pydantic.Field(default=8, alias="backoff-delay")
+    backoff_max_try: float = pydantic.Field(default=10, alias="backoff-max-try")
+    backoff_multiplier: float = pydantic.Field(default=2, alias="backoff-multiplier")
 
 
 class DLSTrigger(CommonService):
@@ -2894,6 +2898,7 @@ def trigger_align_crystal(
     def trigger_strategy(
         self,
         rw: workflows.recipe.RecipeWrapper,
+        message: Dict,
         *,
         parameters: StrategyParameters,
         session: sqlalchemy.orm.session.Session,
@@ -2941,17 +2946,53 @@ def trigger_strategy(
                 AutoProcProgram,
                 AutoProcProgram.autoProcProgramId == AutoProc.autoProcProgramId,
             )
-            .filter(AutoProcProgram.autoProcProgramId == parameters.program_id)
+            .join(
+                ProcessingJob,
+                AutoProcProgram.processingJobId == ProcessingJob.processingJobId,
+            )
+            .filter(ProcessingJob.dataCollectionId == parameters.dcid)
             .filter(AutoProcScalingStatistics.scalingStatisticsType == "overall")
-            .scalar()
+            .all()
         )
 
-        if not resolution:
+        self.log.info(
+            f"Strategy trigger: resolution estimate from ispyb for dcid={parameters.dcid} is {resolution}"
+        )
+        if resolution is None:
+            status = {
+                "ntry": 0,
+            }
+            backoff_delay = parameters.backoff_delay
+            backoff_multiplier = parameters.backoff_multiplier
+            backoff_max_try = parameters.backoff_max_try
+
+            if isinstance(message, dict):
+                status.update(message.get("trigger-status", {}))
+            message_delay = int(backoff_delay * backoff_multiplier ** status["ntry"])
+            status["ntry"] += 1
+
+            if status["ntry"] > backoff_max_try:
+                self.log.info(
+                    f"Skipping strategy trigger: maximum number of retries exceeded for dcid={parameters.dcid}"
+                )
+                return {"success": True}
+            # Send results to myself for next round of processing
             self.log.info(
-                f"Skipping strategy trigger: no resolution estimate found for dcid={parameters.dcid} auto_proc_program_id={parameters.program_id}"
+                f"Strategy trigger: Waiting for a resolution estimate for dcid={parameters.dcid} - Checkpointing message..."
+            )
+            rw.checkpoint(
+                {
+                    "trigger-status": status,
+                },
+                delay=message_delay,
             )
             return {"success": True}
 
+        min_resolution = min(resolution, key=lambda x: x[0])[0]
+        self.log.info(
+            f"Strategy trigger: found minumum resolution {min_resolution} for dcid={parameters.dcid}"
+        )
+
         jp = self.ispyb.mx_processing.get_job_params()
         jp["comments"] = parameters.comment
         jp["datacollectionid"] = parameters.dcid
@@ -2963,10 +3004,13 @@ def trigger_strategy(
 
         strategy_parameters = {
             "beamline": parameters.beamline,
-            "resolution": resolution,
+            "resolution": min_resolution,
             "wavelength": parameters.wavelength,
         }
 
+        if parameters.max_transmission:
+            strategy_parameters["transmission_estimate"] = parameters.max_transmission
+
         for key, value in strategy_parameters.items():
             jpp = self.ispyb.mx_processing.get_job_parameter_params()
             jpp["job_id"] = jobid

src/dlstbx/wrapper/estimate_transmission.py

@@ -20,6 +20,59 @@ class EstimateTransmissionWrapper(Wrapper):
 
     _logger_name = "dlstbx.wrap.estimate_transmission"
 
+    def collect_ispyb_command_list(self, transmission, max_pixel_count_pct):
+        if max_pixel_count_pct < 0.7:
+            warning_level = 0
+        elif max_pixel_count_pct < 0.85:
+            warning_level = 1
+        else:
+            warning_level = 2
+
+        warning_message = {
+            0: "Diffraction spots are unlikely to have detector count rate issues",
+            1: "Some diffraction spots may have detector count rate issues",
+            2: "Some diffraction spots are likely to have detector count rate issues",
+        }.get(warning_level)
+
+        warning_description = f"The most intense pixel is {max_pixel_count_pct * 100}% of the detector's limit"
+        warning_severity = {0: "INFO", 1: "WARNING", 2: "ERROR"}.get(warning_level)
+
+        # Step 0: Add a program message about the count rate warning,
+        # store the autoproc program
+        ispyb_command_list = []
+        d = {
+            "ispyb_command": "add_program_message",
+            "program_id": "$ispyb_autoprocprogram_id",
+            "message": warning_message,
+            "description": warning_description,
+            "severity": warning_severity,
+        }
+        ispyb_command_list.append(d)
+
+        # Step 1: Create screeningOutput record for recipe, linked to the screeningId
+        #         Keep the screeningOutputId
+        d = {
+            "program": "estimate_transmission",
+            "strategysuccess": 1,
+            "ispyb_command": "insert_screening_output",
+            "screening_id": "$ispyb_screening_id",
+            "store_result": "ispyb_screening_output_id",
+        }
+        ispyb_command_list.append(d)
+
+        # Step 2: Store screeningStrategy results, linked to the screeningOutputId
+        #         Keep the screeningStrategyId
+        d = {
+            "program": "estimate_transmission",
+            "ispyb_command": "insert_screening_strategy",
+            "transmission": transmission,
+            "screening_output_id": "$ispyb_screening_output_id",
+        }
+        ispyb_command_list.append(d)
+
+        self.log.info("Sending %s", str(ispyb_command_list))
+        self.recwrap.send_to("ispyb", {"ispyb_command_list": ispyb_command_list})
+
     def run(self):
         """Entrypoint for the estimate_transmission wrapper.
 
@@ -87,43 +140,13 @@ def run(self):
         )
         scale_factor = target_countrate_pct / pixel_countrate_pct
 
-        scaled_transmission = min(1, (transmission * scale_factor) / 100)
+        scaled_transmission = min(100, (transmission * scale_factor))
         self.log.info(f"Scaled transmission is : {scaled_transmission}")
 
-        self.recwrap.send_to(
-            "strategy",
-            {"parameters": {"scaled_transmission": float(scaled_transmission)}},
-        )
-
         max_pixel_count_pct = int(num_counts[-1]) / trusted_range
-        if max_pixel_count_pct < 0.7:
-            warning_level = 0
-        elif max_pixel_count_pct < 0.85:
-            warning_level = 1
-        else:
-            warning_level = 2
-
-        warning_message = {
-            0: "Diffraction spots are unlikely to have detector count rate issues",
-            1: "Some diffraction spots may have detector count rate issues",
-            2: "Some diffraction spots are likely to have detector count rate issues",
-        }.get(warning_level)
-
-        warning_description = f"The most intense pixel is {max_pixel_count_pct * 100}% of the detector's limit"
-        warning_severity = {0: "INFO", 1: "WARNING", 2: "ERROR"}.get(warning_level)
-
-        ispyb_command_list = [
-            {
-                "ispyb_command": "add_program_message",
-                "program_id": "$ispyb_autoprocprogram_id",
-                "message": warning_message,
-                "description": warning_description,
-                "severity": warning_severity,
-            }
-        ]
-
-        self.log.info("Sending %s", str(ispyb_command_list))
-        self.recwrap.send_to("ispyb", {"ispyb_command_list": ispyb_command_list})
+        # Add transmission to environment for use later in the recipe.
+        self.recwrap.environment.update({"max_transmission": scaled_transmission})
+        self.collect_ispyb_command_list(scaled_transmission, max_pixel_count_pct)
 
         results_directory.mkdir(parents=True, exist_ok=True)
         output_file = "dials.find_spots.log"
@@ -140,6 +163,20 @@ def run(self):
         self.save_plot(num_counts, num_pixels, results_directory)
         self.save_hist_to_json(counts_hist, trusted_range, results_directory)
 
+        output_files = {
+            "pixel_intensities.png": "result",
+            "pixel_counts.json": "result",
+            output_file: "log",
+        }
+        for file, filetype in output_files.items():
+            self.record_result_individual_file(
+                {
+                    "file_path": str(results_directory),
+                    "file_name": file,
+                    "file_type": filetype,
+                }
+            )
+
         self.log.info("Done.")
         return True