Add enum and waveform values

src/fastcs/attributes.py

@@ -66,18 +66,16 @@ def __init__(
         access_mode: AttrMode,
         group: str | None = None,
         handler: Any = None,
-        allowed_values: list[T] | None = None,
         description: str | None = None,
     ) -> None:
-        assert datatype.dtype in ATTRIBUTE_TYPES, (
-            f"Attr type must be one of {ATTRIBUTE_TYPES}"
-            f", received type {datatype.dtype}"
+        assert issubclass(datatype.dtype, ATTRIBUTE_TYPES), (
+            f"Attr type must be one of {ATTRIBUTE_TYPES}, "
+            "received type {datatype.dtype}"
         )
         self._datatype: DataType[T] = datatype
         self._access_mode: AttrMode = access_mode
         self._group = group
         self.enabled = True
-        self._allowed_values: list[T] | None = allowed_values
         self.description = description
 
         # A callback to use when setting the datatype to a different value, for example
@@ -100,10 +98,6 @@ def access_mode(self) -> AttrMode:
     def group(self) -> str | None:
         return self._group
 
-    @property
-    def allowed_values(self) -> list[T] | None:
-        return self._allowed_values
-
     def add_update_datatype_callback(
         self, callback: Callable[[DataType[T]], None]
     ) -> None:
@@ -129,15 +123,13 @@ def __init__(
         group: str | None = None,
         handler: Updater | None = None,
         initial_value: T | None = None,
-        allowed_values: list[T] | None = None,
         description: str | None = None,
     ) -> None:
         super().__init__(
             datatype,  # type: ignore
             access_mode,
             group,
             handler,
-            allowed_values=allowed_values,  # type: ignore
             description=description,
         )
         self._value: T = (
@@ -172,15 +164,13 @@ def __init__(
         access_mode=AttrMode.WRITE,
         group: str | None = None,
         handler: Sender | None = None,
-        allowed_values: list[T] | None = None,
         description: str | None = None,
     ) -> None:
         super().__init__(
             datatype,  # type: ignore
             access_mode,
             group,
             handler,
-            allowed_values=allowed_values,  # type: ignore
             description=description,
         )
         self._process_callback: AttrCallback[T] | None = None
@@ -227,7 +217,6 @@ def __init__(
         group: str | None = None,
         handler: Handler | None = None,
         initial_value: T | None = None,
-        allowed_values: list[T] | None = None,
         description: str | None = None,
     ) -> None:
         super().__init__(
@@ -236,7 +225,6 @@ def __init__(
             group=group,
             handler=handler,
             initial_value=initial_value,
-            allowed_values=allowed_values,  # type: ignore
             description=description,
         )
 

src/fastcs/datatypes.py

@@ -1,19 +1,24 @@
 from __future__ import annotations
 
+import enum
 from abc import abstractmethod
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
+from functools import cached_property
 from typing import Generic, TypeVar
 
-T_Numerical = TypeVar("T_Numerical", int, float)
-T = TypeVar("T", int, float, bool, str)
+import numpy as np
+from numpy.typing import DTypeLike
+
+T = TypeVar("T", int, float, bool, str, enum.Enum, np.ndarray)
+
 ATTRIBUTE_TYPES: tuple[type] = T.__constraints__  # type: ignore
 
 
 AttrCallback = Callable[[T], Awaitable[None]]
 
 
-@dataclass(frozen=True)  # So that we can type hint with dataclass methods
+@dataclass(frozen=True)
 class DataType(Generic[T]):
     """Generic datatype mapping to a python type, with additional metadata."""
 
@@ -24,11 +29,18 @@
 
     def validate(self, value: T) -> T:
         """Validate a value against fields in the datatype."""
+        if not isinstance(value, self.dtype):
+            raise ValueError(f"Value {value} is not of type {self.dtype}")
+
         return value
 
     @property
+    @abstractmethod
     def initial_value(self) -> T:
-        return self.dtype()
+        pass
+
+
+T_Numerical = TypeVar("T_Numerical", int, float)
 
 
 @dataclass(frozen=True)
@@ -40,12 +52,17 @@
     max_alarm: int | None = None
 
     def validate(self, value: T_Numerical) -> T_Numerical:
+        super().validate(value)
         if self.min is not None and value < self.min:
             raise ValueError(f"Value {value} is less than minimum {self.min}")
         if self.max is not None and value > self.max:
             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]):
@@ -71,13 +88,14 @@
 class Bool(DataType[bool]):
     """`DataType` mapping to builtin ``bool``."""
 
-    znam: str = "OFF"
-    onam: str = "ON"
-
     @property
     def dtype(self) -> type[bool]:
         return bool
 
+    @property
+    def initial_value(self) -> bool:
+        return False
+
 
 @dataclass(frozen=True)
 class String(DataType[str]):
@@ -86,3 +104,65 @@
     @property
     def dtype(self) -> type[str]:
         return str
+
+    @property
+    def initial_value(self) -> str:
+        return ""
+
+
+T_Enum = TypeVar("T_Enum", bound=enum.Enum)
+
+
+@dataclass(frozen=True)
+class Enum(Generic[T_Enum], DataType[T_Enum]):
+    enum_cls: type[T_Enum]
+
+    def __post_init__(self):
+        if not issubclass(self.enum_cls, enum.Enum):
+            raise ValueError("Enum class has to take an Enum.")
+
+    def index_of(self, value: T_Enum) -> int:
+        return self.members.index(value)
+
+    @cached_property
+    def members(self) -> list[T_Enum]:
+        return list(self.enum_cls)
+
+    @property
+    def dtype(self) -> type[T_Enum]:
+        return self.enum_cls
+
+    @property
+    def initial_value(self) -> T_Enum:
+        return self.members[0]
+
+
+@dataclass(frozen=True)
+class Waveform(DataType[np.ndarray]):
+    array_dtype: DTypeLike
+    shape: tuple[int, ...] = (2000,)
+
+    @property
+    def dtype(self) -> type[np.ndarray]:
+        return np.ndarray
+
+    @property
+    def initial_value(self) -> np.ndarray:
+        return np.zeros(self.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}"
+            )
+        if len(self.shape) != len(value.shape) or any(
+            shape1 > shape2
+            for shape1, shape2 in zip(value.shape, self.shape, strict=True)
+        ):
+            raise ValueError(
+                f"Value shape {value.shape} exceeeds the shape maximum shape "
+                f"{self.shape}"
+            )
+        return value

src/fastcs/transport/epics/gui.py

@@ -25,7 +25,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, 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
 
@@ -42,45 +42,51 @@ 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()
             case Int() | Float():
                 return TextRead()
             case String():
                 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:
-        match attribute.allowed_values:
-            case allowed_values if allowed_values is not None:
-                return ComboBox(choices=allowed_values)
-            case _:
-                pass
-
+    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():
+                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 read_widget is None:
+                    return None
                 return SignalRW(
                     name=name,
                     write_pv=pv,
@@ -90,9 +96,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)}")
@@ -152,6 +162,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: