bec/ophyd_devices
0
0
Fork 0
mirror of https://github.com/bec-project/ophyd_devices.git synced 2025-06-06 03:40:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: restructure bec signals

Browse Source
This commit is contained in:
wakonig_k 2025-06-02 18:44:51 +02:00
parent e47472d886
commit 42a88f7d25
4 changed files with 434 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ophyd_devices/sim/sim_camera.py
View File

@ -39,7 +39,7 @@ class SimCameraControl(Device):
compute_readback=True,
kind=Kind.omitted,
)
preview = Cpt(PreviewSignal, name="preview", ndim=2)
preview = Cpt(PreviewSignal, name="preview", ndim=2, num_rotation_90=0)
write_to_disk = Cpt(SetableSignal, name="write_to_disk", value=False, kind=Kind.config)
def __init__(self, name, *, parent=None, sim_init: dict = None, device_manager=None, **kwargs):

2
ophyd_devices/sim/sim_test_devices.py
View File

@ -329,7 +329,7 @@ class SimCameraWithPSIComponents(SimCamera):
file_event = Cpt(FileEventSignal, doc="File event signal")
progress = Cpt(ProgressSignal, doc="Progress signal")
dynamic_signal = Cpt(
DynamicSignal, doc="Dynamic signals", signal_names=["preview_2d", "preview_1d"]
DynamicSignal, doc="Dynamic signals", signals=["dyn_signal1", "dyn_signal2"]
)
# TODO Handling of AsyncComponents is postponed, issue #104 is created

449
ophyd_devices/utils/bec_signals.py
View File

