WIP - Add Waveform

Author: GDYendell

src/fastcs/datatypes.py

@@ -182,7 +182,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,12 +4,14 @@
 from dataclasses import KW_ONLY, dataclass
 from typing import TypeVar
 
+import numpy as np
+
 from fastcs.attribute_io import AttributeIO
 from fastcs.attribute_io_ref import AttributeIORef
 from fastcs.attributes import AttrR, AttrRW, AttrW
 from fastcs.connections import IPConnection, IPConnectionSettings
 from fastcs.controller import Controller
-from fastcs.datatypes import Enum, Float, Int
+from fastcs.datatypes import Enum, Float, Int, Waveform
 from fastcs.wrappers import command, scan
 
 NumberT = TypeVar("NumberT", int, float)
@@ -68,6 +70,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()
@@ -103,6 +106,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/transport/epics/gui.py

@@ -1,6 +1,7 @@
 from pvi._format.dls import DLSFormatter  # type: ignore
 from pvi.device import (
     LED,
+    ArrayTrace,
     ButtonPanel,
     CheckBox,
     ComboBox,
@@ -68,7 +69,7 @@ def _get_read_widget(self, fastcs_datatype: DataType) -> ReadWidgetUnion | None:
             case Enum():
                 return TextRead(format=TextFormat.string)
             case Waveform():
-                return None
+                return ArrayTrace(axis="x")
             case datatype:
                 raise FastCSError(f"Unsupported type {type(datatype)}: {datatype}")
 

src/fastcs/transport/graphql/graphql.py

@@ -9,11 +9,15 @@
 
 from fastcs.attributes import AttrR, AttrRW, AttrW, T
 from fastcs.controller_api import ControllerAPI
+from fastcs.datatypes import Waveform
 from fastcs.exceptions import FastCSError
 from fastcs.logging import intercept_std_logger
+from fastcs.logging import logger as _logger
 
 from .options import GraphQLServerOptions
 
+logger = _logger.bind(logger_name=__name__)
+
 
 class GraphQLServer:
     """A GraphQL server which handles a controller"""
@@ -60,6 +64,13 @@ def __init__(self, controller_api: ControllerAPI):
     def _process_attributes(self, api: ControllerAPI):
         """Create queries and mutations from api attributes."""
         for attr_name, attribute in api.attributes.items():
+            if isinstance(attribute.datatype, Waveform):
+                logger.warning(
+                    "Waveform attributes are not supported in GraphQL transport",
+                    attribute=attribute,
+                )
+                continue
+
             match attribute:
                 # mutation for server changes https://graphql.org/learn/queries/
                 case AttrRW():

tests/transport/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(controller_api):
         (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, controller_api):