Skip to content

redsun.engine

Run engine

RunEngine 

RunEngine(
    md: dict[str, Any] | None = None,
    *,
    loop: AbstractEventLoop | None = None,
    preprocessors: list[Preprocessor] | None = None,
    md_validator: MDValidator | None = None,
    md_normalizer: MDNormalizer | None = None,
    scan_id_source: MDScanIDSource = default_scan_id_source,
    call_returns_result: bool = True,
)

Bases: RunEngine

Runs plans and emits documents without blocking the calling thread.

Wraps bluesky.run_engine.RunEngine: __call__ runs the plan on a separate thread and returns a concurrent.futures.Future of its result.

Parameters:

Name Type Description Default
md dict[str, Any] | None

Metadata store, a dict by default. Any object with __getitem__, __setitem__ and clear works, such as historydict.HistoryDict, which persists history in a sqlite file.

 None
loop AbstractEventLoop | None

Event loop plans run on. Defaults to the shared background loop.

 None
preprocessors list[Preprocessor] | None

Generator functions modifying a plan's messages, such as the bluesky.plans functions ending in wrapper. [f, g] applies as f(g(plan)).

 None
md_validator MDValidator | None

Raises to prevent a run whose metadata it finds invalid; its return value is ignored.

 None
md_normalizer MDNormalizer | None

Like md_validator, raises for invalid metadata; otherwise returns the normalized metadata.

 None
scan_id_source MDScanIDSource

Function, possibly async, returning the next scan_id. By default scan_id increments by 1.

 default_scan_id_source
call_returns_result bool

What the future __call__ returns holds: a RunEngineResult describing the run if True, a tuple of uids if False.

 True

Attributes:

Name Type Description
md

The metadata store described above.
record_interruptions

False by default. True adds an event stream recording interruptions (pauses, suspensions).
state

One of idle, running, pausing, paused, halting, stopping, aborting, suspending and panicked.
suspenders

Read-only collection of bluesky.suspenders.SuspenderBase objects that suspend and resume execution.
preprocessors

The list of preprocessors described above.
msg_hook

f(msg) called with every bluesky.Msg before it is processed, for logging or debugging. None by default.
state_hook

f(new_state, old_state) called on every state change. The engine sets its own, which feeds sig_state_changed.
waiting_hook

f(status_object) called while waiting for long-running commands (trigger, set, kickoff, complete), for example to show progress.
progress_hook Callable[[list[PlanProgress] | None], None] | None

f(scopes) called with the open progress scopes of the running plan, parents first, or None to clear them. The engine sets its own, which feeds sig_progress.
ignore_callback_exceptions

False by default.
loop

The event loop plans run on, such as one from asyncio.new_event_loop().
max_depth

Maximum stack depth, preventing calls to the RunEngine from inside a function, which breaks introspection. None by default; 2 suits the Python interpreter and 11 IPython (tested on 5.1.0).
pause_msg

Message printed when a run is interrupted, with instructions for changing the RunEngine's state. Empty here; bluesky defaults to bluesky.run_engine.PAUSE_MSG.
commands

The list of commands a Msg can name.

sig_locks_changed class-attribute instance-attribute 

sig_locks_changed = Signal(frozenset)

The names of the locked devices, whenever that set changes.

sig_state_changed class-attribute instance-attribute 

sig_state_changed = Signal(str, str)

The engine's new and old state on every change, named as state names them.

sig_progress class-attribute instance-attribute 

sig_progress = Signal(tuple)

Every progress scope of the running plan, as a tuple of PlanProgress, parents before children, whenever one is declared, updated or finished; an empty tuple when none is left.

The scopes are the engine's own: the engine's thread replaces their update after the emission, so a slot reads it once for each emission and keeps a copy of what it needs later.

locked property 

locked: frozenset[str]

The names of the devices the running plan locks.

__call__ 

__call__(
    plan: Iterable[Msg],
    subs: Subscribers | None = None,
    /,
    **metadata_kw: Any,
) -> Future[RunEngineResult | tuple[str, ...]]

Execute a plan.

Keyword arguments are metadata recorded with every run the plan creates. The plan and optional subscriptions are positional.

Parameters:

Name Type Description Default
plan Iterable[Msg]

A generator yielding Msg objects, or an iterable returning one.

 required
subs Subscribers | None

Callbacks subscribed for this run only, given as:

  • a callable, which will be subscribed to 'all'
  • a list of callables, which again will be subscribed to 'all'
  • a dictionary, mapping specific subscriptions to callables or lists of callables; valid keys are {'all', 'start', 'stop', 'event', 'descriptor'}
 None

resume 

resume() -> Future[RunEngineResult | tuple[str, ...]]

Resume the paused plan on a separate thread.