@ -7,16 +7,58 @@ from typing import Any, Callable, Literal, Type
import numpy as np
from bec_lib import messages
from bec_lib.logger import bec_logger
from ophyd import DeviceStatus, Kind, Signal
from pydantic import ValidationError
from typeguard import typechecked
from pydantic import BaseModel, Field, ValidationError
logger = bec_logger.logger
# pylint: disable=arguments-differ
# pylint: disable=arguments-renamed
# pylint: disable=too-many-arguments
# pylint: disable=signature-differs
class SignalInfo(BaseModel):
"""
Base class for signal information.
This is used to store metadata about the signal.
"""
data_type: Literal["raw", "processed"] = Field(
default="raw",
description="The data type of the signal indicates whether the signal is raw data or processed data.",
)
saved: bool = Field(default=True, description="Indicates whether the signal is saved to disk.")
ndim: Literal[0, 1, 2] | None = Field(
default=None,
description="The number of dimensions of the signal. If None, the signal is not expected to have a shape. "
"If set to 0, the signal is expected to be a scalar. For signals with multiple sub-signals, "
"ndim is expected to be valid for all sub-signals.",
)
scope: Literal["scan", "continuous"] = Field(
default="scan",
description="The scope of the signal indicates whether it is relevant for a specific "
"scan or provides continuous updates, independent of a scan.",
)
role: Literal["main", "preview", "diagnostic", "file event", "progress"] = Field(
default="main",
description="The role of the signal provides context for its usage and allows other components to filter"
" or prioritize signals based on their intended function.",
)
enabled: bool = True
rpc_access: bool = Field(
default=False,
description="Indicates whether the signal is accessible via RPC. If False, the signal is not shown in the RPC interface.",
)
signals: list[tuple[str, int]] | None = Field(
default=None, description="List of sub-signals with their kinds."
)
signal_metadata: dict | None = Field(
default=None,
description="Metadata for the signal, which can include additional information about the signal's properties.",
)
class BECMessageSignal(Signal):
"""
Custom signal class that accepts BECMessage objects as values.
@ -29,7 +71,18 @@ class BECMessageSignal(Signal):
*,
bec_message_type: Type[messages.BECMessage],
value: messages.BECMessage | dict | None = None,
kind: Kind | str = Kind.omitted,
data_type: Literal["raw", "processed"] = "raw",
saved: bool = True,
ndim: Literal[0, 1, 2] | None = None,
scope: Literal["scan", "continuous"] = "scan",
role: Literal["main", "preview", "diagnostic", "file event", "progress"] = "main",
enabled: bool = True,
signals: (
Callable[[], list[str]]
| Callable[[], list[tuple[str, str | Kind]]]
| list[tuple[str, str | Kind] | str]
| None
) = None,
signal_metadata: dict | None = None,
**kwargs,
):
@ -40,7 +93,6 @@ class BECMessageSignal(Signal):
name (str) : The name of the signal.
bec_message_type: type[BECMessage] : The type of BECMessage to accept as values.
value (BECMessage | dict | None) : The value of the signal. Defaults to None.
kind (Kind | str) : The kind of the signal. Defaults to Kind.omitted.
"""
if isinstance(value, dict):
value = bec_message_type(**value)
@ -48,15 +100,73 @@ class BECMessageSignal(Signal):
raise ValueError(
f"Value must be a {bec_message_type.__name__} or a dict for signal {name}"
)
self.signal_metadata = signal_metadata
self._bec_message_type = bec_message_type
kwargs.pop("dtype", None) # Ignore dtype if specified
kwargs.pop("shape", None) # Ignore shape if specified
super().__init__(name=name, value=value, shape=(), dtype=None, kind=kind, **kwargs)
kind = kwargs.pop("kind", None) # Ignore kind if specified
if kind is not None:
logger.warning("The 'kind' argument is ignored for BECMessageSignal. Please remove it.")
super().__init__(name=name, value=value, shape=(), dtype=None, kind=Kind.omitted, **kwargs)
self.data_type = data_type
self.saved = saved
self.ndim = ndim if ndim is not None else 0
self.scope = scope
self.role = role
self.enabled = enabled
self.signals = self._unify_signals(signals)
self.signal_metadata = signal_metadata
self._bec_message_type = bec_message_type
def _unify_signals(
self, signals: Callable[[], list[str]] | list[tuple[str, str | Kind] | str] | str | None
) -> list[tuple[str, int]]:
"""
Unify the signals list to a list of tuples with signal name and kind.
Args:
signals (list[tuple[str, str | Kind] | str]): The list of signals to unify.
Returns:
list[tuple[str, str]]: The unified list of signals.
"""
if isinstance(signals, Callable):
signals = signals()
if signals is None:
return [(self.attr_name or self.name, Kind.hinted.value)]
if isinstance(signals, str):
return [(signals, Kind.hinted.value)]
if not isinstance(signals, list):
raise ValueError(
f"Signals must be a list of tuples or strings, got {type(signals).__name__}."
)
out = []
for signal in signals:
if isinstance(signal, str):
out.append((signal, Kind.normal.value))
elif isinstance(signal, tuple) and len(signal) == 2:
if isinstance(signal[1], Kind):
out.append((signal[0], signal[1].value))
else:
out.append((signal[0], signal[1]))
else:
raise ValueError(
f"Invalid signal format: {signal}. Expected a tuple of (name, kind) or a string."
)
return out
def describe(self):
out = super().describe()
out[self.name]["signal_metadata"] = self.signal_metadata or {}
out[self.name]["signal_info"] = SignalInfo(
data_type=self.data_type, # type: ignore
saved=self.saved,
ndim=self.ndim, # type: ignore
scope=self.scope, # type: ignore
role=self.role, # type: ignore
enabled=self.enabled,
signals=self.signals,
signal_metadata=self.signal_metadata,
).model_dump()
return out
@property
@ -121,9 +231,14 @@ class ProgressSignal(BECMessageSignal):
kwargs.pop("kind", None) # Ignore kind if specified
super().__init__(
name=name,
data_type="raw",
saved=False,
ndim=0,
scope="scan",
role="progress",
signal_metadata=None,
value=value,
bec_message_type=messages.ProgressMessage,
kind=Kind.omitted,
**kwargs,
)
@ -145,21 +260,51 @@ class ProgressSignal(BECMessageSignal):
Otherwise, at least value, max_value and done must be provided.
Args:
msg (ProgressMessage | dict | None) : The progress message.
value (float) : The current progress value.
max_value (float) : The maximum progress value.
done (bool) : Whether the progress is done.
metadata (dict | None) : Additional metadata
msg (ProgressMessage | dict | None): The progress message.
value (float): The current progress value.
max_value (float): The maximum progress value.
done (bool): Whether the progress is done.
metadata (dict | None): Additional metadata
"""
# msg is ProgressMessage or dict
if isinstance(msg, (messages.ProgressMessage, dict)):
if msg is None and (value is None or max_value is None or done is None):
raise ValueError(
"Either msg must be provided or value, max_value and done must be set."
)
if isinstance(msg, messages.ProgressMessage):
if (
value is not None
or max_value is not None
or done is not None
or metadata is not None
):
logger.warning(
"Ignoring value, max_value, done and metadata arguments when msg is provided."
)
return super().put(msg, **kwargs)
if isinstance(msg, dict):
if (
value is not None
or max_value is not None
or done is not None
or metadata is not None
):
logger.warning(
"Ignoring value, max_value, done and metadata arguments when msg is provided as dict."
)
return super().put(msg, **kwargs)
if value is None or max_value is None or done is None:
raise ValueError("If msg is not provided, value, max_value and done must be set.")
try:
msg = messages.ProgressMessage(
value=value, max_value=max_value, done=done, metadata=metadata
)
except ValidationError as exc:
raise ValueError(f"Error setting signal {self.name}: {exc}") from exc
return super().put(msg, **kwargs)
def set(
@ -180,11 +325,11 @@ class ProgressSignal(BECMessageSignal):
Otherwise, at least value, max_value and done must be provided.
Args:
msg (ProgressMessage | dict | None) : The progress message.
value (float) : The current progress value.
max_value (float) : The maximum progress value.
done (bool) : Whether the progress is done.
metadata (dict | None) : Additional metadata
msg (ProgressMessage | dict | None): The progress message.
value (float): The current progress value.
max_value (float): The maximum progress value.
done (bool): Whether the progress is done.
metadata (dict | None): Additional metadata
"""
self.put(msg=msg, value=value, max_value=max_value, done=done, metadata=metadata, **kwargs)
status = DeviceStatus(device=self)
@ -207,9 +352,14 @@ class FileEventSignal(BECMessageSignal):
kwargs.pop("kind", None) # Ignore kind if specified
super().__init__(
name=name,
data_type="raw",
saved=False,
ndim=0,
scope="scan",
role="file event",
signal_metadata=None,
value=value,
bec_message_type=messages.FileMessage,
kind=Kind.omitted,
**kwargs,
)
@ -220,7 +370,6 @@ class FileEventSignal(BECMessageSignal):
file_path: str | None = None,
done: bool | None = None,
successful: bool | None = None,
device_name: str | None = None,
file_type: str = "h5",
hinted_h5_entries: dict[str, str] | None = None,
metadata: dict | None = None,
@ -234,18 +383,53 @@ class FileEventSignal(BECMessageSignal):
Otherwise, at least file_path, done and successful must be provided.
Args:
msg (FileMessage | dict | None) : The file event message.
file_path (str) : The path of the file.
done (bool) : Whether the file event is done.
successful (bool) : Whether the file event was successful.
device_name (str | None) : The name of the device.
file_type (str | None) : The type of the file.
msg (FileMessage | dict | None): The file event message.
file_path (str | None): The path of the file.
done (bool | None): Whether the file event is done.
successful (bool | None): Whether the file writing finished successfully.
file_type (str): The type of the file, defaults to "h5".
hinted_h5_entries (dict[str, str] | None): The hinted h5 entries.
metadata (dict | None) : Additional metadata.
metadata (dict | None): Additional metadata.
"""
if isinstance(msg, (messages.FileMessage, dict)):
device_name = self.root.name if self.root else self.name
if msg is None and (file_path is None or done is None or successful is None):
raise ValueError(
"Either msg must be provided or file_path, done and successful must be set."
)
if isinstance(msg, messages.FileMessage):
if (
file_path is not None
or done is not None
or successful is not None
or file_type != "h5"
or hinted_h5_entries is not None
or metadata is not None
):
logger.warning(
"Ignoring file_path, done, successful, file_type, "
"hinted_h5_entries and metadata arguments when msg is provided."
)
msg.device_name = device_name
return super().put(msg, **kwargs)
# kwargs provided
if isinstance(msg, dict):
if (
file_path is not None
or done is not None
or successful is not None
or file_type != "h5"
or hinted_h5_entries is not None
or metadata is not None
):
logger.warning(
"Ignoring file_path, done, successful, device_name, file_type, "
"hinted_h5_entries and metadata arguments when msg is provided as dict."
)
msg["device_name"] = device_name
return super().put(msg, **kwargs)
if file_path is None or done is None or successful is None:
raise ValueError("If msg is not provided, file_path, done and successful must be set.")
try:
msg = messages.FileMessage(
file_path=file_path,
@ -282,14 +466,14 @@ class FileEventSignal(BECMessageSignal):
Otherwise, at least file_path, done and successful must be provided.
Args:
msg (FileMessage | dict | None) : The file event message.
file_path (str) : The path of the file.
done (bool) : Whether the file event is done.
successful (bool) : Whether the file event was successful.
device_name (str | None) : The name of the device.
file_type (str | None) : The type of the file.
msg (FileMessage | dict | None): The file event message.
file_path (str | None): The path of the file.
done (bool | None): Whether the file event is done.
successful (bool | None): Whether the file writing finished successfully.
device_name (str | None): The name of the device.
file_type (str): The type of the file, defaults to "h5".
hinted_h5_entries (dict[str, str] | None): The hinted h5 entries.
metadata (dict | None) : Additional metadata.
metadata (dict | None): Additional metadata.
"""
self.put(
msg=msg,
@ -321,27 +505,44 @@ class PreviewSignal(BECMessageSignal):
**kwargs,
):
"""
Create a new FileEventSignal object.
Create a new PreviewSignal object.
For 2D data, it can be rotated by 90 degrees counter-clockwise and / or transposed for visualization. These modifications
are applied directly to the data before it is sent to BEC.
Args:
name (str) : The name of the signal.
ndim (Literal[1, 2]) : The number of dimensions.
value (DeviceMonitorMessage | dict | None) : The initial value of the signal. Defaults to None.
name (str): The name of the signal.
ndim (Literal[1, 2]): The number of dimensions.
num_rotation_90 (Literal[0, 1, 2, 3]): The number of 90 degree counter-clockwise rotations to apply to the data for visualization.
transpose (bool): Whether to transpose the data for visualization.
value (DeviceMonitorMessage | dict | None): The initial value of the signal. Defaults to None.
"""
self.num_rotation_90 = num_rotation_90
self.transpose = transpose
kwargs.pop("kind", None)
super().__init__(
name=name,
data_type="raw",
saved=False,
ndim=ndim,
scope="scan",
role="preview",
value=value,
bec_message_type=messages.DevicePreviewMessage,
kind=Kind.omitted,
signal_metadata={
"ndim": ndim,
"num_rotation_90": num_rotation_90,
"transpose": transpose,
},
signal_metadata={"num_rotation_90": num_rotation_90, "transpose": transpose},
**kwargs,
)
def _process_data(self, value: np.ndarray) -> np.ndarray:
if self.ndim == 1:
return value
if self.num_rotation_90:
value = np.rot90(value, k=self.num_rotation_90, axes=(0, 1))
if self.transpose:
value = np.transpose(value)
return value
# pylint: disable=signature-differs
def put(
self,
@ -360,24 +561,38 @@ class PreviewSignal(BECMessageSignal):
value (list | np.ndarray | dict | self._bec_message_type): The preview data. Must be 1D.
metadata (dict | None): Additional metadata. If dict or self._bec_message_type is passed, it will be ignored.
"""
signal_name = self.dotted_name or self.name
device_name = self.parent.name
if isinstance(value, self._bec_message_type):
return super().put(value, **kwargs)
if isinstance(value, dict):
value["device"] = device_name
value["signal"] = signal_name
if isinstance(value, messages.DevicePreviewMessage):
value.data = self._process_data(value.data)
return super().put(value, **kwargs)
if isinstance(value, list):
value = np.array(value)
try:
msg = self._bec_message_type(
data=value, device=device_name, signal=signal_name, metadata=metadata
)
except ValidationError as exc:
raise ValueError(f"Error setting signal {self.name}: {exc}") from exc
super().put(msg, **kwargs)
if isinstance(value, dict):
if "data" not in value:
raise ValueError("Dictionary value must contain 'data' key.")
value["device"] = device_name
value["signal"] = signal_name
value["data"] = self._process_data(value["data"])
return super().put(value, **kwargs)
if isinstance(value, (list, np.ndarray)):
if not isinstance(value, np.ndarray):
value = np.array(value)
value = self._process_data(value)
try:
msg = messages.DevicePreviewMessage(
data=value, device=device_name, signal=signal_name, metadata=metadata
)
except ValidationError as exc:
raise ValueError(f"Error setting signal {self.name}: {exc}") from exc
return super().put(msg, **kwargs)
raise ValueError(
f"Value must be a {self._bec_message_type.__name__}, a dict, a list or a numpy array for signal {self.name}"
)
# pylint: disable=signature-differs
def set(
@ -390,8 +605,8 @@ class PreviewSignal(BECMessageSignal):
if value is a dict, it will be converted to a DevicePreviewMessage.
Args:
value (list | np.ndarray | dict | self._bec_message_type) : The preview data. Must be 1D.
metadata (dict | None) : Additional metadata. If dict or self._bec_message_type is passed, it will be ignored.
value (list | np.ndarray | dict | self._bec_message_type): The preview data. Must be 1D.
metadata (dict | None): Additional metadata. If dict or self._bec_message_type is passed, it will be ignored.
"""
self.put(value=value, metadata=metadata, **kwargs)
status = DeviceStatus(device=self)
@ -402,39 +617,39 @@ class PreviewSignal(BECMessageSignal):
class DynamicSignal(BECMessageSignal):
"""Signal to emit dynamic device data."""
@typechecked
strict_signal_validation = False # Disable strict signal validation
def __init__(
self,
*,
name: str,
signal_names: list[str] | Callable[[], list[str]],
signals: list[str] | Callable[[], list[str]] | None = None,
value: messages.DeviceMessage | dict | None = None,
async_update: dict | None = None,
**kwargs,
):
"""
Create a new DynamicSignal object.
Args:
name (str) : The name of the signal.
signal_names (list[str] | Callable) : The names of all signals. Can be a list or a callable.
value (DeviceMessage | dict | None) : The initial value of the signal. Defaults to None.
name (str): The name of the signal.
signal_names (list[str] | Callable): The names of all signals. Can be a list or a callable.
value (DeviceMessage | dict | None): The initial value of the signal. Defaults to None.
async_update (dict | None): Additional metadata for asynchronous updates. Defaults to None.
"""
_raise = False
if callable(signal_names):
self.signal_names = signal_names()
elif isinstance(signal_names, list):
self.signal_names = signal_names
else:
_raise = True
self.async_update = async_update
if _raise is True or len(self.signal_names) == 0:
raise ValueError(f"No names provided for Dynamic signal {name} via {signal_names}")
kwargs.pop("kind", None) # Ignore kind if specified
super().__init__(
name=name,
data_type=kwargs.pop("data_type", "raw"),
saved=kwargs.pop("saved", True),
ndim=kwargs.pop("ndim", 1),
scope=kwargs.pop("scope", "scan"),
role=kwargs.pop("role", "main"),
signals=signals,
value=value,
bec_message_type=messages.DeviceMessage,
kind=Kind.omitted,
bec_message_type=kwargs.pop("bec_message_type", messages.DeviceMessage),
**kwargs,
)
@ -443,6 +658,7 @@ class DynamicSignal(BECMessageSignal):
value: messages.DeviceMessage | dict[str, dict[Literal["value", "timestamp"], Any]],
*,
metadata: dict | None = None,
async_update: dict | None = None,
**kwargs,
) -> None:
"""
@ -453,13 +669,24 @@ class DynamicSignal(BECMessageSignal):
if value is a dict, it will be converted to a DeviceMessage.
Args:
value (dict | DeviceMessage) : The dynamic device data.
metadata (dict | None) : Additional metadata.
value (dict | DeviceMessage): The dynamic device data.
metadata (dict | None): Additional metadata.
"""
if isinstance(value, messages.DeviceMessage):
if metadata is not None or async_update is not None:
logger.warning(
"Ignoring metadata and async_update arguments when value is a DeviceMessage."
)
self._check_signals(value)
return super().put(value, **kwargs)
try:
metadata = metadata or {}
if "async_update" not in metadata:
if async_update is not None:
metadata["async_update"] = async_update
elif self.async_update is not None:
metadata["async_update"] = self.async_update
msg = messages.DeviceMessage(signals=value, metadata=metadata)
except ValidationError as exc:
raise ValueError(f"Error setting signal {self.name}: {exc}") from exc
@ -468,16 +695,26 @@ class DynamicSignal(BECMessageSignal):
def _check_signals(self, msg: messages.DeviceMessage) -> None:
"""Check if all signals are valid."""
if any(name not in self.signal_names for name in msg.signals):
raise ValueError(
f"Invalid signal name in message {list(msg.signals.keys())} for signals {self.signal_names}"
)
available_signals = [name for name, _ in self.signals]
if self.strict_signal_validation:
if set(msg.signals.keys()) != set(available_signals):
raise ValueError(
f"Signal names in message {list(msg.signals.keys())} do not match expected signals {available_signals}"
)
# If strict validation is disabled, we only check if the names are valid
# and not if all are present
else:
if any(name not in available_signals for name in msg.signals):
raise ValueError(
f"Invalid signal name in message {list(msg.signals.keys())} for signals {available_signals}"
)
def set(
self,
value: messages.DeviceMessage | dict[str, dict[Literal["value"], Any]],
*,
metadata: dict | None = None,
async_update: dict | None = None,
**kwargs,
) -> DeviceStatus:
"""
@ -491,7 +728,43 @@ class DynamicSignal(BECMessageSignal):
value (dict | DeviceMessage) : The dynamic device data.
metadata (dict | None) : Additional metadata.
"""
self.put(value, metadata=metadata, **kwargs)
self.put(value, metadata=metadata, async_update=async_update, **kwargs)
status = DeviceStatus(device=self)
status.set_finished()
return status
class AsyncSignal(DynamicSignal):
"""Signal to emit asynchronous data."""
strict_signal_validation = True
def __init__(
self,
*,
name: str,
ndim: Literal[0, 1, 2],
value: messages.DeviceMessage | dict | None = None,
async_update: dict | None = None,
**kwargs,
):
"""
Create a new AsyncSignal object.
Args:
name (str): The name of the signal.
value (AsyncMessage | dict | None): The initial value of the signal. Defaults to None.
"""
kwargs.pop("kind", None) # Ignore kind if specified
super().__init__(
name=name,
data_type="raw",
saved=True,
ndim=ndim,
scope="scan",
role="main",
value=value,
bec_message_type=messages.DeviceMessage,
async_update=async_update,
**kwargs,
)

