Merge pull request #21 from DiamondLightSource/p45AndAdsim

Add configs for P45 and simulated AreaDetector

Author: callumforrester

pyproject.toml

@@ -13,7 +13,7 @@ classifiers = [
     "Programming Language :: Python :: 3.11",
 ]
 description = "Ophyd devices and other utils that could be used across DLS beamlines"
-dependencies = ["ophyd", "bluesky", "pyepics"]
+dependencies = ["ophyd", "bluesky", "pyepics", "pydantic"]
 dynamic = ["version"]
 license.file = "LICENSE"
 readme = "README.rst"

src/dodal/adsim.py

@@ -0,0 +1,22 @@
+import socket
+
+from pydantic import BaseSettings
+
+from dodal.devices.adsim import SimStage
+from dodal.devices.areadetector import AdSimDetector
+
+
+# Settings can be customized via environment variables
+class Settings(BaseSettings):
+    pv_prefix: str = socket.gethostname().split(".")[0]
+
+
+_settings = Settings()
+
+
+def stage(name: str = "sim_motors") -> SimStage:
+    return SimStage(name=name, prefix=f"{_settings.pv_prefix}-MO-SIM-01:")
+
+
+def det(name: str = "adsim") -> AdSimDetector:
+    return AdSimDetector(name=name, prefix=f"{_settings.pv_prefix}-AD-SIM-01:")

src/dodal/devices/adsim.py

@@ -0,0 +1,13 @@
+from ophyd import Component, EpicsMotor, MotorBundle
+
+
+class SimStage(MotorBundle):
+    """
+    ADSIM EPICS motors
+    """
+
+    x: EpicsMotor = Component(EpicsMotor, "M1")
+    y: EpicsMotor = Component(EpicsMotor, "M2")
+    z: EpicsMotor = Component(EpicsMotor, "M3")
+    theta: EpicsMotor = Component(EpicsMotor, "M4")
+    load: EpicsMotor = Component(EpicsMotor, "M5")

src/dodal/devices/areadetector/__init__.py

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

src/dodal/devices/areadetector/adaravis.py

@@ -0,0 +1,50 @@
+from ophyd import Component as Cpt
+from ophyd import DetectorBase
+from ophyd.areadetector.base import ADComponent as Cpt
+from ophyd.areadetector.detectors import DetectorBase
+
+from .adutils import Hdf5Writer, SingleTriggerV33, SynchronisedAdDriverBase
+
+_ACQUIRE_BUFFER_PERIOD = 0.2
+
+
+class AdAravisDetector(SingleTriggerV33, DetectorBase):
+    cam: SynchronisedAdDriverBase = Cpt(SynchronisedAdDriverBase, suffix="DET:")
+    hdf: Hdf5Writer = Cpt(
+        Hdf5Writer,
+        suffix="HDF5:",
+        root="",
+        write_path_template="",
+    )
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.hdf.kind = "normal"
+
+        self.stage_sigs = {
+            # Get stage to wire up the plugins
+            self.hdf.nd_array_port: self.cam.port_name.get(),
+            # Reset array counter on stage
+            self.cam.array_counter: 0,
+            # Set image mode to multiple on stage so we have the option, can still
+            # set num_images to 1
+            self.cam.image_mode: "Multiple",
+            # For now, this Ophyd device does not support hardware
+            # triggered scanning, disable on stage
+            self.cam.trigger_mode: "Off",
+            **self.stage_sigs,  # type: ignore
+        }
+
+    def stage(self, *args, **kwargs):
+        # We have to manually set the acquire period bcause the EPICS driver
+        # doesn't do it for us. If acquire time is a staged signal, we use the
+        # stage value to calculate the acquire period, otherwise we perform
+        # a caget and use the current acquire time.
+        if self.cam.acquire_time in self.stage_sigs:
+            acquire_time = self.stage_sigs[self.cam.acquire_time]
+        else:
+            acquire_time = self.cam.acquire_time.get()
+        self.stage_sigs[self.cam.acquire_period] = acquire_time + _ACQUIRE_BUFFER_PERIOD
+
+        # Now calling the super method should set the acquire period
+        super(AdAravisDetector, self).stage(*args, **kwargs)

src/dodal/devices/areadetector/adsim.py

