New signal typing

pyproject.toml

@@ -13,12 +13,12 @@ classifiers = [
 description = "Asynchronous Bluesky hardware abstraction code, compatible with control systems like EPICS and Tango"
 dependencies = [
     "networkx>=2.0",
-    "numpy<2.0.0",
+    "numpy",
     "packaging",
     "pint",
     "bluesky>=1.13.0a3",
-    "event_model",
-    "p4p",
+    "event-model @ git+https://github.com/bluesky/event-model@main",
+    "p4p>=4.2.0a3",
     "pyyaml",
     "colorlog",
     "pydantic>=2.0",
@@ -37,10 +37,6 @@ dev = [
     "ophyd_async[pva]",
     "ophyd_async[sim]",
     "ophyd_async[ca]",
-    "black",
-    "flake8",
-    "flake8-isort",
-    "Flake8-pyproject",
     "inflection",
     "ipython",
     "ipywidgets",

src/ophyd_async/core/__init__.py

@@ -19,7 +19,6 @@
 from ._flyer import StandardFlyer, TriggerLogic
 from ._hdf_dataset import HDFDataset, HDFFile
 from ._log import config_ophyd_async_logging
-from ._mock_signal_backend import MockSignalBackend
 from ._mock_signal_utils import (
     callback_on_mock_put,
     get_mock_put,
@@ -62,22 +61,33 @@
     wait_for_value,
 )
 from ._signal_backend import (
-    RuntimeSubsetEnum,
+    Array1D,
     SignalBackend,
-    SubsetEnum,
+    SignalConnector,
+    SignalDatatype,
+    SignalDatatypeT,
+    make_datakey,
+)
+from ._soft_signal_backend import (
+    MockSignalBackend,
+    SignalMetadata,
+    SoftSignalBackend,
+    SoftSignalConnector,
 )
-from ._soft_signal_backend import SignalMetadata, SoftSignalBackend
 from ._status import AsyncStatus, WatchableAsyncStatus, completed_status
 from ._table import Table
 from ._utils import (
     CALCULATE_TIMEOUT,
     DEFAULT_TIMEOUT,
     CalculatableTimeout,
+    Callback,
     NotConnected,
-    ReadingValueCallback,
+    StrictEnum,
+    SubsetEnum,
     T,
     WatcherUpdate,
     get_dtype,
+    get_enum_cls,
     get_unique,
     in_micros,
     is_pydantic_model,
@@ -146,22 +156,29 @@
     "soft_signal_r_and_setter",
     "soft_signal_rw",
     "wait_for_value",
-    "RuntimeSubsetEnum",
+    "Array1D",
     "SignalBackend",
+    "SignalConnector",
+    "make_datakey",
+    "StrictEnum",
     "SubsetEnum",
+    "SignalDatatype",
+    "SignalDatatypeT",
     "SignalMetadata",
     "SoftSignalBackend",
+    "SoftSignalConnector",
     "AsyncStatus",
     "WatchableAsyncStatus",
     "DEFAULT_TIMEOUT",
     "CalculatableTimeout",
+    "Callback",
     "CALCULATE_TIMEOUT",
     "NotConnected",
-    "ReadingValueCallback",
     "Table",
     "T",
     "WatcherUpdate",
     "get_dtype",
+    "get_enum_cls",
     "get_unique",
     "in_micros",
     "is_pydantic_model",

src/ophyd_async/core/_device.py

@@ -1,39 +1,63 @@
-"""Base device"""
+from __future__ import annotations
 
 import asyncio
 import sys
-from collections.abc import Coroutine, Generator, Iterator
+from abc import abstractmethod
+from collections.abc import Callable, Coroutine, Iterator, Mapping
 from functools import cached_property
 from logging import LoggerAdapter, getLogger
-from typing import (
-    Any,
-    Optional,
-    TypeVar,
-)
+from typing import Any, Generic, TypeVar, cast
 
 from bluesky.protocols import HasName
 from bluesky.run_engine import call_in_bluesky_event_loop
 
 from ._utils import DEFAULT_TIMEOUT, NotConnected, wait_for_connection
 
 
-class Device(HasName):
+class DeviceConnector:
+    @abstractmethod
+    async def connect(
+        self, device: Device, mock: bool, timeout: float, force_reconnect: bool
+    ) -> None: ...
+
+
+class DeviceChildConnector(DeviceConnector):
+    async def connect(
+        self, device: Device, mock: bool, timeout: float, force_reconnect: bool
+    ) -> Any:
+        coros = {
+            name: child_device.connect(mock, timeout, force_reconnect)
+            for name, child_device in device.children().items()
+        }
+        await wait_for_connection(**coros)
+
+
+DeviceConnectorType = TypeVar("DeviceConnectorType", bound=DeviceConnector)
+
+
+class Device(HasName, Generic[DeviceConnectorType]):
     """Common base class for all Ophyd Async Devices.
 
     By default, names and connects all Device children.
     """
 
     _name: str = ""
     #: The parent Device if it exists
-    parent: Optional["Device"] = None
+    parent: Device | None = None
     # None if connect hasn't started, a Task if it has
     _connect_task: asyncio.Task | None = None
+    # The value of the mock arg to connect
+    _connect_mock: bool | None = None
+    # The connector to use
+    _connector: DeviceConnectorType = DeviceChildConnector()
 
-    # Used to check if the previous connect was mocked,
-    # if the next mock value differs then we fail
-    _previous_connect_was_mock = None
-
-    def __init__(self, name: str = "") -> None:
+    def __init__(
+        self,
+        name: str = "",
+        connector: DeviceConnectorType | None = None,
+    ) -> None:
+        if connector is not None:
+            self._connector = connector
         self.set_name(name)
 
     @property
@@ -47,10 +71,12 @@ def log(self):
             getLogger("ophyd_async.devices"), {"ophyd_async_device_name": self.name}
         )
 
-    def children(self) -> Iterator[tuple[str, "Device"]]:
-        for attr_name, attr in self.__dict__.items():
-            if attr_name != "parent" and isinstance(attr, Device):
-                yield attr_name, attr
+    def children(self) -> dict[str, Device]:
+        return {
+            attr_name: attr
+            for attr_name, attr in self.__dict__.items()
+            if attr_name != "parent" and isinstance(attr, Device)
+        }
 
     def set_name(self, name: str):
         """Set ``self.name=name`` and each ``self.child.name=name+"-child"``.
@@ -66,7 +92,7 @@ def set_name(self, name: str):
             del self.log
 
         self._name = name
-        for attr_name, child in self.children():
+        for attr_name, child in self.children().items():
             child_name = f"{name}-{attr_name.rstrip('_')}" if name else ""
             child.set_name(child_name)
             child.parent = self
@@ -89,40 +115,27 @@ async def connect(
             Time to wait before failing with a TimeoutError.
         """
 
-        if (
-            self._previous_connect_was_mock is not None
-            and self._previous_connect_was_mock != mock
-        ):
-            raise RuntimeError(
-                f"`connect(mock={mock})` called on a `Device` where the previous "
-                f"connect was `mock={self._previous_connect_was_mock}`. Changing mock "
-                "value between connects is not permitted."
-            )
-        self._previous_connect_was_mock = mock
-
         # If previous connect with same args has started and not errored, can use it
-        can_use_previous_connect = self._connect_task and not (
-            self._connect_task.done() and self._connect_task.exception()
+        can_use_previous_connect = (
+            mock is self._connect_mock
+            and self._connect_task
+            and not (self._connect_task.done() and self._connect_task.exception())
         )
         if force_reconnect or not can_use_previous_connect:
-            # Kick off a connection
-            coros = {
-                name: child_device.connect(
-                    mock, timeout=timeout, force_reconnect=force_reconnect
-                )
-                for name, child_device in self.children()
-            }
-            self._connect_task = asyncio.create_task(wait_for_connection(**coros))
-
+            # Use the connector to make a new connection
+            self._connect_mock = mock
+            self._connect_task = asyncio.create_task(
+                self._connector.connect(self, mock, timeout, force_reconnect)
+            )
         assert self._connect_task, "Connect task not created, this shouldn't happen"
         # Wait for it to complete
         await self._connect_task
 
 
-VT = TypeVar("VT", bound=Device)
+DeviceType = TypeVar("DeviceType", bound=Device)
 
 
-class DeviceVector(dict[int, VT], Device):
+class DeviceVector(Mapping[int, DeviceType], Device):
     """
     Defines device components with indices.
 
@@ -131,10 +144,26 @@ class DeviceVector(dict[int, VT], Device):
     :class:`~ophyd_async.epics.demo.DynamicSensorGroup`
     """
 
-    def children(self) -> Generator[tuple[str, Device], None, None]:
-        for attr_name, attr in self.items():
-            if isinstance(attr, Device):
-                yield str(attr_name), attr
+    def __init__(
+        self,
+        children: dict[int, DeviceType],
+        name: str = "",
+        connector: DeviceConnector | None = None,
+    ) -> None:
+        self._children = children
+        super().__init__(name, connector)
+
+    def __getitem__(self, key: int) -> DeviceType:
+        return self._children[key]
+
+    def __iter__(self) -> Iterator[int]:
+        yield from self._children
+
+    def __len__(self) -> int:
+        return len(self._children)
+
+    def children(self) -> dict[str, Device]:
+        return {str(key): value for key, value in self.items()}
 
 
 class DeviceCollector: