Merlin ophyd-async integration

src/cditools/merlin.py

@@ -129,10 +129,11 @@ class CDIMerlinDetector(CDIModalTrigger, MerlinDetector):
         "HDF1:",
         read_attrs=[],
         configuration_attrs=[],
-        write_path_template="/nsls2/data/tst/legacy/mock-proposals/2025-2/pass-56789/assets/merlin/%Y/%m/%d",
-        root="/nsls2/data/tst/legacy/mock-proposals/2025-2/pass-56789/assets/merlin",
+        write_path_template = '',
+        root="/nsls2/data/cdi/proposals/",
     )
 
+    _asset_path = "merlinES-1"
     proc1 = Cpt(ProcessPlugin, "Proc1:")
     stats1 = Cpt(StatsPlugin, "Stats1:")
     stats2 = Cpt(StatsPlugin, "Stats2:")
@@ -169,17 +170,24 @@ def __init__(
             **kwargs,
         )
 
+    def _update_paths(self):
+        self.write_path_template = self.root_path_str + "%Y/%m/%d/"
+        self.read_path_template = self.root_path_str + "%Y/%m/%d/"
+        self.reg_root = self.root_path_str
+
+    @property
+    def root_path_str(self):
+        return f"{self.root_str}/{self._md['cycle']}/{self._md['data_session']}/assets/{self._asset_path}"
+
     def mode_internal(self) -> None:
         super().mode_internal()
-
         count_time = self.count_time.get()
         if isinstance(count_time, float):
             self.stage_sigs[self.cam.acquire_time] = count_time
             self.stage_sigs[self.cam.acquire_period] = count_time + 0.005
 
     def mode_external(self) -> None:
         super().mode_external()
-
         # NOTE: these values specify a debounce time for external triggering so
         #       they should be set to < 0.5 the expected exposure time, or at
         #       minimum the lowest possible dead time = 1.64ms
@@ -189,3 +197,14 @@ def mode_external(self) -> None:
         self.stage_sigs[self.cam.acquire_period] = expected_exposure + min_dead_time
 
         self.cam.stage_sigs[self.cam.trigger_mode] = "Trigger Enable"
+
+    def stage(self):
+        self._update_paths()
+        _TIMEOUT = 2
+        self.cam.array_counter.set(0, timeout=_TIMEOUT).wait()
+        self.stage_sigs[self.cam.trigger_mode] = 0
+
+        return super().stage()
+
+    def unstage(self):
+        return super().unstage()

src/cditools/merlin_async.py

