diff --git a/ophyd_devices/devices/panda_box/panda_box.py b/ophyd_devices/devices/panda_box/panda_box.py index 117d2b9..b1c42ab 100644 --- a/ophyd_devices/devices/panda_box/panda_box.py +++ b/ophyd_devices/devices/panda_box/panda_box.py @@ -49,7 +49,11 @@ from pandablocks.connections import DataConnection, NeedMoreDataError from pandablocks.responses import Data, EndData, FrameData, ReadyData, StartData from ophyd_devices import DynamicSignal, PSIDeviceBase, StatusBase -from ophyd_devices.devices.panda_box.utils import get_pcap_capture_fields +from ophyd_devices.devices.panda_box.utils import ( + block_name_mapping, + get_pcap_capture_fields, + is_valid_attribute_name, +) if TYPE_CHECKING: # pragma: no cover from bec_lib.devicemanager import ScanInfo @@ -206,7 +210,14 @@ class PandaBox(PSIDeviceBase): mapped to the provided signal names. If data is received for a signal that is not included in the signal_alias, the original name from the PandaBox will be used as the signal name. Signal config should be provided as a dict with keys corresponding to the signal names from the PandaBox, and values corresponding to the desired - signal names to be used in the data frames. + signal names to be used in the data frames. Values of the corresponding signal names must be valid Python + attribute names, and may not contain dots or other special characters. + + Args: + name (str): The name of the device. + host (str): The hostname of the PandaBox to connect to. + signal_alias (dict[str, str], optional): A mapping from PandaBox signal names to desired signal names for the beamline. Defaults to None. + Mapped keys may not contain dots or other special characters. """ data = Cpt( @@ -230,7 +241,7 @@ class PandaBox(PSIDeviceBase): device_manager: DeviceManagerDS | None = None, **kwargs, ) -> None: - self.signal_alias = signal_alias if signal_alias is not None else {} + self.signal_alias = self._convert_signal_aliases(signal_alias) if signal_alias else {} kwargs.pop( "signal_alias", None ) # Remove signal_alias from kwargs to avoid issues with super().__init__() @@ -261,10 +272,23 @@ class PandaBox(PSIDeviceBase): self._stage_timeout_in_s = 3 # Call super().__init__() here to ensure on_init in base class gets called after the PandaBox specific initialization. super().__init__(name=name, scan_info=scan_info, device_manager=device_manager, **kwargs) + self._apply_signal_aliases() - def on_init(self): + def _convert_signal_aliases(self, signal_alias: dict[str, str]) -> dict[str, str]: + """Convert signal names""" + out = {} + for block_key, signal_name in signal_alias.items(): + if not is_valid_attribute_name(signal_name): + raise ValueError( + f"Invalid signal name in signal_alias: '{signal_name}' for block '{block_key}'. " + f"Signal names must be valid Python attribute names and may not contain dots or special characters." + ) + out[block_name_mapping(block_key)] = signal_name + + return out + + def _apply_signal_aliases(self): """Initialize the PandaBox device. This method can be used to perform any additional initialization logic.""" - super().on_init() new_names = [ self.signal_alias.get(original_name, original_name) for original_name, _ in self.data.signals @@ -711,7 +735,11 @@ class PandaBox(PSIDeviceBase): def convert_frame_data(self, frame_data: FrameData) -> dict[str, Any]: """ Convert the data from a FrameData object into a dictionary with expected OPHYD - read format, e.g. signal {signal_name: {"value": [...]}}. + read format, e.g. signal {signal_name: {"value": [...]}}. Please be aware that if + this method is overriden by child classes, you need to make sure that the key names + in the FrameData is converted to the expected names in the data signal. This includes + replacing dots in the original PandaBox keys with underscores using "block_name_mapping" + from the utils, and using the key mapping provided through the signal_alias mapping. Args: frame_data (FrameData): The FrameData object received from the PandaBox. @@ -723,9 +751,13 @@ class PandaBox(PSIDeviceBase): out = {} data = frame_data.data keys = data.dtype.names - # Map keys if mapping is provided + # Converting the data requires us to map the keys of the PandaBox data to follow valid + # Python attribute names, but also to match the renamed signals provided by the signal_alias mapping. + # The mapping is done in two steps: + # I. Remove dots '.' from Pandablock keys as they are not valid for Python attribute names + keys = [block_name_mapping(key) for key in keys] + # Map keys if mapping is provided. We also need to translate all keys received from the mapped_key = [self.signal_alias.get(key, key) for key in keys] - # Initialize lists for each key, consider adjusting names to match for k in mapped_key: out[k] = {"value": [], "timestamp": time.time()} for entry in data: diff --git a/ophyd_devices/devices/panda_box/utils.py b/ophyd_devices/devices/panda_box/utils.py index 4ff185c..024c09b 100644 --- a/ophyd_devices/devices/panda_box/utils.py +++ b/ophyd_devices/devices/panda_box/utils.py @@ -1,3 +1,7 @@ +"""Module containing utility functions for the PandaBox device.""" + +import keyword + PANDA_AVAIL_PCAP_BLOCKS = [ "INENC1.VAL", "INENC2.VAL", @@ -42,13 +46,20 @@ PANDA_AVAIL_PCAP_BLOCKS = [ PANDA_AVAIL_PCAP_CAPTURE_FIELDS = ["Value", "Diff", "Sum", "Mean", "Min", "Max"] +def is_valid_attribute_name(name: str) -> bool: + """Check if a given name is a valid Python attribute name.""" + return name.isidentifier() and not keyword.iskeyword(name) + + +def block_name_mapping(block_name: str) -> str: + """Map block names to a format suitable for use as attribute names.""" + return block_name.replace(".", "_") + + def get_pcap_capture_fields(): out = [] for block in PANDA_AVAIL_PCAP_BLOCKS: for field in PANDA_AVAIL_PCAP_CAPTURE_FIELDS: - # Consider this mapping, and alsock - # block_name = f"{block}.{field}" - # block_name = block.replace(".", "_") - # out.append(block_name) TODO - If applied Adapt 'convert_frame_data' method in panda_box.py to handle this mapping - out.append(f"{block}.{field}") + block_name = block_name_mapping(f"{block}.{field}") + out.append(block_name) return out diff --git a/tests/test_panda.py b/tests/test_panda.py index e69b946..55d3f19 100644 --- a/tests/test_panda.py +++ b/tests/test_panda.py @@ -20,6 +20,7 @@ from ophyd_devices.devices.panda_box.panda_box import ( from ophyd_devices.devices.panda_box.utils import ( PANDA_AVAIL_PCAP_BLOCKS, PANDA_AVAIL_PCAP_CAPTURE_FIELDS, + block_name_mapping, get_pcap_capture_fields, ) @@ -34,6 +35,18 @@ def panda_box(_signal_aliases): return PandaBox(name="panda_box", host="localhost", signal_alias=_signal_aliases) +def test_panda_box_init_invalid_signal_alias(): + """Test that providing invalid signal aliases raises an error.""" + with pytest.raises( + ValueError, match="Invalid signal name in signal_alias: 'Invalid.Signal.Name'" + ): + PandaBox( + name="panda_box", + host="localhost", + signal_alias={"FMC_IN.VAL1.Value": "Invalid.Signal.Name"}, + ) + + def test_panda_box_init(panda_box, _signal_aliases): """Test initialization of PandaBox, including default signal aliases.""" assert panda_box.name == "panda_box" @@ -46,7 +59,9 @@ def test_panda_box_init(panda_box, _signal_aliases): # These signals should be renamed assert _signal_aliases[signal_name] in all_signal_names continue - assert signal_name in all_signal_names, f"Missing signal: {signal_name}" + assert ( + block_name_mapping(signal_name) in all_signal_names + ), f"Missing signal: {signal_name}" def test_panda_wait_for_connection(panda_box): @@ -149,7 +164,7 @@ def test_panda_receive_frame_data(panda_box, _signal_aliases): "value": [np.float64(0), np.float64(1), np.float64(2)], "timestamp": mock.ANY, }, - f"{panda_box.data.name}_COUNTER2.OUT.Value": { + f"{panda_box.data.name}_COUNTER2_OUT_Value": { "value": [np.float64(10), np.float64(11), np.float64(12)], "timestamp": mock.ANY, }, @@ -286,6 +301,7 @@ def test_panda_get_signal_names_configured_for_capture(panda_box): with mock.patch.object(panda_box, "send_raw") as mock_send_raw: mock_send_raw.return_value = return_capture list_of_signals = panda_box._get_signal_names_configured_for_capture() + list_of_signals = [block_name_mapping(name) for name in list_of_signals] mock_send_raw.assert_called_once_with("*CAPTURE?") for signal in list_of_signals: assert signal in possible_signal_names, f"Unexpected signal: {signal}" @@ -315,7 +331,7 @@ def test_panda_get_pcap_capture_fields(): expected_fields = [] for block in PANDA_AVAIL_PCAP_BLOCKS: for field in PANDA_AVAIL_PCAP_CAPTURE_FIELDS: - expected_fields.append(f"{block}.{field}") + expected_fields.append(block_name_mapping(f"{block}.{field}")) actual_fields = get_pcap_capture_fields() assert actual_fields == expected_fields, "PCAP capture fields mismatch"