diff --git a/ophyd_devices/utils/bec_signals.py b/ophyd_devices/utils/bec_signals.py index 8a6639f..178dc14 100644 --- a/ophyd_devices/utils/bec_signals.py +++ b/ophyd_devices/utils/bec_signals.py @@ -29,6 +29,9 @@ __all__ = [ "AsyncMultiSignal", ] +SupportedSignalTypes = (int, float, str, bool, list, np.ndarray) +SupportedSignalTypesUnion = int | float | str | bool | list | np.ndarray + class SignalInfo(BaseModel): """ @@ -716,7 +719,10 @@ class DynamicSignal(BECMessageSignal): @typechecked def put( self, - value: messages.DeviceMessage | dict[str, dict[Literal["value", "timestamp"], Any]], + value: ( + messages.DeviceMessage + | dict[str, dict[Literal["value", "timestamp"], SupportedSignalTypesUnion]] + ), *, metadata: dict | None = None, async_update: dict[Literal["type", "max_shape", "index"], Any] | None = None, @@ -755,6 +761,16 @@ class DynamicSignal(BECMessageSignal): elif self.acquisition_group is not None: metadata["acquisition_group"] = self.acquisition_group + # verify that signal data is of supported type + for signal_name, signal_data in value.items(): + if "value" not in signal_data: + raise ValueError(f"Signal data for {signal_name} must contain 'value' key.") + if not isinstance(signal_data["value"], SupportedSignalTypes): + raise ValueError( + f"Signal data for {signal_name} must be of type {SupportedSignalTypes}, " + f"got {type(signal_data['value']).__name__}." + ) + msg = messages.DeviceMessage(signals=value, metadata=metadata) except ValidationError as exc: raise ValueError(f"Error setting signal {self.name}: {exc}") from exc @@ -973,7 +989,7 @@ class AsyncSignal(DynamicSignal): def put( self, - value: Any, + value: SupportedSignalTypesUnion, timestamp: float | None = None, async_update: dict[Literal["type", "max_shape", "index"], Any] | None = None, acquisition_group: str | None = None, @@ -998,7 +1014,7 @@ class AsyncSignal(DynamicSignal): def set( self, - value: Any, + value: SupportedSignalTypesUnion, timestamp: float | None = None, async_update: dict[Literal["type", "max_shape", "index"], Any] | None = None, acquisition_group: str | None = None,