EPICS: Allow reading partial waveforms (first N elements)

src/ophyd_async/epics/core/_aioca.py

@@ -311,8 +311,17 @@ async def connect(self, timeout: float):
         self.converter = make_converter(self.datatype, self.initial_values)
 
     async def _caget(self, pv: str, format: Format) -> AugmentedValue:
+        # Target: avoid breaking backwards compatability
+        #         perhaps overcautious on how to define element count
+        #         aioca.caget documentation states that element_count=1
+        #         by default. If this is provided values do change
+        # Todo:
+        #         need to check what pvaccess and p4p provide
+        kws = {}
+        if self.options.element_count is not None:
+            kws["count"] = self.options.element_count
         return await caget(
-            pv, datatype=self.converter.read_dbr, format=format, timeout=None
+            pv, datatype=self.converter.read_dbr, format=format, timeout=None, **kws
         )
 
     def _make_reading(self, value: AugmentedValue) -> Reading[SignalDatatypeT]:
@@ -384,12 +393,18 @@ def set_callback(self, callback: Callback[Reading[SignalDatatypeT]] | None) -> N
             self.subscription = None
 
         if callback:
+            # see argumentation of _caget for details
+            kws = {}
+            if self.options.element_count is not None:
+                kws["count"] = self.options.element_count
+
             self.subscription = camonitor(
                 self.read_pv,
                 lambda v: callback(self._make_reading(v)),
                 datatype=self.converter.read_dbr,
                 format=FORMAT_TIME,
                 all_updates=self._all_updates,
+                **kws,
             )
 
 

src/ophyd_async/epics/core/_p4p.py

@@ -371,6 +371,8 @@ def __init__(
         write_pv: str = "",
         options: EpicsOptions | None = None,
     ):
+        # with p4p: single element seems not feasible
+        #           have a look to :meth:`_get_read_pv
         self.converter: PvaConverter = DisconnectedPvaConverter(float)
         self.initial_values: dict[str, Any] = {}
         self.subscription: Subscription | None = None
@@ -383,6 +385,13 @@ async def _store_initial_value(self, pv: str, timeout: float):
         self.initial_values[pv] = await pvget_with_timeout(pv, timeout)
 
     async def connect(self, timeout: float):
