概要

计划是描述数据采集过程的一系列原子操作。每个操作由一个 bluesky.Msg（“消息”）对象表示。计划可以作为简单的消息列表来实现：

In [3]: plan = [Msg('open_run'), Msg('close_run')]

In [4]: RE(plan)
Out[4]: ('6e21a67d-afe5-45f8-b1c6-586aa644e8a8',)

或者作为一次只生成一条消息的生成器:

In [6]: def plan():
   ...:     yield Msg('open_run')
   ...:     yield Msg('close_run')
   ...:

In [7]: RE(plan())
Out[7]: ('b71e6f0d-60fe-4cd9-8958-44c824d18fd1',)

上述示例是等效的。对于更复杂的用途，第二种方式（生成器）更为强大，因为它可以包含循环、条件判断、自适应逻辑——实际上任何 Python 代码都可以。

但关键在于，计划代码本身绝不能与硬件直接通信。（切勿在计划中使用 epics.caput(...)！）相反，每个操作都由一个描述应执行动作的 Msg 对象表示。

在不接触真实硬件的情况下，可以安全地检查计划以进行错误校验、模拟和可视化。例如，我们可以如下打印计划中的每条消息：

In [8]: plan = [Msg('open_run'), Msg('close_run')]

In [9]: for msg in plan:
   ...:     print(msg)
   ...:
Msg('open_run', obj=None, args=(), kwargs={}, run=None)
Msg('close_run', obj=None, args=(), kwargs={}, run=None)

一个Msg包含5个作为属性可访问的成员:

• command
• obj
• args
• kwargs
• run

其中 command 必须是受控命令列表中的一项，obj 是要应用该命令的对象（即设备），如果适用的话，args 和 kwargs 是该命令的参数，而 run 是用户定义的运行键。运行键由运行引擎用于将每条消息与一个打开的运行相关联，管理每个打开运行的状态，并将运行数据路由到一组独立的回调函数（参见多运行计划的文档）。

要执行计划, 运行引擎一次一条消息地消费它:

def very_simple_run_engine(plan):
    for msg in plan:
        # Process the msg.

运行引擎维护着一个注册表，其用于根据 Msg.command 值来分发这些Msg对象。例如，若运行引擎接收到消息 Msg('set', motor, 5)，它将：

1. 识别出该消息的命令为 'set'。
2. 在其命令注册表中查找 'set'，发现它映射到 RunEngine._set。
3. 将 Msg('set', motor, 5) 传递给其 _set 方法。
4. 在 _set 内部，调用 motor.set(5)。（这里才发生与硬件的实际通信。）
5. 更新一些将在之后有用地内部缓存。例如，它会跟踪电机可能正在移动的事实，以便在发生错误时安全地停止它。这说明了另一个重要原因：计划必须始终通过生成消息来与硬件交互，绝不能直接与硬件通信。在计划内部调用 epics.caput 会使 RunEngine 无法感知，从而绕过其在意外退出或错误时将设备置于安全状态的机制。

默认情况下，一组标准命令已注册。按照惯例，命令为 'name' 的消息映射到 RunEngine 上名为 _name 的协程方法，如上例中的 'set' -> RunEngine._set。用户可以注册自己的协程来添加自定义命令，尽管这很少需要。

某些命令不涉及与硬件的通信。例如，Msg('sleep', None, 5) 使 RunEngine 休眠 5 秒。None 是不适用于 'sleep' 命令的“对象”（设备）的占位符。正如计划绝不应直接与硬件通信一样，它们也绝不应使用像 time.sleep() 这样的长时间阻塞调用。相反，'sleep' 命令映射到 RunEngine._sleep，与 RunEngine 的事件循环集成，以非阻塞方式休眠，使得 RunEngine 在此期间保持响应——监视用户中断并可能在后台异步采集数据。

其他命令用于控制元数据和输入/输出。例如，Msg('open_run') 和 Msg('close_run') 界定了一次运行的范围。传递给 'open_run' 消息的任何关键字参数都被解释为元数据，并编码到 RunStart 文档中。

以下是内置命令的全面概述：

命令

以下是“内置”命令，其中一些与 RunEngine 实例的状态紧密相关。

create

此命令告知运行引擎应开始收集read的结果以创建一个事件。如果在它们之间没有save或drop的情况下调用两次，则会引发异常（因为一次不能有多个打开的事件）。这非常依赖于运行引擎的内部状态，不应由用户覆盖。此调用向协程返回 None。它忽略消息中除命令外的所有部分。

