RoboWaiter/robowaiter/scene/ui/pyqt5.py

import importlib
import os

from PyQt5.QtSvg import QGraphicsSvgItem, QSvgWidget
from PyQt5.QtWidgets import QApplication, QMainWindow, QListWidgetItem, QGraphicsView, QGraphicsScene, \
    QGraphicsPixmapItem, QGraphicsProxyWidget
from PyQt5.QtCore import QTimer, QPoint, QRectF
from PyQt5 import QtCore
import sys

from robowaiter.scene.ui.window import Ui_MainWindow
from robowaiter.utils.basic import get_root_path
from PyQt5.QtCore import QThread
import queue
import numpy as np
from PyQt5.QtGui import QImage, QPixmap, QDrag, QPainter
from PyQt5.QtCore import Qt
from PyQt5.QtCore import QThread, pyqtSignal
from robowaiter.scene.ui.scene_ui import SceneUI
from  robowaiter.scene.ui import scene_ui

root_path = get_root_path()

class TaskThread(QThread):
    stop_signal = pyqtSignal()

    def __init__(self, task, scene_cls,robot_cls,scene_queue,ui_queue, *args, **kwargs):
        super(TaskThread, self).__init__(*args, **kwargs)
        self.task = task
        self.scene_cls = scene_cls
        self.robot_cls = robot_cls
        self.scene_queue = scene_queue
        self.ui_queue = ui_queue
        self.stoped = True

    def run(self):
        self.scene = self.task(self.scene_cls,self.robot_cls,self.scene_queue,self.ui_queue)
        self.scene._run()
        # scene._run()

        while not self.isInterruptionRequested():
            self.scene.step()

    # def stop(self):
    #     del self.scene

def run_scene(scene_cls,robot_cls,scene_queue,ui_queue):
    scene = scene_cls(robot_cls(),scene_queue,ui_queue)

    scene.reset()
    # scene.run()
    return scene
    # try:
    #     # 机器人系统的主循环
    #
    # except Exception as e:
    #     # 打印异常信息到命令行
    #     print("Robot system error:", str(e))

example_list = ("AEM","VLN","VLM",'GQA',"OT","AT","reset")
more_example_list = ("VLM_AC","CafeDaily")
dic_more2zh={
    "VLM_AC":"开空调并调节空调温度",
    "CafeDaily":"咖啡厅的一天"
}
more_example_list_zh = [value for value in dic_more2zh.values()]

class GraphicsView(QGraphicsView):
    def __init__(self, parent=None):
        super(GraphicsView, self).__init__(parent)
        self.setRenderHint(QPainter.Antialiasing)
        self.setOptimizationFlag(QGraphicsView.DontAdjustForAntialiasing, True)
        self.setViewportUpdateMode(QGraphicsView.FullViewportUpdate)
        self.setDragMode(QGraphicsView.ScrollHandDrag)
        self.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff)
        self.setVerticalScrollBarPolicy(QtCore.Qt.ScrollBarAlwaysOff)

    def wheelEvent(self, event):
        factor = 1.15
        if event.angleDelta().y() > 0:
            self.scale(factor, factor)
        else:
            self.scale(1.0 / factor, 1.0 / factor)