@@ -0,0 +1,50 @@
+from ophyd import Component as Cpt
+from ophyd.areadetector.base import ADComponent as Cpt
+from ophyd.areadetector.detectors import DetectorBase
+
+from .adutils import Hdf5Writer, SingleTriggerV33, SynchronisedAdDriverBase
+
+
+class AdSimDetector(SingleTriggerV33, DetectorBase):
+    cam: SynchronisedAdDriverBase = Cpt(
+        SynchronisedAdDriverBase, suffix="CAM:", lazy=True
+    )
+    hdf: Hdf5Writer = Cpt(
+        Hdf5Writer,
+        suffix="HDF5:",
+        root="",
+        write_path_template="",
+        lazy=True,
+    )
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__(*args, **kwargs)
+        self.hdf.kind = "normal"
+
+        self.stage_sigs = {
+            # Get stage to wire up the plugins
+            self.hdf.nd_array_port: self.cam.port_name.get(),
+            # Reset array counter on stage
+            self.cam.array_counter: 0,
+            # Set image mode to multiple on stage so we have the option, can still
+            # set num_images to 1
+            self.cam.image_mode: "Multiple",
+            # For now, this Ophyd device does not support hardware
+            # triggered scanning, disable on stage
+            self.cam.trigger_mode: "Internal",
+            **self.stage_sigs,  # type: ignore
+        }
+
+    def stage(self, *args, **kwargs):
+        # We have to manually set the acquire period bcause the EPICS driver
+        # doesn't do it for us. If acquire time is a staged signal, we use the
+        # stage value to calculate the acquire period, otherwise we perform
+        # a caget and use the current acquire time.
+        if self.cam.acquire_time in self.stage_sigs:
+            acquire_time = self.stage_sigs[self.cam.acquire_time]
+        else:
+            acquire_time = self.cam.acquire_time.get()
+        self.stage_sigs[self.cam.acquire_period] = acquire_time
+
+        # Now calling the super method should set the acquire period
+        super(AdSimDetector, self).stage(*args, **kwargs)

src/dodal/devices/areadetector/adutils.py

@@ -0,0 +1,73 @@
+import time as ttime
+
+from ophyd import Component as Cpt
+from ophyd import EpicsSignal, EpicsSignalRO, Staged
+from ophyd.areadetector import ADTriggerStatus, TriggerBase
+from ophyd.areadetector.cam import AreaDetectorCam
+from ophyd.areadetector.filestore_mixins import FileStoreHDF5, FileStoreIterativeWrite
+from ophyd.areadetector.plugins import HDF5Plugin
+
+
+class SingleTriggerV33(TriggerBase):
+    _status_type = ADTriggerStatus
+
+    def __init__(self, *args, image_name=None, **kwargs):
+        super().__init__(*args, **kwargs)
+        if image_name is None:
+            image_name = "_".join([self.name, "image"])
+        self._image_name = image_name
+
+    def trigger(self):
+        "Trigger one acquisition."
+        if self._staged != Staged.yes:
+            raise RuntimeError(
+                "This detector is not ready to trigger."
+                "Call the stage() method before triggering."
+            )
+
+        self._status = self._status_type(self)
+
+        def _acq_done(*args, **kwargs):
+            # TODO sort out if anything useful in here
+            self._status._finished()
+
+        self._acquisition_signal.put(1, use_complete=True, callback=_acq_done)
+        self.dispatch(self._image_name, ttime.time())
+        return self._status
+
+
+class SynchronisedAdDriverBase(AreaDetectorCam):
+    """
+    Base Ophyd device to control an AreaDetector driver and
+    syncrhonise it on other AreaDetector plugins, even non-blocking ones.
+    """
+
+    adcore_version = Cpt(EpicsSignalRO, "ADCoreVersion_RBV", string=True, kind="config")
+    driver_version = Cpt(EpicsSignalRO, "DriverVersion_RBV", string=True, kind="config")
+    wait_for_plugins = Cpt(EpicsSignal, "WaitForPlugins", string=True, kind="config")
+
+    def stage(self, *args, **kwargs):
+        # Makes the detector allow non-blocking AD plugins but makes Ophyd use
+        # the AcquireBusy PV to determine when an acquisition is complete
+        self.ensure_nonblocking()
+        return super().stage(*args, **kwargs)
+
+    def ensure_nonblocking(self):
+        self.stage_sigs["wait_for_plugins"] = "Yes"
+        for c in self.parent.component_names:
+            cpt = getattr(self.parent, c)
+            if cpt is self:
+                continue
+            if hasattr(cpt, "ensure_nonblocking"):
+                cpt.ensure_nonblocking()
+
+
+class Hdf5Writer(HDF5Plugin, FileStoreHDF5, FileStoreIterativeWrite):
+    """ """
+
+    pool_max_buffers = None
+    file_number_sync = None
+    file_number_write = None
+
+    def get_frames_per_point(self):
+        return self.parent.cam.num_images.get()

src/dodal/devices/p45.py

