Merge pull request #1262 from DiamondLightSource/1173_use_event_data_…

…for_nexus_writing

Use event data for nexus writing

src/hyperion/experiment_plans/rotation_scan_plan.py

@@ -208,8 +208,6 @@ def _rotation_scan_plan(
         yield from bps.wait("setup_zebra")
 
         # get some information for the ispyb deposition and trigger the callback
-        yield from read_hardware_for_nexus_writer(composite.eiger)
-
         yield from read_hardware_for_zocalo(composite.eiger)
 
         yield from read_hardware_for_ispyb_pre_collection(
@@ -222,6 +220,7 @@ def _rotation_scan_plan(
         yield from read_hardware_for_ispyb_during_collection(
             composite.attenuator, composite.flux, composite.dcm
         )
+        yield from read_hardware_for_nexus_writer(composite.eiger)
 
         # Get ready for the actual scan
         yield from bps.abs_set(

src/hyperion/external_interaction/callbacks/rotation/nexus_callback.py

@@ -8,7 +8,10 @@
 from hyperion.external_interaction.callbacks.plan_reactive_callback import (
     PlanReactiveCallback,
 )
-from hyperion.external_interaction.nexus.nexus_utils import vds_type_based_on_bit_depth
+from hyperion.external_interaction.nexus.nexus_utils import (
+    create_beam_and_attenuator_parameters,
+    vds_type_based_on_bit_depth,
+)
 from hyperion.external_interaction.nexus.write_nexus import NexusWriter
 from hyperion.log import NEXUS_LOGGER
 from hyperion.parameters.constants import CONST
@@ -48,12 +51,24 @@ def activity_gated_descriptor(self, doc: EventDescriptor):
 
     def activity_gated_event(self, doc: Event):
         event_descriptor = self.descriptors.get(doc["descriptor"])
+        assert isinstance(self.parameters, RotationInternalParameters)
         if event_descriptor is None:
             NEXUS_LOGGER.warning(
                 f"Rotation Nexus handler {self} received event doc {doc} and "
                 "has no corresponding descriptor record"
             )
             return doc
+        if event_descriptor.get("name") == CONST.PLAN.ISPYB_TRANSMISSION_FLUX_READ:
+            NEXUS_LOGGER.info(f"Nexus handler received event from read hardware {doc}")
+            data = doc["data"]
+            assert self.writer, "Nexus writer not initialised"
+            self.writer.beam, self.writer.attenuator = (
+                create_beam_and_attenuator_parameters(
+                    data["dcm_energy_in_kev"],
+                    data["flux_flux_reading"],
+                    data["attenuator_actual_transmission"],
+                )
+            )
         if event_descriptor.get("name") == CONST.PLAN.NEXUS_READ:
             NEXUS_LOGGER.info(f"Nexus handler received event from read hardware {doc}")
             vds_data_type = vds_type_based_on_bit_depth(doc["data"]["eiger_bit_depth"])

src/hyperion/external_interaction/callbacks/xray_centre/nexus_callback.py

@@ -1,10 +1,13 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Dict
 
 from hyperion.external_interaction.callbacks.plan_reactive_callback import (
     PlanReactiveCallback,
 )
+from hyperion.external_interaction.nexus.nexus_utils import (
+    create_beam_and_attenuator_parameters,
+)
 from hyperion.external_interaction.nexus.write_nexus import NexusWriter
 from hyperion.log import NEXUS_LOGGER
 from hyperion.parameters.constants import CONST
@@ -13,7 +16,7 @@
 )
 
 if TYPE_CHECKING:
-    from event_model.documents import EventDescriptor, RunStart
+    from event_model.documents import Event, EventDescriptor, RunStart
 
 
 class GridscanNexusFileCallback(PlanReactiveCallback):
@@ -37,41 +40,49 @@ class GridscanNexusFileCallback(PlanReactiveCallback):
 
     def __init__(self) -> None:
         super().__init__(NEXUS_LOGGER)
-        self.parameters: GridscanInternalParameters | None = None
         self.run_start_uid: str | None = None
         self.nexus_writer_1: NexusWriter | None = None
         self.nexus_writer_2: NexusWriter | None = None
+        self.descriptors: Dict[str, EventDescriptor] = {}
         self.log = NEXUS_LOGGER
 
     def activity_gated_start(self, doc: RunStart):
         if doc.get("subplan_name") == CONST.PLAN.GRIDSCAN_OUTER:
             json_params = doc.get("hyperion_internal_parameters")
             NEXUS_LOGGER.info(
-                f"Nexus writer recieved start document with experiment parameters {json_params}"
+                f"Nexus writer received start document with experiment parameters {json_params}"
             )
-            self.parameters = GridscanInternalParameters.from_json(json_params)
-            self.run_start_uid = doc.get("uid")
-
-    def activity_gated_descriptor(self, doc: EventDescriptor):
-        if doc.get("name") == CONST.PLAN.ISPYB_HARDWARE_READ:
-            assert (
-                self.parameters is not None
-            ), "Nexus callback did not receive parameters before being asked to write!"
-            # TODO instead of ispyb wait for detector parameter reading in plan
-            # https://github.com/DiamondLightSource/python-hyperion/issues/629
-            # and update parameters before creating writers
-
-            NEXUS_LOGGER.info("Initialising nexus writers...")
-            nexus_data_1 = self.parameters.get_nexus_info(1)
-            nexus_data_2 = self.parameters.get_nexus_info(2)
-            self.nexus_writer_1 = NexusWriter(self.parameters, **nexus_data_1)
+            parameters = GridscanInternalParameters.from_json(json_params)
+            nexus_data_1 = parameters.get_nexus_info(1)
+            nexus_data_2 = parameters.get_nexus_info(2)
+            self.nexus_writer_1 = NexusWriter(parameters, **nexus_data_1)
             self.nexus_writer_2 = NexusWriter(
-                self.parameters,
+                parameters,
                 **nexus_data_2,
                 vds_start_index=nexus_data_1["data_shape"][0],
             )
-            self.nexus_writer_1.create_nexus_file()
-            self.nexus_writer_2.create_nexus_file()
-            NEXUS_LOGGER.info(
-                f"Nexus files created at {self.nexus_writer_1.full_filename} and {self.nexus_writer_1.full_filename}"
-            )
+            self.run_start_uid = doc.get("uid")
+
+    def activity_gated_descriptor(self, doc: EventDescriptor):
+        self.descriptors[doc["uid"]] = doc
+
+    def activity_gated_event(self, doc: Event) -> Event | None:
+        event_descriptor = self.descriptors.get(doc["descriptor"])
+        if (
+            event_descriptor
+            and event_descriptor.get("name") == CONST.PLAN.ISPYB_TRANSMISSION_FLUX_READ
+        ):
+            data = doc["data"]
+            for nexus_writer in [self.nexus_writer_1, self.nexus_writer_2]:
+                assert nexus_writer, "Nexus callback did not receive start doc"
+                nexus_writer.beam, nexus_writer.attenuator = (
+                    create_beam_and_attenuator_parameters(
+                        data["dcm_energy_in_kev"],
+                        data["flux_flux_reading"],
+                        data["attenuator_actual_transmission"],
+                    )
+                )
+                nexus_writer.create_nexus_file()
+                NEXUS_LOGGER.info(f"Nexus file created at {nexus_writer.full_filename}")
+
+        return super().activity_gated_event(doc)

src/hyperion/external_interaction/nexus/nexus_utils.py

@@ -9,8 +9,8 @@
 from nexgen.nxs_utils.Axes import TransformationType
 from numpy.typing import DTypeLike
 
-from hyperion.external_interaction.ispyb.ispyb_dataclass import IspybParams
 from hyperion.log import NEXUS_LOGGER
+from hyperion.utils.utils import convert_eV_to_angstrom
 
 
 def vds_type_based_on_bit_depth(detector_bit_depth: int) -> DTypeLike:
@@ -125,17 +125,14 @@ def create_detector_parameters(detector_params: DetectorParams) -> Detector:
 
 
 def create_beam_and_attenuator_parameters(
-    ispyb_params: IspybParams,
+    energy_kev: float, flux: float, transmission_fraction: float
 ) -> tuple[Beam, Attenuator]:
-    """Create beam and attenuator dictionaries that nexgen can understand.
-
-    Args:
-        ispyb_params (IspybParams): An IspybParams object holding all required data.
+    """Create beam and attenuator objects that nexgen can understands
 
     Returns:
         tuple[Beam, Attenuator]: Descriptions of the beam and attenuator for nexgen.
     """
     return (
-        Beam(ispyb_params.wavelength_angstroms, ispyb_params.flux),
-        Attenuator(ispyb_params.transmission_fraction),
+        Beam(convert_eV_to_angstrom(energy_kev * 1000), flux),
+        Attenuator(transmission_fraction),
     )

src/hyperion/external_interaction/nexus/write_nexus.py

@@ -7,15 +7,15 @@
 
 import math
 from pathlib import Path
+from typing import Optional
 
 import numpy as np
 from dodal.utils import get_beamline_name
-from nexgen.nxs_utils import Detector, Goniometer, Source
+from nexgen.nxs_utils import Attenuator, Beam, Detector, Goniometer, Source
 from nexgen.nxs_write.NXmxWriter import NXmxFileWriter
 from numpy.typing import DTypeLike
 
 from hyperion.external_interaction.nexus.nexus_utils import (
-    create_beam_and_attenuator_parameters,
     create_detector_parameters,
     create_goniometer_axes,
     get_start_and_predicted_end_time,
@@ -36,6 +36,8 @@ def __init__(
         run_number: int | None = None,
         vds_start_index: int = 0,
     ) -> None:
+        self.beam: Optional[Beam] = None
+        self.attenuator: Optional[Attenuator] = None
         self.scan_points: dict = scan_points
         self.data_shape: tuple[int, int, int] = data_shape
         hyperion_parameters: HyperionParameters = (
@@ -53,9 +55,6 @@ def __init__(
         self.detector: Detector = create_detector_parameters(
             hyperion_parameters.detector_params
         )
-        self.beam, self.attenuator = create_beam_and_attenuator_parameters(
-            hyperion_parameters.ispyb_params
-        )
         self.source: Source = Source(get_beamline_name("S03"))
         self.directory: Path = Path(hyperion_parameters.detector_params.directory)
         self.filename: str = hyperion_parameters.detector_params.prefix

tests/conftest.py

@@ -292,6 +292,7 @@ def ophyd_pin_tip_detection():
 
 @pytest.fixture
 def robot():
+    RunEngine()  # A RE is needed to start the bluesky loop
     robot = i03.robot(fake_with_ophyd_sim=True)
     set_sim_value(robot.barcode.bare_signal, ["BARCODE"])
     return robot
@@ -304,7 +305,9 @@ def attenuator():
         return_value=Status(done=True, success=True),
         autospec=True,
     ):
-        yield i03.attenuator(fake_with_ophyd_sim=True)
+        attenuator = i03.attenuator(fake_with_ophyd_sim=True)
+        attenuator.actual_transmission.sim_put(0.49118047952)  # type: ignore
+        yield attenuator
 
 
 @pytest.fixture
@@ -318,6 +321,7 @@ def xbpm_feedback(done_status):
 @pytest.fixture
 def dcm():
     dcm = i03.dcm(fake_with_ophyd_sim=True)
+    dcm.energy_in_kev.user_readback.sim_put(12.7)  # type: ignore
     dcm.pitch_in_mrad.user_setpoint._use_limits = False
     dcm.dcm_roll_converter_lookup_table_path = (
         "tests/test_data/test_beamline_dcm_roll_converter.txt"
@@ -514,12 +518,13 @@ def fake_fgs_composite(
     xbpm_feedback,
     aperture_scatterguard,
     zocalo,
+    dcm,
 ):
     fake_composite = FlyScanXRayCentreComposite(
         aperture_scatterguard=aperture_scatterguard,
         attenuator=attenuator,
         backlight=i03.backlight(fake_with_ophyd_sim=True),
-        dcm=i03.dcm(fake_with_ophyd_sim=True),
+        dcm=dcm,
         # We don't use the eiger fixture here because .unstage() is used in some tests
         eiger=i03.eiger(fake_with_ophyd_sim=True),
         fast_grid_scan=i03.fast_grid_scan(fake_with_ophyd_sim=True),