Implement Waveform

Author: GDYendell

src/fastcs/datatypes/waveform.py

@@ -20,7 +20,7 @@ def initial_value(self) -> np.ndarray:
         return np.zeros(self.shape, dtype=self.array_dtype)
 
     def validate(self, value: np.ndarray) -> np.ndarray:
-        _value = super().validate(value)
+        _value = super().validate(np.asarray(value).astype(self.array_dtype))
 
         if self.array_dtype != _value.dtype:
             raise ValueError(

src/fastcs/demo/controllers.py

@@ -4,10 +4,12 @@
 from dataclasses import KW_ONLY, dataclass
 from typing import TypeVar
 
+import numpy as np
+
 from fastcs.attributes import AttributeIO, AttributeIORef, AttrR, AttrRW, AttrW
 from fastcs.connections import IPConnection, IPConnectionSettings
 from fastcs.controllers import Controller
-from fastcs.datatypes import Enum, Float, Int
+from fastcs.datatypes import Enum, Float, Int, Waveform
 from fastcs.methods import command, scan
 
 NumberT = TypeVar("NumberT", int, float)
@@ -66,6 +68,7 @@ async def update(
 class TemperatureController(Controller):
     ramp_rate = AttrRW(Float(), io_ref=TemperatureControllerAttributeIORef(name="R"))
     power = AttrR(Float(), io_ref=TemperatureControllerAttributeIORef(name="P"))
+    voltages = AttrR(Waveform(np.int32, shape=(4,)))
 
     def __init__(self, settings: TemperatureControllerSettings) -> None:
         self.connection = IPConnection()
@@ -101,6 +104,9 @@ async def update_voltages(self):
         voltages = json.loads(
             (await self.connection.send_query(f"{query}\r\n")).strip("\r\n")
         )
+
+        await self.voltages.update(voltages)
+
         for index, controller in enumerate(self._ramp_controllers):
             self.log_event(
                 "Update voltages",

src/fastcs/transports/epics/gui.py

@@ -1,6 +1,7 @@
 from pvi._format.dls import DLSFormatter  # type: ignore
 from pvi.device import (
     LED,
+    ArrayTrace,
     ButtonPanel,
     ComboBox,
     ComponentUnion,
@@ -66,8 +67,12 @@ def _get_read_widget(self, fastcs_datatype: DataType) -> ReadWidgetUnion | None:
                 return TextRead(format=TextFormat.string)
             case Enum():
                 return TextRead(format=TextFormat.string)
-            case Waveform():
-                return None
+            case Waveform() as waveform:
+                if len(waveform.shape) > 1:
+                    logger.warning("EPICS CA transport only supports 1D waveforms")
+                    return None
+
+                return ArrayTrace(axis="x")
             case datatype:
                 raise TypeError(f"Unsupported type {type(datatype)}: {datatype}")
 

src/fastcs/transports/epics/pva/gui.py

@@ -1,12 +1,13 @@
 from pvi.device import (
     CheckBox,
+    ImageRead,
     ReadWidgetUnion,
     TableRead,
     TableWrite,
     WriteWidgetUnion,
 )
 
-from fastcs.datatypes import Bool, DataType, Table, numpy_to_fastcs_datatype
+from fastcs.datatypes import Bool, DataType, Table, Waveform, numpy_to_fastcs_datatype
 from fastcs.transports.epics.gui import EpicsGUI
 
 
@@ -18,31 +19,37 @@ class PvaEpicsGUI(EpicsGUI):
     def _get_pv(self, attr_path: list[str], name: str):
         return f"pva://{super()._get_pv(attr_path, name)}"
 
-    def _get_read_widget(self, fastcs_datatype: DataType) -> ReadWidgetUnion | None:  # noqa: F821
-        if isinstance(fastcs_datatype, Table):
-            fastcs_datatypes = [
-                numpy_to_fastcs_datatype(datatype)
-                for _, datatype in fastcs_datatype.structured_dtype
-            ]
-
-            base_get_read_widget = super()._get_read_widget
-            widgets = [base_get_read_widget(datatype) for datatype in fastcs_datatypes]
-
-            return TableRead(widgets=widgets)  # type: ignore
-        else:
-            return super()._get_read_widget(fastcs_datatype)
+    def _get_read_widget(self, fastcs_datatype: DataType) -> ReadWidgetUnion | None:
+        match fastcs_datatype:
+            case Table():
+                fastcs_datatypes = [
+                    numpy_to_fastcs_datatype(datatype)
+                    for _, datatype in fastcs_datatype.structured_dtype
+                ]
+
+                base_get_read_widget = super()._get_read_widget
+                widgets = [
+                    base_get_read_widget(datatype) for datatype in fastcs_datatypes
+                ]
+
+                return TableRead(widgets=widgets)  # type: ignore
+            case Waveform(shape=(height, width)):
+                return ImageRead(height=height, width=width, grayscale=True)
+            case _:
+                return super()._get_read_widget(fastcs_datatype)
 
     def _get_write_widget(self, fastcs_datatype: DataType) -> WriteWidgetUnion | None:
-        if isinstance(fastcs_datatype, Table):
-            widgets = []
-            for _, datatype in fastcs_datatype.structured_dtype:
-                fastcs_datatype = numpy_to_fastcs_datatype(datatype)
-                if isinstance(fastcs_datatype, Bool):
-                    # Replace with compact version for Table row
-                    widget = CheckBox()
-                else:
-                    widget = super()._get_write_widget(fastcs_datatype)
-                widgets.append(widget)
-            return TableWrite(widgets=widgets)
-        else:
-            return super()._get_write_widget(fastcs_datatype)
+        match fastcs_datatype:
+            case Table():
+                widgets = []
+                for _, datatype in fastcs_datatype.structured_dtype:
+                    fastcs_datatype = numpy_to_fastcs_datatype(datatype)
+                    if isinstance(fastcs_datatype, Bool):
+                        # Replace with compact version for Table row
+                        widget = CheckBox()
+                    else:
+                        widget = super()._get_write_widget(fastcs_datatype)
+                    widgets.append(widget)
+                return TableWrite(widgets=widgets)
+            case _:
+                return super()._get_write_widget(fastcs_datatype)

tests/test_datatypes.py

@@ -34,8 +34,7 @@ class MyIntEnum(IntEnum):
         (Float, {"min": 1}, 0.0),
         (Float, {"max": -1}, 0.0),
         (Enum, {"enum_cls": int}, 0),
-        (Waveform, {"array_dtype": "U64", "shape": (1,)}, np.ndarray([1])),
-        (Waveform, {"array_dtype": "float64", "shape": (1, 1)}, np.ndarray([1])),
+        (Waveform, {"array_dtype": "uint64", "shape": (1, 1)}, np.ndarray([1])),
     ],
 )
 def test_validate(datatype, init_args, value):

tests/transports/epics/ca/test_gui.py

@@ -2,6 +2,7 @@
 import pytest
 from pvi.device import (
     LED,
+    ArrayTrace,
     ButtonPanel,
     ComboBox,
     Group,
@@ -39,7 +40,7 @@ def test_get_pv():
         (Float(), TextRead()),
         (String(), TextRead(format=TextFormat.string)),
         (Enum(ColourEnum), TextRead(format=TextFormat.string)),
-        # (Waveform(array_dtype=np.int32), None),
+        (Waveform(array_dtype=np.int32), ArrayTrace(axis="x")),
     ],
 )
 def test_get_attribute_component_r(datatype, widget):
@@ -78,11 +79,6 @@ def test_get_attribute_component_none(mocker):
     assert gui._get_attribute_component([], "Attr", AttrRW(Int())) is None
 
 
-def test_get_read_widget_none():
-    gui = EpicsGUI(ControllerAPI(), "DEVICE")
-    assert gui._get_read_widget(fastcs_datatype=Waveform(np.int32)) is None
-
-
 def test_get_write_widget_none():
     gui = EpicsGUI(ControllerAPI(), "DEVICE")
     assert gui._get_write_widget(fastcs_datatype=Waveform(np.int32)) is None