save

这是与 create 配对的命令，用于打包并触发Event文档被发出。必须在 create 之后调用，否则扫描将失败并引发 IllegalMessageSequence 异常。此命令高度依赖运行引擎的内部状态，不应被篡改。该调用向协程返回 None。除命令本身外，消息的其他部分均被忽略。

read

这触发对此消息中obj调用读取.

msg.obj.read(*msg.args, **msg.kwargs)

在create和save之间被读取的任何内容将被打包到一个事件中.这高度依赖于运行引擎的内部状态，不应被篡改。将 read 返回的字典返回给协程。消息的 args 和 kwargs 部分会传递给 read 方法。

null

这是一个空消息，被运行引擎忽略。它的存在是为了使逻辑代数成立。向协程返回 None。忽略消息中除命令外的所有值。

set

指示一个Mover进行移动。目前，这模仿了类似 EPICS 的即时运动逻辑。

stage 和 unstage

指示运行引擎对对象进行阶段准备/阶段清理。这会调用 obj.stage()/obj.unstage()。预计的消息对象是:

Msg('stage', object)
Msg('unstage', object)

这产生这些调用:

staged_devices = object.stage()
unstaged_devices = object.unstage()

其中 staged_devices/unstaged_devices 是被（取消）阶段准备的 ophyd.Device 列表，而非状态对象。

有人可能会疑惑为什么返回值是一个设备列表，而不是像 set 及类似Msg中那样的状态对象。这个问题曾讨论过一段时间。阶段准备期间执行的操作应涉及配置调整，且应快速完成。阶段准备不应包含耗时的设置调用。

为什么要返回被阶段准备的对象列表？对设备进行阶段准备会导致其组件设备（如果有）也被阶段准备。所有这些子设备连同 [self] 会被添加到一个列表中，并由 Device.stage() 返回，以便计划能够跟踪哪些设备已被阶段准备，示例如下：

devices_staged = yield Msg('stage', device)

为什么计划需要知道这些？这是为了避免意外地尝试对同一设备进行两次阶段准备，例如，对父设备进行阶段准备后，又试图对其子设备也进行阶段准备。避免这种情况非常重要，因为重复阶段准备会引发错误。

trigger

这将调用obj.trigger方法并且缓存返回的状态对象.

sleep

休眠当前事件循环.

wait

阻塞进程，直到每个被触发或设置了关键字参数 group=<GROUP> 的对象完成操作。

预期的消息对象是：

Msg('wait',group=<GROUP>)

此处<GROUP>是任何可哈希键.

wait_for

指示运行引擎等待此 asyncio.Future 对象完成。这允许从外部对运行引擎进行任意控制。

input

处理一个输入. 允许在运行中用户输入.示例:

Msg('input', None)
Msg('input', None, prompt='>')

checkpoint

指示运行引擎创建一个检查点, 因而如果需要我们可以退到这个点

clear_checkpoint

清除一个设置的检查点

rewindable

pause

请求运行引擎暂停

预计的消息对象是:

Msg('pause', defer=False, name=None, callback=None)

kickoff

启动飞行扫描对象.

collect

收集由飞行器缓存的数据并且发出描述符和事件文档. 这调用obj.collect()方法.

complete

告知飞行器：“随时准备停止采集”。

这会调用给定对象的 obj.complete() 方法。飞行器返回一个状态对象。某些飞行器会响应此命令，停止采集并立即返回一个已完成的状态对象。其他飞行器则会完成其既定过程，无论该命令何时发出，都会在其完成时结束。

configure

配置一个对象. 预计的消息对象是:

Msg('configure', object, *args, **kwargs)

这产生这个调用:

object.configure(*args, **kwargs)

subscribe

在运行开始后添加订阅。

与传递给 __call__ 的订阅一样，这些订阅将在运行结束时由 RunEngine 移除。

预期的消息对象为：

Msg(‘subscribe’, None, callback_function, document_name)

其中 document_name 可以是以下之一：{‘start’, ‘descriptor’, ‘event’, ‘stop’, ‘all’}

并且 callback_function 应具有如下签名：f(name, document)

这里的 name 是上述 document_name 选项之一，document 是事件模型中的文档字典之一。

更多信息请参阅 bluesky.run_engine.Dispatcher.subscribe() 的文档字符串。

