Allow CA/PVA mismatching enums to be bools (#632)

src/ophyd_async/epics/signal/_aioca.py

@@ -175,16 +175,19 @@ def make_converter(
     if is_array and pv_dbr == dbr.DBR_CHAR and datatype is str:
         # Override waveform of chars to be treated as string
         return CaLongStrConverter()
+    elif not is_array and datatype is bool and pv_dbr == dbr.DBR_ENUM:
+        # Database can't do bools, so are often representated as enums of len 2
+        pv_num_choices = get_unique(
+            {k: len(v.enums) for k, v in values.items()}, "number of choices"
+        )
+        if pv_num_choices != 2:
+            raise TypeError(f"{pv} has {pv_num_choices} choices, can't map to bool")
+        return CaBoolConverter()
     elif not is_array and pv_dbr == dbr.DBR_ENUM:
         pv_choices = get_unique(
             {k: tuple(v.enums) for k, v in values.items()}, "choices"
         )
-        if datatype is bool:
-            # Database can't do bools, so are often representated as enums of len 2
-            if len(pv_choices) != 2:
-                raise TypeError(f"{pv} has {pv_choices=}, can't map to bool")
-            return CaBoolConverter()
-        elif enum_cls := get_enum_cls(datatype):
+        if enum_cls := get_enum_cls(datatype):
             # If explicitly requested then check
             return CaEnumConverter(get_supported_values(pv, enum_cls, pv_choices))
         elif datatype in (None, str):

src/ophyd_async/epics/signal/_p4p.py

@@ -212,16 +212,20 @@ def make_converter(datatype: type | None, values: dict[str, Any]) -> PvaConverte
         (typeid, specifier)
     ]
     # Some override cases
-    if typeid == "epics:nt/NTEnum:1.0":
+    if datatype is bool and typeid == "epics:nt/NTEnum:1.0":
+        # Database can't do bools, so are often representated as enums of len 2
+        pv_num_choices = get_unique(
+            {k: len(v["value"]["choices"]) for k, v in values.items()},
+            "number of choices",
+        )
+        if pv_num_choices != 2:
+            raise TypeError(f"{pv} has {pv_num_choices} choices, can't map to bool")
+        return PvaEnumBoolConverter()
+    elif typeid == "epics:nt/NTEnum:1.0":
         pv_choices = get_unique(
             {k: tuple(v["value"]["choices"]) for k, v in values.items()}, "choices"
         )
-        if datatype is bool:
-            # Database can't do bools, so are often representated as enums of len 2
-            if len(pv_choices) != 2:
-                raise TypeError(f"{pv} has {pv_choices=}, can't map to bool")
-            return PvaEnumBoolConverter()
-        elif enum_cls := get_enum_cls(datatype):
+        if enum_cls := get_enum_cls(datatype):
             # We were given an enum class, so make class from that
             return PvaEnumConverter(
                 supported_values=get_supported_values(pv, enum_cls, pv_choices)

tests/epics/demo/test_demo.py

@@ -293,9 +293,23 @@ async def test_assembly_renaming() -> None:
 
 
 async def test_dynamic_sensor_group_disconnected():
-    with pytest.raises(NotConnected):
+    with pytest.raises(NotConnected) as e:
         async with DeviceCollector(timeout=0.1):
             mock_sensor_group_dynamic = demo.SensorGroup("MOCK:SENSOR:")
+    expected = """
+mock_sensor_group_dynamic: NotConnected:
+    sensors: NotConnected:
+        1: NotConnected:
+            value: NotConnected: ca://MOCK:SENSOR:1:Value
+            mode: NotConnected: ca://MOCK:SENSOR:1:Mode
+        2: NotConnected:
+            value: NotConnected: ca://MOCK:SENSOR:2:Value
+            mode: NotConnected: ca://MOCK:SENSOR:2:Mode
+        3: NotConnected:
+            value: NotConnected: ca://MOCK:SENSOR:3:Value
+            mode: NotConnected: ca://MOCK:SENSOR:3:Mode
+"""
+    assert str(e.value) == expected
 
     assert mock_sensor_group_dynamic.name == "mock_sensor_group_dynamic"
 

tests/epics/signal/test_signals.py

@@ -902,6 +902,12 @@ async def test_signals_created_for_not_prec_0_float_cannot_use_int(ioc: IOC):
         await sig.connect()
 
 
+async def test_bool_works_for_mismatching_enums(ioc: IOC):
+    pv_name = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:bool"
+    sig = epics_signal_rw(bool, pv_name, pv_name + "_unnamed")
+    await sig.connect()
+
+
 async def test_can_read_using_ophyd_async_then_ophyd(ioc: IOC):
     oa_read = f"{ioc.protocol}://{PV_PREFIX}:{ioc.protocol}:float_prec_1"
     ophyd_read = f"{PV_PREFIX}:{ioc.protocol}:float_prec_0"