282-standardise-handling-of-read-config-and-hinted-signals-for-standa…

…rddetector (#468)

#282

src/ophyd_async/epics/adcore/__init__.py

@@ -12,7 +12,9 @@
     FileWriteMode,
     ImageMode,
     NDAttributeDataType,
-    NDAttributesXML,
+    NDAttributeParam,
+    NDAttributePv,
+    NDAttributePvDbrType,
     stop_busy_record,
 )
 
@@ -30,7 +32,9 @@
     "ADBaseDataType",
     "FileWriteMode",
     "ImageMode",
+    "NDAttributePv",
+    "NDAttributeParam",
     "NDAttributeDataType",
-    "NDAttributesXML",
     "stop_busy_record",
+    "NDAttributePvDbrType",
 ]

src/ophyd_async/epics/adcore/_core_io.py

@@ -41,7 +41,29 @@ def __init__(self, prefix: str, name: str = "") -> None:
 
 
 class NDPluginStatsIO(NDPluginBaseIO):
-    pass
+    """
+    Plugin for computing statistics from an image or region of interest within an image.
+    Each boolean signal enables or disables all signals in the appropriate Enum class.
+    The enum signals may used in the ScalarSignals kwargs of a HDFWriter, and are also
+    read-only signals on the plugin.
+    """
+
+    def __init__(self, prefix: str, name: str = "") -> None:
+        self.total = epics_signal_rw(float, prefix + "TotalArray")
+        self.statistics = epics_signal_rw(bool, prefix + "ComputeStatistics")
+        self.statistics_background_width = epics_signal_rw(int, prefix + "BgdWidth")
+        self.centroid = epics_signal_rw(bool, prefix + "ComputeCentroid")
+        self.centroid_threshold = epics_signal_rw(float, prefix + "CentroidThreshold")
+        self.profiles = epics_signal_rw(bool, prefix + "ComputeProfiles")
+        self.profile_size_x = epics_signal_rw(int, prefix + "ProfileSizeX")
+        self.profile_cursor_x = epics_signal_rw(int, prefix + "CursorX")
+        self.profile_size_y = epics_signal_rw(int, prefix + "ProfileSizeY")
+        self.profile_cursor_y = epics_signal_rw(int, prefix + "CursorY")
+        self.histogram = epics_signal_rw(bool, prefix + "ComputeHistogram")
+        self.histogram_max = epics_signal_rw(float, prefix + "HistMax")
+        self.histogram_min = epics_signal_rw(float, prefix + "HistMin")
+        self.histogram_size = epics_signal_rw(int, prefix + "HistSize")
+        super().__init__(prefix, name)
 
 
 class DetectorState(str, Enum):

src/ophyd_async/epics/adcore/_hdf_writer.py

@@ -1,6 +1,7 @@
 import asyncio
 from pathlib import Path
 from typing import AsyncGenerator, AsyncIterator, Dict, List, Optional
+from xml.etree import ElementTree as ET
 
 from bluesky.protocols import DataKey, Hints, StreamAsset
 
@@ -18,8 +19,13 @@
     wait_for_value,
 )
 
-from ._core_io import NDFileHDFIO
-from ._utils import FileWriteMode, convert_ad_dtype_to_np
+from ._core_io import NDArrayBaseIO, NDFileHDFIO
+from ._utils import (
+    FileWriteMode,
+    convert_ad_dtype_to_np,
+    convert_param_dtype_to_np,
+    convert_pv_dtype_to_np,
+)
 
 
 class ADHDFWriter(DetectorWriter):
@@ -29,13 +35,14 @@ def __init__(
         path_provider: PathProvider,
         name_provider: NameProvider,
         shape_provider: ShapeProvider,
-        **scalar_datasets_paths: str,
+        *plugins: NDArrayBaseIO,
     ) -> None:
         self.hdf = hdf
         self._path_provider = path_provider
         self._name_provider = name_provider
         self._shape_provider = shape_provider
-        self._scalar_datasets_paths = scalar_datasets_paths
+
+        self._plugins = plugins
         self._capture_status: Optional[AsyncStatus] = None
         self._datasets: List[HDFDataset] = []
         self._file: Optional[HDFFile] = None
@@ -90,16 +97,31 @@ async def open(self, multiplier: int = 1) -> Dict[str, DataKey]:
             )
         ]
         # And all the scalar datasets
