diff --git a/xr_rm_bringup/tools/launcher_ui.py b/xr_rm_bringup/tools/launcher_ui.py index 2b878e4..25bf202 100755 --- a/xr_rm_bringup/tools/launcher_ui.py +++ b/xr_rm_bringup/tools/launcher_ui.py @@ -1,7 +1,10 @@ #!/usr/bin/env python3 from __future__ import annotations +import ctypes +import ctypes.util import os +import re import shlex import shutil import signal @@ -224,13 +227,17 @@ class LauncherApp: 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(2, weight=1) + main_frame.rowconfigure(1, weight=1) - title_lbl = ttk.Label(main_frame, text="XR-RM Teleop Launcher", font=("Helvetica", 12, "bold")) + 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(main_frame) - mode_frame.grid(row=1, column=0, sticky="ew", pady=(8, 6)) + 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]) @@ -238,14 +245,14 @@ class LauncherApp: mode_frame, textvariable=self.mode_var, values=MODES, - width=24, + 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=2, column=0, sticky="nsew") + 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) @@ -257,13 +264,13 @@ class LauncherApp: 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=3, column=0, sticky="w", pady=(8, 2)) + 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=4, column=0, sticky="ew") + 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=5, column=0, sticky="ew", pady=(8, 0)) + 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") @@ -599,14 +606,169 @@ class LauncherApp: def topic_monitor_rect(self) -> tuple[int, int, int, int]: screen_width = max(self.root.winfo_screenwidth(), 1) screen_height = max(self.root.winfo_screenheight(), 1) - width = min(screen_width - 80, max(900, int(screen_width * 0.75))) - height = min(screen_height - 80, max(560, int(screen_height * (2.0 / 3.0)))) - width = max(width, min(screen_width, 640)) - height = max(height, min(screen_height, 420)) + width = min(screen_width - 160, max(960, int(screen_width * 0.75))) + height = min(screen_height - 160, max(560, int(screen_height * 0.66))) + width = max(width, min(screen_width, 900)) + height = max(height, min(screen_height, 520)) left = max((screen_width - width) // 2, 0) top = max((screen_height - height) // 2, 0) return width, height, left, top + 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, attempts_left: int = 40) -> None: + if attempts_left <= 0 or not self.root.winfo_exists(): + return + + window_ids = self.window_ids_matching_title(title) + if window_ids: + width, height, left, top = self.topic_monitor_rect() + self.apply_x11_window_geometry(window_ids, left, top, width, height) + + self.root.after(250, lambda: self.force_monitor_window_geometry(title, attempts_left - 1)) + + 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) -> Path: width, height, left, top = self.topic_monitor_rect() left_script, right_script = [ @@ -713,10 +875,6 @@ class LauncherApp: config_file.write(config_text) return Path(config_file.name) - def topic_monitor_geometry(self) -> str: - width, height, left, top = self.topic_monitor_rect() - return f"{width}x{height}+{left}+{top}" - def launch_topic_monitor(self) -> None: terminal = shutil.which("x-terminal-emulator") if terminal is None: @@ -746,7 +904,13 @@ class LauncherApp: "xr_rm_topic_monitor", ] try: - subprocess.Popen(command, cwd=str(self.workspace_root)) + subprocess.Popen( + command, + cwd=str(self.workspace_root), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + self.root.after(150, lambda: self.force_monitor_window_geometry(TOPIC_MONITOR_TITLE)) 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}") @@ -781,7 +945,13 @@ class LauncherApp: "xr_rm_ros_graph_monitor", ] try: - subprocess.Popen(command, cwd=str(self.workspace_root)) + subprocess.Popen( + command, + cwd=str(self.workspace_root), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + self.root.after(150, lambda: self.force_monitor_window_geometry(ROS_GRAPH_MONITOR_TITLE)) 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}")