Presenter¶

def create_plan_spec(
    plan: cabc.Callable[..., cabc.Generator[Any, Any, Any]],
    devices: cabc.Mapping[str, OADevice],
) -> PlanSpec:
    """Inspect *plan* and return a ``PlanSpec`` with one ``ParamDescription`` per parameter.

    Parameters
    ----------
    plan : Callable[..., Any]
        The plan function (or bound method) to inspect.
        Must be a generator function whose return annotation is a ``MsgGenerator``.
    devices : Mapping[str, OADevice]
        Registry of active devices; used to compute ``choices`` for parameters
        annotated with a ``OADevice`` subtype.

    Returns
    -------
    PlanSpec
        Fully populated plan specification.

    Raises
    ------
    TypeError
        If *plan* is not a generator function or its return type is not a
        ``MsgGenerator`` (``Generator[Msg, Any, Any]``).
    RuntimeError
        If an unexpected ``inspect.Parameter.kind`` is encountered.
    """
    func_obj: cabc.Callable[..., cabc.Generator[Any, Any, Any]] = getattr(
        plan, "__func__", plan
    )

    if not inspect.isgeneratorfunction(func_obj):
        raise TypeError(f"Plan {func_obj.__name__!r} must be a generator function.")

    sig = signature(func_obj)
    type_hints = get_type_hints(func_obj, include_extras=True)
    return_type = type_hints.get("return", None)

    if return_type is None:
        raise TypeError(
            f"Plan {func_obj.__name__!r} must have a return type annotation."
        )

    ret_origin = get_origin(return_type)
    is_generator = ret_origin is not None and _safe_issubclass(
        ret_origin, cabc.Generator
    )
    if not is_generator:
        raise TypeError(
            f"Plan {func_obj.__name__!r} must have a MsgGenerator return type; "
            f"got {return_type!r}."
        )

    params: list[ParamDescription] = []

    for name, param in _iterate_signature(sig):
        raw_ann: Any = type_hints.get(name, param.annotation)
        if raw_ann is _empty:
            raw_ann = Any

        if get_origin(raw_ann) is Annotated:
            ann_args = get_args(raw_ann)
            ann: Any = ann_args[0] if ann_args else Any
        else:
            ann = raw_ann

        actions_meta = _extract_action_meta(param, ann)

        pkind = _PARAM_KIND_MAP.get(param.kind)
        if pkind is None:
            raise RuntimeError(f"Unexpected parameter kind: {param.kind!r}")

        # Action parameters never get a widget, so they skip dispatch
        if actions_meta is not None:
            fields = _FieldsFromAnnotation()
        else:
            fields = _dispatch_annotation(ann, pkind, devices)

        # probe the magicgui path now: failing here is clearer than a broken
        # widget or a crash once the plan runs
        is_required = param.default is _empty
        needs_widget_probe = (
            actions_meta is None
            and is_required
            and pkind is not ParamKind.VAR_KEYWORD
            and fields.choices is None
        )
        if needs_widget_probe and not _is_magicgui_resolvable(ann):
            raise UnresolvableAnnotationError(func_obj.__name__, name, ann)

        params.append(
            ParamDescription(
                name=name,
                kind=pkind,
                annotation=ann,
                default=param.default,
                choices=fields.choices,
                multiselect=fields.multiselect,
                actions=actions_meta,
                device_proto=fields.device_proto,
                hidden=False,
            )
        )

    togglable = bool(getattr(func_obj, "__togglable__", False))
    pausable = bool(getattr(func_obj, "__pausable__", False))

    return PlanSpec(
        name=func_obj.__name__,
        docs=inspect.getdoc(func_obj) or "No documentation available.",
        parameters=params,
        togglable=togglable,
        pausable=pausable,
    )