-        for ds_name, ds_path in self._scalar_datasets_paths.items():
-            self._datasets.append(
-                HDFDataset(
-                    f"{name}-{ds_name}",
-                    f"/entry/instrument/NDAttributes/{ds_path}",
-                    (),
-                    "",
-                    multiplier,
-                )
-            )
+        for plugin in self._plugins:
+            maybe_xml = await plugin.nd_attributes_file.get_value()
+            # This is the check that ADCore does to see if it is an XML string
+            # rather than a filename to parse
+            if "<Attributes>" in maybe_xml:
+                root = ET.fromstring(maybe_xml)
+                for child in root:
+                    datakey = child.attrib["name"]
+                    if child.attrib.get("type", "EPICS_PV") == "EPICS_PV":
+                        np_datatye = convert_pv_dtype_to_np(
+                            child.attrib.get("dbrtype", "DBR_NATIVE")
+                        )
+                    else:
+                        np_datatye = convert_param_dtype_to_np(
+                            child.attrib.get("datatype", "INT")
+                        )
+                    self._datasets.append(
+                        HDFDataset(
+                            datakey,
+                            f"/entry/instrument/NDAttributes/{datakey}",
+                            (),
+                            np_datatye,
+                            multiplier,
+                        )
+                    )
 
         describe = {
             ds.data_key: DataKey(
@@ -148,8 +170,8 @@ async def collect_stream_docs(
 
     async def close(self):
         # Already done a caput callback in _capture_status, so can't do one here
-        await self.hdf.capture.set(0, wait=False)
-        await wait_for_value(self.hdf.capture, 0, DEFAULT_TIMEOUT)
+        await self.hdf.capture.set(False, wait=False)
+        await wait_for_value(self.hdf.capture, False, DEFAULT_TIMEOUT)
         if self._capture_status:
             # We kicked off an open, so wait for it to return
             await self._capture_status

src/ophyd_async/epics/adcore/_utils.py

@@ -1,8 +1,9 @@
+from dataclasses import dataclass
 from enum import Enum
 from typing import Optional
-from xml.etree import cElementTree as ET
 
 from ophyd_async.core import DEFAULT_TIMEOUT, SignalRW, T, wait_for_value
+from ophyd_async.core._signal import SignalR
 
 
 class ADBaseDataType(str, Enum):
@@ -16,6 +17,7 @@ class ADBaseDataType(str, Enum):
     UInt64 = "UInt64"
     Float32 = "Float32"
     Float64 = "Float64"
+    Double = "DOUBLE"
 
 
 def convert_ad_dtype_to_np(ad_dtype: ADBaseDataType) -> str:
@@ -34,6 +36,45 @@ def convert_ad_dtype_to_np(ad_dtype: ADBaseDataType) -> str:
     return ad_dtype_to_np_dtype[ad_dtype]
 
 
+def convert_pv_dtype_to_np(datatype: str) -> str:
+    _pvattribute_to_ad_datatype = {
+        "DBR_SHORT": ADBaseDataType.Int16,
+        "DBR_ENUM": ADBaseDataType.Int16,
+        "DBR_INT": ADBaseDataType.Int32,
+        "DBR_LONG": ADBaseDataType.Int32,
+        "DBR_FLOAT": ADBaseDataType.Float32,
+        "DBR_DOUBLE": ADBaseDataType.Float64,
+    }
+    if datatype in ["DBR_STRING", "DBR_CHAR"]:
+        np_datatype = "s40"
+    elif datatype == "DBR_NATIVE":
+        raise ValueError("Don't support DBR_NATIVE yet")
+    else:
+        try:
+            np_datatype = convert_ad_dtype_to_np(_pvattribute_to_ad_datatype[datatype])
+        except KeyError:
+            raise ValueError(f"Invalid dbr type {datatype}")
+    return np_datatype
+
+
+def convert_param_dtype_to_np(datatype: str) -> str:
+    _paramattribute_to_ad_datatype = {
+        "INT": ADBaseDataType.Int32,
+        "INT64": ADBaseDataType.Int64,
+        "DOUBLE": ADBaseDataType.Float64,
+    }
+    if datatype in ["STRING"]:
+        np_datatype = "s40"
+    else:
+        try:
+            np_datatype = convert_ad_dtype_to_np(
+                _paramattribute_to_ad_datatype[datatype]
+            )
+        except KeyError:
+            raise ValueError(f"Invalid datatype {datatype}")
+    return np_datatype
+
+
 class FileWriteMode(str, Enum):
     single = "Single"
     capture = "Capture"
@@ -52,75 +93,34 @@ class NDAttributeDataType(str, Enum):
     STRING = "STRING"
 
 
-class NDAttributesXML:
-    """Helper to make NDAttributesFile XML for areaDetector"""
-
-    _dbr_types = {
-        None: "DBR_NATIVE",
-        NDAttributeDataType.INT: "DBR_LONG",
-        NDAttributeDataType.DOUBLE: "DBR_DOUBLE",
-        NDAttributeDataType.STRING: "DBR_STRING",
-    }
-
-    def __init__(self):
-        self._root = ET.Element("Attributes")
-
-    def add_epics_pv(
-        self,
-        name: str,
-        pv: str,
-        datatype: Optional[NDAttributeDataType] = None,
-        description: str = "",
-    ):
-        """Add a PV to the attribute list
-
-        Args:
-            name: The attribute name
-            pv: The pv to get from
-            datatype: An override datatype, otherwise will use native EPICS type
-            description: A description that appears in the HDF file as an attribute
-        """
-        ET.SubElement(
-            self._root,
-            "Attribute",
-            name=name,
-            type="EPICS_PV",
-            source=pv,
-            datatype=self._dbr_types[datatype],
-            description=description,
-        )
-
-    def add_param(
-        self,
-        name: str,
-        param: str,
-        datatype: NDAttributeDataType,
-        addr: int = 0,
-        description: str = "",
-    ):
-        """Add a driver or plugin parameter to the attribute list
-
-        Args:
-            name: The attribute name
-            param: The parameter string as seen in the INP link of the record
-            datatype: The datatype of the parameter
-            description: A description that appears in the HDF file as an attribute
-        """
-        ET.SubElement(
-            self._root,
-            "Attribute",
-            name=name,
-            type="PARAM",
-            source=param,
-            addr=str(addr),
-            datatype=datatype.value,
-            description=description,
-        )
-
-    def __str__(self) -> str:
-        """Output the XML pretty printed"""
-        ET.indent(self._root, space="    ", level=0)
-        return ET.tostring(self._root, xml_declaration=True, encoding="utf-8").decode()
+class NDAttributePvDbrType(str, Enum):
+    DBR_SHORT = "DBR_SHORT"
+    DBR_ENUM = "DBR_ENUM"
+    DBR_INT = "DBR_INT"
+    DBR_LONG = "DBR_LONG"
+    DBR_FLOAT = "DBR_FLOAT"
+    DBR_DOUBLE = "DBR_DOUBLE"
+    DBR_STRING = "DBR_STRING"
+    DBR_CHAR = "DBR_CHAR"
+
+
+@dataclass
+class NDAttributePv:
+    name: str  # name of attribute stamped on array, also scientifically useful name
+    # when appended to device.name
+    signal: SignalR  # caget the pv given by signal.source and attach to each frame
+    dbrtype: NDAttributePvDbrType
+    description: str = ""  # A description that appears in the HDF file as an attribute
+
+
+@dataclass
+class NDAttributeParam:
+    name: str  # name of attribute stamped on array, also scientifically useful name
+    # when appended to device.name
+    param: str  # The parameter string as seen in the INP link of the record
+    datatype: NDAttributeDataType  # The datatype of the parameter
+    addr: int = 0  # The address as seen in the INP link of the record
+    description: str = ""  # A description that appears in the HDF file as an attribute
 
 
 async def stop_busy_record(

src/ophyd_async/plan_stubs/__init__.py

@@ -4,10 +4,13 @@
     prepare_static_seq_table_flyer_and_detectors_with_same_trigger,
     time_resolved_fly_and_collect_with_static_seq_table,
 )
+from ._nd_attributes import setup_ndattributes, setup_ndstats_sum
 
 __all__ = [
     "fly_and_collect",
     "prepare_static_seq_table_flyer_and_detectors_with_same_trigger",
     "time_resolved_fly_and_collect_with_static_seq_table",
     "ensure_connected",
+    "setup_ndattributes",
+    "setup_ndstats_sum",
 ]

src/ophyd_async/plan_stubs/_nd_attributes.py

@@ -0,0 +1,63 @@
+from typing import Sequence
+from xml.etree import cElementTree as ET
+
+import bluesky.plan_stubs as bps
+
+from ophyd_async.core._device import Device
+from ophyd_async.epics.adcore._core_io import NDArrayBaseIO
+from ophyd_async.epics.adcore._utils import (
+    NDAttributeDataType,
+    NDAttributeParam,
+    NDAttributePv,
+)
+
+
+def setup_ndattributes(
+    device: NDArrayBaseIO, ndattributes: Sequence[NDAttributePv | NDAttributeParam]
+):
+    xml_text = ET.Element("Attributes")
+
+    for ndattribute in ndattributes:
+        if isinstance(ndattribute, NDAttributeParam):
+            ET.SubElement(
+                xml_text,
+                "Attribute",
+                name=ndattribute.name,
+                type="PARAM",
+                source=ndattribute.param,
+                addr=str(ndattribute.addr),
+                datatype=ndattribute.datatype.value,
+                description=ndattribute.description,
+            )
+        elif isinstance(ndattribute, NDAttributePv):
+            ET.SubElement(
+                xml_text,
+                "Attribute",
+                name=ndattribute.name,
+                type="EPICS_PV",
+                source=ndattribute.signal.source.split("ca://")[-1],
+                dbrtype=ndattribute.dbrtype.value,
+                description=ndattribute.description,
+            )
+        else:
+            raise ValueError(
+                f"Invalid type for ndattributes: {type(ndattribute)}. "
+                "Expected NDAttributePv or NDAttributeParam."
+            )
+    yield from bps.mv(device.nd_attributes_file, xml_text)
+
+
+def setup_ndstats_sum(detector: Device):
+    yield from (
+        setup_ndattributes(
+            detector.hdf,
+            [
+                NDAttributeParam(
+                    name=f"{detector.name}-sum",
+                    param="NDPluginStatsTotal",
+                    datatype=NDAttributeDataType.DOUBLE,
+                    description="Sum of the array",
+                )
+            ],
+        )
+    )