Add Jungfrau rotation scan plan

pyproject.toml

@@ -48,7 +48,7 @@ dependencies = [
     "ophyd >= 1.10.5",
     "ophyd-async >= 0.14.0",
     "bluesky >= 1.14.6",
-    "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@62960e0e587bf86943ce1b581848fa131ef884d5",
+    "dls-dodal @ git+https://github.com/DiamondLightSource/dodal.git@50e72219dd379dd05e1fb1173397852407832f75",
 ]
 
 

src/mx_bluesky/beamlines/i24/jungfrau_commissioning/__init__.py

@@ -0,0 +1,13 @@
+from mx_bluesky.beamlines.i24.jungfrau_commissioning.experiment_plans.do_darks import (
+    do_non_pedestal_darks,
+    do_pedestal_darks,
+)
+from mx_bluesky.beamlines.i24.jungfrau_commissioning.experiment_plans.rotation_scan_plan import (
+    single_rotation_plan,
+)
+
+__all__ = [
+    "do_pedestal_darks",
+    "do_non_pedestal_darks",
+    "single_rotation_plan",
+]

src/mx_bluesky/beamlines/i24/jungfrau_commissioning/callbacks/metadata_writer.py

@@ -0,0 +1,81 @@
+import json
+
+from bluesky.callbacks import CallbackBase
+from dodal.devices.i24.commissioning_jungfrau import JungfrauCommissioningWriter
+
+from mx_bluesky.beamlines.i24.parameters.constants import PlanNameConstants
+from mx_bluesky.common.parameters.rotation import SingleRotationScan
+from mx_bluesky.common.utils.log import LOGGER
+
+READING_DUMP_FILENAME = "collection_info.json"
+
+
+class JsonMetadataWriter(CallbackBase):
+    """Callback class to handle the creation of metadata json files for commissioning.
+
+    To use, subscribe the Bluesky RunEngine to an instance of this class.
+    E.g.:
+        metadata_writer_callback = JsonMetadataWriter(parameters)
+        RE.subscribe(metadata_writer_callback)
+    Or decorate a plan using bluesky.preprocessors.subs_decorator.
+
+    See: https://blueskyproject.io/bluesky/callbacks.html#ways-to-invoke-callbacks
+
+    """
+
+    def __init__(self, jf_writer: JungfrauCommissioningWriter):
+        self.jf_writer = jf_writer  # For path information
+        self.wavelength_in_a = None
+        self.energy_in_kev = None
+        self.detector_distance_mm = None
+        self.descriptors: dict[str, dict] = {}
+        self.transmission: float | None = None
+        self.parameters: SingleRotationScan | None = None
+
+        super().__init__()
+
+    def start(self, doc: dict):  # type: ignore
+        if doc.get("subplan_name") == PlanNameConstants.ROTATION_MAIN:
+            json_params = doc.get("rotation_scan_params")
+            assert json_params is not None
+            LOGGER.info(
+                f"Metadata writer recieved start document with experiment parameters {json_params}"
+            )
+            self.parameters = SingleRotationScan(**json.loads(json_params))
+            self.run_start_uid = doc.get("uid")
+
+    def descriptor(self, doc: dict):  # type: ignore
+        self.descriptors[doc["uid"]] = doc
+
+    def event(self, doc: dict):  # type: ignore
+        event_descriptor = self.descriptors[doc["descriptor"]]
+
+        if event_descriptor.get("name") == PlanNameConstants.ROTATION_DEVICE_READ:
+            assert self.parameters is not None
+            data = doc.get("data")
+            assert data is not None
+            self.wavelength_in_a = data.get("dcm-wavelength_in_a")
+            self.energy_in_kev = data.get("dcm-energy_in_keV")
+            self.detector_distance_mm = data.get("detector_motion-z")
+
+            LOGGER.info(
+                f"Metadata writer received parameters, energy_in_kev: {self.energy_in_kev}, wavelength: {self.wavelength_in_a}, det_distance_mm: {self.detector_distance_mm}"
+            )
+
+    def stop(self, doc: dict):  # type: ignore
+        assert self.parameters is not None
+        if (
+            self.run_start_uid is not None
+            and doc.get("run_start") == self.run_start_uid
+        ):
+            with open(self.jf_writer.final_path / READING_DUMP_FILENAME, "w") as f:
+                f.write(
+                    json.dumps(
+                        {
+                            "wavelength_in_a": self.wavelength_in_a,
+                            "energy_kev": self.energy_in_kev,
+                            "angular_increment_deg": self.parameters.rotation_increment_deg,
+                            "detector_distance_mm": self.detector_distance_mm,
+                        }
+                    )
+                )