Returns a future of the resumed plan's result. Pausing completes the future __call__ returned, so this is a new one.

cancel_pause 

cancel_pause() -> None

Withdraw a deferred pause that no checkpoint has reached yet.

stop 

stop() -> Future[RunEngineResult | tuple[str, ...]]

Stop the plan and mark it successful, on a thread of its own.

A paused plan runs its cleanup there, not on the caller's thread.

abort 

abort(
    reason: str = "",
) -> Future[RunEngineResult | tuple[str, ...]]

Stop the plan and mark it aborted, on a thread of its own.

halt 

halt() -> Future[RunEngineResult | tuple[str, ...]]

Stop the plan with no cleanup, on a thread of its own.

Deferrals 

Deferrals(engine: RunEngine)

Changes to apply before the next message of the running plan.

Every change runs on the engine's loop. One asked for while a plan runs waits for the message under way to complete, then runs before the next message is sent, or before the plan returns if that message was its last; the plan is neither suspended nor rewound. One asked for while no plan runs is applied at once. A change that raises is logged, and the ones after it still run.

request 

request(
    apply: Callable[[], Awaitable[None]],
) -> Future[None]

Ask for apply to run, and return a future done once it did.

Safe from any thread. A caller on a loop must not block on the future.

wrap 

wrap(plan: MsgGenerator[Any]) -> MsgGenerator[Any]

Run the changes queued so far before each message of plan and before it returns.

PlanProgress 

PlanProgress(
    name: str, *, parent: PlanProgress[Any] | None = None
)

Bases: Generic[T]

A progress scope a plan declared, passing each update to its watchers.

Watchers receive every field of a WatcherUpdate as keywords, those that are None included, and the values themselves, not copies. name is always the name of the scope.

name instance-attribute 

name = name

Name the plan declared the scope under.

parent instance-attribute 

parent = parent

Scope this one is nested under, or None.

done property 

done: bool

Whether the scope has finished.

update property 

update: WatcherUpdate[T | None] | None

The last update, or None before the first.

watch 

watch(watcher: Watcher[T]) -> None

Pass every update to watcher, starting with the last one, if any.

finish 

finish() -> None

Mark the scope done, reporting it complete to the watchers once more.

register_bound_command 

register_bound_command(
    engine: RunEngine,
    command: Callable[[RunEngine, Msg], Any],
) -> None

Register command in engine, under the command's __name__.

Unlike RunEngine.register_command, binds the command to engine: it is called with the engine first, then the Msg.

RunEngineResult is the class of bluesky, RunEngineResult.

Actions

Decorators and types for continuous, interactive plans.

A continuous plan loops until stopped, and may be paused and resumed and take actions the user triggers while it runs.

  • SRLatch: a set-reset latch synchronising a plan with outside signals.
  • continuous: marks a plan as continuous, recording whether it is pausable.
  • PlanAction: what a plan declares of an action: its name, its description and the labels of its button.
  • ActionManager: the actions a running plan offers, asked for from outside and reported as they change state.

PlanAction dataclass 

PlanAction(
    *,
    name: str,
    description: str = "",
    toggle_states: tuple[str, str] | None = None,
)

An action a user triggers while a continuous plan runs.

A declaration only. A plan names it as the default of a parameter, from which its button is made, and waits for it through an ActionManager.

Subclass it to add fields; the subclass is a frozen dataclass too.

name instance-attribute 

name: str

Name of the action.

description class-attribute instance-attribute 

description: str = ''

Short description of the action, usable as a tooltip.

toggle_states class-attribute instance-attribute 

toggle_states: tuple[str, str] | None = None

Labels of a button that stays pressed until it is released.

The first is shown while it is released and the second while it is pressed. None for a button that is clicked.

ActionManager 

ActionManager()

The actions a running plan offers, and the state each is in.

Whoever owns the plans owns one. A plan waits on it with wait, a user asks through request, and sig_changed reports every change of state. Only the plan changes a state, so the changes are reported in the order they happen.

sig_changed class-attribute instance-attribute 

sig_changed = Signal(str, str)

Emitted with the name of an action and its new ActionState.

request 

request(name: str, on: bool = True) -> None

Ask for the action name, or with on=False ask a running one to end.

Safe from any thread. Asking for an action no plan offers, or asking one that is not running to end, changes nothing and raises nothing: it is logged as a warning.

wait 

wait(
    *actions: PlanAction, poll_interval: float = SIXTY_FPS
) -> MsgGenerator[str]

Offer actions, wait until one is asked for, and return its name.

Every call offers latches of its own, so a request left from an earlier wait cannot start an action of this one. The action returned runs until done. The others go back to idle, as all of them do when the plan is stopped while it waits. A checkpoint is yielded every poll_interval seconds, so it cannot be used between create and save.

