API source snapshot · 2026-09-24
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """MTC Python Ophyd Device for Bluesky Integration This module provides an Ophyd device wrapper for beambot orchestrator, enabling Bluesky experiment orchestration with MTC robot tasks. """ from ophyd.status import DeviceStatus from rclpy.node import Node from rclpy.action import ActionClient from rosidl_runtime_py.utilities import get_action from bluesky.protocols import Movable import rclpy class ActionStatus(DeviceStatus): """Track the status of a ROS Action for Bluesky""" def __init__(self, device, **kwargs): super().__init__(device, **kwargs) def _handle_failure(self): self.device.cancel_goal() class MTCExecutionDevice(Node, Movable): """Ophyd device for MTC task execution via Bluesky""" def __init__(self, name="mtc_executor", **kwargs): super().__init__(name, **kwargs) # Ophyd attributes Bluesky expects on a Movable (merge_cycler reads .parent) self.name = name self.parent = None self.action_type = get_action('beambot_interfaces/MTCExecution') self._action_client = ActionClient(self, self.action_type, 'beambot_execution') self._goal_handle = None self._bluesky_status = None self._send_goal_future = None self._get_result_future = None def construct_goal_message(self, json_path_or_string): """Construct goal from JSON file path or JSON string Args: json_path_or_string: Either a path to a .json file or a JSON string Returns: MTCExecution.Goal with full_json populated """ goal = self.action_type.Goal() # If it's a file path, read the file if isinstance(json_path_or_string, str) and json_path_or_string.endswith('.json'): with open(json_path_or_string, 'r') as f: goal.full_json = f.read() else: # Assume it's already a JSON string goal.full_json = json_path_or_string return goal def set(self, json_path_or_string): """Execute MTC task""" self._bluesky_status = ActionStatus(self) # Construct and send goal goal_msg = self.construct_goal_message(json_path_or_string) self.get_logger().info("Waiting for action server...") self._action_client.wait_for_server(timeout_sec=10.0) self.get_logger().info("Sending goal...") self._send_goal_future = self._action_client.send_goal_async( goal_msg, feedback_callback=self._feedback_callback ) self._send_goal_future.add_done_callback(self._goal_response_callback) # Spin until complete while not self._bluesky_status.done: rclpy.spin_once(self, timeout_sec=0.1) return self._bluesky_status def _feedback_callback(self, feedback_msg): """Handle execution feedback""" fb = feedback_msg.feedback self.get_logger().info( f"Step {fb.current_step}: {fb.current_action} " f"({fb.progress_percentage:.1f}%) - {fb.status_message}" ) def _goal_response_callback(self, future): """Handle goal acceptance/rejection""" goal_handle = future.result() if not goal_handle.accepted: self.get_logger().error("Goal rejected") self._bluesky_status.set_exception(Exception("Goal rejected")) return self.get_logger().info("Goal accepted") self._goal_handle = goal_handle self._get_result_future = goal_handle.get_result_async() self._get_result_future.add_done_callback(self._result_callback) def _result_callback(self, future): """Handle final result""" result = future.result() if result.status == 4: # SUCCEEDED self.get_logger().info( f"Task completed: {result.result.completed_steps}/{result.result.total_steps} steps" ) self._bluesky_status.set_finished() elif result.status == 5: # ABORTED self.get_logger().error(f"Task aborted: {result.result.error_message}") self._bluesky_status.set_exception(Exception(result.result.error_message)) elif result.status == 6: # CANCELED self.get_logger().warning("Task canceled") self._bluesky_status.set_exception(Exception("Canceled")) else: self.get_logger().error(f"Unknown result status: {result.status}") self._bluesky_status.set_exception(Exception("Unknown status")) def cancel_goal(self): """Cancel the current goal""" if self._goal_handle is not None: self.get_logger().info("Canceling goal...") cancel_future = self._goal_handle.cancel_goal_async() rclpy.spin_until_future_complete(self, cancel_future, timeout_sec=5.0) |