WIP: added waveform

need to pass dimensions to pytango

src/fastcs/datatypes.py

@@ -6,8 +6,9 @@
 from dataclasses import dataclass, field
 from functools import cached_property
 from typing import Generic, TypeVar
+import numpy as np
 
-T = TypeVar("T", int, float, bool, str, enum.IntEnum)
+T = TypeVar("T", int, float, bool, str, enum.IntEnum, np.ndarray)
 
 ATTRIBUTE_TYPES: tuple[type] = T.__constraints__  # type: ignore
 
@@ -44,8 +45,9 @@ def validate(self, value: T) -> T:
         return value
 
     @property
+    @abstractmethod
     def initial_value(self) -> T:
-        return self.dtype()
+        pass
 
 
 T_Numerical = TypeVar("T_Numerical", int, float)
@@ -67,6 +69,10 @@ def validate(self, value: T_Numerical) -> T_Numerical:
             raise ValueError(f"Value {value} is greater than maximum {self.max}")
         return value
 
+    @property
+    def initial_value(self) -> T_Numerical:
+        return self.dtype(0)
+
 
 @dataclass(frozen=True)
 class Int(_Numerical[int]):
@@ -103,6 +109,10 @@ class Bool(DataType[bool]):
     def dtype(self) -> type[bool]:
         return bool
 
+    @property
+    def initial_value(self) -> bool:
+        return False
+
 
 @dataclass(frozen=True)
 class String(DataType[str]):
@@ -114,6 +124,10 @@ class String(DataType[str]):
     def dtype(self) -> type[str]:
         return str
 
+    @property
+    def initial_value(self) -> str:
+        return ""
+
 
 T_Enum = TypeVar("T_Enum", bound=enum.IntEnum)
 
@@ -143,3 +157,26 @@ def dtype(self) -> type[enum.IntEnum]:
     @property
     def initial_value(self) -> enum.IntEnum:
         return self.members[0]
+
+
+@dataclass(frozen=True)
+class WaveForm(DataType[np.ndarray]):
+    array_dtype: np.typing.DTypeLike
+    array_shape: tuple[int, ...] = (2000,)
+
+    @property
+    def dtype(self) -> type[np.ndarray]:
+        return np.ndarray
+
+    @property
+    def initial_value(self) -> np.ndarray:
+        return np.ndarray(self.array_shape, dtype=self.array_dtype)
+
+    def validate(self, value: np.ndarray) -> np.ndarray:
+        super().validate(value)
+        if self.array_dtype != value.dtype:
+            raise ValueError(
+                f"Value dtype {value.dtype} is not the same as the array dtype "
+                f"{self.array_dtype}"
+            )
+        return value

src/fastcs/transport/epics/gui.py

@@ -27,7 +27,7 @@
 from fastcs.attributes import Attribute, AttrR, AttrRW, AttrW
 from fastcs.controller import Controller, SingleMapping, _get_single_mapping
 from fastcs.cs_methods import Command
-from fastcs.datatypes import Bool, Enum, Float, Int, String
+from fastcs.datatypes import Bool, Enum, Float, Int, String, WaveForm
 from fastcs.exceptions import FastCSException
 from fastcs.util import snake_to_pascal
 
@@ -44,7 +44,7 @@ def _get_pv(self, attr_path: list[str], name: str):
         return f"{attr_prefix}:{name.title().replace('_', '')}"
 
     @staticmethod
-    def _get_read_widget(attribute: AttrR) -> ReadWidgetUnion:
+    def _get_read_widget(attribute: AttrR) -> ReadWidgetUnion | None:
         match attribute.datatype:
             case Bool():
                 return LED()
@@ -54,43 +54,41 @@ def _get_read_widget(attribute: AttrR) -> ReadWidgetUnion:
                 return TextRead(format=TextFormat.string)
             case Enum():
                 return TextRead(format=TextFormat.string)
+            case WaveForm():
+                return None
             case datatype:
                 raise FastCSException(f"Unsupported type {type(datatype)}: {datatype}")
 
     @staticmethod
-    def _get_write_widget(attribute: AttrW) -> WriteWidgetUnion:
+    def _get_write_widget(attribute: AttrW) -> WriteWidgetUnion | None:
         match attribute.datatype:
             case Bool():
                 return ToggleButton()
             case Int() | Float():
                 return TextWrite()
             case String():
                 return TextWrite(format=TextFormat.string)