Raises:

Type Description
ValueError

If no action is given.

wait_released 

wait_released(
    action: PlanAction, poll_interval: float = SIXTY_FPS
) -> MsgGenerator[None]

Wait until the running action is asked to end.

Raises:

Type Description
ValueError

If action is not running.

done 

done(name: str) -> None

Say the plan finished the action name, which goes back to idle.

ActionState

Bases: StrEnum

The state of an action, as ActionManager.sig_changed reports it.

IDLE class-attribute instance-attribute 

IDLE = 'idle'

No plan waits for it and none runs it.

OFFERED class-attribute instance-attribute 

OFFERED = 'offered'

A plan waits for it.

RUNNING class-attribute instance-attribute 

RUNNING = 'running'

The plan took it, and has not finished it.

SRLatch 

SRLatch()

A set-reset latch a coroutine waits on, settable from any thread.

A coroutine waits for either the set or the reset state, on whatever loop it runs. A new latch is reset.

changed_at property 

changed_at: float

When the latch last changed state, as time.monotonic reads it.

0.0 for a latch that never changed.

set 

set() -> None

Set the latch, waking every coroutine in wait_for_set.

Does nothing if already set.

reset 

reset() -> None

Reset the latch, waking every coroutine in wait_for_reset.

Does nothing if already reset.

is_set 

is_set() -> bool

Return whether the latch is set.

wait_for_set async 

wait_for_set() -> None

Wait until the latch is set; return at once if it already is.

wait_for_reset async 

wait_for_reset() -> None

Wait until the latch is reset; return at once if it already is.

Continuous dataclass 

Continuous(pausable: bool = False)

What continuous records on the plan it marks.

pausable class-attribute instance-attribute 

pausable: bool = False

Whether the run engine can pause and resume the plan.

continuous 

continuous(func: F) -> F

continuous(*, pausable: bool = False) -> Callable[[F], F]

continuous(
    func: F | None = None, /, *, pausable: bool = False
) -> F | Callable[[F], F]

Mark a plan as continuous: it loops until it is stopped.

A continuous plan gets a toggle to start and stop it, and with pausable a button to pause and resume it. Usable with or without arguments:

@continuous
def my_plan() -> MsgGenerator[None]: ...


@continuous(pausable=True)
def my_plan(detectors: Sequence[DetectorProtocol]) -> MsgGenerator[None]: ...

The signature is untouched; what was asked is stored on the function as __continuous__, a Continuous.

Waiting on a latch

ActionManager makes one SRLatch for each action it offers, and waits on it. A latch is set and reset from any thread, and a coroutine waits for either state on whatever loop it runs:

latch = SRLatch()

# in a coroutine:
await latch.wait_for_set()  # blocks until set()
await latch.wait_for_reset()  # blocks until reset()

latch.changed_at is when the latch last changed state, as time.monotonic reads it, and 0.0 for a latch that never changed.

For a wait_for_actions message, the RunEngine runs one wait_for_set (or wait_for_reset) task per latch, for poll_interval seconds at most. Of several latches in the wanted state, it returns the one that reached that state first. Of those that reached it together, it returns the first in the map. If no latch is in the wanted state when the interval ends, the stub yields its next checkpoint and sends the message again.

Plan stubs

redsun.engine.plan_stubs holds steps to use inside larger plans, beside those of bluesky.plan_stubs.

Descriptor stubs

import redsun.engine.plan_stubs as rps

# gather descriptors from Readable / Collectable devices inside a plan
descriptor = yield from rps.describe(readable)
descriptors = yield from rps.describe_collect(collectable)

Lock stubs

A plan locks the devices it must not have disturbed, and views disable their controls while those devices are locked. The run engine keeps the locks and announces them on RunEngine.sig_locks_changed, which a view connects to.

import redsun.engine.plan_stubs as rps

# lock the stage and camera while the inner plan runs, however it ends
yield from rps.lock_wrapper(scan(stage, camera), stage, camera)

Action flow-control stubs

wait_for_actions waits on a map of names to latches. ActionManager.wait and ActionManager.wait_released use it, so a plan written against an ActionManager does not call it.

import redsun.engine.plan_stubs as rps
from redsun.engine.actions import SRLatch

latches = {"snap": SRLatch()}

# wait until a latch in the map is set, however long that takes,
# with a checkpoint every `poll_interval` seconds
name, latch = yield from rps.wait_for_actions(latches, poll_interval=0.016)

The stub does not time out. It waits as long as it takes for a latch to be set, or reset with wait_for="reset", and returns at once if one already is. While it waits it yields a checkpoint every poll_interval seconds, 1/60 s by default. A checkpoint is where the plan can be paused, so the stub cannot sit between create and save. An empty map raises ValueError.

