Typed Command with multiple input values to replace SignalX

[burkeds]

Addresses issue #1087

New Command with R, RW, W, and X flavours.

SoftCommandBackend enforces typematching between callback signature and command_args. command_args must be a list of types in order of the callback signature.

MockCommandBackend included for testing.

Example use of soft signal.

from ophyd_async.core import CommandRW, soft_command_rw

my_async_command: CommandRW[[int], float]

async def async_compute(x: int) -> float:
    return float(x) * 2.0

my_async_command = soft_command_rw(
    command_args=[int],
    command_return=float,
    command_cb=async_compute,
    name="async_compute_gain",
)

await my_async_command.call(1)

[burkeds]

@coretl Could I get a preliminary review of the core classes before I implement Tango and/or EPICS versions/

[coretl]

I think we can't specify them in this way and still get static typing. We have to do something like:
https://pyright-play.net/?code=GYJw9gtgBALgngBwJYDsDmUkQWEMoAqiApgCYAiSAxjADSEkBqAhiPQOLErEjX0CqKBMyoBregGFmAG2nMARtOL0ACq2YQAygmJUAUHpVQAvFDUgN23QAoARCtsBKPQRMMdLEHYJODVOQDOAVASkBDMKKTWnNy8VADaKvQEALqOAFx6UNlQpMTAUAD6hahIMMXWAcTSwPTAAK4oVOkhMnKKxInJaZk5fVAAdENZOXkFxVRtFVU19ABUrGgBLSoDiwHzc6IA7usrAzvrjlAAtAB8hL392UMDBmNQwNYAHi2odFBwLQEwIMfnj2kYGYMBatz8blCEHCkWswGcQA

This means we would do something like this:

class MyCommandType(Protocol):
    async def command_type(x: int, y: str) -> float: ...

class MyDevice(TangoDevice):
    my_command: Command[MyCommandType]

a_float = await MyDevice().my_command.trigger(x=0, y="foo")  # type checks as float

or

async def async_compute(x: int) -> float:
    return float(x) * 2.0

my_async_command = soft_command_rw(
    async_compute,
    name="async_compute_gain",
)
a_float = await my_async_command.trigger(x=0)  # type checks as float

but either way we are forced to make at least a stub function. I also think there shouldn't be any async executor anywhere in this PR, maybe in the Tango one, but we should expect to be called from inside the event loop.

Let me know if you'd like to talk this over on zoom

[coretl]

Actually if you don't care about keyword args you could do:

class MyDevice(TangoDevice):
    my_command: Command[Callable[[int, str], Awaitable[float]]]

a_float = await MyDevice().my_command.trigger(0, "foo")  # type checks as float

but I can't see how to avoid having to put Awaitable in the type string.

[burkeds]

@coretl I refactored Command to use a Protocol for the backend and made the Command a generic over P and T. So far it's just the core code, I didn't write a Tango or EPICS backend yet. What do you think of this version?

[coretl]

I think we should use the bluesky verb

Suggested change

	async def __call__(self, *args: P.args, **kwargs: P.kwargs) -> T:
	async def trigger(self, *args: P.args, **kwargs: P.kwargs) -> T:

[burkeds]

@runtime_checkable
class Triggerable(Protocol):
    @abstractmethod
    def trigger(self) -> Status:
        """Return a ``Status`` that is marked done when the device is done triggering."""
        ...

The problem here is that it won't match the protocol signature.

[coretl]

Good point, Triggerable doesn't take args and kwargs because it was difficult to type hint at the time, but now we ParamSpec then maybe we can open that up again?

What's the intended use of Tango commands, would they be triggered directly from plans, or always wrapped in some other device that calls them?

[burkeds]

They should be callable directly from plans. There are no "rules" or best practices that are adopted community-wide in Tango so commands could do anything and everything. I would propose altering Trlggerable to take optional arguments rather than implementing a new Bluesky verb.

[coretl]

Why return AsyncStatus rather than being a plain async def? The Status interface was never meant to return results, so I think it's clearer to leave it as an async def.

[burkeds]

If we don't return a status we lose a lot of the functionality of SignalX. It is still useful to be able to choose to await a status or not, call groups of object concurrently with something like bluesky.plans.call(...). Without the status object used by other bluesky verbs, the interface to bluesky will be very different specifically for call.

Does having a status return a result break anything? Tests seem to all pass ok.

[coretl]

Yes, the type hinting I proposed in #1138 (comment) will only work if CommandBackend.__call__ has the same signature as Command.__call__.

If we do this then we need to add a return value to the Status protocol.
@tacaswell what do you think?

[coretl]

@claude comment on #1138 (comment)

[claude]

Claude finished @coretl's task in 54s —— View job

---

Analysis complete

• Read the code at src/ophyd_async/core/_command.py
• Analyze the bind usage vs direct parameter passing
• Provide technical analysis

Summary

I've added an inline comment on the specific line in question. The key finding:

@coretl is correct - the bind() mechanism is redundant here because:

1. The callback is invoked with the original *args, **kwargs (line 302), not with bound.arguments
2. Python will re-bind parameters natively when calling self._command_cb(*args, **kwargs)
3. The ParamSpec[P] generic already enforces signature matching at type-check time
4. The runtime type validation duplicates Python's native parameter handling

