Skip to content

redsun.presenter

Positioner

PositionerPresenter dataclass 

PositionerPresenter(
    name: str,
    *,
    devices: DeviceMapping,
    include: list[str] | None = None,
    timeout: float = 10.0,
)

Bases: Loggable

Move the axes of devices by hand, one move at a time per device.

Axes are grouped by the device holding them: an axis is a row of the box of its parent, or a box of its own when it has none. Every box is kept, or only those named in include. Each axis is described and followed from the start, and its readbacks are relayed on sig_readback; an axis that cannot be, within timeout, is left out with a warning.

A step starts from the setpoint, so steps add up exactly, and from the readback after a stop or a failure, when the setpoint no longer says where the axis is. Every target passes check before it is sent. The limits of an axis are read again after each configuration write to its device and after a move that is refused or fails, and before each move of an axis that is not Checkable; a change is reported on sig_limits.

sig_readback class-attribute instance-attribute 

sig_readback = Signal(str, str, float)

Device, axis and position, whenever the axis reads back.

sig_moving class-attribute instance-attribute 

sig_moving = Signal(str, bool)

Device, and whether it starts or ends a move.

sig_failed class-attribute instance-attribute 

sig_failed = Signal(str, str)

Device, and why a move or a stop failed.

sig_configuration class-attribute instance-attribute 

sig_configuration = Signal(str, object)

Key and value of a configuration signal, whenever it changes and after a write.

sig_limits class-attribute instance-attribute 

sig_limits = Signal(str, str, object, object)

Device, axis, and its new low and high limits; None for no bound.

name instance-attribute 

name: str

Name of the presenter in its session.

devices class-attribute instance-attribute 

devices: DeviceMapping = field(repr=False)

Devices of the session, searched for axes, with the devices below them.

include class-attribute instance-attribute 

include: list[str] | None = None

Names of the boxes to keep; None keeps every box.

timeout class-attribute instance-attribute 

timeout: float = 10.0

Seconds an axis may take to be described and followed at the start.

axes property 

axes: dict[str, dict[str, AxisInfo]]

The axes of each box, by row, by the name of the device holding them.

configuration property 

configuration: Configuration

The configuration of every axis.

An entry whose signal cannot be written has :readonly appended to its source.

check 

check(device: str, axis: str, target: float) -> None

Refuse target for axis of device, or return.

A target that is not a finite number, or that falls outside the limits the axis last reported, is refused. A subclass refusing more calls super().check first.

Raises:

Type Description
ValueError

With the reason, if the target is refused.

move async 

move(device: str, axis: str, delta: float) -> None

Move axis of device by delta, in the axis' units.

Skipped while device is still moving, so steps asked faster than the device moves do not queue up behind it.

Raises:

Type Description
KeyError

If the presenter does not hold device.

move_to async 

move_to(
    device: str, positions: Mapping[str, float]
) -> None

Move the axes of device in positions there, one after another.

Waits for a move of device already running, and is dropped if device is stopped, or taken by a plan, meanwhile.

Raises:

Type Description
KeyError

If the presenter does not hold device.

stop async 

stop(device: str) -> None

Stop device and each of its axes that can be stopped.

Every one is asked at once, and one that fails does not keep the rest from stopping: the failures are reported on sig_failed. A move ended this way is not reported as failed, and a go-to waiting for device is dropped.

Raises:

Type Description
KeyError

If the presenter does not hold device.

set_locked 

set_locked(names: frozenset[str]) -> None

Hold the devices in names, and every device below them, for a plan.

No hand move, waiting go-to or configuration write reaches a held axis.

configure async 

configure(key: str, value: object) -> None

Write value to the configuration signal key, then report it.

sig_configuration carries what the signal reads back, also after a write that was refused, or not made because a plan holds the device.

Raises:

Type Description
KeyError

If key names no writable configuration signal.

shutdown 

shutdown() -> None

Stop the devices still moving, then stop following the axes.

DescribesAxes

Bases: Protocol

A component describing the axes it moves.

axes property 

axes: Mapping[str, Mapping[str, AxisInfo]]

The axes of each box, by row, by the name of the device holding them.

configuration property 

configuration: Configuration

The configuration of every axis.

Acquisition

AcquisitionPresenter dataclass 

AcquisitionPresenter(name: str, *, devices: DeviceMapping)

Bases: Loggable

