Bluesky CLI adapter
Run task files from a Bluesky plan through the existing EROBS command-line client.
bluesky_ros.simple_mtc_bluesky · Source
Examples and demos · Guide · Symbol index
How it is used
This example executes a sequence of task files through the existing beambot_client.py command. simple_mtc_plan() yields the Bluesky messages; MTCDevice.set() launches one blocking subprocess per file. The returned Ophyd status is already resolved when set() returns.
Use this page to understand the subprocess adapter. The direct ROS devices provide a different integration path; this module does not share their goal handles or background executor.
Start here
Callable |
Purpose |
|---|---|
Iterate task paths in order, yield abs_set for each file and then a Bluesky wait message, printing task progress. |
|
Run ros2 run beambot beambot_client.py for task_params.json_file, wait up to 300 seconds and complete an Ophyd Status from the subprocess exit code. |
|
Initialize ROS and RunEngine, locate workspace from EROBS_WORKSPACE or script parents, resolve task paths (or default CMS sequence) and run the sequential plan. |
Functions
simple_mtc_plan()
Iterate task paths in order, yield abs_set for each file and then a Bluesky wait message, printing task progress.
Function · Source
simple_mtc_plan(mtc_device, json_files)
Parameter |
Meaning |
|---|---|
|
Device implementing set({json_file: path}); normally MTCDevice. |
|
Sized iterable of task JSON paths; len() is used for progress. |
Returns: Bluesky generator yielding plan messages; generator return is None.
Errors: RunEngine/device failures propagate through iteration; this function does not validate files.
Side effects: Prints as iterated and causes task execution when run by a RunEngine; merely creating generator does not execute it.
Referenced in source (1)
main— bluesky_ros.simple_mtc_bluesky
main()
Initialize ROS and RunEngine, locate workspace from EROBS_WORKSPACE or script parents, resolve task paths (or default CMS sequence) and run the sequential plan.
Entry point · Source
main()
Returns: None, including early missing-workspace/default-file returns and caught errors; does not set a nonzero process exit code for caught task failure.
Errors: Run/setup Exceptions in the try print traceback; rclpy.init and final shutdown errors are outside its handler. Explicit task paths are not prevalidated.
Side effects: Reads environment/path existence, starts ROS, creates RunEngine and executes subprocess-backed robot tasks; finally shuts down ROS and prints completion.
MTCDevice
Adapt a task JSON filename to a blocking CLI subprocess and translate its exit status into an Ophyd status.
MTCDevice.set()
Run ros2 run beambot beambot_client.py for task_params.json_file, wait up to 300 seconds and complete an Ophyd Status from the subprocess exit code.
Method · Source
set(self, task_params)
Parameter |
Meaning |
|---|---|
|
Mapping requiring json_file, a path argument passed to the client process. Other keys are unused. |
Returns: Finished Status on exit code zero or exceptional Status for nonzero/timeout/launch errors; call blocks until outcome.
Errors: Missing json_file raises before the try/status handler. Timeout and other subprocess Exceptions are converted to status errors; stderr becomes failure detail. Killing a timed-out client does not guarantee its remote robot goal is canceled.
Side effects: Spawns a robot-task client, captures its stdout/stderr and prints summary lines; may initiate physical task execution.
MTCDevice.__init__()
Store a Bluesky device name and expose a get_logger lambda returning print.
Lifecycle · Source
__init__(self, name='mtc_device')
Parameter |
Meaning |
|---|---|
|
Device name visible to the plan/RunEngine. |
Errors: No explicit validation or error handling.
Side effects: Sets in-memory attributes only; no ROS client.
Referenced in source (1)
main— bluesky_ros.simple_mtc_bluesky