-            case Enum(enum_cls=enum_cls):
-                match enum_cls:
-                    case enum_cls if issubclass(enum_cls, enum.Enum):
-                        return ComboBox(
-                            choices=[
-                                member.name for member in attribute.datatype.members
-                            ]
-                        )
-                    case _:
-                        raise FastCSException(
-                            f"Unsupported Enum type {type(enum_cls)}: {enum_cls}"
-                        )
+            case Enum():
+                return ComboBox(
+                    choices=[member.name for member in attribute.datatype.members]
+                )
+            case WaveForm():
+                return None
             case datatype:
                 raise FastCSException(f"Unsupported type {type(datatype)}: {datatype}")
 
     def _get_attribute_component(
         self, attr_path: list[str], name: str, attribute: Attribute
-    ) -> SignalR | SignalW | SignalRW:
+    ) -> SignalR | SignalW | SignalRW | None:
         pv = self._get_pv(attr_path, name)
         name = name.title().replace("_", "")
 
         match attribute:
             case AttrRW():
                 read_widget = self._get_read_widget(attribute)
                 write_widget = self._get_write_widget(attribute)
+                if write_widget is None or write_widget is None:
+                    return None
                 return SignalRW(
                     name=name,
                     write_pv=pv,
@@ -100,9 +98,13 @@ def _get_attribute_component(
                 )
             case AttrR():
                 read_widget = self._get_read_widget(attribute)
+                if read_widget is None:
+                    return None
                 return SignalR(name=name, read_pv=pv, read_widget=read_widget)
             case AttrW():
                 write_widget = self._get_write_widget(attribute)
+                if write_widget is None:
+                    return None
                 return SignalW(name=name, write_pv=pv, write_widget=write_widget)
             case _:
                 raise FastCSException(f"Unsupported attribute type: {type(attribute)}")
@@ -161,6 +163,9 @@ def extract_mapping_components(self, mapping: SingleMapping) -> Tree:
                 print(f"Invalid name:\n{e}")
                 continue
 
+            if signal is None:
+                continue
+
             match attribute:
                 case Attribute(group=group) if group is not None:
                     if group not in groups:

src/fastcs/transport/epics/ioc.py

@@ -7,9 +7,10 @@
 from softioc.asyncio_dispatcher import AsyncioDispatcher
 from softioc.pythonSoftIoc import RecordWrapper
 
+from fastcs import attributes
 from fastcs.attributes import AttrR, AttrRW, AttrW
 from fastcs.controller import BaseController, Controller
-from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String, T
+from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String, WaveForm, T
 from fastcs.exceptions import FastCSException
 from fastcs.transport.epics.util import (
     MBB_MAX_CHOICES,
@@ -200,41 +201,30 @@ def _get_input_record(pv: str, attribute: AttrR) -> RecordWrapper:
             )
         case Enum():
             if len(attribute.datatype.members) > MBB_MAX_CHOICES:
-                if attribute.datatype.is_string_enum:
-                    replacement_record, replacement_str = (
-                        builder.longStringIn,
-                        "longStringIn",
-                    )
-                else:
-                    replacement_record, replacement_str = builder.longIn, "longIn"
-
-                warnings.warn(
-                    f"Received an enum datatype on attribute {attribute} "
+                raise RuntimeError(
+                    f"Received an `Enum` datatype on attribute {attribute} "
                     f"with more elements than the epics limit `{MBB_MAX_CHOICES}` "
-                    f"for mbbIn, will use a {replacement_str} record instead. "
-                    "To stop with warning use a different datatype with "
-                    "`allowed_values`",
-                    stacklevel=1,
+                    f"for `mbbIn`. Use an `Int or `String with `allowed_values`."
                 )
-                record = replacement_record(
-                    pv,
-                    **get_record_metadata_from_datatype(attribute.datatype),
-                    **get_record_metadata_from_attribute(attribute),
-                )
-            else:
-                state_keys = dict(
-                    zip(
-                        MBB_STATE_FIELDS,
-                        [member.name for member in attribute.datatype.members],
-                        strict=False,
-                    )
-                )
-                record = builder.mbbIn(
-                    pv,
-                    **state_keys,
-                    **get_record_metadata_from_datatype(attribute.datatype),
-                    **get_record_metadata_from_attribute(attribute),
+            state_keys = dict(
+                zip(
+                    MBB_STATE_FIELDS,
+                    [member.name for member in attribute.datatype.members],
+                    strict=False,
                 )
+            )
+            record = builder.mbbIn(
+                pv,
+                **state_keys,
+                **get_record_metadata_from_datatype(attribute.datatype),
+                **get_record_metadata_from_attribute(attribute),
+            )
+        case WaveForm():
+            record = builder.WaveformIn(
+                pv,
+                **get_record_metadata_from_datatype(attribute.datatype),
+                **get_record_metadata_from_attribute(attribute),
+            )
         case _:
             raise FastCSException(
                 f"Unsupported type {type(attribute.datatype)}: {attribute.datatype}"
@@ -302,48 +292,37 @@ def _get_output_record(pv: str, attribute: AttrW, on_update: Callable) -> Any:
                 **get_record_metadata_from_datatype(attribute.datatype),
                 **get_record_metadata_from_attribute(attribute),
             )
-        case Enum(enum_cls=enum_cls):
-            members = list(enum_cls)
-            if len(members) > MBB_MAX_CHOICES:
-                if attribute.datatype.is_string_enum:
-                    replacement_record, replacement_str = (
-                        builder.longStringOut,
-                        "longStringOut",
-                    )
-                else:
-                    replacement_record, replacement_str = builder.longOut, "longOut"
-
-                warnings.warn(
-                    f"Received an enum datatype on attribute {attribute} "
+        case Enum():
+            if len(attribute.datatype.members) > MBB_MAX_CHOICES:
+                raise RuntimeError(
+                    f"Received an `Enum` datatype on attribute {attribute} "
                     f"with more elements than the epics limit `{MBB_MAX_CHOICES}` "
-                    f"for mbbOut, will use a {replacement_str} record instead. "
-                    "To stop with warning use a different datatype with "
-                    "`allowed_values`",
-                    stacklevel=1,
+                    f"for `mbbOut`. Use an `Int or `String with `allowed_values`."
                 )
-                record = replacement_record(
-                    pv,
-                    always_update=True,
-                    on_update=on_update,
-                    **get_record_metadata_from_datatype(attribute.datatype),
-                    **get_record_metadata_from_attribute(attribute),
-                )
-            else:
-                state_keys = dict(
-                    zip(
-                        MBB_STATE_FIELDS,
-                        [member.name for member in members],
-                        strict=False,
-                    )
-                )
-                record = builder.mbbOut(
-                    pv,
-                    **state_keys,
-                    always_update=True,
-                    on_update=on_update,
-                    **get_record_metadata_from_datatype(attribute.datatype),
-                    **get_record_metadata_from_attribute(attribute),
+
+            state_keys = dict(
+                zip(
+                    MBB_STATE_FIELDS,
+                    [member.name for member in attribute.datatype.members],
+                    strict=False,
                 )
+            )
+            record = builder.mbbOut(
+                pv,
+                **state_keys,
+                always_update=True,
+                on_update=on_update,
+                **get_record_metadata_from_datatype(attribute.datatype),
+                **get_record_metadata_from_attribute(attribute),
+            )
+        case WaveForm():
+            record = builder.WaveformOut(
+                pv,
+                always_update=True,
+                on_update=on_update,
+                **get_record_metadata_from_datatype(attribute.datatype),
+                **get_record_metadata_from_attribute(attribute),
+            )
 
         case _:
             raise FastCSException(

src/fastcs/transport/epics/util.py

@@ -2,7 +2,7 @@
 from dataclasses import asdict
 
 from fastcs.attributes import Attribute
-from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String, T
+from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String, WaveForm, T
 
 _MBB_FIELD_PREFIXES = (
     "ZR",
@@ -28,7 +28,7 @@
 MBB_MAX_CHOICES = len(_MBB_FIELD_PREFIXES)
 
 
-EPICS_ALLOWED_DATATYPES = (Bool, DataType, Enum, Float, Int, String)
+EPICS_ALLOWED_DATATYPES = (Bool, DataType, Enum, Float, Int, String, WaveForm)
 
 DATATYPE_FIELD_TO_RECORD_FIELD = {
     "prec": "PREC",
@@ -39,6 +39,7 @@
     "max_alarm": "HOPR",
     "znam": "ZNAM",
     "onam": "ONAM",
+    "array_shape": "length",
 }
 
 

src/fastcs/transport/tango/util.py

@@ -1,8 +1,8 @@
 from collections.abc import Callable
 
-from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String, T
+from fastcs.datatypes import Bool, DataType, Enum, Float, Int, String, WaveForm, T
 
-TANGO_ALLOWED_DATATYPES = (Bool, DataType, Enum, Float, Int, String)
+TANGO_ALLOWED_DATATYPES = (Bool, DataType, Enum, Float, Int, String, WaveForm)
 
 
 def get_cast_method_to_tango_type(datatype: DataType[T]) -> Callable[[T], object]:

tests/conftest.py

@@ -1,5 +1,6 @@
 import copy
 import enum
+import numpy as np
 import os
 import random
 import string
@@ -15,7 +16,7 @@
 
 from fastcs.attributes import AttrR, AttrRW, AttrW, Handler, Sender, Updater
 from fastcs.controller import Controller, SubController
-from fastcs.datatypes import Bool, Enum, Float, Int, String
+from fastcs.datatypes import Bool, Enum, Float, Int, String, WaveForm
 from fastcs.wrappers import command, scan
 
 DATA_PATH = Path(__file__).parent / "data"
@@ -81,6 +82,8 @@ def __init__(self) -> None:
     write_bool: AttrW = AttrW(Bool(), handler=TestSender())
     read_string: AttrRW = AttrRW(String())
     enum: AttrRW = AttrRW(Enum(enum.IntEnum("Enum", {"RED": 0, "GREEN": 1, "BLUE": 2})))
+    ond_d_waveform: AttrRW = AttrRW(WaveForm(np.int32, (10,)))
+    two_d_waveform: AttrRW = AttrRW(WaveForm(np.int32, (10, 10)))
     big_enum: AttrR = AttrR(
         Int(
             allowed_values=list(range(17)),