Run the plans the components of a session offer, one at a time.

The plans of every component with a plan_map are gathered in setup; one whose signature no plan widget can show is left out with a warning. A run gets the callbacks its plan lists, then those attached to it. Its files are named after the plan while it runs. Its end is reported on sig_plan_done when it finishes, on sig_plan_stopped when it is stopped, or on sig_plan_failed, with the error logged, when it raises.

sig_plan_started class-attribute instance-attribute 

sig_plan_started = Signal(str, dict)

Plan, and the values it was launched with, once the engine has started it.

sig_plan_done class-attribute instance-attribute 

sig_plan_done = Signal(str)

Plan, once it has finished.

sig_plan_stopped class-attribute instance-attribute 

sig_plan_stopped = Signal(str)

Plan, once it has been stopped.

sig_plan_failed class-attribute instance-attribute 

sig_plan_failed = Signal(str, str)

Plan, and why it failed.

sig_progress class-attribute instance-attribute 

sig_progress = Signal(tuple)

Every progress scope of the running plan, as the engine reports them.

sig_locks_changed class-attribute instance-attribute 

sig_locks_changed = Signal(frozenset)

Names of the devices the running plan holds, whenever they change.

sig_action_changed class-attribute instance-attribute 

sig_action_changed = Signal(str, object)

Action and its state, from the actions of every component.

sig_base_dir_changed class-attribute instance-attribute 

sig_base_dir_changed = Signal(object)

The directory runs now write under.

name instance-attribute 

name: str

Name of the presenter in its session.

devices class-attribute instance-attribute 

devices: DeviceMapping = field(repr=False)

Devices of the session, which plan arguments name.

plans property 

plans: Mapping[str, PlanSpec]

The specification of each plan, by name.

callbacks property 

callbacks: Mapping[str, CallbackType]

The document callbacks a user may attach to a run, by name.

plan_callbacks property 

plan_callbacks: Mapping[str, Sequence[CallbackType]]

The document callbacks each plan runs with, by plan name.

base_dir property 

base_dir: Path | None

The directory runs write under; None before setup.

engine 

engine() -> RunEngine

Return the engine every plan runs on, shared with any component asking for one.

deferrals 

deferrals() -> Deferrals

Return the deferrals of the engine, shared with any component asking for them.

setup 

setup(
    plan_sources: Mapping[str, HasPlans],
    callbacks: Mapping[str, CallbackType],
    paths: SessionPathProvider,
) -> None

Gather the plans of every provider, the callbacks, and the path provider.

launch 

launch(
    plan: str,
    values: dict[str, Any],
    attached: Sequence[str] = (),
) -> None

Run plan with the parameter values, attaching the callbacks named.

Refused, with a warning, while another plan runs. Values that cannot be turned into the plan's arguments are reported on sig_plan_failed.

pause 

pause() -> None

Pause the running plan at its next checkpoint.

resume 

resume() -> None

Resume the paused plan, or withdraw a pause it has not reached yet.

stop 

stop() -> None

Stop the running or paused plan, which ends as stopped.

request_action 

request_action(action: str, on: bool) -> None

Ask for action of the running plan, or ask it to end.

set_base_dir 

set_base_dir(path: Path) -> None

Make runs write under path.

set_label 

set_label(label: str) -> None

Lead the file names of every later run with label; empty for none.

shutdown 

shutdown() -> None

Abort a running or paused plan, reporting neither its end nor a failure.

Returns once the plan's cleanup has run, or after ten seconds.

DescribesPlans

Bases: Protocol

A component describing the plans it runs.

plans property 

plans: Mapping[str, PlanSpec]

The specification of each plan, by name.

callbacks property 

callbacks: Mapping[str, CallbackType]

The document callbacks a user may attach to a run, by name.

plan_callbacks property 

plan_callbacks: Mapping[str, Sequence[CallbackType]]

The document callbacks each plan runs with, by plan name.

base_dir property 

base_dir: Path | None

The directory runs write under; None before it is known.

Lights

LightPresenter dataclass 

LightPresenter(
    name: str,
    *,
    devices: DevicesOf[Light],
    include: list[str] | None = None,
    timeout: float = 10.0,
)

Bases: Loggable

Switch light sources on and off, and set their intensity.

Every light is described and followed from the start; one that cannot be, within timeout, is left out with a warning. Writes to a light a plan holds are refused. While an intensity is being written, only the newest one asked for meanwhile is written after it.

