[#207] Add VMXm fast grid scan devices

src/dodal/beamlines/beamline_utils.py

@@ -13,11 +13,23 @@
 
 ACTIVE_DEVICES: Dict[str, AnyDevice] = {}
 BL = ""
+PREFIX: BeamlinePrefix = None  # type: ignore
 
 
-def set_beamline(beamline: str):
-    global BL
+def set_beamline(
+    beamline: str,
+    suffix: Optional[str] = None,
+    beamline_prefix: Optional[str] = None,
+    insertion_prefix: Optional[str] = None,
+):
+    global BL, PREFIX
     BL = beamline
+    PREFIX = BeamlinePrefix(
+        ixx=beamline,
+        suffix=suffix,
+        beamline_prefix=beamline_prefix,
+        insertion_prefix=insertion_prefix,
+    )
 
 
 def clear_devices():
@@ -95,9 +107,7 @@ def device_instantiation(
     if already_existing_device is None:
         device_instance = device_factory(
             name=name,
-            prefix=f"{(BeamlinePrefix(BL).beamline_prefix)}{prefix}"
-            if bl_prefix
-            else prefix,
+            prefix=f"{PREFIX.beamline_prefix}{prefix}" if bl_prefix else prefix,
             **kwargs,
         )
         ACTIVE_DEVICES[name] = device_instance

src/dodal/beamlines/vmxm.py

@@ -0,0 +1,142 @@
+from typing import Optional
+
+from dodal.beamlines.beamline_utils import device_instantiation
+from dodal.beamlines.beamline_utils import set_beamline as set_utils_beamline
+from dodal.devices.backlight import VmxmBacklight
+from dodal.devices.detector import DetectorParams
+from dodal.devices.eiger import EigerDetector
+from dodal.devices.fast_grid_scan_2d import FastGridScan2D
+from dodal.devices.synchrotron import Synchrotron
+from dodal.devices.vmxm.vmxm_attenuator import VmxmAttenuator
+from dodal.devices.vmxm.vmxm_sample_motors import VmxmSampleMotors
+from dodal.devices.zebra import Zebra
+from dodal.log import set_beamline as set_log_beamline
+from dodal.utils import get_beamline_name, skip_device
+
+SIM_BEAMLINE_NAME = "S02-1"
+
+BL = get_beamline_name(SIM_BEAMLINE_NAME)
+set_log_beamline(BL)
+set_utils_beamline(
+    BL, suffix="J", beamline_prefix="BL02J", insertion_prefix="SR-DI-J02"
+)
+
+
+@skip_device(lambda: BL == SIM_BEAMLINE_NAME)
+def eiger(
+    wait_for_connection: bool = True,
+    fake_with_ophyd_sim: bool = False,
+    params: Optional[DetectorParams] = None,
+) -> EigerDetector:
+    """Get the i24 Eiger device, instantiate it if it hasn't already been.
+    If this is called when already instantiated, it will return the existing object.
+    If called with params, will update those params to the Eiger object.
+    """
+
+    def set_params(eiger: EigerDetector):
+        if params is not None:
+            eiger.set_detector_parameters(params)
+
+    return device_instantiation(
+        device_factory=EigerDetector,
+        name="eiger",
+        prefix="-EA-EIGER-01:",
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+        post_create=set_params,
+    )
+
+
+@skip_device(lambda: BL == SIM_BEAMLINE_NAME)
+def fast_grid_scan(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> FastGridScan2D:
+    """Get the vmxm fast_grid_scan device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in vmxm, it will return the existing object.
+    """
+    return device_instantiation(
+        device_factory=FastGridScan2D,
+        name="fast_grid_scan",
+        prefix="-MO-SAMP-11:FGS:",
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )
+
+
+@skip_device(lambda: BL == SIM_BEAMLINE_NAME)
+def zebra(wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False) -> Zebra:
+    """Get the vmxm zebra device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in vmxm, it will return the existing object.
+    """
+    return device_instantiation(
+        Zebra,
+        "zebra",
+        "-EA-ZEBRA-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+@skip_device(lambda: BL == SIM_BEAMLINE_NAME)
+def attenuator(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> VmxmAttenuator:
+    """Get the vmxm attenuator device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in vmxm, it will return the existing object.
+    """
+    return device_instantiation(
+        VmxmAttenuator,
+        "attenuator",
+        "-OP-ATTN-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )
+
+
+@skip_device(lambda: BL == SIM_BEAMLINE_NAME)
+def backlight(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> VmxmBacklight:
+    """Get the VMXm backlight device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in VMXm, it will return the existing object.
+    """
+    return device_instantiation(
+        device_factory=VmxmBacklight,
+        name="backlight",
+        prefix="",
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )
+
+
+@skip_device(lambda: BL == SIM_BEAMLINE_NAME)
+def synchrotron(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Synchrotron:
+    """Get the VMXm synchrotron device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in VMXm, it will return the existing object.
+    """
+    return device_instantiation(
+        Synchrotron,
+        "synchrotron",
+        "",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+        bl_prefix=False,
+    )
+
+
+@skip_device(lambda: BL == SIM_BEAMLINE_NAME)
+def sample_motors(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> VmxmSampleMotors:
+    """Get the VMXm sample_motors device, instantiate it if it hasn't already been.
+    If this is called when already instantiated in VMXm, it will return the existing object.
+    """
+    return device_instantiation(
+        VmxmSampleMotors,
+        "sample_motors",
+        "-MO-SAMP-01:",
+        wait_for_connection,
+        fake_with_ophyd_sim,
+    )

src/dodal/devices/backlight.py

@@ -2,13 +2,12 @@
 
 
 class Backlight(Device):
-    """Simple device to trigger the pneumatic in/out."""
+    """Simple device for i03 to trigger the pneumatic in/out."""
 
     OUT = 0
     IN = 1
 
     pos: EpicsSignal = Component(EpicsSignal, "-EA-BL-01:CTRL")
-    # Toggle to switch it On or Off
     toggle: EpicsSignal = Component(EpicsSignal, "-EA-BLIT-01:TOGGLE")
 
     def set(self, position: int) -> StatusBase:
@@ -18,3 +17,21 @@ def set(self, position: int) -> StatusBase:
         else:
             status &= self.toggle.set("On")
         return status
+
+
+class VmxmBacklight(Device):
+    """Simple device for VMXm to control the backlight."""
+
+    OUT = 0
+    IN = 1
+
+    pos: EpicsSignal = Component(EpicsSignal, "-DI-IOC-02:LED:INOUT")
+    toggle: EpicsSignal = Component(EpicsSignal, "-EA-OAV-01:FZOOM:TOGGLE")
+
+    def set(self, position: int) -> StatusBase:
+        status = self.pos.set(position)
+        if position == self.OUT:
+            status &= self.toggle.set("Off")
+        else:
+            status &= self.toggle.set("On")
+        return status

src/dodal/devices/eiger.py

@@ -1,4 +1,5 @@
 from enum import Enum
+from time import sleep
 from typing import Optional
 
 from ophyd import Component, Device, EpicsSignalRO, Signal
@@ -35,7 +36,9 @@ def set(self, value, *, timeout=None, settle_time=None, **kwargs):
 
     STALE_PARAMS_TIMEOUT = 60
     GENERAL_STATUS_TIMEOUT = 10
-    ALL_FRAMES_TIMEOUT = 120
+    ALL_FRAMES_TIMEOUT = (
+        600  # FIXME: VMXm hack - we need to get this passed in a a param probably?
+    )
     ARMING_TIMEOUT = 60
 
     filewriters_finished: SubscriptionStatus
@@ -103,7 +106,9 @@ def stage(self):
             self.async_stage().wait(timeout=self.ARMING_TIMEOUT)
 
     def stop_odin_when_all_frames_collected(self):
-        LOGGER.info("Waiting on all frames")
+        LOGGER.info(
+            f"Waiting on all frames, expected {self.detector_params.full_number_of_images}"
+        )
         try:
             await_value(
                 self.odin.file_writer.num_captured,
@@ -152,6 +157,10 @@ def change_roi_mode(self, enable: bool) -> Status:
             else self.detector_params.detector_size_constants.det_size_pixels
         )
 
+        LOGGER.info(
+            f"Setting height and width on odin to {detector_dimensions.height}, {detector_dimensions.width}"
+        )
+
         status = self.cam.roi_mode.set(
             1 if enable else 0, timeout=self.GENERAL_STATUS_TIMEOUT
         )
@@ -242,7 +251,6 @@ def set_detector_threshold(self, energy: float, tolerance: float = 0.1) -> Statu
             tolerance (float, optional): If the energy is already set to within
                 this tolerance it is not set again. Defaults to 0.1eV.
         """
-
         current_energy = self.cam.photon_energy.get()
         if abs(current_energy - energy) > tolerance:
             return self.cam.photon_energy.set(
@@ -258,7 +266,6 @@ def set_num_triggers_and_captures(self) -> Status:
         during the datacollection. The number of images is the number of images per
         trigger.
         """
-
         assert self.detector_params is not None
         status = self.cam.num_images.set(
             self.detector_params.num_images_per_trigger,
@@ -286,6 +293,9 @@ def set_num_triggers_and_captures(self) -> Status:
 
     def _wait_for_odin_status(self) -> Status:
         self.forward_bit_depth_to_filewriter()
+        sleep(
+            2
+        )  # FIXME - VMXm hack - on VMXm ODIN will *occasionally* fail to start without this sleep. Need a better solution.
         status = self.odin.file_writer.capture.set(
             1, timeout=self.GENERAL_STATUS_TIMEOUT
         )
@@ -306,16 +316,20 @@ def _finish_arm(self) -> Status:
 
     def forward_bit_depth_to_filewriter(self):
         bit_depth = self.bit_depth.get()
-        self.odin.file_writer.data_type.put(f"UInt{bit_depth}")
+        self.odin.file_writer.data_type.set(f"UInt{bit_depth}").wait(
+            self.GENERAL_STATUS_TIMEOUT
+        )
 
     def disarm_detector(self):
         self.cam.acquire.put(0)
 
     def do_arming_chain(self) -> Status:
         functions_to_do_arm = []
         detector_params: DetectorParams = self.detector_params
-        if detector_params.use_roi_mode:
-            functions_to_do_arm.append(lambda: self.change_roi_mode(enable=True))
+
+        functions_to_do_arm.append(
+            lambda: self.change_roi_mode(detector_params.use_roi_mode)
+        )
 
         functions_to_do_arm.extend(
             [