Typed Command with multiple input values to replace SignalX

docs/conf.py

@@ -139,6 +139,9 @@ def setup(app: application.Sphinx):
 # domain name if present. Example entries would be ('py:func', 'int') or
 # ('envvar', 'LD_LIBRARY_PATH').
 obj_ignore = [
+    "ophyd_async.core._command.P",
+    "ophyd_async.core._command.T",
+    "ophyd_async.core._command.T_co",
     "ophyd_async.core._derived_signal_backend.RawT",
     "ophyd_async.core._derived_signal_backend.DerivedT",
     "ophyd_async.core._detector.DetectorControllerT",

src/ophyd_async/core/__init__.py

@@ -1,5 +1,20 @@
 """The building blocks for making devices."""
 
+from ._command import (
+    Command,
+    CommandBackend,
+    CommandConnector,
+    CommandError,
+    ConnectionError,
+    ConnectionTimeoutError,
+    ExecutionError,
+    MockCommandBackend,
+    SoftCommandBackend,
+    soft_command_r,
+    soft_command_rw,
+    soft_command_w,
+    soft_command_x,
+)
 from ._derived_signal import (
     DerivedSignalFactory,
     derived_signal_r,
@@ -283,4 +298,18 @@ def __getattr__(name):
     "OnOff",
     "YesNo",
     "TableSubclass",
+    # Command
+    "Command",
+    "CommandBackend",
+    "CommandConnector",
+    "SoftCommandBackend",
+    "soft_command_r",
+    "soft_command_w",
+    "soft_command_x",
+    "soft_command_rw",
+    "CommandError",
+    "ExecutionError",
+    "ConnectionError",
+    "ConnectionTimeoutError",
+    "MockCommandBackend",
 ]

src/ophyd_async/core/_command.py

@@ -0,0 +1,340 @@
+from __future__ import annotations
+
+import asyncio
+import inspect
+from abc import abstractmethod
+from collections.abc import Awaitable, Callable, Sequence
+from typing import (
+    Generic,
+    ParamSpec,
+    Protocol,
+    TypeVar,
+    cast,
+    get_args,
+    get_origin,
+    get_type_hints,
+)
+from unittest.mock import AsyncMock
+
+from ._device import Device, DeviceConnector, LazyMock
+from ._utils import DEFAULT_TIMEOUT
+
+P = ParamSpec("P")
+T_co = TypeVar("T_co", covariant=True)
+T = TypeVar("T")
+
+
+class CommandError(Exception):
+    """Base class for all Command related errors."""
+
+
+class ConnectionError(CommandError):
+    """Raised when a Command cannot connect to its backend."""
+
+
+class ConnectionTimeoutError(ConnectionError):
+    """Raised when a Command connection times out."""
+
+
+class ExecutionError(CommandError):
+    """Raised when a Command fails during execution."""
+
+
+class CommandBackend(Protocol[P, T_co]):
+    """A backend for a Command."""
+
+    @abstractmethod
+    def source(self, name: str, read: bool) -> str:
+        """Return source of command."""
+
+    @abstractmethod
+    async def connect(self, timeout: float) -> None:
+        """Connect to underlying hardware."""
+
+    @abstractmethod
+    async def call(self, *args: P.args, **kwargs: P.kwargs) -> T_co:
+        """Execute the command and return its result."""
+
+
+class CommandConnector(DeviceConnector):
+    """A connector for a Command."""
+
+    def __init__(self, backend: CommandBackend):
+        self.backend = backend
+
+    async def connect_mock(self, device: Device, mock: LazyMock):
+        """Connect the backend in mock mode."""
+        self.backend = MockCommandBackend(self.backend, mock)
+
+    async def connect_real(self, device: Device, timeout: float, force_reconnect: bool):
+        """Connect the backend to real hardware."""
+        await self.backend.connect(timeout)
+
+
+class Command(Device, Generic[P, T]):
+    """A Device that can be called to execute a command.
+
+    :param backend: The backend for executing the command.
+    :param timeout: The default timeout for calling the command.
+    :param name: The name of the command.
+    """
+
+    _connector: CommandConnector
+
+    def __init__(
+        self,
+        backend: CommandBackend[P, T],
+        timeout: float | None = DEFAULT_TIMEOUT,
+        name: str = "",
+    ):
+        super().__init__(name=name, connector=CommandConnector(backend))
+        self._timeout = timeout
+
+    @property
+    def source(self) -> str:
+        """Returns the source of the command."""
+        return self._connector.backend.source(self.name, True)
+
+    async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T:
+        """Call the command."""
+        coro = self._connector.backend.call(*args, **kwargs)
+        try:
+            return await asyncio.wait_for(coro, self._timeout)
+        except TimeoutError as exc:
+            raise ConnectionTimeoutError(
+                f"Timeout calling {self.name} after {self._timeout}s"
+            ) from exc
+        except (TypeError, CommandError):
+            raise
+        except Exception as exc:
+            raise ExecutionError(f"Command execution failed: {exc}") from exc
+
+
+def soft_command_r(
+    command_return: type[T],
+    command_cb: Callable[[], T | Awaitable[T]],
+    units: str | None = None,
+    precision: int | None = None,
+    name: str = "",
+    timeout: float | None = DEFAULT_TIMEOUT,
+) -> Command[[], T]:
+    """Create a read-only Command with a [](#SoftCommandBackend).
+
+    :param command_return: The type of the value the command returns.
+    :param command_cb: The callback function to execute when the command is called.
+    :param units: Units of the return value.
+    :param precision: Precision of the return value.
+    :param name: The name of the command.
+    :param timeout: The default timeout for calling the command.
+    """
+    backend = SoftCommandBackend(None, command_return, command_cb, units, precision)
+    return Command(backend, timeout, name)
+
+
+def soft_command_w(
+    command_args: Sequence[type],
+    command_cb: Callable[P, None | Awaitable[None]],
+    units: str | None = None,
+    precision: int | None = None,
+    name: str = "",
+    timeout: float | None = DEFAULT_TIMEOUT,
+) -> Command[P, None]:
+    """Create a write-only Command with a [](#SoftCommandBackend).
+
+    :param command_args: Types of the arguments the command takes.
+    :param command_cb: The callback function to execute when the command is called.
+    :param units: Units of the return value.
+    :param precision: Precision of the return value.
+    :param name: The name of the command.
+    :param timeout: The default timeout for calling the command.
+    """
+    backend = SoftCommandBackend(command_args, None, command_cb, units, precision)
+    return Command(backend, timeout, name)
+
+
+def soft_command_rw(
+    command_args: Sequence[type],
+    command_return: type[T],
+    command_cb: Callable[P, T | Awaitable[T]],
+    units: str | None = None,
+    precision: int | None = None,
+    name: str = "",
+    timeout: float | None = DEFAULT_TIMEOUT,
+) -> Command[P, T]:
+    """Create a read-writable Command with a [](#SoftCommandBackend).
+
+    :param command_args: Types of the arguments the command takes.
+    :param command_return: The type of the value the command returns.
+    :param command_cb: The callback function to execute when the command is called.
+    :param units: Units of the return value.
+    :param precision: Precision of the return value.
+    :param name: The name of the command.
+    :param timeout: The default timeout for calling the command.
+    """
+    backend = SoftCommandBackend(
+        command_args, command_return, command_cb, units, precision
+    )
+    return Command(backend, timeout, name)
+
+
+def soft_command_x(
+    command_cb: Callable[[], None | Awaitable[None]],
+    units: str | None = None,
+    precision: int | None = None,
+    name: str = "",
+    timeout: float | None = DEFAULT_TIMEOUT,
+) -> Command[[], None]:
+    """Create a no-arg/no-return Command with a [](#SoftCommandBackend).
+
+    :param command_cb: The callback function to execute when the command is called.
+    :param units: Units of the return value.
+    :param precision: Precision of the return value.
+    :param name: The name of the command.
+    :param timeout: The default timeout for calling the command.
+    """
+    backend = SoftCommandBackend(None, None, command_cb, units, precision)
+    return Command(backend, timeout, name)
+
+
+class SoftCommandBackend(CommandBackend[P, T]):
+    """A backend for a Command that uses a Python callback."""
+
+    def __init__(
+        self,
+        command_args: Sequence[type] | None,
+        command_return: type[T] | None,
+        command_cb: Callable[P, T | Awaitable[T]],
+        units: str | None = None,
+        precision: int | None = None,
+    ):
+        self._command_args = command_args or []
+        self._command_return = command_return
+        self._command_cb = command_cb
+        self._units = units
+        self._precision = precision
+        self._last_return_value: T | None = None
+        self._lock = asyncio.Lock()
+
+        # Validate callback signature
+        sig = inspect.signature(command_cb)
+        params = list(sig.parameters.values())
+
+        if len(params) != len(self._command_args):
+            raise TypeError(
+                f"Number of command_args ({len(self._command_args)}) does not match "
+                f"callback arguments ({len(params)})"
+            )
+
+        hints = get_type_hints(command_cb)
+        for i, param in enumerate(params):
+            expected_type = self._command_args[i]
+            actual_type = hints.get(param.name)
+            if actual_type and actual_type is not expected_type:
+                if not (
+                    (
+                        hasattr(actual_type, "__origin__")
+                        and actual_type.__origin__ is expected_type
+                    )
+                    or actual_type == expected_type
+                ):
+                    raise TypeError(
+                        f"command_args type {expected_type} does not match "
+                        f"callback parameter '{param.name}' type {actual_type}"
+                    )
+
+        if self._command_return is not None:
+            actual_return = hints.get("return")
+            if actual_return and actual_return is not self._command_return:
+                if (
+                    get_origin(actual_return) is Awaitable
+                    or get_origin(actual_return) is asyncio.Future
+                ):
+                    actual_return = get_args(actual_return)[0]
+
+                if not (
+                    actual_return is self._command_return
+                    or actual_return == self._command_return
+                ):
+                    raise TypeError(
+                        f"command_return type {self._command_return} does not match "
+                        f"callback return type {actual_return}"
+                    )
+
+    def source(self, name: str, read: bool) -> str:
+        """Return the source of the command."""
+        return f"softcmd://{name}"
+
+    async def connect(self, timeout: float):
+        """No-op for SoftCommandBackend."""
+        pass
+
+    async def call(self, *args: P.args, **kwargs: P.kwargs) -> T:
+        """Execute the configured callback and return its result."""
+        if len(args) != len(self._command_args):
+            raise TypeError(
+                f"Expected {len(self._command_args)} arguments, got {len(args)}"
+            )
+
+        for i, (arg, expected_type) in enumerate(
+            zip(args, self._command_args, strict=True)
+        ):
+            if expected_type is not None:
+                origin = get_origin(expected_type) or expected_type
+                if origin is Sequence:
+                    if not isinstance(arg, Sequence):
+                        sig = inspect.signature(self._command_cb)
+                        param_name = list(sig.parameters.keys())[i]
+                        raise TypeError(
+                            f"Argument '{param_name}' should be {expected_type}, "
+                            f"got {type(arg)}"
+                        )
+                elif not isinstance(arg, origin):
+                    sig = inspect.signature(self._command_cb)
+                    param_name = list(sig.parameters.keys())[i]
+                    raise TypeError(
+                        f"Argument '{param_name}' should be {expected_type}, "
+                        f"got {type(arg)}"
+                    )
+
+        async with self._lock:
+            try:
+                result = self._command_cb(*args, **kwargs)
+                if inspect.isawaitable(result):
+                    result = await result
+                self._last_return_value = result
+                return cast(T, result)
+            except (TypeError, CommandError):
+                raise
+            except Exception as exc:
+                raise ExecutionError(f"Command execution failed: {exc}") from exc
+
+    def _async_lock(self):
+        return self._lock
+
+
+class MockCommandBackend(CommandBackend[P, T]):
+    """A backend for a Command that uses a mock for testing."""
+
+    def __init__(self, initial_backend: CommandBackend[P, T], mock: LazyMock):
+        self._initial_backend = initial_backend
+        self._mock = mock
+
+        async_mock = AsyncMock()
+        self.call_mock: Callable[P, Awaitable[T]] = cast(
+            Callable[P, Awaitable[T]], async_mock
+        )
+
+        # Attach to the device mock
+        self._mock().attach_mock(async_mock, "call")
+
+    def source(self, name: str, read: bool) -> str:
+        """Return the source of the mocked command."""
+        return f"mock+{self._initial_backend.source(name, read)}"
+
+    async def connect(self, timeout: float):
+        """Mock backend does not support real connection."""
+        raise ConnectionError("It is not possible to connect a MockCommandBackend")
+
+    async def call(self, *args: P.args, **kwargs: P.kwargs) -> T:
+        """Call the mock command."""
+        return await self.call_mock(*args, **kwargs)