#!/usr/bin/env python3 """XR-RM 桌面调试启动器。 提供 Tkinter 图形界面，按“仿真/左臂/右臂/双臂/诊断”组织常用 ROS2 launch、sample_udp_sender、topic 监控和环境检查命令，降低现场调试时的命令输入成本。 """ from __future__ import annotations import ctypes import ctypes.util import os import re import shlex import shutil import signal import subprocess import tempfile import time from pathlib import Path try: import tkinter as tk from tkinter import messagebox, ttk except ImportError as exc: print(f"Missing dependency: {exc}") print("Install Tkinter with: sudo apt-get install python3-tk") raise SystemExit(1) DEFAULT_LEFT_IP = "192.168.192.18" DEFAULT_RIGHT_IP = "192.168.192.19" DEFAULT_PORT = "15000" SAMPLE_SENDER_SECONDS = "30" SAMPLE_SENDER_STAGGERED_SECONDS = "60" SAMPLE_SENDER_ARGS = ( f"--host 127.0.0.1 --port {DEFAULT_PORT} " "--pattern axis_sweep --amplitude 0.5 " "--hold-seconds 4.0 --center-seconds 1.2 --initial-center-seconds 1.0" ) TERMINAL_TITLE_PREFIX = "XR-RM Terminal - " TOPIC_MONITOR_TITLE = "XR-RM Topic Monitor" TOPIC_MONITOR_ACTION = "__xr_rm_topic_monitor__" ROS_GRAPH_MONITOR_TITLE = "XR-RM ROS Graph Monitor" ROS_GRAPH_MONITOR_ACTION = "__xr_rm_ros_graph_monitor__" TOPIC_MONITORS = [ ("Left Controller", "/xr/left_controller"), ("Right Controller", "/xr/right_controller"), ] ROS_GRAPH_MONITORS = [ ("ROS Topic List", "ros2 topic list"), ("ROS Node List", "ros2 node list"), ] MODES = [ "Simulation", "Left Arm", "Right Arm", "Dual Arm", "Diagnostics", ] def _quote(value: str | Path) -> str: return shlex.quote(str(value)) def _config_quote(value: str) -> str: return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"' def _with_index(items: list[tuple[str, str]]) -> list[tuple[str, str]]: return [(f"{idx}. {title}", cmd) for idx, (title, cmd) in enumerate(items, start=1)] def _sample_udp_sender_command( hand: str, seconds: str = SAMPLE_SENDER_SECONDS, both_mode: str = "synchronized", ) -> str: command = ( f"ros2 run xr_rm_input sample_udp_sender --hand {hand} " f"{SAMPLE_SENDER_ARGS} --seconds {seconds}" ) if hand == "both": command += f" --both-mode {both_mode}" return command def _find_workspace_root() -> Path: # 优先使用显式环境变量，便于从安装目录、源码目录或桌面快捷方式启动。 env_workspace = os.environ.get("XR_RM_WS") if env_workspace: return Path(env_workspace).expanduser().resolve() here = Path(__file__).resolve() for parent in (here.parent, *here.parents): if (parent / "install" / "setup.bash").exists() and (parent / "src").is_dir(): return parent if parent.name == "src" and (parent / "xr_rm_bringup" / "package.xml").exists(): return parent.parent return Path.cwd().resolve() def _source_lines(workspace_root: Path) -> list[str]: # 每个新终端都重新 source ROS 和工作空间，避免依赖启动器自身的 shell 环境。 ros_setup = Path("/opt/ros/humble/setup.bash") install_setup = workspace_root / "install" / "setup.bash" lines = [ "set +e", f"cd {_quote(workspace_root)}", ] if ros_setup.exists(): lines.append(f"source {_quote(ros_setup)}") else: lines.append("echo 'Warning: /opt/ros/humble/setup.bash was not found.'") lines.extend([ ( f"if [ -f {_quote(install_setup)} ]; then " f"source {_quote(install_setup)}; " "else echo 'Warning: workspace install/setup.bash was not found. Run colcon build first.'; fi" ), "echo", ]) return lines def _one_click_mock(arm: str, hand: str) -> str: sender_seconds = SAMPLE_SENDER_STAGGERED_SECONDS if hand == "both" else SAMPLE_SENDER_SECONDS both_mode = "staggered" if hand == "both" else "synchronized" return "\n".join([ f"ros2 launch xr_rm_bringup arm_debug.launch.py arm:={arm} use_mock:=true &", "launch_pid=$!", "sleep 2", _sample_udp_sender_command(hand, sender_seconds, both_mode), "echo", "echo 'Sample sender finished. The launch process is still running in this terminal.'", "echo 'Press Ctrl-C here, or use the cleanup button in the launcher, to stop it.'", "wait \"$launch_pid\"", ]) def _diagnostic_commands() -> list[tuple[str, str]]: return [ ("Open ROS Topic/Node List Monitor", ROS_GRAPH_MONITOR_ACTION), ] def _topic_monitor_item() -> tuple[str, str]: return ("Open Controller Topic Monitor", TOPIC_MONITOR_ACTION) def _finalize_items( items: list[tuple[str, str]], one_click: tuple[str, str] | None = None, ) -> list[tuple[str, str]]: final_items = items + _diagnostic_commands() + [_topic_monitor_item()] if one_click is not None: final_items.append(one_click) return _with_index(final_items) def build_commands_by_mode(mode: str) -> list[tuple[str, str]]: # UI 列表只维护命令模板；真正执行时统一套上工作空间 source 和终端包装。 if mode == "Simulation": items = [ ("Left Arm Mock Launch", "ros2 launch xr_rm_bringup arm_debug.launch.py arm:=left use_mock:=true"), ("Right Arm Mock Launch", "ros2 launch xr_rm_bringup arm_debug.launch.py arm:=right use_mock:=true"), ("Dual Arm Mock Launch", "ros2 launch xr_rm_bringup arm_debug.launch.py arm:=both use_mock:=true"), ( "Sample UDP Sender (Left, 30s)", _sample_udp_sender_command("left"), ), ( "Sample UDP Sender (Right, 30s)", _sample_udp_sender_command("right"), ), ( "Sample UDP Sender (Both Staggered, 60s)", _sample_udp_sender_command("both", SAMPLE_SENDER_STAGGERED_SECONDS, "staggered"), ), ("One-Click Left Mock Demo", _one_click_mock("left", "left")), ("One-Click Right Mock Demo", _one_click_mock("right", "right")), ("One-Click Dual Mock Demo", _one_click_mock("both", "both")), ] one_click = None elif mode == "Left Arm": items = [ ("Ping Left RM75", f"ping -c 4 {DEFAULT_LEFT_IP}"), ( "Left Arm RealMan Launch", "ros2 launch xr_rm_bringup arm_debug.launch.py arm:=left use_mock:=false " f"left_robot_ip:={DEFAULT_LEFT_IP} " "move_to_initial_pose_on_connect:=false", ), ( "Sample UDP Sender (Left, 30s)", _sample_udp_sender_command("left"), ), ] one_click = None elif mode == "Right Arm": items = [ ("Ping Right RM75", f"ping -c 4 {DEFAULT_RIGHT_IP}"), ( "Right Arm RealMan Launch", "ros2 launch xr_rm_bringup arm_debug.launch.py arm:=right use_mock:=false " f"right_robot_ip:={DEFAULT_RIGHT_IP} " "move_to_initial_pose_on_connect:=false", ), ( "Sample UDP Sender (Right, 30s)", _sample_udp_sender_command("right"), ), ] one_click = None elif mode == "Dual Arm": items = [ ("Ping Left RM75", f"ping -c 4 {DEFAULT_LEFT_IP}"), ("Ping Right RM75", f"ping -c 4 {DEFAULT_RIGHT_IP}"), ( "Dual Arm RealMan Launch", "ros2 launch xr_rm_bringup arm_debug.launch.py arm:=both use_mock:=false " f"left_robot_ip:={DEFAULT_LEFT_IP} right_robot_ip:={DEFAULT_RIGHT_IP} " "move_to_initial_pose_on_connect:=false", ), ( "Sample UDP Sender (Both Staggered, 60s)", _sample_udp_sender_command("both", SAMPLE_SENDER_STAGGERED_SECONDS, "staggered"), ), ] one_click = None else: items = [ ("ROS Doctor Report", "ros2 doctor --report"), ("XR-RM Bringup Prefix", "ros2 pkg prefix xr_rm_bringup"), ("XR-RM Input Prefix", "ros2 pkg prefix xr_rm_input"), ("XR-RM Teleop Prefix", "ros2 pkg prefix xr_rm_teleop"), ] one_click = None return _finalize_items(items, one_click) class LauncherApp: """XR-RM 调试启动器主窗口，负责命令列表、终端启动和监控窗口管理。""" def __init__(self, root: tk.Tk) -> None: self.root = root self.workspace_root = _find_workspace_root() self.current_commands: list[tuple[str, str]] = [] self.root.title("XR-RM Teleop Launcher") self.root.geometry("760x640") self.root.minsize(640, 560) self.root.columnconfigure(0, weight=1) self.root.rowconfigure(0, weight=1) main_frame = ttk.Frame(root, padding=(10, 8, 10, 8)) main_frame.grid(row=0, column=0, sticky="nsew") main_frame.columnconfigure(0, weight=1) main_frame.rowconfigure(1, weight=1) header_frame = ttk.Frame(main_frame) header_frame.grid(row=0, column=0, sticky="ew", pady=(0, 8)) header_frame.columnconfigure(0, weight=1) title_lbl = ttk.Label(header_frame, text="XR-RM Teleop Launcher", font=("Helvetica", 12, "bold")) title_lbl.grid(row=0, column=0, sticky="w") mode_frame = ttk.Frame(header_frame) mode_frame.grid(row=0, column=1, sticky="e") mode_frame.columnconfigure(1, weight=1) ttk.Label(mode_frame, text="Mode:", font=("Helvetica", 10)).grid(row=0, column=0, sticky="w", padx=(0, 6)) self.mode_var = tk.StringVar(value=MODES[0]) self.mode_combo = ttk.Combobox( mode_frame, textvariable=self.mode_var, values=MODES, width=16, state="readonly", ) self.mode_combo.grid(row=0, column=1, sticky="ew") self.mode_combo.bind("<>", self.on_mode_changed) list_frame = ttk.Frame(main_frame) list_frame.grid(row=1, column=0, sticky="nsew") list_frame.columnconfigure(0, weight=1) list_frame.rowconfigure(0, weight=1) self.listbox = tk.Listbox(list_frame, selectmode=tk.SINGLE, font=("Courier", 10), height=8, width=1) self.listbox.grid(row=0, column=0, sticky="nsew") list_scroll = ttk.Scrollbar(list_frame, orient=tk.VERTICAL, command=self.listbox.yview) list_scroll.grid(row=0, column=1, sticky="ns") self.listbox.configure(yscrollcommand=list_scroll.set) self.listbox.bind("", self.run_command) self.listbox.bind("<>", self.on_selection_changed) preview_label = ttk.Label(main_frame, text="Command Preview", anchor=tk.W, font=("Helvetica", 10, "bold")) preview_label.grid(row=2, column=0, sticky="w", pady=(8, 2)) self.preview = tk.Text(main_frame, height=3, width=1, wrap=tk.WORD, font=("Courier", 9)) self.preview.grid(row=3, column=0, sticky="ew") self.preview.configure(state=tk.DISABLED) button_frame = ttk.Frame(main_frame) button_frame.grid(row=4, column=0, sticky="ew", pady=(8, 0)) for index in range(3): button_frame.columnconfigure(index, weight=1, uniform="actions") run_btn = tk.Button( button_frame, text="Run Selected", command=self.run_command, bg="#2e7d32", fg="white", font=("Helvetica", 10, "bold"), height=1, ) run_btn.grid(row=0, column=0, sticky="ew", padx=(0, 5)) check_btn = tk.Button( button_frame, text="Check Env", command=self.check_prerequisites, bg="#1976d2", fg="white", font=("Helvetica", 10, "bold"), height=1, ) check_btn.grid(row=0, column=1, sticky="ew", padx=5) kill_btn = tk.Button( button_frame, text="Stop All", command=self.kill_launched_processes, bg="#c62828", fg="white", font=("Helvetica", 10, "bold"), height=1, ) kill_btn.grid(row=0, column=2, sticky="ew", padx=(5, 0)) self.status = tk.Label(root, text="Ready", bd=1, relief=tk.SUNKEN, anchor=tk.W) self.status.grid(row=1, column=0, sticky="ew") self.refresh_command_list() def refresh_command_list(self) -> None: self.current_commands = build_commands_by_mode(self.mode_var.get()) self.listbox.delete(0, tk.END) for name, _ in self.current_commands: self.listbox.insert(tk.END, name) if self.current_commands: self.listbox.select_set(0) self.update_preview() def on_mode_changed(self, event=None) -> None: self.refresh_command_list() self.status.config(text=f"Mode: {self.mode_var.get()}") def on_selection_changed(self, event=None) -> None: self.update_preview() def selected_command(self) -> tuple[str, str] | None: selection = self.listbox.curselection() if not selection: return None return self.current_commands[selection[0]] def update_preview(self) -> None: selected = self.selected_command() text = selected[1] if selected else "" if text == TOPIC_MONITOR_ACTION: topics = ", ".join(topic for _title, topic in TOPIC_MONITORS) text = f"Open one Terminator window split into controller topic panes:\n{topics}" elif text == ROS_GRAPH_MONITOR_ACTION: text = "Open one Terminator window split into 2 horizontal panes:\nros2 topic list | ros2 node list" self.preview.configure(state=tk.NORMAL) self.preview.delete("1.0", tk.END) self.preview.insert(tk.END, text) self.preview.configure(state=tk.DISABLED) def build_terminal_script(self, title: str, command: str) -> str: window_title = TERMINAL_TITLE_PREFIX + title lines = _source_lines(self.workspace_root) lines.extend([ "export XR_RM_LAUNCHER_SESSION=1", f"printf '\\033]0;%s\\007' {_quote(window_title)}", command, "exit_code=$?", "echo", "echo \"Command exited with code ${exit_code}.\"", "read -r -p 'Press Enter to close...'", "exit ${exit_code}", ]) return "\n".join(lines) def terminal_command(self, title: str, script: str) -> list[str] | None: window_title = TERMINAL_TITLE_PREFIX + title if shutil.which("x-terminal-emulator"): return ["x-terminal-emulator", "--no-dbus", "--title", window_title, "-e", "bash", "-ic", script] if shutil.which("gnome-terminal"): return ["gnome-terminal", f"--title={window_title}", "--", "bash", "-ic", script] if shutil.which("konsole"): return ["konsole", "--new-tab", "-p", f"tabtitle={window_title}", "-e", "bash", "-ic", script] if shutil.which("xfce4-terminal"): return ["xfce4-terminal", "--title", window_title, "--command", f"bash -ic {_quote(script)}"] if shutil.which("xterm"): return ["xterm", "-T", window_title, "-e", "bash", "-ic", script] return None def run_command(self, event=None) -> None: selected = self.selected_command() if selected is None: messagebox.showwarning("No Selection", "Please select a command first.") return name, command = selected if command == TOPIC_MONITOR_ACTION: self.launch_topic_monitor() return if command == ROS_GRAPH_MONITOR_ACTION: self.launch_ros_graph_monitor() return terminal_cmd = self.terminal_command(name, self.build_terminal_script(name, command)) if terminal_cmd is None: messagebox.showerror( "Terminal Not Found", "No supported terminal emulator was found. Install gnome-terminal, konsole, xfce4-terminal, or xterm.", ) return try: subprocess.Popen(terminal_cmd, cwd=str(self.workspace_root)) self.status.config(text=f"Launched: {name}") except Exception as exc: messagebox.showerror("Launch Failed", f"Failed to launch command:\n{exc}") self.status.config(text="Launch failed") def _ros_package_available(self, package: str) -> bool: script = "\n".join(_source_lines(self.workspace_root) + [f"ros2 pkg prefix {package} >/dev/null 2>&1"]) try: result = subprocess.run( ["bash", "-lc", script], cwd=str(self.workspace_root), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=8, check=False, ) except Exception: return False return result.returncode == 0 def check_prerequisites(self) -> None: # 环境检查只做轻量探测，避免在 UI 线程里执行耗时的网络或硬件连接动作。 ok: list[str] = [] warnings: list[str] = [] errors: list[str] = [] if Path("/opt/ros/humble/setup.bash").exists(): ok.append("[OK] ROS2 Humble setup found.") else: errors.append("[MISS] /opt/ros/humble/setup.bash not found.") if (self.workspace_root / "src" / "xr_rm_bringup" / "package.xml").exists(): ok.append(f"[OK] Workspace root: {self.workspace_root}") else: warnings.append(f"[WARN] Workspace root guess may be wrong: {self.workspace_root}") if (self.workspace_root / "install" / "setup.bash").exists(): ok.append("[OK] Workspace install/setup.bash found.") else: warnings.append("[WARN] Workspace has not been built yet, or install/setup.bash is missing.") if self.terminal_command("test", "true") is not None: ok.append("[OK] Supported terminal emulator found.") else: errors.append("[MISS] No supported terminal emulator found.") if shutil.which("x-terminal-emulator"): ok.append("[OK] x-terminal-emulator found.") else: errors.append("[MISS] x-terminal-emulator not found. Split monitors require it.") terminal_target = self.x_terminal_target() if terminal_target == "terminator": ok.append("[OK] x-terminal-emulator uses Terminator split-layout support.") elif terminal_target: warnings.append( f"[WARN] x-terminal-emulator points to {terminal_target}. " "The split monitors are designed for Terminator." ) else: warnings.append("[WARN] Could not identify x-terminal-emulator target.") for package in ("xr_rm_bringup", "xr_rm_input", "xr_rm_teleop"): if self._ros_package_available(package): ok.append(f"[OK] ROS package available: {package}") else: warnings.append(f"[WARN] ROS package not discoverable yet: {package}") try: import Robotic_Arm # noqa: F401 ok.append("[OK] Robotic_Arm Python package found for RealMan mode.") except ImportError: warnings.append("[WARN] Robotic_Arm Python package not found. Mock mode is still usable.") sections = [] if errors: sections.append("Errors:\n" + "\n".join(errors)) if warnings: sections.append("Warnings:\n" + "\n".join(warnings)) if ok: sections.append("Passed:\n" + "\n".join(ok)) message = "\n\n".join(sections) if sections else "No checks were run." if errors: self.show_text_dialog("Prerequisite Check", "Errors Found", message) self.status.config(text="Prerequisite check failed") elif warnings: self.show_text_dialog("Prerequisite Check", "Warnings Found", message) self.status.config(text="Prerequisite check passed with warnings") else: self.show_text_dialog("Prerequisite Check", "All Checks Passed", message) self.status.config(text="Prerequisite check passed") def show_text_dialog(self, title: str, heading: str, message: str) -> None: dialog = tk.Toplevel(self.root) dialog.title(title) dialog.geometry("760x520") dialog.minsize(620, 360) dialog.transient(self.root) dialog.grab_set() dialog.columnconfigure(0, weight=1) dialog.rowconfigure(0, weight=1) frame = ttk.Frame(dialog, padding=(12, 10, 12, 12)) frame.grid(row=0, column=0, sticky="nsew") frame.columnconfigure(0, weight=1) frame.rowconfigure(1, weight=1) ttk.Label(frame, text=heading, font=("Helvetica", 12, "bold")).grid(row=0, column=0, sticky="w") text_frame = ttk.Frame(frame) text_frame.grid(row=1, column=0, sticky="nsew", pady=(8, 10)) text_frame.columnconfigure(0, weight=1) text_frame.rowconfigure(0, weight=1) text = tk.Text(text_frame, width=88, height=22, wrap=tk.WORD, font=("Courier", 10)) text.grid(row=0, column=0, sticky="nsew") y_scroll = ttk.Scrollbar(text_frame, orient=tk.VERTICAL, command=text.yview) y_scroll.grid(row=0, column=1, sticky="ns") text.configure(yscrollcommand=y_scroll.set) text.insert("1.0", message) text.configure(state=tk.DISABLED) ok_btn = ttk.Button(frame, text="OK", command=dialog.destroy) ok_btn.grid(row=2, column=0, sticky="e") ok_btn.focus_set() dialog.bind("", lambda _event: dialog.destroy()) dialog.bind("", lambda _event: dialog.destroy()) @staticmethod def x_terminal_target() -> str: terminal = shutil.which("x-terminal-emulator") if terminal is None: return "" try: return Path(terminal).resolve().name except OSError: return Path(terminal).name def topic_pane_command(self, title: str, topic: str) -> str: pane_title = f"{TOPIC_MONITOR_TITLE} - {title}" lines = _source_lines(self.workspace_root) lines.extend([ "export XR_RM_LAUNCHER_SESSION=1", f"printf '\\033]0;%s\\007' {_quote(pane_title)}", "clear", f"echo {_quote('=== ' + title + ' ===')}", f"echo {_quote('ros2 topic echo ' + topic)}", "echo", f"ros2 topic echo {topic}", "echo", "echo 'Topic echo stopped. This pane is left open for inspection.'", "exec bash", ]) return "\n".join(lines) def ros_graph_pane_command(self, title: str, command: str) -> str: pane_title = f"{ROS_GRAPH_MONITOR_TITLE} - {title}" lines = _source_lines(self.workspace_root) lines.extend([ "export XR_RM_LAUNCHER_SESSION=1", f"printf '\\033]0;%s\\007' {_quote(pane_title)}", "trap 'echo; echo Monitor stopped.; exec bash' INT", "while true; do", " clear", f" echo {_quote('=== ' + title + ' ===')}", f" echo {_quote(command + ' (refreshes every 1s)')}", " date '+%H:%M:%S'", " echo", f" {command}", " sleep 1", "done", ]) return "\n".join(lines) def write_topic_monitor_script(self, title: str, topic: str) -> Path: script_file = tempfile.NamedTemporaryFile( mode="w", prefix="xr_rm_topic_monitor_", suffix=".sh", delete=False, encoding="utf-8", ) with script_file: script_file.write(self.topic_pane_command(title, topic)) script_file.write("\n") os.chmod(script_file.name, 0o755) return Path(script_file.name) def write_ros_graph_monitor_script(self, title: str, command: str) -> Path: script_file = tempfile.NamedTemporaryFile( mode="w", prefix="xr_rm_ros_graph_monitor_", suffix=".sh", delete=False, encoding="utf-8", ) with script_file: script_file.write(self.ros_graph_pane_command(title, command)) script_file.write("\n") os.chmod(script_file.name, 0o755) return Path(script_file.name) def topic_monitor_rect(self) -> tuple[int, int, int, int]: # 监控窗口尽量放在启动器所在显示器，双屏现场调试时更容易找回窗口。 monitor_left, monitor_top, monitor_width, monitor_height = self.active_monitor_rect() width = min(monitor_width - 120, max(960, int(monitor_width * 0.75))) height = min(monitor_height - 120, max(560, int(monitor_height * 0.66))) width = max(width, min(monitor_width, 900)) height = max(height, min(monitor_height, 520)) left = monitor_left + max((monitor_width - width) // 2, 0) top = monitor_top + max((monitor_height - height) // 2, 0) return width, height, left, top def active_monitor_rect(self) -> tuple[int, int, int, int]: monitors = self.xrandr_monitor_rects() if not monitors: return (0, 0, max(self.root.winfo_screenwidth(), 1), max(self.root.winfo_screenheight(), 1)) self.root.update_idletasks() root_center = ( self.root.winfo_rootx() + max(self.root.winfo_width(), 1) // 2, self.root.winfo_rooty() + max(self.root.winfo_height(), 1) // 2, ) pointer = (self.root.winfo_pointerx(), self.root.winfo_pointery()) for point_x, point_y in (root_center, pointer): for left, top, width, height, _primary in monitors: if left <= point_x < left + width and top <= point_y < top + height: return (left, top, width, height) for left, top, width, height, primary in monitors: if primary: return (left, top, width, height) left, top, width, height, _primary = monitors[0] return (left, top, width, height) @staticmethod def xrandr_monitor_rects() -> list[tuple[int, int, int, int, bool]]: if not shutil.which("xrandr"): return [] try: output = subprocess.check_output( ["xrandr", "--query"], text=True, stderr=subprocess.DEVNULL, ) except Exception: return [] monitors: list[tuple[int, int, int, int, bool]] = [] pattern = re.compile( r"^\S+ connected(?P primary)? " r"(?P\d+)x(?P\d+)\+(?P-?\d+)\+(?P-?\d+)" ) for line in output.splitlines(): match = pattern.match(line) if not match: continue monitors.append(( int(match.group("x")), int(match.group("y")), int(match.group("w")), int(match.group("h")), bool(match.group("primary")), )) return monitors def window_ids_matching_title(self, title: str) -> list[int]: if not shutil.which("xprop"): return [] try: root_props = subprocess.check_output( ["xprop", "-root", "_NET_CLIENT_LIST"], text=True, stderr=subprocess.DEVNULL, ) except Exception: return [] window_ids = [] for xid_text in re.findall(r"0x[0-9a-fA-F]+", root_props): try: props = subprocess.check_output( ["xprop", "-id", xid_text, "_NET_WM_NAME", "_NET_WM_VISIBLE_NAME", "WM_NAME"], text=True, stderr=subprocess.DEVNULL, ) except Exception: continue if title in props: window_ids.append(int(xid_text, 16)) return window_ids def force_monitor_window_geometry( self, title: str, target_rect: tuple[int, int, int, int], attempts_left: int = 30, ) -> None: if attempts_left <= 0 or not self.root.winfo_exists(): return window_ids = self.window_ids_matching_title(title) if not window_ids: self.root.after( 80, lambda: self.force_monitor_window_geometry(title, target_rect, attempts_left - 1), ) return if any(self.monitor_window_needs_geometry(window_id, target_rect) for window_id in window_ids): width, height, left, top = target_rect self.apply_x11_window_geometry(window_ids, left, top, width, height) def monitor_window_needs_geometry(self, window_id: int, target_rect: tuple[int, int, int, int]) -> bool: props = self.x11_window_props(window_id) if "_NET_WM_STATE_FULLSCREEN" in props or "_NET_WM_STATE_MAXIMIZED_HORZ" in props: return True if "_NET_WM_STATE_MAXIMIZED_VERT" in props: return True geometry = self.x11_window_geometry(window_id) if geometry is None: return True actual_left, actual_top, actual_width, actual_height = geometry target_width, target_height, target_left, target_top = target_rect if abs(actual_width - target_width) > 180 or abs(actual_height - target_height) > 180: return True if abs(actual_left - target_left) > 180 or abs(actual_top - target_top) > 180: return True return False @staticmethod def x11_window_props(window_id: int) -> str: if not shutil.which("xprop"): return "" try: return subprocess.check_output( ["xprop", "-id", hex(window_id), "_NET_WM_STATE"], text=True, stderr=subprocess.DEVNULL, ) except Exception: return "" @staticmethod def x11_window_geometry(window_id: int) -> tuple[int, int, int, int] | None: if not shutil.which("xwininfo"): return None try: output = subprocess.check_output( ["xwininfo", "-id", hex(window_id)], text=True, stderr=subprocess.DEVNULL, ) except Exception: return None values: dict[str, int] = {} patterns = { "left": r"Absolute upper-left X:\s+(-?\d+)", "top": r"Absolute upper-left Y:\s+(-?\d+)", "width": r"Width:\s+(\d+)", "height": r"Height:\s+(\d+)", } for key, pattern in patterns.items(): match = re.search(pattern, output) if match: values[key] = int(match.group(1)) if len(values) != 4: return None return values["left"], values["top"], values["width"], values["height"] def apply_x11_window_geometry(self, window_ids: list[int], left: int, top: int, width: int, height: int) -> None: lib_path = ctypes.util.find_library("X11") if not lib_path: return xlib = ctypes.cdll.LoadLibrary(lib_path) xlib.XOpenDisplay.argtypes = [ctypes.c_char_p] xlib.XOpenDisplay.restype = ctypes.c_void_p xlib.XDefaultRootWindow.argtypes = [ctypes.c_void_p] xlib.XDefaultRootWindow.restype = ctypes.c_ulong xlib.XInternAtom.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_int] xlib.XInternAtom.restype = ctypes.c_ulong xlib.XSendEvent.argtypes = [ctypes.c_void_p, ctypes.c_ulong, ctypes.c_int, ctypes.c_long, ctypes.c_void_p] xlib.XSendEvent.restype = ctypes.c_int xlib.XMoveResizeWindow.argtypes = [ ctypes.c_void_p, ctypes.c_ulong, ctypes.c_int, ctypes.c_int, ctypes.c_uint, ctypes.c_uint, ] xlib.XMoveResizeWindow.restype = ctypes.c_int xlib.XFlush.argtypes = [ctypes.c_void_p] xlib.XCloseDisplay.argtypes = [ctypes.c_void_p] display = xlib.XOpenDisplay(None) if not display: return try: root = xlib.XDefaultRootWindow(display) wm_state = xlib.XInternAtom(display, b"_NET_WM_STATE", False) moveresize = xlib.XInternAtom(display, b"_NET_MOVERESIZE_WINDOW", False) fullscreen = xlib.XInternAtom(display, b"_NET_WM_STATE_FULLSCREEN", False) maximized_horz = xlib.XInternAtom(display, b"_NET_WM_STATE_MAXIMIZED_HORZ", False) maximized_vert = xlib.XInternAtom(display, b"_NET_WM_STATE_MAXIMIZED_VERT", False) for window_id in window_ids: self.send_x11_client_message( xlib, display, root, window_id, wm_state, [0, fullscreen, 0, 1, 0], ) self.send_x11_client_message( xlib, display, root, window_id, wm_state, [0, maximized_horz, maximized_vert, 1, 0], ) xlib.XMoveResizeWindow(display, window_id, left, top, width, height) # EWMH _NET_MOVERESIZE_WINDOW with x/y/w/h flags set. self.send_x11_client_message( xlib, display, root, window_id, moveresize, [(1 << 8) | (1 << 9) | (1 << 10) | (1 << 11), left, top, width, height], ) xlib.XFlush(display) finally: xlib.XCloseDisplay(display) @staticmethod def send_x11_client_message( xlib, display, root: int, window_id: int, message_type: int, data: list[int], ) -> None: class ClientMessageData(ctypes.Union): _fields_ = [ ("b", ctypes.c_char * 20), ("s", ctypes.c_short * 10), ("l", ctypes.c_long * 5), ] class XClientMessageEvent(ctypes.Structure): _fields_ = [ ("type", ctypes.c_int), ("serial", ctypes.c_ulong), ("send_event", ctypes.c_int), ("display", ctypes.c_void_p), ("window", ctypes.c_ulong), ("message_type", ctypes.c_ulong), ("format", ctypes.c_int), ("data", ClientMessageData), ] class XEvent(ctypes.Union): _fields_ = [ ("xclient", XClientMessageEvent), ("pad", ctypes.c_long * 24), ] event = XEvent() event.xclient.type = 33 # ClientMessage event.xclient.serial = 0 event.xclient.send_event = True event.xclient.display = display event.xclient.window = window_id event.xclient.message_type = message_type event.xclient.format = 32 for index, value in enumerate(data[:5]): event.xclient.data.l[index] = int(value) event_mask = (1 << 19) | (1 << 20) # SubstructureNotifyMask | SubstructureRedirectMask xlib.XSendEvent(display, root, False, event_mask, ctypes.byref(event)) def write_topic_monitor_config(self, target_rect: tuple[int, int, int, int] | None = None) -> Path: width, height, left, top = target_rect or self.topic_monitor_rect() left_script, right_script = [ self.write_topic_monitor_script(title, topic) for title, topic in TOPIC_MONITORS ] layout_name = "xr_rm_topic_monitor" config_text = "\n".join([ "[global_config]", " geometry_hinting = False", " window_state = normal", " suppress_multiple_term_dialog = True", "[keybindings]", "[profiles]", " [[default]]", " scrollback_infinite = True", "[layouts]", f" [[{layout_name}]]", " [[[window0]]]", " type = Window", " parent = \"\"", f" title = {_config_quote(TOPIC_MONITOR_TITLE)}", f" position = {_config_quote(f'{left}:{top}')}", f" size = {width}, {height}", " maximised = False", " fullscreen = False", " [[[split0]]]", " type = HPaned", " parent = window0", " order = 0", " ratio = 0.5", " [[[terminal_left]]]", " type = Terminal", " parent = split0", " order = 0", f" command = {_config_quote('bash ' + str(left_script))}", " [[[terminal_right]]]", " type = Terminal", " parent = split0", " order = 1", f" command = {_config_quote('bash ' + str(right_script))}", "[plugins]", "", ]) config_file = tempfile.NamedTemporaryFile( mode="w", prefix="xr_rm_topic_monitor_", suffix=".conf", delete=False, encoding="utf-8", ) with config_file: config_file.write(config_text) return Path(config_file.name) def write_ros_graph_monitor_config(self, target_rect: tuple[int, int, int, int] | None = None) -> Path: width, height, left, top = target_rect or self.topic_monitor_rect() topic_script, node_script = [ self.write_ros_graph_monitor_script(title, command) for title, command in ROS_GRAPH_MONITORS ] layout_name = "xr_rm_ros_graph_monitor" config_text = "\n".join([ "[global_config]", " geometry_hinting = False", " window_state = normal", " suppress_multiple_term_dialog = True", "[keybindings]", "[profiles]", " [[default]]", " scrollback_infinite = True", "[layouts]", f" [[{layout_name}]]", " [[[window0]]]", " type = Window", " parent = \"\"", f" title = {_config_quote(ROS_GRAPH_MONITOR_TITLE)}", f" position = {_config_quote(f'{left}:{top}')}", f" size = {width}, {height}", " maximised = False", " fullscreen = False", " [[[split0]]]", " type = HPaned", " parent = window0", " order = 0", " ratio = 0.5", " [[[terminal_topic_list]]]", " type = Terminal", " parent = split0", " order = 0", f" command = {_config_quote('bash ' + str(topic_script))}", " [[[terminal_node_list]]]", " type = Terminal", " parent = split0", " order = 1", f" command = {_config_quote('bash ' + str(node_script))}", "[plugins]", "", ]) config_file = tempfile.NamedTemporaryFile( mode="w", prefix="xr_rm_ros_graph_monitor_", suffix=".conf", delete=False, encoding="utf-8", ) with config_file: config_file.write(config_text) return Path(config_file.name) def launch_topic_monitor(self) -> None: terminal = shutil.which("x-terminal-emulator") if terminal is None: messagebox.showerror( "Terminal Not Found", "x-terminal-emulator was not found. Install a terminal emulator to use the topic monitor.", ) return if self.x_terminal_target() != "terminator": messagebox.showwarning( "Topic Monitor", "The controller topic monitor uses Terminator's split-layout support. " "Please set x-terminal-emulator to Terminator.", ) return target_rect = self.topic_monitor_rect() config_path = self.write_topic_monitor_config(target_rect) command = [ terminal, "--no-dbus", "--title", TOPIC_MONITOR_TITLE, "--config", str(config_path), "--layout", "xr_rm_topic_monitor", ] try: subprocess.Popen( command, cwd=str(self.workspace_root), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) self.root.after(60, lambda: self.force_monitor_window_geometry(TOPIC_MONITOR_TITLE, target_rect)) self.status.config(text="Opened Terminator controller topic monitor.") except Exception as exc: messagebox.showerror("Topic Monitor Failed", f"Failed to launch topic monitor:\n{exc}") self.status.config(text="Topic monitor launch failed") def launch_ros_graph_monitor(self) -> None: terminal = shutil.which("x-terminal-emulator") if terminal is None: messagebox.showerror( "Terminal Not Found", "x-terminal-emulator was not found. Install a terminal emulator to use the ROS graph monitor.", ) return if self.x_terminal_target() != "terminator": messagebox.showwarning( "ROS Graph Monitor", "The ROS graph monitor uses Terminator's split-layout support. " "Please set x-terminal-emulator to Terminator.", ) return target_rect = self.topic_monitor_rect() config_path = self.write_ros_graph_monitor_config(target_rect) command = [ terminal, "--no-dbus", "--title", ROS_GRAPH_MONITOR_TITLE, "--config", str(config_path), "--layout", "xr_rm_ros_graph_monitor", ] try: subprocess.Popen( command, cwd=str(self.workspace_root), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) self.root.after(60, lambda: self.force_monitor_window_geometry(ROS_GRAPH_MONITOR_TITLE, target_rect)) self.status.config(text="Opened Terminator ROS graph monitor.") except Exception as exc: messagebox.showerror("ROS Graph Monitor Failed", f"Failed to launch ROS graph monitor:\n{exc}") self.status.config(text="ROS graph monitor launch failed") def close_related_terminal_windows(self) -> int: title_patterns = (TERMINAL_TITLE_PREFIX, TOPIC_MONITOR_TITLE, ROS_GRAPH_MONITOR_TITLE) closed = 0 if shutil.which("wmctrl"): try: output = subprocess.check_output(["wmctrl", "-l"], text=True) except Exception: output = "" for line in output.splitlines(): parts = line.split(None, 3) if len(parts) < 4: continue window_id, title = parts[0], parts[3] if any(pattern in title for pattern in title_patterns): subprocess.run(["wmctrl", "-ic", window_id], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) closed += 1 if closed == 0 and shutil.which("xdotool"): for pattern in title_patterns: try: output = subprocess.check_output(["xdotool", "search", "--name", pattern], text=True) except subprocess.CalledProcessError: continue except Exception: continue for window_id in output.splitlines(): subprocess.run(["xdotool", "windowclose", window_id], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) closed += 1 return closed def kill_launched_processes(self) -> None: confirm = messagebox.askyesno( "Confirm Stop", "Stop XR-RM launcher terminals, topic monitors, and ROS nodes started from this workspace?", ) if not confirm: return # 只清理由启动器常用命令创建的 ROS/监控进程，并保护当前 UI 进程。 patterns = [ "ros2 launch xr_rm_bringup", "ros2 run xr_rm_input", "ros2 run xr_rm_teleop", "XR_RM_LAUNCHER_SESSION=1", TERMINAL_TITLE_PREFIX, TOPIC_MONITOR_TITLE, ROS_GRAPH_MONITOR_TITLE, "xr_rm_topic_monitor_", "xr_rm_ros_graph_monitor_", "udp_controller_receiver", "sample_udp_sender", "single_arm_velocity_teleop", "ros2 topic echo /xr/left_controller", "ros2 topic echo /xr/right_controller", "ros2 topic list", "ros2 node list", ] protected = {os.getpid(), os.getppid()} killed: set[int] = set() for pattern in patterns: try: output = subprocess.check_output(["pgrep", "-f", pattern], text=True) except subprocess.CalledProcessError: continue except Exception as exc: print(f"Failed to search process pattern {pattern!r}: {exc}") continue for pid_text in output.splitlines(): try: pid = int(pid_text) except ValueError: continue if pid in protected or pid in killed: continue try: os.kill(pid, signal.SIGTERM) killed.add(pid) except OSError: pass time.sleep(0.3) closed_windows = self.close_related_terminal_windows() self.status.config( text=f"Stop requested: {len(killed)} matching processes, {closed_windows} terminal windows." ) messagebox.showinfo( "Stop Complete", "Stop requested for:\n" f"- {len(killed)} matching processes, including launcher terminal wrappers\n" f"- {closed_windows} related terminal windows", ) def main() -> None: root = tk.Tk() LauncherApp(root) root.mainloop() if __name__ == "__main__": main()