805 move p99 detectors into dodal

src/dodal/beamlines/p99.py

@@ -1,4 +1,13 @@
-from dodal.common.beamlines.beamline_utils import device_instantiation, set_beamline
+from pathlib import Path
+
+from ophyd_async.core import AutoIncrementFilenameProvider, StaticPathProvider
+from ophyd_async.epics.adcore import SingleTriggerDetector
+
+from dodal.common.beamlines.beamline_utils import (
+    device_instantiation,
+    set_beamline,
+)
+from dodal.devices.areadetector import Andor2
 from dodal.devices.motors import XYZPositioner
 from dodal.devices.p99.sample_stage import FilterMotor, SampleAngleStage
 from dodal.log import set_beamline as set_log_beamline
@@ -12,7 +21,7 @@
 def sample_angle_stage(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> SampleAngleStage:
-    """Sample stage for p99"""
+    """sample angle stage (old currently on the side)"""
 
     return device_instantiation(
         SampleAngleStage,
@@ -26,8 +35,6 @@ def sample_angle_stage(
 def sample_stage_filer(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> FilterMotor:
-    """Sample stage for p99"""
-
     return device_instantiation(
         FilterMotor,
         prefix="-MO-STAGE-02:MP:SELECT",
@@ -41,7 +48,7 @@ def sample_xyz_stage(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> XYZPositioner:
     return device_instantiation(
-        FilterMotor,
+        XYZPositioner,
         prefix="-MO-STAGE-02:",
         name="sample_xyz_stage",
         wait=wait_for_connection,
@@ -53,9 +60,49 @@ def sample_lab_xyz_stage(
     wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
 ) -> XYZPositioner:
     return device_instantiation(
-        FilterMotor,
+        XYZPositioner,
         prefix="-MO-STAGE-02:LAB:",
         name="sample_lab_xyz_stage",
         wait=wait_for_connection,
         fake=fake_with_ophyd_sim,
     )
+
+
+andor_data_path = StaticPathProvider(
+    filename_provider=AutoIncrementFilenameProvider(base_filename="andor2"),
+    directory_path=Path("/dls/p99/data/2024/cm37284-2/processing/writenData"),
+)
+
+
+def andor2_det(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> Andor2:
+    return device_instantiation(
+        Andor2,
+        prefix="-EA-DET-03:",
+        name="andor2_det",
+        path_provider=andor_data_path,
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )
+
+
+def andor2_point(
+    wait_for_connection: bool = True, fake_with_ophyd_sim: bool = False
+) -> SingleTriggerDetector:
+    """
+    This is the same physical detector as andor2_det but it will not save
+    any image as we only care about the detector mean count.
+    This behave very much like a oversize/overpirce photo diode when use in this mode.
+    """
+    return device_instantiation(
+        SingleTriggerDetector,
+        drv=andor2_det(wait_for_connection, fake_with_ophyd_sim).drv,
+        read_uncached=(
+            [andor2_det(wait_for_connection, fake_with_ophyd_sim).drv.stat_mean]
+        ),
+        prefix="",
+        name="andor2_point",
+        wait=wait_for_connection,
+        fake=fake_with_ophyd_sim,
+    )

src/dodal/devices/areadetector/__init__.py

@@ -1,10 +1,12 @@
 from .adaravis import AdAravisDetector
 from .adsim import AdSimDetector
 from .adutils import Hdf5Writer, SynchronisedAdDriverBase
+from .andor2 import Andor2
 
 __all__ = [
     "AdSimDetector",
     "SynchronisedAdDriverBase",
     "Hdf5Writer",
     "AdAravisDetector",
+    "Andor2",
 ]

src/dodal/devices/areadetector/andor2.py

@@ -0,0 +1,49 @@
+from bluesky.protocols import Hints
+from ophyd_async.core import PathProvider, StandardDetector
+from ophyd_async.epics.adcore import ADBaseDatasetDescriber, ADHDFWriter, NDFileHDFIO
+
+from dodal.devices.areadetector.andor2_epics import Andor2Controller, Andor2DriverIO
+
+
+class Andor2(StandardDetector):
+    """
+    Andor 2 area detector device (CCD detector 56fps with full chip readout).
+    Andor model:DU897_BV.
+    """
+
+    _controller: Andor2Controller
+    _writer: ADHDFWriter
+
+    def __init__(
+        self,
+        prefix: str,
+        path_provider: PathProvider,
+        name: str,
+    ):
+        """
+        Parameters
+        ----------
+        prefix: str
+            Epic Pv,
+        path_provider: PathProvider
+            Path provider for hdf writer
+        name: str
+            Name of the device
+        """
+        self.drv = Andor2DriverIO(prefix + "CAM:")
+        self.hdf = NDFileHDFIO(prefix + "HDF5:")
+        super().__init__(
+            Andor2Controller(self.drv),
+            ADHDFWriter(
+                hdf=self.hdf,
+                path_provider=path_provider,
+                name_provider=lambda: self.name,
+                dataset_describer=ADBaseDatasetDescriber(self.drv),
+            ),
+            config_sigs=[self.drv.acquire_time],
+            name=name,
+        )
+
+    @property
+    def hints(self) -> Hints:
+        return self._writer.hints

src/dodal/devices/areadetector/andor2_epics/__init__.py

@@ -0,0 +1,4 @@
+from .andor2_controller import Andor2Controller
+from .andor2_io import Andor2DriverIO, Andor2TriggerMode, ImageMode
+
+__all__ = ["Andor2Controller", "Andor2TriggerMode", "Andor2DriverIO", "ImageMode"]

src/dodal/devices/areadetector/andor2_epics/andor2_controller.py

@@ -0,0 +1,79 @@
+import asyncio
+
+from ophyd_async.core import DetectorController, DetectorTrigger
+from ophyd_async.core._detector import TriggerInfo
+from ophyd_async.epics import adcore
+from ophyd_async.epics.adcore import (
+    DEFAULT_GOOD_STATES,
+    DetectorState,
+    stop_busy_record,
+)
+
+from .andor2_io import (
+    Andor2DriverIO,
+    Andor2TriggerMode,
+    ImageMode,
+)
+
+MIN_DEAD_TIME = 0.1
+DEFAULT_MAX_NUM_IMAGE = 999_999
+
+
+class Andor2Controller(DetectorController):
+    _supported_trigger_types = {
+        DetectorTrigger.internal: Andor2TriggerMode.internal,
+        DetectorTrigger.edge_trigger: Andor2TriggerMode.ext_trigger,
+    }
+
+    def __init__(
+        self,
+        driver: Andor2DriverIO,
+        good_states: set[DetectorState] | None = None,
+    ) -> None:
+        if good_states is None:
+            good_states = set(DEFAULT_GOOD_STATES)
+        self._drv = driver
+        self.good_states = good_states
+
+    def get_deadtime(self, exposure: float | None) -> float:
+        if exposure is None:
+            return MIN_DEAD_TIME
+        return exposure + MIN_DEAD_TIME
+
+    async def prepare(self, trigger_info: TriggerInfo):
+        if trigger_info.livetime is not None:
+            await adcore.set_exposure_time_and_acquire_period_if_supplied(
+                self, self._drv, trigger_info.livetime
+            )
+        await asyncio.gather(
+            self._drv.trigger_mode.set(self._get_trigger_mode(trigger_info.trigger)),
+            self._drv.num_images.set(
+                DEFAULT_MAX_NUM_IMAGE
+                if trigger_info.total_number_of_triggers == 0
+                else trigger_info.total_number_of_triggers
+            ),
+            self._drv.image_mode.set(ImageMode.multiple),
+        )
+
+    async def arm(self) -> None:
+        # Standard arm the detector and wait for the acquire PV to be True
+        self._arm_status = await adcore.start_acquiring_driver_and_ensure_status(
+            self._drv
+        )
+
+    async def wait_for_idle(self):
+        if self._arm_status:
+            await self._arm_status
+
+    @classmethod
+    def _get_trigger_mode(cls, trigger: DetectorTrigger) -> Andor2TriggerMode:
+        if trigger not in cls._supported_trigger_types.keys():
+            raise ValueError(
+                f"{cls.__name__} only supports the following trigger "
+                f"types: {cls._supported_trigger_types.keys()} but was asked to "
+                f"use {trigger}"
+            )
+        return cls._supported_trigger_types[trigger]
+
+    async def disarm(self):
+        await stop_busy_record(self._drv.acquire, False, timeout=1)

src/dodal/devices/areadetector/andor2_epics/andor2_io.py

@@ -0,0 +1,52 @@
+from enum import Enum
+
+from ophyd_async.epics.adcore._core_io import ADBaseIO
+from ophyd_async.epics.signal import (
+    epics_signal_r,
+    epics_signal_rw,
+    epics_signal_rw_rbv,
+)
+
+
+class Andor2TriggerMode(str, Enum):
+    internal = "Internal"
+    ext_trigger = "External"
+    ext_start = "External Start"
+    ext_exposure = "External Exposure"
+    ext_FVP = "External FVP"
+    soft = "Software"
+
+
+class ImageMode(str, Enum):
+    single = "Single"
+    multiple = "Multiple"
+    continuous = "Continuous"
+    fast_kinetics = "Fast Kinetics"
+
+
+class ADBaseDataType(str, Enum):
+    UInt16 = "UInt16"
+    UInt32 = "UInt32"
+    Float32 = "Float32"
+    Float64 = "Float64"
+    _ = ""
+    """
+    #This is needed due to all empty enum from epics will reduce to a single
+    "" and throw a typeerror as the blank "" is unmatched.
+    """
+
+
+class Andor2DriverIO(ADBaseIO):
+    """
+    Epics pv for andor model:DU897_BV as deployed on p99
+    """
+
+    def __init__(self, prefix: str) -> None:
+        super().__init__(prefix)
+        self.trigger_mode = epics_signal_rw(Andor2TriggerMode, prefix + "TriggerMode")
+        self.data_type = epics_signal_r(ADBaseDataType, prefix + "DataType_RBV")
+        self.accumulate_period = epics_signal_r(
+            float, prefix + "AndorAccumulatePeriod_RBV"
+        )
+        self.image_mode = epics_signal_rw_rbv(ImageMode, prefix + "ImageMode")
+        self.stat_mean = epics_signal_r(int, prefix[:-4] + "STAT:MeanValue_RBV")