src/mx_bluesky/beamlines/i24/jungfrau_commissioning/composites.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import pydantic
+from dodal.devices.attenuator.attenuator import EnumFilterAttenuator
+from dodal.devices.hutch_shutter import HutchShutter
+from dodal.devices.i24.aperture import Aperture
+from dodal.devices.i24.beamstop import Beamstop
+from dodal.devices.i24.commissioning_jungfrau import CommissioningJungfrau
+from dodal.devices.i24.dcm import DCM
+from dodal.devices.i24.dual_backlight import DualBacklight
+from dodal.devices.i24.vgonio import VerticalGoniometer
+from dodal.devices.motors import YZStage
+from dodal.devices.synchrotron import Synchrotron
+from dodal.devices.xbpm_feedback import XBPMFeedback
+from dodal.devices.zebra.zebra import Zebra
+from dodal.devices.zebra.zebra_controlled_shutter import ZebraShutter
+
+
+@pydantic.dataclasses.dataclass(config={"arbitrary_types_allowed": True})
+class RotationScanComposite:
+    """All devices which are directly or indirectly required by this plan"""
+
+    aperture: Aperture
+    attenuator: EnumFilterAttenuator
+    jungfrau: CommissioningJungfrau
+    gonio: VerticalGoniometer
+    synchrotron: Synchrotron
+    sample_shutter: ZebraShutter
+    zebra: Zebra
+    xbpm_feedback: XBPMFeedback
+    hutch_shutter: HutchShutter
+    beamstop: Beamstop
+    det_stage: YZStage
+    backlight: DualBacklight
+    dcm: DCM

src/mx_bluesky/beamlines/i24/jungfrau_commissioning/experiment_plans/do_darks.py

@@ -13,7 +13,6 @@
 
 from mx_bluesky.beamlines.i24.jungfrau_commissioning.plan_stubs.plan_utils import (
     fly_jungfrau,
-    override_file_path,
 )
 from mx_bluesky.common.utils.log import LOGGER
 
@@ -25,8 +24,8 @@ def do_pedestal_darks(
     exp_time_s: float = 0.001,
     pedestal_frames: PositiveInt = 20,
     pedestal_loops: PositiveInt = 200,
+    filename: str = "pedestal_darks",
     jungfrau: CommissioningJungfrau = inject("jungfrau"),
-    path_of_output_file: str | None = None,
 ) -> MsgGenerator:
     """Acquire darks in pedestal mode, using dynamic gain mode. This calibrates the offsets
     for the jungfrau, and must be performed before acquiring real data in dynamic gain mode.
@@ -46,18 +45,19 @@ def do_pedestal_darks(
         exp_time_s: Length of detector exposure for each frame.
         pedestal_frames: Number of frames acquired per pedestal loop.
         pedestal_loops: Number of times to acquire a set of pedestal_frames
+        filename: Name of output file
         jungfrau: Jungfrau device
-        path_of_output_file: Absolute path of the detector file output, including file name. If None, then use the PathProvider
-            set during Jungfrau device instantiation
     """
 
     @bpp.set_run_key_decorator(PEDESTAL_DARKS_RUN)
-    @bpp.run_decorator(md={"subplan_name": PEDESTAL_DARKS_RUN})
+    @bpp.run_decorator(
+        md={
+            "subplan_name": PEDESTAL_DARKS_RUN,
+            "detector_file_template": filename,
+        }
+    )
     @bpp.stage_decorator([jungfrau])
     def _do_decorated_plan():
-        if path_of_output_file:
-            override_file_path(jungfrau, path_of_output_file)
-
         trigger_info = create_jungfrau_pedestal_triggering_info(
             exp_time_s, pedestal_frames, pedestal_loops
         )
@@ -67,12 +67,11 @@ def _do_decorated_plan():
         yield from bps.mv(
             jungfrau.drv.acquisition_type,
             AcquisitionType.PEDESTAL,
-            jungfrau.drv.gain_mode,
-            GainMode.DYNAMIC,
         )
         yield from fly_jungfrau(
             jungfrau,
             trigger_info,
+            GainMode.DYNAMIC,
             wait=True,
             log_on_percentage_prefix="Jungfrau pedestal dynamic gain mode darks triggers received",
         )
@@ -84,8 +83,8 @@ def do_non_pedestal_darks(
     gain_mode: GainMode,
     exp_time_s: float = 0.001,
     total_triggers: PositiveInt = 1000,
+    filename: str = "darks",
     jungfrau: CommissioningJungfrau = inject("jungfrau"),
-    path_of_output_file: str | None = None,
 ) -> MsgGenerator:
     """Internally take a set of images at a given gain mode.
 
@@ -96,26 +95,26 @@ def do_non_pedestal_darks(
         exp_time_s: Length of detector exposure for each trigger.
         total_triggers: Total triggers for the dark scan.
         jungfrau: Jungfrau device
-        path_of_output_file: Absolute path of the detector file output, including file name. If None, then use the PathProvider
-            set during Jungfrau device instantiation
+        filename: Name of output file
     """
 
     @bpp.set_run_key_decorator(STANDARD_DARKS_RUN)
-    @bpp.run_decorator(md={"subplan_name": STANDARD_DARKS_RUN})
+    @bpp.run_decorator(
+        md={
+            "subplan_name": STANDARD_DARKS_RUN,
+            "detector_file_template": filename,
+        }
+    )
     @bpp.stage_decorator([jungfrau])
     def _do_decorated_plan():
-        if path_of_output_file:
-            override_file_path(jungfrau, path_of_output_file)
-
         trigger_info = create_jungfrau_internal_triggering_info(
             total_triggers, exp_time_s
         )
 
-        yield from bps.mv(jungfrau.drv.gain_mode, gain_mode)
-
         yield from fly_jungfrau(
             jungfrau,
             trigger_info,
+            gain_mode,
             wait=True,
             log_on_percentage_prefix=f"Jungfrau {gain_mode} gain mode darks triggers received",
         )