83
tests/test_utils.py
View File

@ -221,7 +221,17 @@ def test_utils_bec_message_signal():
"source": "BECMessageSignal:bec_message_signal",
"dtype": "GUIInstructionMessage",
"shape": [],
"signal_metadata": {},
"signal_info": {
"data_type": "raw",
"saved": True,
"ndim": 0,
"scope": "scan",
"role": "main",
"enabled": True,
"rpc_access": False,
"signals": [("bec_message_signal", 5)],
"signal_metadata": None,
},
}
}
# Put works with Message
@ -246,20 +256,28 @@ def test_utils_bec_message_signal():
def test_utils_dynamic_signal():
"""Test DynamicSignal"""
dev = Device(name="device")
signal = DynamicSignal(
name="dynamic_signal", signal_names=["sig1", "sig2"], value=None, parent=dev
)
signal = DynamicSignal(name="dynamic_signal", signals=["sig1", "sig2"], value=None, parent=dev)
assert signal.parent == dev
assert signal._bec_message_type == messages.DeviceMessage
assert signal._readback is None
assert signal.name == "dynamic_signal"
assert signal.signal_names == ["sig1", "sig2"]
assert signal.signals == [("sig1", 1), ("sig2", 1)]
assert signal.describe() == {
"dynamic_signal": {
"source": "BECMessageSignal:dynamic_signal",
"dtype": "DeviceMessage",
"shape": [],
"signal_metadata": {},
"signal_info": {
"data_type": "raw",
"saved": True,
"ndim": 1,
"scope": "scan",
"role": "main",
"enabled": True,
"rpc_access": False,
"signals": [("sig1", 1), ("sig2", 1)],
"signal_metadata": None,
},
}
}
@ -295,9 +313,20 @@ def test_utils_file_event_signal():
"source": "BECMessageSignal:file_event_signal",
"dtype": "FileMessage",
"shape": [],
"signal_metadata": {},
"signal_info": {
"data_type": "raw",
"saved": False,
"ndim": 0,
"scope": "scan",
"role": "file event",
"enabled": True,
"rpc_access": False,
"signals": [("file_event_signal", 5)],
"signal_metadata": None,
},
}
}
# Test put works with FileMessage
msg_dict = {"file_path": "/path/to/another/file.txt", "done": False, "successful": True}
msg = messages.FileMessage(**msg_dict)
@ -325,7 +354,7 @@ def test_utils_preview_1d_signal():
"""Test Preview1DSignal"""
dev = Device(name="device")
signal = PreviewSignal(name="preview_1d_signal", ndim=1, value=None, parent=dev)
assert signal.signal_metadata.get("ndim") == 1
assert signal.ndim == 1
assert signal.parent == dev
assert signal._bec_message_type == messages.DevicePreviewMessage
assert signal._readback is None
@ -335,7 +364,17 @@ def test_utils_preview_1d_signal():
"source": "BECMessageSignal:preview_1d_signal",
"dtype": "DevicePreviewMessage",
"shape": [],
"signal_metadata": {"ndim": 1, "num_rotation_90": 0, "transpose": False},
"signal_info": {
"data_type": "raw",
"saved": False,
"ndim": 1,
"scope": "scan",
"role": "preview",
"enabled": True,
"rpc_access": False,
"signals": [("preview_1d_signal", 5)],
"signal_metadata": {"num_rotation_90": 0, "transpose": False},
},
}
}
# Put works with Message
@ -377,7 +416,7 @@ def test_utils_preview_2d_signal():
"""Test Preview2DSignal"""
dev = Device(name="device")
signal = PreviewSignal(name="preview_2d_signal", ndim=2, value=None, parent=dev)
assert signal.signal_metadata.get("ndim") == 2
assert signal.ndim == 2
assert signal.parent == dev
assert signal._bec_message_type == messages.DevicePreviewMessage
assert signal._readback is None
@ -387,7 +426,17 @@ def test_utils_preview_2d_signal():
"source": "BECMessageSignal:preview_2d_signal",
"dtype": "DevicePreviewMessage",
"shape": [],
"signal_metadata": {"ndim": 2, "num_rotation_90": 0, "transpose": False},
"signal_info": {
"data_type": "raw",
"saved": False,
"ndim": 2,
"scope": "scan",
"role": "preview",
"enabled": True,
"rpc_access": False,
"signals": [("preview_2d_signal", 5)],
"signal_metadata": {"num_rotation_90": 0, "transpose": False},
},
}
}
# Put works with Message
@ -442,7 +491,17 @@ def test_utils_progress_signal():
"source": "BECMessageSignal:progress_signal",
"dtype": "ProgressMessage",
"shape": [],
"signal_metadata": {},
"signal_info": {
"data_type": "raw",
"saved": False,
"ndim": 0,
"scope": "scan",
"role": "progress",
"enabled": True,
"rpc_access": False,
"signals": [("progress_signal", 5)],
"signal_metadata": None,
},
}
}
# Put works with Message