Adds ability to read the lookup table for beam_centre X and Y

src/i19_bluesky/parameters/serial_parameters.py

@@ -12,6 +12,7 @@
     VisitParameters,
     ZebraRotationParams,
 )
+from i19_bluesky.parameters.constants import DetectorConstants, EigerConstants
 
 
 class SerialExperiment(VisitParameters):
@@ -77,3 +78,9 @@ def panda_rotation_params(self) -> PandaRotationParams:
             scan_steps=self.images_per_well,
             exposure_time_s=self.exposure_time_s,
         )
+
+    @property
+    def detector_constants(self) -> EigerConstants:
+        match self.detector_type:
+            case DetectorType.EIGER:
+                return DetectorConstants.EIGER

src/i19_bluesky/serial/device_setup_plans/eiger_metadata.py

@@ -1,25 +1,84 @@
 import bluesky.plan_stubs as bps
+from daq_config_server.models.lookup_tables import DetectorXYLookupTable
+from dodal.common.beamlines.beamline_utils import (
+    get_config_client,
+)
+from dodal.devices.detector.det_dim_constants import DetectorSize, DetectorSizeConstants
+from dodal.devices.util.lookup_tables import linear_interpolation_lut
 from ophyd_async.fastcs.eiger import EigerDetector
 
+from i19_bluesky.parameters.serial_parameters import SerialExperimentEh2
+
+BEAM_XY_TABLE_PATH = (
+    "/dls_sw/i19-2/software/daq_configuration/lookup/DetDistToBeamXYConverterE4M.txt"
+)
+
+
+def _convert_beam_centre_to_pixels(
+    beam_centre_mm: tuple[float, float],
+    image_size_mm: DetectorSize,
+    image_size_px: DetectorSize,
+) -> tuple[float, float]:
+    beam_x_px = beam_centre_mm[0] * image_size_px.width / image_size_mm.width
+    beam_y_px = beam_centre_mm[1] * image_size_px.height / image_size_mm.height
+
+    return (beam_x_px, beam_y_px)
+
+
+def _read_converter_lut():
+    config_client = get_config_client()
+    lut_contents = config_client.get_file_contents(
+        BEAM_XY_TABLE_PATH, DetectorXYLookupTable
+    )
+    return lut_contents.columns
+
+
+def calculate_beam_centre_from_lut(
+    detector_distance_mm: float, det_size_constants: DetectorSizeConstants
+) -> tuple[float, float]:
+    """Use the lookup table to find the beam centre position in mm and return it in
+    pixels.
+
+    Args:
+        detector_distance_mm (float): The detector distance in mm from the parameters.
+        det_size_constants (DetectorSizeConstants): The detector size parameters.
+
+    Returns:
+        tuple[float, float]: x and y positions of beam centre, in pixels.
+    """
+    lut_columns = _read_converter_lut()
+
+    interpolate_x = linear_interpolation_lut(lut_columns[0], lut_columns[1])
+    beam_x_mm = interpolate_x(detector_distance_mm)
+    interpolate_y = linear_interpolation_lut(lut_columns[0], lut_columns[2])
+    beam_y_mm = interpolate_y(detector_distance_mm)
+
+    beam_centre_px = _convert_beam_centre_to_pixels(
+        (beam_x_mm, beam_y_mm),
+        det_size_constants.det_dimension,
+        det_size_constants.det_size_pixels,
+    )
+    return beam_centre_px
+
 
 def write_eiger_params(
-    detector_distance_mm: float,
-    two_theta_deg: float,
-    phi_start: float,
-    phi_increment: float,
-    beam_center: tuple,
+    parameters: SerialExperimentEh2,
     energy: float,
     wavelength: float,
     eiger: EigerDetector,
     wait: bool,
     group: str = "eiger_metadata",
 ):