@@ -0,0 +1,140 @@
+"""
+Ophyd Async implementation for the Merlin Detector
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Annotated as A
+
+from ophyd_async.core import (
+    DetectorTriggerLogic,
+    PathProvider,
+    SignalDict,
+    SignalR,
+    SignalRW,
+    StrictEnum,
+    SubsetEnum,
+    soft_signal_rw
+)
+from ophyd_async.epics.adcore import (
+    ADArmLogic,
+    ADBaseIO,
+    ADWriterType,
+    AreaDetector,
+    NDPluginBaseIO,
+    ADBaseDataType,
+    prepare_exposures,
+    trigger_info_from_num_images,
+)
+from ophyd_async.epics.core import PvSuffix
+
+__all__ = [
+    "MerlinDetector",
+    "MerlinDriverIO",
+    "MerlinTriggerLogic",
+]
+
+_MIN_DEAD_TIME = 0.002
+
+
+class MerlinTriggerMode(SubsetEnum):
+    """Trigger modes for the Merlin detector"""
+
+    INTERNAL = "Internal"
+    TRIGGER_ENABLE = "Trigger Enable"
+    TRIGGER_START_RISING = "Trigger start rising"
+    TRIGGER_START_FALLING = "Trigger start falling"
+    TRIGGER_BOTH_RISING = "Trigger both rising"
+    SOFTWARE = "Software"
+
+# class MerlinTriggerModeRBV(SubsetEnum):
+#     """Trigger modes for the Merlin detector"""
+
+#     INTERNAL = "Internal"
+#     TRIGGER_ENABLE = "Trigger Enable"
+#     TRIGGER_START_RISING = "Trigger start rising"
+#     TRIGGER_START_FALLING = "Trigger start falling"
+#     TRIGGER_BOTH_RISING = "Trigger both rising "
+#     SOFTWARE = "Software"
+
+class MerlinDriverIO(ADBaseIO):
+    """Driver for merlin model:DU897_BV as deployed on p99.
+
+    This mirrors the interface provided by ADMerlin/db/merlin.template.
+    https://github.com/areaDetector/ADMerlin/blob/master/merlinApp/Db/merlin.template
+    """
+
+    trigger_mode: A[SignalRW[MerlinTriggerMode], PvSuffix.rbv("TriggerMode")]
+
+    # Since ADMerlin doesn't set the data type readback correctly, but is always uint16, 
+    # just turn it into a static soft signal
+    def __init__(self, prefix: str, name: str = ""):
+        super().__init__(prefix, name=name)
+        self.data_type = soft_signal_rw(ADBaseDataType, ADBaseDataType.UINT16, name="data_type")
+
+
+# The deadtime of an Merlin controller varies depending on the exact model of camera.
+# Ideally we would maximize performance by dynamically retrieving the deadtime at
+# runtime. See https://github.com/bluesky/ophyd-async/issues/308
+@dataclass
+class MerlinTriggerLogic(DetectorTriggerLogic):
+    """Trigger logic for MerlinDriverIO."""
+
+    driver: MerlinDriverIO
+
+    def get_deadtime(self, config_values: SignalDict) -> float:  # noqa: ARG002
+        return _MIN_DEAD_TIME
+
+    async def prepare_internal(self, num: int, livetime: float, deadtime: float):
+        await self.driver.trigger_mode.set(MerlinTriggerMode.INTERNAL)
+        await prepare_exposures(self.driver, num, livetime, deadtime)
+
+    async def prepare_edge(self, num: int, livetime: float):
+        # Is this the right trigger mode?
+        await self.driver.trigger_mode.set(MerlinTriggerMode.TRIGGER_START_RISING)
+        await prepare_exposures(self.driver, num, livetime)
+
+    async def default_trigger_info(self):
+        return await trigger_info_from_num_images(self.driver)
+
+
+class MerlinDetector(AreaDetector[MerlinDriverIO]):
+    """Create an ADMerlin AreaDetector instance.
+
+    :param prefix: EPICS PV prefix for the detector
+    :param path_provider: Provider for file paths during acquisition
+    :param driver_suffix: Suffix for the driver PV, defaults to "cam1:"
+    :param writer_type: Type of file writer (HDF or TIFF)
+    :param writer_suffix: Suffix for the writer PV
+    :param plugins: Additional areaDetector plugins to include
+    :param config_sigs: Additional signals to include in configuration
+    :param name: Name for the detector device
+    """
+
+    def __init__(
+        self,
+        prefix: str,
+        path_provider: PathProvider | None = None,
+        driver_suffix="cam1:",
+        writer_type: ADWriterType | None = ADWriterType.HDF,
+        writer_suffix: str | None = None,
+        plugins: dict[str, NDPluginBaseIO] | None = None,
+        config_sigs: Sequence[SignalR] = (),
+        name: str = "",
+    ) -> None:
+        driver = MerlinDriverIO(prefix + driver_suffix)
+        super().__init__(
+            prefix=prefix,
+            driver=driver,
+            arm_logic=ADArmLogic(driver),
+            trigger_logic=MerlinTriggerLogic(driver),
+            path_provider=path_provider,
+            writer_type=writer_type,
+            writer_suffix=writer_suffix,
+            plugins=plugins,
+            config_sigs=config_sigs,
+            name=name,
+        )
+