Supported pva dicts and fixed tests

src/ophyd_async/epics/_backend/_p4p.py

@@ -1,6 +1,7 @@
 import asyncio
 import atexit
 import logging
+import time
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Dict, List, Optional, Sequence, Type, Union
@@ -119,9 +120,7 @@ def value(self, value):
 
     def descriptor(self, source: str, value) -> Descriptor:
         choices = [e.value for e in self.enum_class]
-        return dict(
-            source=source, dtype="string", shape=[], choices=choices
-        )  # type: ignore
+        return dict(source=source, dtype="string", shape=[], choices=choices)
 
 
 class PvaEnumBoolConverter(PvaConverter):
@@ -142,9 +141,14 @@ def descriptor(self, source: str, value) -> Descriptor:
 
 
 class PvaDictConverter(PvaConverter):
+    def reading(self, value):
+        value = value.todict()
+        ts = time.time()
+        # Alarm severity is vacuously 0 for a table
+        return dict(value=value, timestamp=ts, alarm_severity=0)
 
-    def value(self, value):
-        return value
+    def value(self, value: Value):
+        return value.todict()
 
     def descriptor(self, source: str, value) -> Descriptor:
         raise NotImplementedError("Describing Dict signals not currently supported")

tests/epics/test_signals.py

@@ -20,13 +20,13 @@
     Type,
     TypedDict,
 )
+from unittest.mock import ANY
 
 import numpy as np
 import numpy.typing as npt
 import pytest
 from aioca import purge_channel_caches
 from bluesky.protocols import Reading
-
 from ophyd_async.core import SignalBackend, T, get_dtype, load_from_yaml, save_to_yaml
 from ophyd_async.core.utils import NotConnected
 from ophyd_async.epics.signal._epics_transport import EpicsTransport
@@ -100,17 +100,12 @@ async def assert_updates(self, expected_value):
             "timestamp": pytest.approx(time.time(), rel=0.1),
             "alarm_severity": 0,
         }
+        backend_reading = await asyncio.wait_for(self.backend.get_reading(), timeout=5)
         reading, value = await asyncio.wait_for(self.updates.get(), timeout=5)
-        assert (
-            value
-            == expected_value
-            == await asyncio.wait_for(self.backend.get_value(), timeout=5)
-        )
-        assert (
-            reading
-            == expected_reading
-            == await asyncio.wait_for(self.backend.get_reading(), timeout=5)
-        )
+        backend_value = await asyncio.wait_for(self.backend.get_value(), timeout=5)
+
+        assert value == expected_value == backend_value
+        assert reading == expected_reading == backend_reading
 
     def close(self):
         self.backend.set_callback(None)
@@ -350,8 +345,10 @@ async def test_pvi_structure(ioc: IOC) -> None:
         return
     # Make and connect the backend
     backend = await ioc.make_backend(Dict[str, Any], "pvi")
+
     # Make a monitor queue that will monitor for updates
     q = MonitorQueue(backend)
+
     expected = {
         "pvi": {
             "width": {
@@ -360,8 +357,10 @@ async def test_pvi_structure(ioc: IOC) -> None:
             "height": {
                 "rw": f"{PV_PREFIX}:{ioc.protocol}:height",
             },
-        }
+        },
+        "record": ANY,
     }
+
     try:
         # Check descriptor
         with pytest.raises(NotImplementedError):