+    beam_centre = calculate_beam_centre_from_lut(
+        parameters.detector_distance_mm,
+        parameters.detector_constants.DET_SIZE_CONSTANTS,
+    )
     yield from bps.abs_set(eiger.detector.photon_energy, energy, group=group)
     yield from bps.abs_set(
-        eiger.detector.detector_distance, detector_distance_mm, group=group
+        eiger.detector.detector_distance, parameters.detector_distance_mm, group=group
     )
-    yield from bps.abs_set(eiger.detector.beam_center_x, beam_center[0], group=group)
-    yield from bps.abs_set(eiger.detector.beam_center_y, beam_center[1], group=group)
+    yield from bps.abs_set(eiger.detector.beam_center_x, beam_centre[0], group=group)
+    yield from bps.abs_set(eiger.detector.beam_center_y, beam_centre[1], group=group)
     yield from bps.abs_set(eiger.detector.omega_start, 0, group=group)
     yield from bps.abs_set(eiger.detector.omega_increment, 0, group=group)
     yield from bps.abs_set(
@@ -29,17 +88,17 @@ def write_eiger_params(
     )
     yield from bps.abs_set(
         eiger.detector.two_theta,  # type:ignore
-        two_theta_deg,
+        parameters.two_theta_deg,
         group=group,
     )
     yield from bps.abs_set(
         eiger.detector.phi_start,  # type:ignore
-        phi_start,
+        parameters.rot_axis_start,
         group=group,
     )
     yield from bps.abs_set(
         eiger.detector.phi_increment,  # type:ignore
-        phi_increment,
+        parameters.rot_axis_increment,
         group=group,
     )
     yield from bps.abs_set(

tests/unit_tests/parameters/test_serial_parameters.py

@@ -3,6 +3,7 @@
 import pytest
 
 from i19_bluesky.parameters.components import PandaRotationParams, ZebraRotationParams
+from i19_bluesky.parameters.constants import EigerConstants
 from i19_bluesky.parameters.serial_parameters import (
     SerialExperiment,
     SerialExperimentEh2,
@@ -119,7 +120,7 @@ def test_split_wells_into_run_list_for_collection(
     assert len(wells_per_run) == expected_run_num
 
 
-def test_serial_parameter_model_for_eh2(dummy_serial_params_eh2):
+def test_serial_parameter_model_for_eh2_with_eiger(dummy_serial_params_eh2):
     eh2_params = SerialExperimentEh2(**dummy_serial_params_eh2)
 
     assert eh2_params.total_num_images == 20
@@ -128,6 +129,7 @@ def test_serial_parameter_model_for_eh2(dummy_serial_params_eh2):
 
     assert isinstance(eh2_params.zebra_rotation_params, ZebraRotationParams)
     assert isinstance(eh2_params.panda_rotation_params, PandaRotationParams)
+    assert isinstance(eh2_params.detector_constants, EigerConstants)
 
     assert eh2_params.zebra_rotation_params.scan_start_deg == eh2_params.rot_axis_start
     assert eh2_params.zebra_rotation_params.rotation_direction == "Positive"

tests/unit_tests/serial/device_setup_plans/test_eiger_metadata.py

@@ -2,45 +2,96 @@
 
 import pytest
 from bluesky.run_engine import RunEngine
+from daq_config_server.models.lookup_tables import DetectorXYLookupTable
 from ophyd_async.fastcs.eiger import EigerDetector
 
+from i19_bluesky.parameters.serial_parameters import SerialExperimentEh2
 from i19_bluesky.serial.device_setup_plans.eiger_metadata import (
+    _convert_beam_centre_to_pixels,
+    _read_converter_lut,
+    calculate_beam_centre_from_lut,
     write_eiger_params,
 )
 
+LUT_FILE = """
+#Table giving centre position of beam X and Y as a function of detector distance
+#
+# Columns: detector distance, x-centre, y-centre
+Units mm mm mm
+85  69.66 81.09
+585 69.66 81.09
+"""
+
+lut_columns = [[85.0, 585.0], [69.66, 69.66], [81.09, 81.09]]
+
+
+def test_convert_bc_to_pix(parameters):
+    det_size_mm = parameters.detector_constants.DET_SIZE_CONSTANTS.det_dimension
+    det_size_px = parameters.detector_constants.DET_SIZE_CONSTANTS.det_size_pixels
+    beam_x, beam_y = _convert_beam_centre_to_pixels((20, 10), det_size_mm, det_size_px)
+
+    assert beam_x == pytest.approx(266.67, 1e-2)
+    assert beam_y == pytest.approx(133.33, 1e-2)
+
+
+@patch("i19_bluesky.serial.device_setup_plans.eiger_metadata.get_config_client")
+def test_read_converter_lut(mock_config_client: MagicMock):
+    file_contents = DetectorXYLookupTable.from_contents(LUT_FILE)
+    mock_config_client.return_value.get_file_contents.return_value = file_contents
+    cols = _read_converter_lut()
+
+    assert cols == lut_columns
+
+
+@patch("i19_bluesky.serial.device_setup_plans.eiger_metadata._read_converter_lut")
+def test_calculate_beam_centre_from_lut(
+    mock_read_lut: MagicMock, parameters: SerialExperimentEh2
+):
+    mock_read_lut.return_value = lut_columns
+
+    (beam_centre_x, beam_centre_y) = calculate_beam_centre_from_lut(
+        parameters.detector_distance_mm,
+        parameters.detector_constants.DET_SIZE_CONSTANTS,
+    )
+
+    assert beam_centre_x == pytest.approx(928.80, 1e-2)
+    assert beam_centre_y == pytest.approx(1081.2, 1e-2)
+
 
 @pytest.mark.parametrize("wait", [(False, True)])
 @patch("i19_bluesky.serial.device_setup_plans.eiger_metadata.bps.wait")
+@patch(
+    "i19_bluesky.serial.device_setup_plans.eiger_metadata.calculate_beam_centre_from_lut"
+)
 async def test_write_eiger_params(
+    mock_bc_from_lut: MagicMock,
     mock_bps_wait: MagicMock,
     eh2_eiger: EigerDetector,
+    parameters: SerialExperimentEh2,
     RE: RunEngine,
     wait: bool,
 ):
+    mock_bc_from_lut.return_value = (100, 200)
 
     RE(
         write_eiger_params(
-            100,
-            0,
-            0,
-            1,
-            (3, 2),
-            10,
-            100,
+            parameters,
+            17,
+            0.6,
             eh2_eiger,
             wait=wait,
         )
     )
-    assert await eh2_eiger.detector.detector_distance.get_value() == 100
-    assert await eh2_eiger.detector.beam_center_x.get_value() == 3
-    assert await eh2_eiger.detector.beam_center_y.get_value() == 2
-    assert await eh2_eiger.detector.photon_energy.get_value() == 10
+    assert await eh2_eiger.detector.detector_distance.get_value() == 320
+    assert await eh2_eiger.detector.beam_center_x.get_value() == 100
+    assert await eh2_eiger.detector.beam_center_y.get_value() == 200
+    assert await eh2_eiger.detector.photon_energy.get_value() == 17
     assert await eh2_eiger.detector.omega_start.get_value() == 0
     assert await eh2_eiger.detector.omega_increment.get_value() == 0
-    assert await eh2_eiger.detector.wavelength.get_value() == 100  # type:ignore
+    assert await eh2_eiger.detector.wavelength.get_value() == 0.6  # type:ignore
     assert await eh2_eiger.detector.two_theta.get_value() == 0  # type:ignore
     assert await eh2_eiger.detector.phi_start.get_value() == 0  # type:ignore
-    assert await eh2_eiger.detector.phi_increment.get_value() == 1  # type:ignore
+    assert await eh2_eiger.detector.phi_increment.get_value() == 0.1  # type:ignore
     assert await eh2_eiger.detector.chi_start.get_value() == 0  # type:ignore
     assert await eh2_eiger.detector.chi_increment.get_value() == 0  # type:ignore
     assert await eh2_eiger.detector.kappa_start.get_value() == 0  # type:ignore