+        # options are only available at connetion time
+        if self.options and self.options.element_count is not None:
+            if self.options.element_count <= 1:
+                raise ValueError(
+                    f'"{self.read_pv}": p4p can only support epics option'
+                    " element_count >=2"
+                )
         if self.read_pv != self.write_pv:
             # Different, need to connect both
             await wait_for_connection(
@@ -414,8 +423,14 @@ async def put(self, value: SignalDatatypeT | None):
             wait = self.options.wait
         await context().put(self.write_pv, {"value": write_value}, wait=wait)
 
+    def _get_read_pv(self) -> str:
+        """Read pv with subarray index when requested."""
+        if self.options.element_count is None:
+            return self.read_pv
+        return f"{self.read_pv}.[0:{self.options.element_count - 1:d}]"
+
     async def get_datakey(self, source: str) -> DataKey:
-        value = await context().get(self.read_pv)
+        value = await context().get(self._get_read_pv())
         metadata = _metadata_from_value(self.converter.datatype, value)
         return make_datakey(
             self.converter.datatype, self.converter.value(value), source, metadata
@@ -425,12 +440,12 @@ async def get_reading(self) -> Reading:
         request = _pva_request_string(
             self.converter.value_fields + self.converter.reading_fields
         )
-        value = await context().get(self.read_pv, request=request)
+        value = await context().get(self._get_read_pv(), request=request)
         return self._make_reading(value)
 
     async def get_value(self) -> SignalDatatypeT:
         request = _pva_request_string(self.converter.value_fields)
-        value = await context().get(self.read_pv, request=request)
+        value = await context().get(self._get_read_pv(), request=request)
         return self.converter.value(value)
 
     async def get_setpoint(self) -> SignalDatatypeT:
@@ -456,7 +471,7 @@ async def async_callback(v):
                 self.converter.value_fields + self.converter.reading_fields
             )
             self.subscription = context().monitor(
-                self.read_pv, async_callback, request=request
+                self._get_read_pv(), async_callback, request=request
             )
 
 

src/ophyd_async/epics/core/_signal.py

@@ -160,6 +160,7 @@ def epics_signal_r(
     read_pv: str,
     name: str = "",
     timeout: float = DEFAULT_TIMEOUT,
+    element_count: int | None = None,
 ) -> SignalR[SignalDatatypeT]:
     """Create a `SignalR` backed by 1 EPICS PV.
 
@@ -168,7 +169,9 @@ def epics_signal_r(
     :param name: The name of the signal (defaults to empty string)
     :param timeout: A timeout to be used when reading (not connecting) this signal
     """
-    backend = _epics_signal_backend(datatype, read_pv, read_pv)
+    backend = _epics_signal_backend(
+        datatype, read_pv, read_pv, options=EpicsOptions(element_count=element_count)
+    )
     return SignalR(backend, name=name, timeout=timeout)
 
 

src/ophyd_async/epics/core/_util.py

@@ -40,6 +40,15 @@ class EpicsOptions(Generic[SignalDatatypeT]):
     as it causes a deadlock.
     """
 
+    element_count: None | int = None
+    """Epics allows to specify the maximum number of elements to transfer
+
+    Fast devices provide buffers that acquire large data sets. Typically
+    one only needs the beginning of this buffer.
+
+    None is used as standard argument: transfer as many elements as provided
+    """
+
 
 def get_pv_basename_and_field(pv: str) -> tuple[str, str | None]:
     """Split PV into record name and field."""

src/ophyd_async/epics/testing/__init__.py

@@ -4,6 +4,7 @@
     CA_PVA_RECORDS,
     PVA_RECORDS,
     EpicsTestCaDevice,
+    EpicsTestCaDeviceInitMustFail,
     EpicsTestEnum,
     EpicsTestPvaDevice,
     EpicsTestPviDevice,
@@ -52,4 +53,5 @@
     "EpicsTestTable",
     "generate_random_pv_prefix",
     "start_ioc",
+    "EpicsTestCaDeviceInitMustFail",
 ]

src/ophyd_async/epics/testing/_devices.py

@@ -9,6 +9,7 @@
     SignalR,
     SignalRW,
     SignalW,
+    StandardReadable,
     StrictEnum,
     SubsetEnum,
     SupersetEnum,
@@ -17,6 +18,7 @@
 )
 from ophyd_async.epics.core import (
     EpicsDevice,
+    EpicsOptions,
     PvSuffix,
 )
 
@@ -66,7 +68,7 @@ class EpicsTestTable(Table):
     a_enum: Sequence[EpicsTestEnum]
 
 
-class EpicsTestCaDevice(EpicsDevice):
+class EpicsTestCaDevice(StandardReadable, EpicsDevice):
     """Device for use in a channel access test IOC."""
 
     a_int: A[SignalRW[int], PvSuffix("int")]
@@ -90,6 +92,15 @@ class EpicsTestCaDevice(EpicsDevice):
     int16a: A[SignalRW[Array1D[np.int16]], PvSuffix("int16a")]
     int32a: A[SignalRW[Array1D[np.int32]], PvSuffix("int32a")]
     float32a: A[SignalRW[Array1D[np.float32]], PvSuffix("float32a")]
+    # todo: check if it can be combined with int32a above
+    #       if its user tolerate a length of 5 with elements set
+    float32al5: A[SignalRW[Array1D[np.float32]], PvSuffix("float32al5")]
+    # need a separate entry to be compatible with signal test
+    float32al5o3: A[
+        SignalRW[Array1D[np.float32]],
+        PvSuffix("float32al5"),
+        EpicsOptions(element_count=3),
+    ]
     float64a: A[SignalRW[Array1D[np.float64]], PvSuffix("float64a")]
     stra: A[SignalRW[Sequence[str]], PvSuffix("stra")]
     mbb_direct_bit_r: A[SignalR[bool], PvSuffix("mbb_direct.B0")]
@@ -145,3 +156,17 @@ class EpicsTestPviDisagreeingSuffixDevice(EpicsDevice):
     """
 
     overridden_float: A[SignalRW[float], PvSuffix("float_prec_1")]
+
+
+class EpicsTestCaDeviceInitMustFail(StandardReadable, EpicsDevice):
+    """Dedicated device for signals that fail at initialisation.
+
+    Don't add them to :class:`EpicsTestCaDevice`. as this will break
+    test_epics_signal_lifecycle
+    """
+
+    float32al5o1: A[
+        SignalRW[Array1D[np.float32]],
+        PvSuffix("float32al5"),
+        EpicsOptions(element_count=1),
+    ]

src/ophyd_async/epics/testing/_epics_test_ca_records.db

@@ -145,6 +145,15 @@ record(waveform, "$(device)float32a") {
   field(PINI, "YES")
 }
 
+# could reuse float32a ... if its ok to extend the array
+# and add different values to the extra fields
+record(waveform, "$(device)float32al5") {
+  field(NELM, "5")
+  field(FTVL, "FLOAT")
+  field(INP, {const:[0.000002, -123.123, 3, 4, 5]})
+  field(PINI, "YES")
+}
+
 record(waveform, "$(device)float64a") {
   field(NELM, "3")
   field(FTVL, "DOUBLE")
@@ -387,6 +396,15 @@ record("*", "$(device)float32a") {
     })
 }
 
