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, a generator function annotated to
        return a ``MsgGenerator``.
    devices : Mapping[str, OADevice]
        The session's devices, giving ``choices`` for parameters annotated
        with an ``OADevice`` subtype.

    Returns
    -------
    PlanSpec
        The plan specification.

    Raises
    ------
    TypeError
        If *plan* is not a generator function or its return type is not a
        ``MsgGenerator`` (``Generator[Msg, Any, Any]``).
    UnresolvableAnnotationError
        If an annotation names something missing at runtime, or no view can
        build a control for it.
    RuntimeError
        On an unexpected ``inspect.Parameter.kind``.
    """
    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, unresolved = _resolve_annotations(func_obj)

    if "return" in unresolved:
        raise UnresolvableAnnotationError(
            func_obj.__name__, "return", unresolved["return"]
        )

    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):
        if name in unresolved:
            raise UnresolvableAnnotationError(func_obj.__name__, name, unresolved[name])

        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)

        # refuse now: failing here is clearer than a broken control or a
        # crash once the plan runs
        is_required = param.default is _empty
        needs_control = (
            actions_meta is None
            and is_required
            and pkind is not ParamKind.VAR_KEYWORD
            and fields.choices is None
        )
        if needs_control and not _is_renderable(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,
            )
        )

    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 the ``(args, kwargs)`` calling a plan, from its ``PlanSpec``.

    Parameters
    ----------
    spec : PlanSpec
        The plan specification.
    values : Mapping[str, Any]
        Resolved values by parameter name.

    Returns
    -------
    tuple[tuple[Any, ...], dict[str, Any]]
        Positional and keyword arguments for 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]:
    """Turn parameter values from the interface into values a plan takes.

    * **Action parameters** are filled from the spec when the interface lacks
      them.
    * **Device parameters**: names become ``OADevice`` instances from
      ``devices``.
    * **Everything else** passes unchanged.

    Parameters
    ----------
    spec : PlanSpec
        The plan specification.
    param_values : Mapping[str, Any]
        Parameter values from the interface.
    devices : Mapping[str, OADevice]
        The session's devices.

    Returns
    -------
    dict[str, Any]
        Resolved arguments 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]:
    """Return the devices named in *choices* that are instances of *proto*.

    Parameters
    ----------
    devices : Mapping[str, OADevice]
        Devices by name.
    proto : type[D]
        Class checked with ``isinstance``.
    choices : Sequence[str]
        Names of the devices to consider.

    Returns
    -------
    list[D]
        The matching devices.
    """
    return [
        model
        for name, model in devices.items()
        if isinstance(model, proto) and name in choices
    ]

def isdevice(ann: Any) -> bool:
    """Return True if the annotation *ann* is a [`Device`][ophyd_async.core.Device] subclass."""
    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."""
    return issequence(ann) and _single_device_arg(ann)

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

    Notes
    -----
    ``str`` and ``bytes`` are sequences, but not generic aliases
    (``get_origin(str)`` is ``None``), so they are excluded.
    """
    return _origin_subclasses(ann, Sequence)