Progress stubs

A plan opens a progress scope, moves it and finishes it. The run engine announces every open scope on RunEngine.sig_progress, which a view connects to. How to show a plan's progress shows the whole path.

import redsun.engine.plan_stubs as rps

yield from rps.declare_progress("series")
yield from rps.update_progress("series", current=3, initial=0, target=10, unit="frames")
yield from rps.update_progress("series", done=True)
yield from rps.monitor_progress("frames", status)  # follow a device status

Plan stubs adding action flow control to bluesky.plan_stubs.

wait_for_actions waits on user actions; lock, unlock and lock_wrapper lock devices against the user. Every stub is a generator yielding Msg objects, used inside larger plans with yield from.

wait_for_actions 

wait_for_actions(
    events: Mapping[str, SRLatch],
    poll_interval: float = SIXTY_FPS,
    wait_for: Literal["set", "reset"] = "set",
) -> MsgGenerator[tuple[str, SRLatch]]

Wait until one of the given latches is in the wanted state, and return it.

Returns the name and the latch as soon as a latch is set, or reset with wait_for="reset", whether it was already or changed meanwhile. Of several in the wanted state, the one that reached it first is returned, and of those that reached it together, the first in events. It waits as long as that takes. A checkpoint is yielded every poll_interval seconds, where the plan can be paused, so it cannot be used between create and save.

Parameters:

Name Type Description Default
events Mapping[str, SRLatch]

Mapping of action names to their SRLatch objects.

 required
poll_interval float

Seconds between two checkpoints, 1/60 s by default.

 SIXTY_FPS
wait_for Literal['set', 'reset']

Whether to wait for a latch to be set or reset.

 'set'

Raises:

Type Description
ValueError

If events is empty.

describe 

describe(
    obj: Readable[Any],
) -> MsgGenerator[dict[str, Descriptor]]

Return what obj.describe() returns, from inside a plan.

describe_collect 

describe_collect(
    obj: Collectable,
) -> MsgGenerator[
    dict[str, Descriptor] | dict[str, dict[str, Descriptor]]
]

Return what obj.describe_collect() returns, from inside a plan.

lock 

lock(*devices: HasName) -> MsgGenerator[str]

Lock devices, so views disable their controls, and return the lock's token.

Prefer lock_wrapper, which unlocks however the plan ends.

unlock 

unlock(token: str) -> MsgGenerator[None]

Release the lock that returned token.

lock_wrapper 

lock_wrapper(
    plan: MsgGenerator[T], *devices: HasName
) -> MsgGenerator[T]

Run plan with devices locked, unlocking them however it ends.

def scan(stage: Stage, camera: Camera) -> MsgGenerator[None]:
    yield from lock_wrapper(bp.count([camera], 10), stage, camera)

declare_progress 

declare_progress(
    name: str, *, parent: str | None = None
) -> MsgGenerator[PlanProgress]

Open a progress scope named name, nested under parent when given.

Returns the scope. A view shows it while the plan runs, and the engine finishes it when the plan ends.

update_progress 

update_progress(
    name: str,
    *,
    current: Any = None,
    initial: Any = None,
    target: Any = None,
    unit: str = "unit",
    precision: int | None = None,
    fraction: float | None = None,
    time_elapsed: float | None = None,
    time_remaining: float | None = None,
    done: bool = False,
) -> MsgGenerator[None]

Report how far the scope name has got, or finish it with done.

Parameters:

Name Type Description Default
current Any

Where the scope is, started and ends, in unit; target left out when the end is not known.

 None
initial Any

Where the scope is, started and ends, in unit; target left out when the end is not known.

 None
target Any

Where the scope is, started and ends, in unit; target left out when the end is not known.

 None
fraction float | None

How far the scope has got, from 0 to 1, in place of the three above.

 None
precision int | None

Decimals to show the numbers with.

 None

monitor_progress 

monitor_progress(
    name: str,
    status: WatchableAsyncStatus[T],
    *,
    parent: str | None = None,
) -> MsgGenerator[PlanProgress[T]]

monitor_progress(
    name: str, status: Status, *, parent: str | None = None
) -> MsgGenerator[PlanProgress]

monitor_progress(
    name: str, status: Status, *, parent: str | None = None
) -> MsgGenerator[PlanProgress]

Open the progress scope name and keep it in step with status.

A status reporting updates, such as a detector's complete or a motor's set in ophyd-async, fills the scope; any other shows it without an end. The scope finishes when the status is done, whether it succeeds or not; the plan still waits on the status itself, and does not update or finish the scope with update_progress. Returns the scope, which reports values of the type the status reports.