@@ -0,0 +1,44 @@
+from ophyd import Component as Cpt
+from ophyd import EpicsMotor, MotorBundle
+from ophyd.areadetector.base import ADComponent as Cpt
+
+
+class SampleY(MotorBundle):
+    """
+    Motors for controlling the sample's y position and stretch in the y axis.
+    """
+
+    base: EpicsMotor = Cpt(EpicsMotor, "CS:Y")
+    stretch: EpicsMotor = Cpt(EpicsMotor, "CS:Y:STRETCH")
+    top: EpicsMotor = Cpt(EpicsMotor, "Y:TOP")
+    bottom: EpicsMotor = Cpt(EpicsMotor, "Y:BOT")
+
+
+class SampleTheta(MotorBundle):
+    """
+    Motors for controlling the sample's theta position and skew
+    """
+
+    base: EpicsMotor = Cpt(EpicsMotor, "THETA:POS")
+    skew: EpicsMotor = Cpt(EpicsMotor, "THETA:SKEW")
+    top: EpicsMotor = Cpt(EpicsMotor, "THETA:TOP")
+    bottom: EpicsMotor = Cpt(EpicsMotor, "THETA:BOT")
+
+
+class TomoStageWithStretchAndSkew(MotorBundle):
+    """
+    Grouping of motors for the P45 tomography stage
+    """
+
+    x: EpicsMotor = Cpt(EpicsMotor, "X")
+    y: SampleY = Cpt(SampleY, "")
+    theta: SampleTheta = Cpt(SampleTheta, "")
+
+
+class Choppers(MotorBundle):
+    """
+    Grouping for the P45 chopper motors
+    """
+
+    x: EpicsMotor = Cpt(EpicsMotor, "ENDAT")
+    y: EpicsMotor = Cpt(EpicsMotor, "BISS")

src/dodal/p45.py

@@ -0,0 +1,30 @@
+from pydantic import BaseSettings
+
+from dodal.devices.areadetector import AdAravisDetector
+from dodal.devices.p45 import Choppers, TomoStageWithStretchAndSkew
+
+
+# Settings can be customized via environment variables
+class Settings(BaseSettings):
+    pv_prefix: str = "BL45P"
+
+
+_settings = Settings()
+
+
+def sample_sample(name: str = "sample_stage") -> TomoStageWithStretchAndSkew:
+    return TomoStageWithStretchAndSkew(
+        name=name, prefix=f"{_settings.pv_prefix}-MO-STAGE-01:"
+    )
+
+
+def choppers(name: str = "chopper") -> Choppers:
+    return Choppers(name=name, prefix=f"{_settings.pv_prefix}-MO-CHOP-01:")
+
+
+def det(name: str = "det") -> AdAravisDetector:
+    return AdAravisDetector(name=name, prefix=f"{_settings.pv_prefix}-EA-MAP-01:")
+
+
+def diff(name: str = "diff") -> AdAravisDetector:
+    return AdAravisDetector(name=name, prefix=f"{_settings.pv_prefix}-EA-DIFF-01:")

src/dodal/util.py

@@ -0,0 +1,66 @@
+import inspect
+from importlib import import_module
+from inspect import signature
+from types import ModuleType
+from typing import Any, Callable, Dict, Iterable, Type, Union
+
+from bluesky.protocols import (
+    Checkable,
+    Configurable,
+    Flyable,
+    HasHints,
+    HasName,
+    HasParent,
+    Movable,
+    Pausable,
+    Readable,
+    Stageable,
+    Stoppable,
+    Subscribable,
+    Triggerable,
+    WritesExternalAssets,
+)
+
+#: Protocols defining interface to hardware
+BLUESKY_PROTOCOLS = [
+    Checkable,
+    Flyable,
+    HasHints,
+    HasName,
+    HasParent,
+    Movable,
+    Pausable,
+    Readable,
+    Stageable,
+    Stoppable,
+    Subscribable,
+    WritesExternalAssets,
+    Configurable,
+    Triggerable,
+]
+
+
+def make_all_devices(module: Union[str, ModuleType, None] = None) -> Dict[str, Any]:
+    if isinstance(module, str) or module is None:
+        module = import_module(module or __name__)
+    factories = collect_factories(module)
+    return {device.name: device for device in map(lambda factory: factory(), factories)}
+
+
+def collect_factories(module: ModuleType) -> Iterable[Callable[..., Any]]:
+    for var in module.__dict__.values():
+        if callable(var) and is_device_factory(var):
+            yield var
+
+
+def is_device_factory(func: Callable[..., Any]) -> bool:
+    return_type = signature(func).return_annotation
+    return is_device_type(return_type)
+
+
+def is_device_type(obj: Type[Any]) -> bool:
+    is_class = inspect.isclass(obj)
+    follows_protocols = any(
+        map(lambda protocol: isinstance(obj, protocol), BLUESKY_PROTOCOLS)
+    )
+    return is_class and follows_protocols