Support for typed SignalX

[burkeds]

In Tango it is standard practice for commands to be able to handle an input and/or output type. These types do not have to be the same but they must be one of the enumerated Tango types. Currently, we do a workaround where the device filler treats these commands as SignalR, SignalRW, or SignalW as appropriate.

This results in some awkward looking devices such as a motor where the Stop is a SignalR which has to be "read" in order for the command to be triggered because in Tango it is a command that returns an integer.

To support this common Tango practice, we would need to modify SignalX or a new Signal class which can handle types.

[coretl]

Here's a sketch of a design:

from abc import abstractmethod
from typing import Generic, TypeVar

# Remove SignalX and replace with Command

ArgumentDatatype = TypeVar("ArgumentDatatype", bound=SignalDatatype | None)
ReturnDatatype = TypeVar("ReturnDatatype", bound=SignalDatatype | None)


class CommandBackend(Generic[ArgumentDatatype, ReturnDatatype]):
    @abstractmethod
    async def connect(self, timeout: float):
        """Connect to underlying hardware."""
        raise NotImplementedError()

    @abstractmethod
    async def call(self, argument: ArgumentDatatype = None) -> ReturnDatatype:
        """Call the command with the given argument."""
        raise NotImplementedError()


# Need to add a MockCommandBackend for testing

# Need to add CommandConnector that calls CommandBackend.connect()


class Command(Device, Generic[ArgumentDatatype, ReturnDatatype]):
    """A Device with the concept of a value, with R, RW, W and X flavours.

    :param backend: The backend for providing Signal values.
    :param timeout: The default timeout for operations on the Signal.
    :param name: The name of the signal.
    """

    _connector: CommandConnector
    _child_devices = _ChildrenNotAllowed()  # type: ignore

    def __init__(
        self,
        backend: CommandBackend[ArgumentDatatype, ReturnDatatype],
        timeout: float | None = DEFAULT_TIMEOUT,
        name: str = "",
        attempts: int = 1,
    ) -> None:
        super().__init__(name=name, connector=CommandConnector(backend))
        self._timeout = timeout
        self._attempts = attempts

    @AsyncStatus.wrap
    async def trigger(
        self,
        argument: ArgumentDatatype = None,
        timeout: CalculatableTimeout = CALCULATE_TIMEOUT,
    ) -> ReturnDatatype:
        """Trigger the action and return a status saying when it's done.

        :param wait: If True, wait for the trigger to complete.
        :param timeout: The timeout for the trigger.
        """
        if timeout == CALCULATE_TIMEOUT:
            timeout = self._timeout
        result = await _wait_for(self._connector.backend.call(None), timeout)
        return result

[coretl]

Limitations:

• Only a single position argument of the given type (or None)
• Only a single return value of the given type (or None)
• The bluesky trigger verb currently warns if you pass args or kwargs: https://github.com/bluesky/bluesky/blob/8adfe3ab1266fbb2555de7e3f906f79cdeb2d704/src/bluesky/run_engine.py#L2254-L2259

We might need to add source to this, and need to decide if trigger is the right verb for this

@tacaswell thoughts?

[Tom-Willemsen]

Would this also be a step towards eventual support for PVA remote procedure calls?

[coretl]

Would this also be a step towards eventual support for PVA remote procedure calls?

In which case as we support arbitrary named and typed args and kwargs then maybe we should widen the interface out:

from abc import abstractmethod
from typing import Generic, ParamSpec, TypeVar

# Remove SignalX and replace with Command

CommandArguments = ParamSpec("CommandArguments")
CommandReturn = TypeVar("CommandReturn")


class CommandBackend(Generic[CommandArguments, CommandReturn]):
    @abstractmethod
    async def connect(self, timeout: float):
        """Connect to underlying hardware."""
        raise NotImplementedError()

    @abstractmethod
    async def call(
        *args: CommandArguments.args, **kwargs: CommandArguments.kwargs
    ) -> CommandReturn:
        """Call the command with the given argument."""
        raise NotImplementedError()


# Need to add a MockCommandBackend for testing

# Need to add CommandConnector that calls CommandBackend.connect()


class Command(Device, Generic[CommandArguments, CommandReturn]):
    """A Device with the concept of a value, with R, RW, W and X flavours.

    :param backend: The backend for providing Signal values.
    :param timeout: The default timeout for operations on the Signal.
    :param name: The name of the signal.
    """

    _connector: CommandConnector
    _child_devices = _ChildrenNotAllowed()  # type: ignore

    def __init__(
        self,
        backend: CommandBackend[CommandArguments, CommandReturn],
        name: str = "",
    ) -> None:
        super().__init__(name=name, connector=CommandConnector(backend))

    @AsyncStatus.wrap
    async def trigger(
        self,
        *args: CommandArguments.args,
        **kwargs: CommandArguments.kwargs,
    ) -> CommandReturn:
        """Trigger the action and return a status saying when it's done."""
        return await self._connector.backend.call(*args, **kwargs)