optimize launcher_ui.py
This commit is contained in:
@ -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("<<ComboboxSelected>>", 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("<<ListboxSelect>>", 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}")
|
||||
|
||||
Reference in New Issue
Block a user