initial commit

src/ophyd_async/epics/areadetector/__init__.py

@@ -6,7 +6,6 @@
     FileWriteMode,
     ImageMode,
     NDAttributeDataType,
-    NDAttributesXML,
 )
 from .vimba import VimbaDetector
 

src/ophyd_async/epics/areadetector/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 FileWriteMode(str, Enum):
@@ -23,75 +24,25 @@ 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()
+@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
+    datatype: Optional[NDAttributeDataType] = (
+        None  # An override datatype, otherwise will use native EPICS type
+    )
+    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/epics/areadetector/writers/hdf_writer.py

@@ -1,4 +1,5 @@
 import asyncio
+from collections.abc import Sequence
 from pathlib import Path
 from typing import AsyncGenerator, AsyncIterator, Dict, List, Optional
 
@@ -18,7 +19,7 @@
 
 from .general_hdffile import _HDFDataset, _HDFFile
 from .nd_file_hdf import FileWriteMode, NDFileHDF
-from .nd_plugin import convert_ad_dtype_to_np
+from .nd_plugin import NDArrayBase, convert_ad_dtype_to_np
 
 
 class HDFWriter(DetectorWriter):
@@ -28,13 +29,15 @@ def __init__(
         path_provider: PathProvider,
         name_provider: NameProvider,
         shape_provider: ShapeProvider,
-        **scalar_datasets_paths: str,
+        plugins: Sequence[NDArrayBase],
+        # **scalar_datasets_paths: str,
     ) -> 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._scalar_datasets_paths = scalar_datasets_paths
         self._capture_status: Optional[AsyncStatus] = None
         self._datasets: List[_HDFDataset] = []
         self._file: Optional[_HDFFile] = None
@@ -89,7 +92,7 @@ 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():
+        for ds_name, ds_path in self.plugins.items():
             self._datasets.append(
                 _HDFDataset(
                     f"{name}-{ds_name}",

src/ophyd_async/epics/areadetector/writers/nd_plugin.py

@@ -65,4 +65,25 @@ def __init__(self, prefix: str, name: str = "") -> None:
 
 
 class NDPluginStats(NDPluginBase):
-    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.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)

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,48 @@
+from typing import Sequence
+from xml.etree import cElementTree as ET
+
+import bluesky.plan_stubs as bps
+
+from ophyd_async.epics.areadetector.utils import (
+    NDAttributeDataType,
+    NDAttributeParam,
+    NDAttributePv,
+)
+from ophyd_async.epics.areadetector.writers.nd_plugin import NDArrayBase, NDPluginStats
+
+
+def setup_ndattributes(device: NDArrayBase,
+                       ndattributes: Sequence[NDAttributePv | NDAttributeParam]):
+    xml_text = ET.Element("Attributes")
+    _dbr_types = {
+        None: "DBR_NATIVE",
+        NDAttributeDataType.INT: "DBR_LONG",
+        NDAttributeDataType.DOUBLE: "DBR_DOUBLE",
+        NDAttributeDataType.STRING: "DBR_STRING",
+    }
+    if isinstance(ndattributes,NDAttributeParam):
+        ET.SubElement(
+            xml_text,
+            "Attribute",
+            name=ndattributes.name,
+            type="PARAM",
+            source=ndattributes.param,
+            addr=str(ndattributes.addr),
+            datatype=_dbr_types[ndattributes.datatype],
+            description=ndattributes.description,
+        )
+    elif isinstance(ndattributes,NDAttributePv):
+        ET.SubElement(
+            xml_text,
+            "Attribute",
+            name=ndattributes.name,
+            type="EPICS_PV",
+            source=ndattributes.signal.source,
+            datatype=_dbr_types[ndattributes.datatype],
+            description=ndattributes.description,
+        )
+    yield from bps.abs_set(device.nd_attributes_file, xml_text)
+
+def setup_ndstats_sum(stats: NDPluginStats):
+    pass
+    #NDAttributeParam(name=f"{stats.parent.name}-sum", ...)