sig_enabled class-attribute instance-attribute 

sig_enabled = Signal(str, bool)

Device, and whether it is on, whenever it reads back.

sig_intensity class-attribute instance-attribute 

sig_intensity = Signal(str, float)

Device and intensity, whenever it reads back.

sig_failed class-attribute instance-attribute 

sig_failed = Signal(str, str)

Device, and why a write was refused or failed.

sig_configuration class-attribute instance-attribute 

sig_configuration = Signal(str, object)

Key and value of a configuration signal, when it changes and after a write.

name instance-attribute 

name: str

Name of the presenter in its session.

devices class-attribute instance-attribute 

devices: DevicesOf[Light] = field(repr=False)

Devices of the session with an enabled signal.

include class-attribute instance-attribute 

include: list[str] | None = None

Names of the lights to keep; None keeps every light.

timeout class-attribute instance-attribute 

timeout: float = 10.0

Seconds a light may take to be described, followed, or written.

lights property 

lights: dict[str, LightInfo]

Each light's state and intensity range, by device name.

configuration property 

configuration: Configuration

The configuration of every light.

An entry whose signal cannot be written has :readonly appended to its source.

set_enabled async 

set_enabled(device: str, on: bool) -> None

Switch device on or off.

Raises:

Type Description
KeyError

If the presenter does not hold device.

set_intensity async 

set_intensity(device: str, value: float) -> None

Set the intensity of device, in its units.

Raises:

Type Description
KeyError

If the presenter does not hold device.

configure async 

configure(key: str, value: object) -> None

Write value to the configuration signal key, then report it.

Raises:

Type Description
KeyError

If key names no writable configuration signal.

set_locked 

set_locked(names: frozenset[str]) -> None

Hold the lights in names, or below a device in it, for a plan.

No write reaches a held light.

shutdown 

shutdown() -> None

Stop following the lights.

DescribesLights

Bases: Protocol

A component describing the lights it switches.

lights property 

lights: Mapping[str, LightInfo]

Each light's state and intensity range, by device name.

configuration property 

configuration: Configuration

The configuration of every light.

Shared parts

DeviceConfiguration 

DeviceConfiguration(name: str)

Bases: Loggable

The configuration of a set of devices: read, followed, and written.

Each device is added under an owner, the name the session knows it by, and every key it holds belongs to that owner. A value changed on a device, or read back after a write, is reported on sig_changed.

Hold no device yet; name names the owner in log records.

sig_changed class-attribute instance-attribute 

sig_changed = Signal(str, object)

Key and value of a configuration signal, when it changes and after a write.

configuration property 

configuration: Configuration

Everything added so far.

An entry whose signal cannot be written has :readonly appended to its source.

add async 

add(
    owner: str,
    device: AsyncConfigurable,
    *,
    owner_device: Device | None = None,
) -> Configuration

Read device's configuration, follow its signals, and file it under owner.

Returns the configuration of device alone. Nothing is kept when reading fails or the call is cancelled. A key already added is followed once, and stays with its first owner.

Parameters:

Name Type Description Default
owner_device Device | None

The device owner names, when it is not device.

 None

configure async 

configure(key: str, value: object) -> str

Write value to key, report what it reads back, and return its owner.

A write to a signal that is held, or that sits below a held device, is not made, and the value is still read back and reported.

Raises:

Type Description
KeyError

If key names no writable configuration signal.

set_locked 

set_locked(names: frozenset[str]) -> None

Refuse writes to the signals in names and below the devices in it.

remove_all 

remove_all() -> None

Stop following every signal added.

Plan specification

Describe a plan's signature as a PlanSpec.

create_plan_spec inspects a bluesky MsgGenerator function and returns a PlanSpec describing its parameters, from which a view builds the controls of the plan.

_fields_from_annotation turns a Literal or device annotation into ParamDescription fields (choices, device_proto, multiselect).

ParamKind

Bases: IntEnum

inspect._ParameterKind as a public IntEnum.

Usable in match/case without importing private standard library names.

POSITIONAL_ONLY class-attribute instance-attribute 

POSITIONAL_ONLY = 0

Given by position only.

POSITIONAL_OR_KEYWORD class-attribute instance-attribute 

POSITIONAL_OR_KEYWORD = 1

Given by position or by name.

VAR_POSITIONAL class-attribute instance-attribute 

