Support numpy floating and integer dtypes in SimConverter

test numpy dtype with SimSignalBackend

src/ophyd_async/core/signal.py

@@ -60,9 +60,7 @@ def set_name(self, name: str = ""):
 
     async def connect(self, sim=False, timeout=DEFAULT_TIMEOUT):
         if sim:
-            self._backend = SimSignalBackend(
-                datatype=self._init_backend.datatype
-            )
+            self._backend = SimSignalBackend(datatype=self._init_backend.datatype)
             _sim_backends[self] = self._backend
         else:
             self._backend = self._init_backend
@@ -254,30 +252,30 @@ def set_sim_callback(signal: Signal[T], callback: ReadingValueCallback[T]) -> No
 
 
 def soft_signal_rw(
-    datatype: Optional[Type[T]],
-    name: str,
+    datatype: Optional[Type[T]] = None,
     initial_value: Optional[T] = None,
+    name: Optional[str] = None,
 ) -> SignalRW[T]:
     """Creates a read-writable Signal with a SimSignalBackend"""
-    signal = SignalRW(
-        SimSignalBackend(datatype, initial_value)
-    )
-    signal.set_name(name)
+    signal = SignalRW(SimSignalBackend(datatype, initial_value))
+    if name is not None:
+        signal.set_name(name)
     return signal
 
 
 def soft_signal_r_and_backend(
-    datatype: Optional[Type[T]],
-    name: str,
+    datatype: Optional[Type[T]] = None,
     initial_value: Optional[T] = None,
+    name: Optional[str] = None,
 ) -> Tuple[SignalR[T], SimSignalBackend]:
     """Returns a tuple of a read-only Signal and its SimSignalBackend through
     which the signal can be internally modified within the device. Use
     soft_signal_rw if you want a device that is externally modifiable
     """
     backend = SimSignalBackend(datatype, initial_value)
     signal = SignalR(backend)
-    signal.set_name(name)
+    if name is not None:
+        signal.set_name(name)
     return (signal, backend)
 
 

src/ophyd_async/core/sim_signal_backend.py

@@ -8,6 +8,7 @@
 from enum import Enum
 from typing import Any, Dict, Generic, Optional, Type, Union, cast, get_origin
 
+import numpy as np
 from bluesky.protocols import Descriptor, Dtype, Reading
 
 from .signal_backend import SignalBackend
@@ -36,11 +37,16 @@ def reading(self, value: T, timestamp: float, severity: int) -> Reading:
         )
 
     def descriptor(self, source: str, value) -> Descriptor:
+        dtype = type(value)
+        if np.issubdtype(dtype, np.integer):
+            dtype = int
+        elif np.issubdtype(dtype, np.floating):
+            dtype = float
         assert (
-            type(value) in primitive_dtypes
+            dtype in primitive_dtypes
         ), f"invalid converter for value of type {type(value)}"
-        dtype = primitive_dtypes[type(value)]
-        return {"source": source, "dtype": dtype, "shape": []}
+        dtype_name = primitive_dtypes[dtype]
+        return {"source": source, "dtype": dtype_name, "shape": []}
 
     def make_initial_value(self, datatype: Optional[Type[T]]) -> T:
         if datatype is None:

tests/core/test_signal.py

@@ -2,6 +2,7 @@
 import re
 import time
 
+import numpy
 import pytest
 
 from ophyd_async.core import (
@@ -132,9 +133,9 @@ async def test_create_soft_signal(signal_method, signal_class):
     SIGNAL_NAME = "TEST-PREFIX:SIGNAL"
     INITIAL_VALUE = "INITIAL"
     if signal_method == soft_signal_r_and_backend:
-        signal, backend = signal_method(str, SIGNAL_NAME, INITIAL_VALUE)
+        signal, backend = signal_method(str, INITIAL_VALUE, SIGNAL_NAME)
     elif signal_method == soft_signal_rw:
-        signal = signal_method(str, SIGNAL_NAME, INITIAL_VALUE)
+        signal = signal_method(str, INITIAL_VALUE, SIGNAL_NAME)
         backend = signal._backend
     assert signal.source == f"soft://{SIGNAL_NAME}"
     assert isinstance(signal, signal_class)
@@ -143,3 +144,12 @@ async def test_create_soft_signal(signal_method, signal_class):
     assert (await signal.get_value()) == INITIAL_VALUE
     # connecting with sim=False uses existing SimSignalBackend
     assert signal._backend is backend
+
+
+async def test_soft_signal_numpy():
+    float_signal = soft_signal_rw(numpy.float64, numpy.float64(1), "float_signal")
+    int_signal = soft_signal_rw(numpy.int32, numpy.int32(1), "int_signal")
+    await float_signal.connect()
+    await int_signal.connect()
+    assert (await float_signal.describe())["float_signal"]["dtype"] == "number"
+    assert (await int_signal.describe())["int_signal"]["dtype"] == "integer"

tests/core/test_sim.py

@@ -100,10 +100,9 @@ async def test_backend_get_put_monitor(
     try:
         # Check descriptor
         source = "soft://test"
-        assert (
-            dict(source=source, **descriptor(initial_value))
-            == await backend.get_descriptor(source)
-        )
+        assert dict(
+            source=source, **descriptor(initial_value)
+        ) == await backend.get_descriptor(source)
         # Check initial value
         await q.assert_updates(
             pytest.approx(initial_value) if initial_value != "" else initial_value

tests/panda/test_writer.py

@@ -69,12 +69,8 @@ async def sim_writer(tmp_path, sim_panda) -> PandaHDFWriter:
 async def test_get_capture_signals_gets_all_signals(sim_panda):
     async with DeviceCollector(sim=True):
         sim_panda.test_seq = Device("seq")
-        sim_panda.test_seq.seq1_capture = SignalR(
-            backend=SimSignalBackend(str)
-        )
-        sim_panda.test_seq.seq2_capture = SignalR(
-            backend=SimSignalBackend(str)
-        )
+        sim_panda.test_seq.seq1_capture = SignalR(backend=SimSignalBackend(str))
+        sim_panda.test_seq.seq2_capture = SignalR(backend=SimSignalBackend(str))
         await asyncio.gather(
             sim_panda.test_seq.connect(),
             sim_panda.test_seq.seq1_capture.connect(),