Merge pull request #616 from ZLLentz/fix_enum_handling

FIX: metadata handling for enums

docs/source/upcoming_release_notes/616-fix_enum_handling.rst

@@ -0,0 +1,23 @@
+616 fix_enum_handling
+#################
+
+API Breaks
+----------
+- N/A
+
+Features
+--------
+- N/A
+
+Bugfixes
+--------
+- Fix various issues with enum handling in the SignalPlugin.
+
+Maintenance
+-----------
+- N/A
+
+Contributors
+------------
+- canismarko
+- zllentz

typhos/plugins/core.py

@@ -80,9 +80,11 @@ class SignalConnection(PyDMConnection):
     def __init__(self, channel, address, protocol=None, parent=None):
         # Create base connection
         super().__init__(channel, address, protocol=protocol, parent=parent)
-        self._connection_open = True
-        self.signal_type = None
-        self.is_float = False
+        self._connection_open: bool = True
+        self.signal_type: type | None = None
+        self.is_float: bool = False
+        self.enum_strs: tuple[str, ...] = ()
+
         # Collect our signal
         self.signal = self.find_signal(address)
         # Subscribe to updates from Ophyd
@@ -143,7 +145,26 @@ def cast(self, value):
                          dtype, self.signal.name, self.signal_type)
 
         logger.debug("Casting %r to %r", value, self.signal_type)
-        if self.signal_type is np.ndarray:
+        if self.enum_strs:
+            # signal_type is either int or str
+            # use enums to cast type
+            if self.signal_type is int:
+                # Get the index
+                try:
+                    value = self.enum_strs.index(value)
+                except (TypeError, ValueError, AttributeError):
+                    value = int(value)
+            elif self.signal_type is str:
+                # Get the enum string
+                try:
+                    value = self.enum_strs[value]
+                except (TypeError, ValueError):
+                    value = str(value)
+            else:
+                raise TypeError(
+                    f"Invalid combination: enum_strs={self.enum_strs} with signal_type={self.signal_type}"
+                )
+        elif self.signal_type is np.ndarray:
             value = np.asarray(value)
         else:
             value = self.signal_type(value)
@@ -227,6 +248,7 @@ def send_new_meta(
             self.unit_signal.emit(units)
         if enum_strs is not None:
             self.enum_strings_signal.emit(enum_strs)
+            self.enum_strs = enum_strs
 
         # Special handling for severity
         if severity is None:
@@ -266,9 +288,9 @@ def add_listener(self, channel):
         else:
             self.is_float = False
 
-        # Report new value
-        self.send_new_value(signal_val)
+        # Report new meta for context, then value
         self.send_new_meta(**signal_meta)
+        self.send_new_value(signal_val)
         # If the channel is used for writing to PVs, hook it up to the
         # 'put' methods.
         if channel.value_signal is not None:

typhos/tests/plugins/test_core.py

@@ -5,8 +5,9 @@
 import pytest
 from ophyd import Component as Cpt
 from ophyd import Device, Signal
+from ophyd.sim import EnumSignal
 from pydm import PyDMApplication
-from pydm.widgets import PyDMLineEdit
+from pydm.widgets import PyDMChannel, PyDMLineEdit
 from pytestqt.qtbot import QtBot
 
 from typhos.plugins.core import (SignalConnection, register_signal,
@@ -211,3 +212,53 @@ def test_array_signal_put_value(qapp, qtbot):
     widget.send_value_signal[np.ndarray].emit(np.zeros(4))
     qapp.processEvents()
     assert all(sig.get() == np.zeros(4))
+
+
+def test_add_listener_order(qapp):
+    sig = EnumSignal(name="my_listener", value=0, enum_strings=("zero", "one", "two"))
+    register_signal(sig)
+
+    order = []
+
+    def new_value(*args, **kwargs):
+        order.append("value")
+
+    def new_meta(*args, **kwargs):
+        order.append("meta")
+
+    chan = PyDMChannel(address="sig://my_listener", value_slot=new_value, enum_strings_slot=new_meta)
+    _ = SignalConnection(chan, "my_listener", "sig")
+    qapp.processEvents()
+
+    assert order == ["meta", "value"]
+
+
+def test_enum_casts(qapp):
+    sig = Signal(name="my_enum_caster", value=0)
+    register_signal(sig)
+
+    chan = PyDMChannel(address="sig://my_enum_caster")
+    conn = SignalConnection(chan, "my_enum_caster", "sig")
+    qapp.processEvents()
+
+    conn.enum_strs = ("1", "2", "5")
+
+    # First, keep type as an int
+    assert conn.cast("1") == 0
+    assert conn.cast("2") == 1
+    assert conn.cast("5") == 2
+    assert conn.cast(0) == 0
+    assert conn.cast(1) == 1
+    assert conn.cast(2) == 2
+
+    # Try str next
+    conn.signal_type = None
+    sig.put("2")
+    qapp.processEvents()
+
+    assert conn.cast("1") == "1"
+    assert conn.cast("2") == "2"
+    assert conn.cast("5") == "5"
+    assert conn.cast(0) == "1"
+    assert conn.cast(1) == "2"
+    assert conn.cast(2) == "5"