VAR_POSITIONAL = 2

Any number of values given by position, as *args.

KEYWORD_ONLY class-attribute instance-attribute 

KEYWORD_ONLY = 3

Given by name only.

VAR_KEYWORD class-attribute instance-attribute 

VAR_KEYWORD = 4

Any number of values given by name, as **kwargs.

UnresolvableAnnotationError 

UnresolvableAnnotationError(
    plan_name: str, param_name: str, annotation: Any
)

Bases: TypeError

Raised when a plan parameter's annotation maps to no widget.

Parameters:

Name Type Description Default
plan_name str

Name of the plan.

 required
param_name str

Name of the parameter.

 required
annotation Any

The unresolvable annotation.

 required

create_plan_spec 

create_plan_spec(
    plan: Callable[..., Generator[Any, Any, Any]],
    devices: Mapping[str, Device],
) -> PlanSpec

Inspect plan and return a PlanSpec with one ParamDescription per parameter.

Parameters:

Name Type Description Default
plan Callable[..., Generator[Any, Any, Any]]

The plan function or bound method, a generator function annotated to return a MsgGenerator.

 required
devices Mapping[str, Device]

The session's devices, giving choices for parameters annotated with an OADevice subtype.

 required

Raises:

Type Description
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.
ValueError

If plan declares two actions of one name, or a parameter's default breaks its own limits.

collect_arguments 

collect_arguments(
    spec: PlanSpec, values: Mapping[str, Any]
) -> tuple[tuple[Any, ...], dict[str, Any]]

Build the (args, kwargs) calling a plan, from its PlanSpec.

A positional parameter with no value takes its default when a later positional argument is given, so every argument keeps its place.

Parameters:

Name Type Description Default
values Mapping[str, Any]

Resolved values by parameter name.

 required

Raises:

Type Description
ValueError

If a positional parameter with no value and no default comes before a positional argument that is given.
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.

resolve_arguments 

resolve_arguments(
    spec: PlanSpec,
    param_values: Mapping[str, Any],
    devices: Mapping[str, Device],
) -> 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.
  • Hidden parameters get their default, so the parameters after them keep their places.
  • Device parameters: names become OADevice instances from devices.
  • Everything else passes unchanged.

Parameters:

Name Type Description Default
spec PlanSpec

The plan specification.

 required
param_values Mapping[str, Any]

Parameter values from the interface.

 required

Raises:

Type Description
ValueError

If a value breaks its parameter's limits, naming every one.

ParamDescription dataclass 

ParamDescription(
    name: str,
    kind: ParamKind,
    annotation: Any,
    default: Any,
    choices: list[Any] | None = None,
    labels: dict[str, str] = dict(),
    multiselect: bool = False,
    hidden: bool = False,
    actions: Sequence[PlanAction]
    | PlanAction
    | None = None,
    device_proto: type[Any] | None = None,
    annotated: Any = _empty,
)

Description of one plan parameter.

name instance-attribute 

name: str

Name of the parameter in the plan signature.

kind instance-attribute 

kind: ParamKind

Kind of the parameter, as inspect.Parameter.kind.

annotation instance-attribute 

annotation: Any

Type annotation, without Annotated metadata.

default instance-attribute 

default: Any

Default value of the parameter, or inspect.Parameter.empty if none.

choices class-attribute instance-attribute 

choices: list[Any] | None = None

Selectable values: a Literal's own values, or device names.

labels class-attribute instance-attribute 

labels: dict[str, str] = field(default_factory=dict)

What a view shows for each device name in choices, such as scope / xy.

multiselect class-attribute instance-attribute 

multiselect: bool = False

Whether several values can be selected, as for Sequence[OADevice].

hidden class-attribute instance-attribute 

hidden: bool = False

Whether no input can show the parameter.

A view leaves a hidden parameter out, and the plan keeps its default.

actions class-attribute instance-attribute 

actions: Sequence[PlanAction] | PlanAction | None = None

Actions taken from the parameter's default value, if any.

device_proto class-attribute instance-attribute 

device_proto: type[Any] | None = None

Device class or runtime-checkable protocol of a device parameter, used to look devices up when resolving arguments.

annotated class-attribute instance-attribute 

annotated: Any = _empty

The full annotation, Annotated metadata kept at every level; annotation when not given.

has_default property 

has_default: bool

Return True if the parameter has a default.