+record("*", "$(device)float32al5") {
+    info(Q:group, {
+        "$(device)PVI": {
+            "value.float32al5.rw": {"+channel": "NAME", "+type": "plain"},
+            "value.float32al5o3.rw": {"+channel": "NAME", "+type": "plain"},
+        }
+    })
+}
+
 record("*", "$(device)float64a") {
     info(Q:group, {
         "$(device)PVI": {

tests/system_tests/epics/core/test_epics_signal_mechanisms.py

@@ -67,6 +67,7 @@
 from ophyd_async.epics.testing import (
     IOC,
     EpicsTestCaDevice,
+    EpicsTestCaDeviceInitMustFail,
     EpicsTestEnum,
     EpicsTestPvaDevice,
     EpicsTestPviDevice,
@@ -99,8 +100,8 @@ def __init__(self):
         self.prefix = generate_random_pv_prefix()
         ca_prefix = f"{self.prefix}ca:"
         pva_prefix = f"{self.prefix}pva:"
-        self.ca_device = EpicsTestCaDevice(f"ca://{ca_prefix}")
-        self.pva_device = EpicsTestPvaDevice(f"pva://{pva_prefix}")
+        self.ca_device = EpicsTestCaDevice(f"ca://{ca_prefix}", name="test_ca")
+        self.pva_device = EpicsTestPvaDevice(f"pva://{pva_prefix}", name="test_pva")
         self.pvi_device = EpicsTestPviDevice(pva_prefix, with_pvi=True)
 
     def get_device(self, protocol: str) -> EpicsTestCaDevice | EpicsTestPvaDevice:
@@ -818,3 +819,92 @@ async def test_pvi_adds_undeclared_signal_dynamically(pvi_device: EpicsTestPviDe
     extra_int = pvi_device.extra_int  # type: ignore[attr-defined]
     assert isinstance(extra_int, SignalRW)
     assert await extra_int.get_value() == 42
+
+
+async def wf_verify_data(sig, identifier, expected_len):
+    des = await sig.describe()
+    assert list(des) == [identifier]
+
+    shape = tuple(des[identifier]["shape"])
+    assert shape == (expected_len,)
+
+    t_data = await sig.get_value()
+    assert len(t_data) == expected_len
+
+    data = await sig.read()
+    assert list(data) == [identifier]
+
+    wf_data = data[identifier]["value"]
+    assert len(wf_data) == shape[0]
+
+
+@pytest.mark.parametrize("protocol", get_args(Protocol))
+async def test_waveform_different_length(
+    ioc_devices: MechanismIocAndDevices, protocol: str
+):
+    sig = ioc_devices.get_signal(protocol, "float32al5")
+    await sig.connect()
+    await wf_verify_data(sig, sig.name, expected_len=5)
+
+    sig_lim = ioc_devices.get_signal(protocol, "float32al5o3")
+    await sig_lim.connect()
+    await wf_verify_data(sig_lim, sig_lim.name, expected_len=3)
+
+
+@pytest.mark.parametrize("protocol", get_args(Protocol))
+@pytest.mark.parametrize(
+    "signal_name, expected_len",
+    [
+        ("float32al5", 5),
+        ("float32al5o3", 3),
+    ],
+)
+async def test_waveform_cb(
+    ioc_devices: MechanismIocAndDevices,
+    protocol: str,
+    signal_name: str,
+    expected_len: int,
+):
+    sig = ioc_devices.get_signal(protocol, signal_name)
+    await sig.connect()
+
+    got_len = None
+    event = asyncio.Event()
+
+    def cb(d):
+        nonlocal got_len
+        data = d[f"test_{protocol}-{signal_name}"]["value"]
+        got_len = len(data)
+        event.set()
+
+    sig.subscribe(cb)
+
+    try:
+        await asyncio.wait_for(event.wait(), timeout=4.0)
+    finally:
+        sig.clear_sub(cb)
+
+    assert got_len == expected_len
+
+
+@pytest.mark.parametrize("protocol", get_args(Protocol))
+async def test_waveform_invalid_length(
+    ioc_devices: MechanismIocAndDevices, protocol: str
+):
+    dev = EpicsTestCaDeviceInitMustFail(
+        f"{protocol}://{ioc_devices.prefix}{protocol}:", name="test_fail"
+    )
+    sig = dev.float32al5o1
+    assert sig.source.startswith(f"{protocol}://")
+    if protocol == "ca":
+        await sig.connect()
+    elif protocol == "pva":
+        chk, pv_name = sig.source.split("pva://")
+        assert chk == ""
+        with pytest.raises(
+            ValueError,
+            match=f'"{pv_name}": p4p can only support epics option element_count >=2',
+        ):
+            await sig.connect()
+    else:
+        raise NotImplementedError(f"not handling {protocol}")

tests/system_tests/epics/core/test_yaml_save_ca.yaml

@@ -1,22 +1,24 @@
+a_bool: true
+a_float: 3.141
+a_int: 42
+a_str: hello
 bool_unnamed: true
 enum: Bbb
 enum2: Bbb
 enum_str_fallback: Bbb
-slowseq: 0
 float32a: [1.9999999949504854e-06, -123.12300109863281]
+float32al5: [1.9999999949504854e-06, -123.12300109863281, 3.0]
+float32al5o3: [1.9999999949504854e-06, -123.12300109863281, 3.0]
 float64a: [0.1, -12345678.123]
 float_prec_0: 3
 int16a: [-32768, 32767]
 int32a: [-2147483648, 2147483647]
 lessint: 42
 longstr: a string that is just longer than forty characters
 longstr2: a string that is just longer than forty characters
-mbb_direct_bit: False
-a_bool: true
-a_float: 3.141
-a_int: 42
-a_str: hello
+mbb_direct_bit: false
 partialint: 42
+slowseq: 0
 stra:
 - five
 - six