Datatypes

FastCS uses a datatype system to map Python types to attributes with additional metadata for validation, serialization, and transport handling.

Supported Types

FastCS defines DType as the union of supported Python types:

DType = (
    int  # Int
    | float  # Float
    | bool  # Bool
    | str  # String
    | enum.Enum  # Enum
    | np.ndarray  # Waveform / Table
)

Each has a corresponding DataType class.

Scalar Datatypes

Int and Float

Both inherit from _Numeric, which adds support for bounds and alarm limits:

@dataclass(frozen=True)
class _Numeric(DataType[Numeric_T]):
    """Base class for numeric FastCS DataType classes"""

    units: str | None = None
    """The units of the numeric value"""
    min: Numeric_T | None = None
    """The minimum allowed value - values below this will raise an exception"""
    max: Numeric_T | None = None
    """The maximum allowed value - values above this will raise an exception"""
    min_alarm: Numeric_T | None = None
    """The minimum alarm limit - values below this will be set with an alarm state"""
    max_alarm: Numeric_T | None = None

Bool

Maps to Python bool. Initial value is False.

@dataclass(frozen=True)
class Bool(DataType[bool]):
    """`DataType` mapping to builtin ``bool``."""

    @property
    def dtype(self) -> type[bool]:
        return bool

    @property
    def initial_value(self) -> bool:
        return False

String

Maps to Python str. Has an optional length field that truncates values during validation. It is also used as a hint by some transports to configure the size of string records (e.g. EPICS CA string waveform records).

@dataclass(frozen=True)
class String(DataType[str]):
    """`DataType` mapping to builtin ``str``."""

    length: int | None = None
    """Maximum length of string to display in transports. Must be >=1 or None."""

    def __post_init__(self):
        if self.length is not None and self.length < 1:
            raise ValueError("String length must be >= 1")

    @property
    def dtype(self) -> type[str]:
        return str

    @property
    def initial_value(self) -> str:
        return ""

    def validate(self, value: Any) -> str:
        """Truncate string to maximum length

        Returns:
            The string, truncated to the maximum length if set

        """
        return super().validate(value)[: self.length]

Enum Datatype

Wraps a Python enum.Enum class:

@dataclass(frozen=True)
class Enum(Generic[Enum_T], DataType[Enum_T]):
    enum_cls: type[Enum_T]

    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: Enum_T) -> int:
        return self.members.index(value)

    @cached_property
    def members(self) -> list[Enum_T]:
        return list(self.enum_cls)

    @cached_property
    def names(self) -> list[str]:
        return [member.name for member in self.members]

    @property
    def dtype(self) -> type[Enum_T]:
        return self.enum_cls

    @property
    def initial_value(self) -> Enum_T:
        return self.members[0]

The Enum datatype provides helper properties:

• members: List of enum values

• names: List of enum member names

• index_of(value): Get the index of a value in the members list

Note

FastCS uses enum member names (not values) when exposing choices to transports and PVI. This means member names are the user-friendly UI strings while values are the strings sent to the device:

class DetectorStatus(StrEnum):
    Idle = "IDLE_STATE"
    Running = "RUNNING_STATE"
    Error = "ERROR_STATE"

Clients will see the choices as ["Idle", "Running", "Error"].

For UI strings with spaces, use the functional enum.Enum API with a dict:

import enum
from fastcs.datatypes import Enum

DetectorStatus = Enum(enum.Enum("DetectorStatus", {"Run Finished": "RUN_FINISHED", "In Progress": "IN_PROGRESS"}))

Clients will see the choices as ["Run Finished", "In Progress"].

Array Datatypes

Waveform

For homogeneous numpy arrays (spectra, images):

@dataclass(frozen=True)
class Waveform(DataType[np.ndarray]):
    array_dtype: DTypeLike
    """Numpy array dtype"""
    shape: tuple[int, ...] = (2000,)
    """Numpy array shape"""

    @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:
        _value = super().validate(np.asarray(value).astype(self.array_dtype))

        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

    @staticmethod
    def equal(value1: np.ndarray, value2: np.ndarray) -> bool:
        return np.array_equal(value1, value2)

Validation ensures the array fits within the declared shape and has the correct dtype.

Table

For structured numpy arrays with named columns:

@dataclass(frozen=True)
class Table(DataType[np.ndarray]):
    structured_dtype: list[tuple[str, DTypeLike]]
    """The structured dtype for numpy array

    See docs for more information:
    https://numpy.org/devdocs/user/basics.rec.html#structured-datatype-creation
    """

    @property
    def dtype(self) -> type[np.ndarray]:
        return np.ndarray

    @property
    def initial_value(self) -> np.ndarray:
        return np.array([], dtype=self.structured_dtype)

    def validate(self, value: Any) -> np.ndarray:
        _value = super().validate(value)

        if self.structured_dtype != _value.dtype:
            raise ValueError(
                f"Value dtype {_value.dtype.descr} is not the same as the structured "
                f"dtype {self.structured_dtype}"
            )

        return _value

    @staticmethod
    def equal(value1: np.ndarray, value2: np.ndarray) -> bool:
        return np.array_equal(value1, value2)