def collect_arguments(
    spec: PlanSpec,
    values: cabc.Mapping[str, Any],
) -> tuple[tuple[Any, ...], dict[str, Any]]:
    """Build ``(args, kwargs)`` for calling a plan, driven by a ``PlanSpec``.

    Parameters
    ----------
    spec : PlanSpec
        The plan specification.
    values : Mapping[str, Any]
        Mapping of parameter names to their resolved values.

    Returns
    -------
    tuple[tuple[Any, ...], dict[str, Any]]
        Positional and keyword arguments ready to be splatted into the plan.

    Notes
    -----
    * ``POSITIONAL_ONLY`` and ``POSITIONAL_OR_KEYWORD`` -> ``args``, in
      declaration order.
    * ``KEYWORD_ONLY`` -> ``kwargs``.
    * ``VAR_POSITIONAL`` (``*args``) -> sequence expanded into ``args``.
    * ``VAR_KEYWORD`` (``**kwargs``) -> mapping merged into ``kwargs``.
    """
    args: list[Any] = []
    kwargs: dict[str, Any] = {}

    for p in spec.parameters:
        if p.name not in values:
            continue
        value = values[p.name]

        match p.kind:
            case ParamKind.VAR_POSITIONAL:
                if isinstance(value, cabc.Sequence) and not isinstance(
                    value, (str, bytes)
                ):
                    args.extend(value)
                else:
                    args.append(value)
            case ParamKind.VAR_KEYWORD:
                if isinstance(value, cabc.Mapping):
                    kwargs.update(value)
                else:
                    raise TypeError(
                        f"Value for **{p.name} must be a Mapping, got {type(value)!r}"
                    )
            case ParamKind.POSITIONAL_ONLY | ParamKind.POSITIONAL_OR_KEYWORD:
                args.append(value)
            case ParamKind.KEYWORD_ONLY:
                kwargs[p.name] = value

    return tuple(args), kwargs

def resolve_arguments(
    spec: PlanSpec,
    param_values: Mapping[str, Any],
    devices: Mapping[str, OADevice],
) -> dict[str, Any]:
    """Resolve raw UI parameter values into plan-callable values.

    Handles:
    * **Action parameters** - injected from metadata when absent from the UI.
    * **Model-backed parameters** - string labels are resolved to live
      ``OADevice`` instances via the ``devices`` registry.
    * **Everything else** - passed through unchanged.

    Parameters
    ----------
    spec : PlanSpec
        The plan specification containing parameter metadata.
    param_values : Mapping[str, Any]
        Raw parameter values from the UI.
    devices : Mapping[str, OADevice]
        Active device registry.

    Returns
    -------
    dict[str, Any]
        Resolved arguments ready for ``collect_arguments``.
    """
    values: dict[str, Any] = dict(param_values)

    # action parameters have no widget, so their values never come from the UI
    for p in spec.parameters:
        if p.actions is not None and p.name not in values:
            values[p.name] = p.actions

    resolved: dict[str, Any] = {}

    for p in spec.parameters:
        if p.name not in values:
            continue
        val = values[p.name]

        if p.choices is not None and p.device_proto is not None:
            if isinstance(val, str):
                labels = [val]
            elif isinstance(val, (cabc.Sequence, cabc.Set)) and not isinstance(
                val, (str, bytes)
            ):
                labels = [str(v) for v in val]
            else:
                labels = [str(val)]

            device_list = get_choice_list(devices, p.device_proto, labels)

            if p.kind is ParamKind.VAR_POSITIONAL or isdevicesequence(p.annotation):
                resolved[p.name] = device_list
            elif isdeviceset(p.annotation):
                resolved[p.name] = set(device_list)
            else:
                resolved[p.name] = device_list[0] if device_list else None
        else:
            resolved[p.name] = val

    return resolved

def get_choice_list(
    devices: Mapping[str, OADevice], proto: type[D], choices: Sequence[str]
) -> list[D]:
    """Filter a device registry to those that match a protocol and are in *choices*.

    Parameters
    ----------
    devices : Mapping[str, OADevice]
        Mapping of device names to device instances.
    proto : type[D]
        Class to match against via ``isinstance``.
    choices : Sequence[str]
        Subset of device names to consider.

    Returns
    -------
    list[D]
        Device instances whose name is in *choices* and that satisfy *proto*.
    """
    return [
        model
        for name, model in devices.items()
        if isinstance(model, proto) and name in choices
    ]

def isdevice(ann: Any) -> bool:
    """Return True if *ann* is a class that subclasses [`Device`][ophyd_async.core.Device].

    Operates on type annotations (the class itself), not on instances.
    """
    return _is_device_annotation(ann)

def isdevicesequence(ann: Any) -> bool:
    """Return True if *ann* is ``Sequence[T]`` where *T* is a [`Device`][ophyd_async.core.Device] subtype."""
    if not issequence(ann):
        return False
    args = get_args(ann)
    return len(args) == 1 and _is_device_annotation(args[0])

def issequence(ann: Any) -> bool:
    """Return True if *ann* is a ``Sequence[...]`` generic alias.

    Notes
    -----
    ``str`` and ``bytes`` are sequences in the stdlib sense, but their
    annotations are not generic aliases (``get_origin(str)`` is ``None``),
    so they are naturally excluded.
    """
    origin = get_origin(ann)
    if origin is None:
        return False
    try:
        return issubclass(origin, Sequence)
    except TypeError:
        return False