unsubscribe

在调用过程中移除订阅——适用于多运行调用中某些运行需要订阅而其他运行不需要的场景。

预计的消息对象是:

Msg('unsubscribe', None, TOKEN)
Msg('unsubscribe', token=TOKEN)

此处TOKEN是从RunEngine._subscribe()返回的值.

open_run

指示RunEngine开始一个新"运行", 预计的消息对象是:

Msg('open_run', None, **kwargs)

此处**kwargs是应该进入到RunStart文档的任何其它元数据

close_run

指示运行引擎编写RunStop文档, 预计的消息对象是:

Msg('close_run', None, exit_status=None, reason=None)

如果未提供 exit_stats 和 reason，则使用运行引擎中存储的值。

drop

在不发出已完成的事件文档的情况下, 丢弃一组读数。

此命令用于放弃先前 create 和 read 命令的结果，而不发出事件。它可用于丢弃已知的坏事件（例如没有光束），保持事件文档流的整洁。在执行 drop 之后，可以安全地开始新的 create、read、save 序列。

必须在 create 之后调用，否则扫描将失败并引发 IllegalMessageSequence 异常。

此调用向协程返回 None。除命令本身外，消息的所有其他部分均被忽略。

monitor

监控一个信号。异步发出事件文档。

立即发出一个描述符文档。随后，定义一个闭包，该闭包从独立线程中发出与该描述符关联的事件文档。此过程与主要的事件打包流程（create/read/save）无关。

预期的消息对象为：

Msg('monitor', obj, **kwargs)
Msg('monitor', obj, name='event-stream-name', **kwargs)

此处kwargs被传递到obj.subscribe().

unmonitor

停止监视, 即, 移除发出事件文档的回调. 预计的消息对象是:

Msg('unmonitor', obj)

stop

停止一个设备,预计的消息对象是:

Msg('stop', obj)

这等价于调用obj.stop().

注册自定义的命令

运行引擎可以注册任何新命令. 可以使用以下方法注册它们.

RunEngine.register_command(name, func)

注册新的消息命令.

参数:

• name: str
• func: callable, 这可以是一个函数或方法. 签名为f(msg)

RunEngine.unregister_command(name)

注销一条消息命令. 参数: name: str

RunEngine.commands

对Msg可用的命令列表.

In [41]: RE.commands
Out[41]:
['declare_stream',
 'create',
 'save',
 'drop',
 'read',
 'locate',
 'monitor',
 'unmonitor',
 'null',
 'RE_class',
 'stop',
 'set',
 'trigger',
 'sleep',
 'wait',
 'checkpoint',
 'clear_checkpoint',
 'rewindable',
 'pause',
 '_resume_from_suspender',
 '_start_suspender',
 'prepare',
 'collect',
 'kickoff',
 'complete',
 'configure',
 'stage',
 'unstage',
 'subscribe',
 'unsubscribe',
 'open_run',
 'close_run',
 'wait_for',
 'input',
 'install_suspender',
 'remove_suspender']

RunEngine.print_command_registry(verbose=False)

这便捷地打印出可用命令的命令注册表.

参数:

• verbose: bool, 可选, 详细打印, 默认是Fasle.

In [45]: print(RE.print_command_registry())
List of available commands
declare_stream : Trigger the run engine to start bundling future obj.describe() calls for
create : Trigger the run engine to start bundling future obj.read() calls for
save : Save the event that is currently being bundled
drop : Drop the event that is currently being bundled
read :
locate :
monitor :
unmonitor :
null :
RE_class :
stop :
set :
trigger :
sleep :
wait : Block progress until every object that was triggered or set
checkpoint : Instruct the RunEngine to create a checkpoint so that we can rewind
clear_checkpoint : Clear a set checkpoint
rewindable : Set rewindable state of RunEngine
pause : Request the run engine to pause
_resume_from_suspender : Request the run engine to resume
_start_suspender :
prepare : Prepare a flyer for a flyscan
collect :
kickoff : Start a flyscan object
complete :
configure : Configure an object
stage : Instruct the RunEngine to stage the object
unstage : Instruct the RunEngine to unstage the object
subscribe :
unsubscribe :
open_run : Instruct the RunEngine to start a new "run"
close_run : Instruct the RunEngine to write the RunStop document
wait_for : Instruct the RunEngine to wait for futures and return the resulting tasks.
input :
install_suspender :
remove_suspender :