Options:

• Option A: Remove bind entirely and rely on ParamSpec[P] for type safety (consistent with Signal)
• Option B: Keep bind if explicit runtime parameter type checking is desired (but this would be inconsistent with how Signal.set works, which doesn't do this level of validation)

The current code essentially validates twice: once with bind+type-check, then again when Python natively binds the parameters to the callback.

---

[burkeds]

@coretl per our conversation in the dev call, should I add a subclass of AsyncStatus which returns its value? If so, what should I call it?

[coretl]

@coretl per our conversation in the dev call, should I add a subclass of AsyncStatus which returns its value? If so, what should I call it?

I think we should prototype this with plain async functions all the way to a Tango implementation, then revisit at the end to see if we want to make it AsyncStatus. I'll raise an issue on bluesky in the meantime to discuss and link it here.

[coretl]

@burkeds I've written bluesky/bluesky#1988. I think we should add value to the base AsyncStatus and make it always return self.task.result() which will do for now.

[burkeds]

@burkeds I've written bluesky/bluesky#1988. I think we should add value to the base AsyncStatus and make it always return self.task.result() which will do for now.

Done. I made the Awaitable in AsyncStatusBase generic over T since the value property now is typed to return something.

[coretl]

Please could you change to execute from __call__? Then I'll be interested to see how the tango implementation goes. I've updated the playground link:

Code sample in pyright playground

import inspect
from collections.abc import Awaitable, Callable
from typing import Any, Concatenate, Generic, ParamSpec, Protocol, TypeVar

P = ParamSpec("P")
T = TypeVar("T")


class CommandBackend(Generic[P, T]):
    def source(self, name: str) -> str: ...
    async def execute(self, *args: P.args, **kwargs: P.kwargs) -> T: ...


class SoftCommandBackend(CommandBackend[P, T]):
    def __init__(self, func: Callable[P, Awaitable[T]]):
        self.func = func

    async def execute(self, *args: P.args, **kwargs: P.kwargs) -> T:
        return await self.func(*args, **kwargs)


class TangoCommandBackend(CommandBackend[P, T]):
    def __init__(self, protocol_call_func: Callable[Concatenate[Any, P], Awaitable[T]]):
        self.signature = inspect.signature(protocol_call_func)
        # Need to pop self out of this

    async def execute(self, *args: P.args, **kwargs: P.kwargs) -> T: ...


class CommandProtocol(Protocol):
    def source(self) -> str: ...


class Command(Generic[P, T]):
    def __init__(self, backend: CommandBackend[P, T]):
        self.backend = backend

    def source(self) -> str:
        return self.backend.source("name")

    async def execute(self, *args: P.args, **kwargs: P.kwargs) -> T:
        print("Do some logging")
        return await self.backend.execute(*args, **kwargs)


async def f(x: int, y: str) -> float:
    return float(x + float(y))


def soft_command(func: Callable[P, Awaitable[T]]) -> Command[P, T]:
    return Command(SoftCommandBackend(func))


def tango_command(
    protocol_call_func: Callable[Concatenate[Any, P], Awaitable[T]],
) -> Command[P, T]:
    return Command(TangoCommandBackend(protocol_call_func))


# mydevice.py


# Specify by kwargs
class F(CommandProtocol, Protocol):
    async def execute(self, x: int, y: str) -> float: ...


class MyDevice:
    c1: Command[[int, str], float]  # Specify by position
    c2: F  # Specify by kwargs


# end mydevice.py


async def test_mydevice():
    # Check they match types
    d_soft = MyDevice()
    d_soft.c1 = soft_command(f)
    d_soft.c2 = soft_command(f)
    d_tango = MyDevice()
    d_tango.c1 = tango_command(F.execute)
    d_tango.c2 = tango_command(F.execute)
    for d in [d_soft, d_tango]:
        assert await d.c1.execute(1, ".1") == 1.1
        assert await d.c2.execute(3, ".2") == 3.2
        assert await d.c2.execute(x=1, y=".2") == 1.2
        assert d.c1.source()
        assert d.c2.source()

[coretl]

Looking good, couple of minor points. I'm interested to see how the TangoCommandBackend looks, then I'll have a go at the EPICS one...

[coretl]

Probably want to add something like this here:

ophyd-async/src/ophyd_async/core/_mock_signal_backend.py

Lines 68 to 70 in beba1c8

	side_effect=self._mock_put_callback
	if self._mock_put_callback
	else lambda v: None,

but we need to manufacture a value to return. If we made a converter for the return value type like the input types then we could use converter.write_value(None) to make it for us like:

ophyd-async/src/ophyd_async/core/_soft_signal_backend.py

Line 141 in beba1c8

self.initial_value = self.converter.write_value(initial_value)

[burkeds]

I didn't see a way to do this without having the return type be retrievable from the backend. I added get_return_type to CommandBackend. What do you think? It is very possible that I have misunderstood the issue.

    @abstractmethod
    def get_return_type(self) -> type[T_co] | None:
        """Return the return type of the command, or None if it returns None."""