The structured_dtype field is a list of (name, dtype) tuples following numpy’s structured array conventions.

Validation

Built-in Numeric Validation

Int and Float datatypes support min/max limits and alarm thresholds:

from fastcs.attributes import AttrRW
from fastcs.datatypes import Int, Float

# Integer with bounds
count = AttrRW(Int(min=0, max=100))

# Float with units and alarm limits
temperature = AttrRW(Float(
    units="degC",
    min=-273.15,           # Absolute minimum
    max=1000.0,            # Absolute maximum
    min_alarm=-50.0,       # Warning below this
    max_alarm=200.0,       # Warning above this
))

Validation Behavior

temp = Float(min=0.0, max=100.0)

temp.validate(50.0)   # Returns 50.0
temp.validate(-10.0)  # Raises ValueError: "Value -10.0 is less than minimum 0.0"
temp.validate(150.0)  # Raises ValueError: "Value 150.0 is greater than maximum 100.0"

String Length

Limit the display length of strings:

from fastcs.datatypes import String

# Limit display to 40 characters
status = AttrR(String(length=40))

Note

The length parameter truncates values during validation and is also used by some transports to configure their records, for example the EPICS CA transport uses it to set the length of string waveform records.

Type Coercion

All datatypes automatically coerce compatible types:

from fastcs.datatypes import Int, Float

int_type = Int()
int_type.validate("42")     # Returns 42 (str -> int)
int_type.validate(3.7)      # Returns 3 (float -> int, truncated)

float_type = Float()
float_type.validate("3.14") # Returns 3.14 (str -> float)
float_type.validate(42)     # Returns 42.0 (int -> float)

When Validation Runs

Validation runs automatically when:

1. Attribute update: await attr.update(value) validates before storing

2. Put request: await attr.put(value) validates before sending to device

3. Initial value: Values passed to initial_value are validated on creation

from fastcs.attributes import AttrRW
from fastcs.datatypes import Int

attr = AttrRW(Int(min=0, max=10), initial_value=5)

# Updates are validated
await attr.update(7)    # OK
await attr.update(15)   # Raises ValueError

# Puts are validated
await attr.put(3)       # OK
await attr.put(-1)      # Raises ValueError

Transport Handling

Transports are responsible for serializing datatypes appropriately for their protocol. Each transport must handle all supported datatypes. The datatype’s dtype property and class type are used to determine serialization:

• Scalars (Int, Float, Bool, String) serialize directly

• Enum values are typically serialized as integers (index) or strings (name)

• Waveform and Table arrays are serialized as lists or protocol-specific array types

Creating Custom Datatypes

All datatypes inherit from DataType[DType_T], a generic frozen dataclass that defines the interface for type handling:

@dataclass(frozen=True)
class DataType(Generic[DType_T]):
    """Generic datatype mapping to a python type, with additional metadata."""

    @property
    @abstractmethod
    def dtype(self) -> type[DType_T]:  # Using property due to lack of Generic ClassVars
        """Underlying python type"""
        raise NotImplementedError()

Required Properties

To create a custom datatype, subclass DataType or one of the existing datatypes and implement the required properties:

dtype: Returns the underlying Python type. This is used for type coercion in validate() and for transport serialization.

initial_value: Returns the default value used when an attribute is created without an explicit initial value.

Overriding validate()

The base validate() implementation attempts to cast incoming values to the target type:

    def validate(self, value: Any) -> DType_T:
        """Validate a value against the datatype.

        The base implementation is to try the cast and raise a useful error if it fails.

        Child classes can implement logic before calling ``super.validate(value)`` to
        modify the value passed in and help the cast succeed or after to perform further
        validation of the coerced type.

        Args:
            value: The value to validate

        Returns:
            The validated value

        Raises:
            ValueError: If the value cannot be coerced

        """
        if isinstance(value, self.dtype):
            return value

        try:
            return self.dtype(value)
        except (ValueError, TypeError) as e:
            raise ValueError(f"Failed to cast {value} to type {self.dtype}") from e

Subclasses can override this to add validation logic. The pattern is

1. Coerce input to help type casting succeed - e.g. Waveform calls numpy.asarray(...)

2. Call super().validate(value) to call parent implementation and perform the type cast

3. Perform any additional validation such as checking limits - e.g. _Numeric adds min/max validation:

    def validate(self, value: Any) -> Numeric_T:
        _value = 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

Overriding equal()

The equal() method is used by the always flag in attribute callbacks to determine if a value has changed. The default uses Python’s == operator, but array types override this to use numpy.array_equal():

    @staticmethod
    def equal(value1: np.ndarray, value2: np.ndarray) -> bool:
        return np.array_equal(value1, value2)

Transport Compatibility

When creating a new datatype, existing transports will need to be updated to handle it, unless the datatype inherits from a supported type. In the latter case, the transport will use the parent class handling, while the custom datatype can add validation or other behaviour on top.