| import sys
|
| import os
|
| import json
|
| from PySide6.QtCore import QObject, Slot
|
|
|
|
|
| PROJECT_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
|
| if PROJECT_SRC not in sys.path:
|
| sys.path.insert(0, PROJECT_SRC)
|
|
|
| from frontend.threads.llm_thread import LLMThread |
| from frontend.threads.simulation_thread import SimulationThread |
| from llm_panda.conversation import ConversationManager |
| from llm_panda.plan_verifier import PlanVerifier |
|
|
| class UIController(QObject):
|
| """
|
| Main controller coordinates widgets and background threads (LLM Planning & PyBullet Simulation).
|
| """
|
| def __init__(self, main_window):
|
| super().__init__()
|
| self.main_window = main_window
|
|
|
|
|
| self.viewport_widget = None
|
| self.chat_widget = None
|
| self.log_widget = None
|
| self.status_widget = None
|
| self.score_modal = None
|
|
|
|
|
| self.sim_thread = None |
| self.llm_thread = None |
| self.last_scene_state = None |
| self.conversation_manager = ConversationManager() |
|
|
| def start(self):
|
| """Initializes and starts the simulation background thread."""
|
| self.viewport_widget = self.main_window.viewport_widget
|
| self.chat_widget = self.main_window.chat_widget
|
| self.log_widget = self.main_window.log_widget
|
| self.status_widget = self.main_window.status_widget
|
| self.score_modal = self.main_window.score_modal
|
|
|
|
|
| self.viewport_widget.camera_rotated.connect(self.handle_camera_rotate)
|
| self.viewport_widget.camera_panned.connect(self.handle_camera_pan)
|
| self.viewport_widget.camera_zoomed.connect(self.handle_camera_zoom)
|
| self.viewport_widget.viewport_clicked.connect(self.handle_viewport_clicked)
|
|
|
|
|
| self.chat_widget.command_submitted.connect(self.handle_command_submit)
|
| self.chat_widget.preset_clicked.connect(self.handle_preset_click)
|
| self.chat_widget.reset_clicked.connect(self.handle_reset_click)
|
| self.chat_widget.gui_toggled.connect(self.handle_gui_toggle)
|
|
|
|
|
| self.sim_thread = SimulationThread()
|
|
|
|
|
| self.sim_thread.frame_ready.connect(self.viewport_widget.update_frame)
|
| self.sim_thread.log_message.connect(self.log_widget.append_log)
|
| self.sim_thread.fps_updated.connect(self.status_widget.update_fps)
|
| self.sim_thread.step_completed.connect(self.handle_step_completed)
|
| self.sim_thread.finished.connect(self.handle_execution_finished)
|
| self.sim_thread.init_completed.connect(self.handle_sim_init_completed)
|
|
|
| self.sim_thread.start()
|
|
|
| def handle_sim_init_completed(self):
|
| self.status_widget.update_connection_badge(True)
|
|
|
| def handle_camera_rotate(self, dx, dy):
|
| if self.sim_thread:
|
| self.sim_thread.queue_command("camera_rotate", (dx, dy))
|
|
|
| def handle_camera_pan(self, dx, dy):
|
| if self.sim_thread:
|
| self.sim_thread.queue_command("camera_pan", (dx, dy))
|
|
|
| def handle_camera_zoom(self, delta):
|
| if self.sim_thread:
|
| self.sim_thread.queue_command("camera_zoom", delta)
|
|
|
| def handle_viewport_clicked(self, x, y):
|
| if self.sim_thread:
|
| w = self.viewport_widget.width()
|
| h = self.viewport_widget.height()
|
| self.sim_thread.queue_command("viewport_click", (x, y, w, h))
|
|
|
| def handle_preset_click(self, prompt_text):
|
| self.chat_widget.set_prompt_text(prompt_text)
|
|
|
| def handle_reset_click(self): |
| if self.sim_thread: |
| self.conversation_manager.clear() |
| self.log_widget.clear_logs() |
| self.chat_widget.clear_plan()
|
| self.status_widget.reset_steps()
|
| self.sim_thread.queue_command("reset")
|
|
|
| def handle_gui_toggle(self, checked):
|
| if self.sim_thread:
|
| self.sim_thread.queue_command("gui_toggle", checked)
|
|
|
| def handle_command_submit(self, prompt, use_mock):
|
| if not prompt:
|
| return
|
|
|
| self.log_widget.clear_logs()
|
| self.chat_widget.clear_plan()
|
| self.status_widget.reset_steps()
|
|
|
| self.log_widget.append_log(f"Gửi yêu cầu: \"{prompt}\"", "info")
|
| self.chat_widget.lock_inputs()
|
|
|
|
|
| scene_state = {}
|
| with self.sim_thread.pybullet_lock:
|
|
|
| def reachability_fn(oid):
|
| cls = self.sim_thread.sim.scene.get_class(oid)
|
| try:
|
| grasp_pos, grasp_ori = self.sim_thread.sim.grasp_planner.compute_grasp_pose(oid, cls)
|
| return self.sim_thread.sim.controller.is_reachable(grasp_pos, grasp_ori)
|
| except Exception:
|
| return False
|
|
|
| carried_name = self.sim_thread.sim.scene.get_name(self.sim_thread.sim._carried) if self.sim_thread.sim._carried is not None else None
|
| scene_state = self.sim_thread.sim.scene.get_scene_state(reachability_fn=reachability_fn, carried_object=carried_name)
|
|
|
| self.last_scene_state = scene_state
|
|
|
|
|
| self.status_widget.set_step_active("llm")
|
| self.llm_thread = LLMThread( |
| prompt, |
| scene_state, |
| use_mock, |
| conversation_manager=self.conversation_manager, |
| ) |
| self.llm_thread.plan_ready.connect(self.handle_plan_ready)
|
| self.llm_thread.log_message.connect(self.log_widget.append_log)
|
| self.llm_thread.failed.connect(self.handle_llm_failed)
|
| self.llm_thread.start()
|
|
|
| def handle_llm_failed(self, error_msg):
|
| self.status_widget.set_step_error("llm")
|
| self.chat_widget.unlock_inputs()
|
|
|
| def handle_plan_ready(self, plan_steps): |
| self.status_widget.set_step_completed("llm") |
| |
| if not plan_steps: |
| self.log_widget.append_log("⚠️ [1] Kế hoạch rỗng.", "warn")
|
| self.status_widget.set_step_active("score")
|
|
|
| self.sim_thread.queue_command("execute_plan", []) |
| return |
|
|
| first_step = plan_steps[0] |
| if first_step.get("skill") in ("clarify", "reject"): |
| message = first_step.get("args", {}).get( |
| "message", |
| "Yêu cầu cần được làm rõ.", |
| ) |
| self.chat_widget.add_message(message, is_user=False) |
| self.log_widget.append_log(f"Robot: {message}", "info") |
| self.chat_widget.clear_plan() |
| self.chat_widget.unlock_inputs() |
| return |
|
|
| self.chat_widget.display_plan(plan_steps) |
|
|
|
|
| self.status_widget.set_step_active("val")
|
| self.log_widget.append_log("⏳ [2/5] Đang kiểm tra kế hoạch (Validator)...", "info")
|
|
|
| try:
|
| warnings = PlanVerifier.check(plan_steps, self.last_scene_state)
|
| if not warnings:
|
| self.log_widget.append_log("✅ [2] Kế hoạch hợp lệ.", "success")
|
| self.status_widget.set_step_completed("val")
|
| else:
|
| self.log_widget.append_log(f"⚠️ [2] Validator phát hiện {len(warnings)} cảnh báo.", "warn")
|
| for w in warnings:
|
| self.log_widget.append_log(f" - {w}", "warn")
|
| self.status_widget.set_step_active("val")
|
| except Exception as e:
|
| self.log_widget.append_log(f"❌ [2] Lỗi kiểm tra kế hoạch: {e}", "error")
|
| self.status_widget.set_step_error("val")
|
| self.chat_widget.unlock_inputs()
|
| return
|
|
|
|
|
| self.status_widget.set_step_active("exec")
|
| self.sim_thread.queue_command("execute_plan", plan_steps)
|
|
|
| def handle_step_completed(self, step_idx):
|
| self.chat_widget.mark_step_completed(step_idx)
|
|
|
| def handle_execution_finished(self, success, score_md, score_val):
|
| if success:
|
| self.status_widget.set_step_completed("exec")
|
| else:
|
| self.status_widget.set_step_error("exec")
|
|
|
| self.status_widget.set_step_completed("score")
|
|
|
|
|
| self.score_modal.show_score(score_md)
|
|
|
|
|
| self.chat_widget.unlock_inputs()
|
|
|
| def shutdown(self):
|
| """Safely stops background threads."""
|
| if self.llm_thread and self.llm_thread.isRunning():
|
| self.llm_thread.terminate()
|
| self.llm_thread.wait()
|
| if self.sim_thread:
|
| self.sim_thread.shutdown()
|
|
|