class UI(QMainWindow, Ui_MainWindow):
    scene = None
    history_dict = {
        "System": ""
    }

    def __init__(self,robot_cls):
        app = QApplication(sys.argv)
        MainWindow = QMainWindow()
        super(UI, self).__init__(MainWindow)
        # 创建线程安全的队列
        self.scene_queue = queue.Queue()
        self.ui_queue = queue.Queue()

        self.scene_cls = SceneUI
        self.robot_cls = robot_cls

        self.setupUi(MainWindow)  # 初始化UI

        # 绑定说话按钮
        self.btn_say.clicked.connect(self.btn_say_on_click)
        self.verticalLayout_2.removeWidget(self.img_view_bt)
        self.img_view_bt = GraphicsView(self.centralwidget)
        self.img_view_bt.setObjectName("img_view_bt")
        self.verticalLayout_2.addWidget(self.img_view_bt)
        self.img_view_bt.setDragMode(QGraphicsView.ScrollHandDrag)


        # 下拉菜单绑定函数
        # 下拉菜单.seleted.connect(self.你写的函数)

        # # BT树的显示
        # self.setScaledContents(True)  # 图片将根据label的大小自动缩放
        # self.label_moved = False  # 用于判断是否移动了标签
        # self.drag_start_position = QPoint()  # 记录鼠标按下时的位置

        self.cb_task.setCurrentIndex(-1)
        self.cb_task.setPlaceholderText("请选择更多任务")

        # 多个添加条目
        self.cb_task.addItems(more_example_list_zh)
        # 当下拉索引发生改变时发射信号触发绑定的事件
        self.cb_task.currentIndexChanged.connect(self.cb_selectionchange)

        # 绑定任务演示按钮
        for example in example_list:
            btn = getattr(self,f"btn_{example}")
            btn.clicked.connect(self.create_example_click(example))

        # 设置一个定时器，每隔100ms检查一次队列
        timer = QTimer()
        timer.setInterval(100)  # 设置检查队列的间隔时间
        timer.timeout.connect(self.handle_queue_messages)  # 当定时器超时时，处理队列中的消息
        timer.start()  # 启动定时器

        MainWindow.show()

        self.thread = TaskThread(run_scene,self.scene_cls,self.robot_cls,self.scene_queue,self.ui_queue)
        self.thread.start()
        self.list_customer.itemClicked.connect(self.show_customer_history)

        sys.exit(app.exec_())

    def cb_selectionchange(self, i):
        print(i)
        self.create_example_click(more_example_list[i])()

    def get_semantic_info(self, semantic_info_str):
        # self.textEdit_5.clear()
        # self.textEdit_5.append(semantic_info_str)
        self.textBrowser_5.clear()
        self.textBrowser_5.append(semantic_info_str)

    def new_history(self,customer_name,chat):
        role = chat["role"]
        content = chat["content"].strip()
        function_call = chat.get("function_call",None)
        if function_call:
            return
        if role == "function":
            new_chat = f'Robot:\n 工具调用-{chat["name"]}\n'
        elif role == "user":
            new_chat = f'{customer_name}:\n {content}\n'
        else:
            new_chat = f'Robot:\n {content}\n'

        self.history_dict[customer_name] += new_chat + "\n"
        self.history_dict["Global"] += new_chat + "\n"
        items = self.list_customer.findItems(customer_name, Qt.MatchExactly)
        if items:
            index = self.list_customer.indexFromItem(items[0])
            self.list_customer.setCurrentIndex(index)
        self.edit_local_history.clear()
        self.edit_local_history.append(self.history_dict["Global"])

    def reset(self):
        self.history_dict = {
            "Global":"",
            "System": "",
        }
        self.edit_local_history.clear()
        self.list_customer.clear()
        item = QListWidgetItem("Global")
        self.list_customer.addItem(item)
        item = QListWidgetItem("System")
        self.list_customer.addItem(item)




    def create_example_click(self,name):
        def btn_example_on_click():
            self.reset()

            self.thread.requestInterruption()
            self.thread.wait()  # 等待线程安全退出

            self.scene_queue = queue.Queue()
            self.ui_queue = queue.Queue()
            self.thread = TaskThread(run_scene, self.scene_cls, self.robot_cls,self.scene_queue,self.ui_queue)
            self.thread.start()

            self.scene_func((f"run_{name}",))
        return btn_example_on_click

    def add_walker(self,name):
        self.history_dict[name] = ""
        item = QListWidgetItem(name)
        self.list_customer.addItem(item)


    def show_customer_history(self, item):
        # 此处为示例，实际上你可能需要从数据库或其他地方获取聊天记录
        name = item.text()
        self.edit_local_history.clear()
        self.edit_local_history.append(self.history_dict[name])

    def btn_say_on_click(self):
        question = self.edit_say.text()
        if question[-1] == ")":
            print(f"设置目标:{question}")
            self.scene_func(("new_set_goal",question))
            # self.scene.new_set_goal(question)
        else:
            self.scene_func(("customer_say","System",question))
            # self.scene.customer_say("System", question)

    def scene_func(self,args):
        self.scene_queue.put(args)

    def handle_queue_messages(self):
        while not self.ui_queue.empty():
            message = self.ui_queue.get()
            function_name = message[0]
            function = getattr(self, function_name, None)

            args = []
            if len(message)>1:
                args = message[1:]

            result = function(*args)

    def setupUi(self, MainWindow):
        Ui_MainWindow.setupUi(self,MainWindow)

        # dir_path = os.path.dirname(__file__)

        # 加载并显示图片
        # img_path = os.path.join(root_path, "robowaiter/utils/draw_bt/test.png")
        # self.draw_from_file("img_label_bt",img_path)
        # self.label.setAlignment(Qt.AlignCenter)  # 图片居中显示

    def draw_from_file(self,control_name,file_name):

        control = getattr(self,control_name,None)
        # 加载并显示图片
        pixmap = QPixmap(file_name) # 替换为你的图片路径

        if control_name == "img_view_bt":
            # self.img_view_bt = ImageView()
            scene = QGraphicsScene(self.img_view_bt)
            self.img_view_bt.setScene(scene)
            pixmap_item = QGraphicsPixmapItem(pixmap)
            scene.addItem(pixmap_item)


            # widget = QSvgWidget(file_name)
            # # widget.resize(10000, 1600)
            # svg_width = widget.width()
            # widget.resize(svg_width*2, 1600)
            # proxy = QGraphicsProxyWidget()
            # proxy.setWidget(widget)
            # scene.addItem(+proxy)


        elif control_name == "img_label_obj":
            return
            # self.label.setPixmap(pixmap)
        else:
            control.setPixmap(self.scale_pixmap_to_label(pixmap, control))

    def draw_img(self,control_name,img):
        control = getattr(self,control_name,None)
        # # 加载并显示图片
        # print(img)
        # img = self.ndarray_to_qimage(img)
        # print(img)

        # image = Image.open(io.BytesIO(img))
        # print(image)
        pixmap = QPixmap.fromImage(QImage.fromData(img))

        # self.label.setPixmap(pixmap)
        control.setPixmap(self.scale_pixmap_to_label(pixmap, control))
        # control.setPixmap(pixmap)


    def draw_canvas(self,control_name,canvas):
        control = getattr(self,control_name,None)
        # 加载并显示图片

        pixmap = QApplication.grabWidget(canvas)

        # self.label.setPixmap(pixmap)
        control.setPixmap(self.scale_pixmap_to_label(pixmap, control))


    def ndarray_to_qimage(self,array: np.ndarray) -> QImage:
        # 假设你的ndarray是uint8类型的，且是RGB格式（三个通道）
        height, width, channels = array.shape
        if channels ==3:
            bytes_per_line = channels * width
            image = QImage(array.data, width, height, bytes_per_line, QImage.Format_RGB888)
        else:
            # 对于灰度图像
            image = QImage(array.data, width, height, QImage.Format_Grayscale8)

        return image

    def scale_pixmap_to_label(self, pixmap, label):
        # 获取QLabel的大小
        label_width = label.width()
        label_height = label.height()

        # 保持图片的长宽比，同时确保图片的一个维度适应QLabel的大小
        scaled_pixmap = pixmap.scaledToWidth(label_width, Qt.SmoothTransformation)
        if scaled_pixmap.height() > label_height:
            scaled_pixmap = pixmap.scaledToHeight(label_height, Qt.SmoothTransformation)

        return scaled_pixmap

    def set_scene(self, scene):
        self.scene = scene


if __name__ == "__main__":
    # app = QApplication(sys.argv)
    # MainWindow = QMainWindow()
    ui = UI()
    # ui.setupUi(MainWindow)
    # MainWindow.show()
    # sys.exit(app.exec_())