problems 

problems(value: Any) -> list[str]

Return a line for each limit value breaks, located inside it: points[2]: ....

Empty when the annotation carries no limit. Each item of a *args given as a list or tuple, and each value of a **kwargs given as a mapping, is checked on its own.

PlanSpec dataclass 

PlanSpec(
    name: str,
    docs: str,
    parameters: list[ParamDescription],
    continuous: bool = False,
    pausable: bool = False,
)

Description of a plan's signature and type hints.

name instance-attribute 

name: str

Plan name, the callable's __name__.

docs instance-attribute 

docs: str

Plan docstring, or a default message without one.

parameters instance-attribute 

parameters: list[ParamDescription]

One description per parameter, in order.

continuous class-attribute instance-attribute 

continuous: bool = False

Whether the plan loops until stopped with a toggle button.

pausable class-attribute instance-attribute 

pausable: bool = False

Whether a running continuous plan can be paused and resumed.

How an annotation is read

Annotations map to ParamDescription fields, first match wins:

  1. Literal["a", "b"] -> choices=["a", "b"]
  2. Sequence[MyDevice] -> multi-select, choices=<matching device names>
  3. *args: MyDevice (VAR_POSITIONAL) -> multi-select
  4. MyDevice (bare protocol) -> single-select
  5. Everything else -> a plain value, left to the view layer to render

Step 5 accepts an annotation an input can show:

Annotation Shown when
int, float, str, bool, bytes, range, Path, datetime, date, time, timedelta, an Enum, a Literal always
list[T], Sequence[T], Iterable[T], tuple[T, ...], set[T], frozenset[T], or any one-element container a list, tuple, set or frozenset satisfies T can be shown
tuple[A, B] of fixed length every member can be shown
dict[K, V], Mapping[K, V], or any mapping a dict satisfies K is a plain type, an Enum or a Literal, and V can be shown
X \| None X can be shown
A \| B every member can be shown

Annotated metadata is set aside when deciding, at every level, so list[Annotated[float, ...]] is read as list[float]. A device inside any of these shapes cannot be shown, and neither can Any, a Callable, a container without type arguments, or a container no built-in satisfies, such as deque[T] or OrderedDict[K, V]: the plan would receive a list or a dict, not the class it names.

annotated-types limits in the metadata are kept in ParamDescription.annotated and checked by ParamDescription.problems; see Limit the values a parameter takes.

Parameter Outcome
an input can show it an input, starting from the default
no input can show it, and it has a default hidden=True: a view leaves it out and the plan keeps its default
no input can show it, and it has no default UnresolvableAnnotationError

The inputs each annotation gets are pictured in How to choose the inputs of a plan.

From the values to a call

Once the user fills in the plan widget, the presenter turns the values into a plan call:

from redsun.presenter.plan_spec import collect_arguments, resolve_arguments

# 1. Resolve: string device names -> live device instances, hidden parameters -> their defaults
resolved = resolve_arguments(spec, widget_values, devices)

# 2. Collect: build (args, kwargs) matching the plan signature
args, kwargs = collect_arguments(spec, resolved)

# 3. Run
engine(my_plan(*args, **kwargs))

Utilities

Predicates and helpers inspecting plan parameters.

create_plan_spec uses them to classify annotations, and resolve_arguments to turn device names into Device instances.

device_class 

device_class(ann: Any) -> Any

Return what devices are checked against for ann: its origin when it has type arguments.

Readable[Any] gives Readable, since isinstance takes no type arguments; any other annotation is returned unchanged.

get_choice_list 

get_choice_list(
    devices: Mapping[str, Device],
    proto: type[D],
    choices: Sequence[str],
) -> list[D]

Return the devices named in choices that are instances of proto.

Parameters:

Name Type Description Default
devices Mapping[str, Device]

Devices by name.

 required
proto type[D]

Class checked with isinstance.

 required
choices Sequence[str]

Names of the devices to consider.

 required

isdevice 

isdevice(ann: Any) -> bool

Return True if the annotation ann is a Device subclass.

isdevicesequence 

isdevicesequence(ann: Any) -> bool

Return True if ann is Sequence[T] where T is a Device subtype.

isdeviceset 

isdeviceset(ann: Any) -> bool

Return True if ann is Set[T] (or AbstractSet[T], FrozenSet[T]) where T is a Device subtype.

issequence 

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.