#!/usr/bin/env python # -*- coding: utf-8 -*- ''' # @Time : 2025/11/10 11:29 # @Author : reenrr # @Description : 派单任务的详情按钮点击之后弹出, 显示派单任务的详情 ''' from PySide6.QtWidgets import ( QApplication, QDialog, QVBoxLayout, QHBoxLayout, QGridLayout, QLabel, QWidget, QPushButton, ) from PySide6.QtGui import QPixmap, QFont, QPainter, QIcon from PySide6.QtCore import Qt, QEvent, QSize, QTimer, Slot from PySide6.QtNetwork import QTcpSocket, QAbstractSocket import sys import json from datetime import datetime # ----------- # 参数配置 # ----------- tcp_server_host = "127.0.0.1" tcp_server_port = 8888 MAX_RECONNECT = 3 # 最大重连次数 RECONNECT_INTERVAL = 2000 # 重连间隔（毫秒） class DispatchDetailsDialog(QDialog): """ 派单任务的界面 """ def __init__(self, parent=None): super().__init__(parent) self.setAttribute(Qt.WA_TranslucentBackground) self.is_running = False # 系统运行状态标记 self.latest_json_data = {"erp_id":None,"artifact_id":None} # 缓存服务端发送的最新JSON数据 self.statusWidgets = [] # 存储api_field + 对应的value标签 # --------------- # TCP客户端核心配置 # --------------- self.tcp_socket = QTcpSocket(self) # TCP socket实例 self.tcp_server_host = tcp_server_host self.tcp_server_port = tcp_server_port self.is_tcp_connected = False # TCP连接状态标记 self.has_connected_once = False # 连接至服务器至少一次标记（区别首次连接和断开后重连） self.reconnect_count = 0 # 重连次数计数器 # 重连定时器，每隔RECONNECT_INTERVAL毫秒重连一次 self.reconnect_timer = QTimer(self) self.reconnect_timer.setInterval(RECONNECT_INTERVAL) self.reconnect_timer.timeout.connect(self._reconnect_to_server) # 绑定重连函数 # 绑定TCP信号与槽（事件驱动） self._bind_tcp_signals() # 初始化存储需要修改的控件 self.id_value_label = None # 对应管片ID值标签 self.rows = [] # 所有行的单元格列表（包含label、value） self._init_ui() # ---------------- # 客户端自动后自动连接服务端 # ---------------- print(f"客户端启动，自动连接服务端{self.tcp_server_host}:{self.tcp_server_port}...") self._connect_to_server() # --------------------- # TCP连接相关函数 # --------------------- def _connect_to_server(self): """主动发起连接（仅在未连接状态下有效""" if not self.is_tcp_connected: self.tcp_socket.abort() # 终止现有连接 self.tcp_socket.connectToHost(self.tcp_server_host, self.tcp_server_port) def _reconnect_to_server(self): """重连执行函数：仅在未连接且未达最大次数时触发""" if not self.is_tcp_connected and self.reconnect_count < MAX_RECONNECT: self.reconnect_count += 1 print(f"第{self.reconnect_count}次重连（共{MAX_RECONNECT}次尝试）...") self._connect_to_server() elif self.reconnect_count >= MAX_RECONNECT: self.reconnect_timer.stop() # 停止重连定时器 print(f"已达最大重连次数（{MAX_RECONNECT}次），停止重连，请检查服务端状态") def _bind_tcp_signals(self): """绑定TCP socket的核心信号（连接、断开、接收数据、错误）""" # 连接成功信号 self.tcp_socket.connected.connect(self._on_tcp_connected) # 断开连接信号 self.tcp_socket.disconnected.connect(self._on_tcp_disconnected) # 接收数据信号（有新数据时触发） self.tcp_socket.readyRead.connect(self._on_tcp_data_received) # 错误信号（连接/通信出错时触发） self.tcp_socket.errorOccurred.connect(self._on_tcp_error) # ----------------------- # 界面初始化函数 # ----------------------- def _init_ui(self): self.setWindowFlags(Qt.FramelessWindowHint) self._load_background() main_layout = QVBoxLayout(self) main_layout.setContentsMargins(32, 20, 32, 50) main_layout.setSpacing(0) # 1. 顶部区域（标题 + 关闭按钮） self._add_top_area(main_layout) # 2. 对应管片ID区域 self._add_segment_id_area(main_layout) # 3. 网格信息区域（单列6行） self._add_grid_info_area(main_layout) main_layout.addSpacing(30) # 4. 操作按钮区域 self._init_operation_buttons(main_layout) def _init_operation_buttons(self, parent_layout): """初始化操作按钮（下料完成/生产异常/生产取消）""" buttonContainer = QWidget() buttonLayout = QHBoxLayout(buttonContainer) buttonLayout.setSpacing(100) self.errorButton = QPushButton("生产异常") self.cancelButton = QPushButton("生产取消") self.errorButton.setStyleSheet(""" QPushButton { color: #1E1E1E; font-size: 18px; font-weight: bold; padding: 15px 50px; border: 1px solid; background-color: #FF0000; border-color: transparent; } QPushButton:hover { opacity: 0.9; background-color: #B71C1C; } QPushButton:pressed { opacity: 0.8; } """) self.cancelButton.setStyleSheet(""" QPushButton { color: #1E1E1E; font-size: 18px; font-weight: bold; padding: 15px 50px; border: 1px solid; background-color: #FFD700; border-color: transparent; } QPushButton:hover { opacity: 0.9; background-color: #FFB300 } QPushButton:pressed { opacity: 0.8; } """) self.errorButton.clicked.connect(self.on_error_clicked) self.cancelButton.clicked.connect(self.on_cancel_clicked) # 初始禁用“停止”和“异常”按钮（未连接时不可用） self.errorButton.setDisabled(True) self.cancelButton.setDisabled(True) buttonLayout.addStretch(1) buttonLayout.addWidget(self.errorButton) buttonLayout.addWidget(self.cancelButton) buttonLayout.addStretch(1) parent_layout.addWidget(buttonContainer) def _load_background(self): self.bg_pixmap = QPixmap("详情弹出背景.png") # 修改为派单任务背景图 if self.bg_pixmap.isNull(): print("错误：派单任务背景.png 加载失败，请检查路径！") self.setFixedSize(800, 500) else: self.setFixedSize(self.bg_pixmap.size()) def _add_top_area(self, parent_layout): top_layout = QHBoxLayout() top_layout.setContentsMargins(0, 0, 0, 36) top_layout.addStretch() # 标题改为“任务派单” title_label = QLabel("任务派单") font = QFont() font.setPixelSize(24) font.setLetterSpacing(QFont.AbsoluteSpacing, 2) font.setBold(True) title_label.setFont(font) title_label.setStyleSheet("color: #13fffc; font-weight: Bold;") title_label.setAlignment(Qt.AlignCenter) top_layout.addWidget(title_label, alignment=Qt.AlignTop) # 关闭按钮 top_layout.addStretch() parent_layout.addLayout(top_layout) def _add_segment_id_area(self, parent_layout): id_layout = QHBoxLayout() id_label = QLabel("对应管片ID") # 标签文字修改 id_label.setFixedSize(318, 32) id_font = QFont() id_font.setPixelSize(18) id_font.setLetterSpacing(QFont.AbsoluteSpacing, 2) id_font.setBold(True) id_label.setFont(id_font) id_label.setStyleSheet( """ background-image: url("详情标题.png"); background-repeat: no-repeat; background-position: center; color: #13ffff; """ ) id_label.setContentsMargins(16, 0, 0, 0) id_label.setAlignment(Qt.AlignLeft | Qt.AlignVCenter) self.id_value_label = QLabel("") # 初始管片ID值 value_font = QFont() value_font.setPixelSize(18) value_font.setBold(True) value_font.setLetterSpacing(QFont.AbsoluteSpacing, 2) self.id_value_label.setFont(value_font) self.id_value_label.setStyleSheet("color: #13ffff;") self.id_value_label.setAlignment(Qt.AlignRight | Qt.AlignVCenter) id_layout.addWidget(id_label) id_layout.addStretch() id_layout.addWidget(self.id_value_label) id_layout.setContentsMargins(0, 0, 0, 16) parent_layout.addLayout(id_layout) parent_layout.setAlignment(id_layout, Qt.AlignTop) def _add_grid_info_area(self, parent_layout): grid_layout = QGridLayout() grid_layout.setSpacing(12) # 初始化信息条目（6行） info_items = [ {"name":"任务单号", "value": "", "api_field": "task_id"}, {"name": "工程名称", "value": "", "api_field": "project_name"}, {"name": "配比编号", "value": "", "api_field": "produce_mix_id"}, {"name": "要料状态", "value": "", "api_field": "flag"}, {"name": "砼强度", "value": "", "api_field": "beton_grade"}, {"name": "要料方量", "value": "", "api_field": "adjusted_volume"}, ] self.rows.clear() self.statusWidgets.clear() # 清空缓存 for row, item in enumerate(info_items): cell_widget = self._create_info_cell(item["name"], item["value"]) self.rows.append(cell_widget) grid_layout.addWidget(cell_widget, row, 0) self.statusWidgets.append({ 'api_field': item["api_field"], 'valueLabel': cell_widget.value # 绑定实际UI控件 } ) parent_layout.addLayout(grid_layout) def _create_info_cell(self, label_text, value_text): cell_widget = QWidget() cell_bg = QPixmap("派单任务信息栏1.png") # 正常背景图 cell_widget.setObjectName("infoCell") if not cell_bg.isNull(): cell_widget.setFixedSize(cell_bg.size()) cell_widget.setStyleSheet( """ QWidget { background-image: url(派单任务信息栏1.png); background-repeat: no-repeat; background-position: Center; } QWidget:hover { background-image: url(派单任务信息栏2.png); } QWidget QLabel#valueLabel { color: #9fbfd4; background: none; } """ ) cell_layout = QHBoxLayout(cell_widget) cell_layout.setContentsMargins(0, 0, 0, 0) # 左侧标签 label = QLabel(label_text) label.setFixedSize(136, 60) label_font = QFont() label_font.setPixelSize(16) label_font.setLetterSpacing(QFont.AbsoluteSpacing, 2) label.setFont(label_font) label.setStyleSheet("background: none;color: #fffffd; font-weight:Bold;") label.setAlignment(Qt.AlignCenter) cell_widget.label = label # 右侧值标签（设置objectName以便样式选择） value = QLabel(value_text) value.setObjectName("valueLabel") value_font = QFont() value_font.setPixelSize(20) value.setFont(value_font) value.setAlignment(Qt.AlignCenter) cell_widget.value = value cell_layout.addWidget(label) # 左侧的标题标签 cell_layout.addSpacing(60) cell_layout.addWidget(value) # 右侧的值标签 cell_widget.installEventFilter(self) return cell_widget def eventFilter(self, obj, event): """ 实现事件过滤器，动态修改右侧值颜色 """ # 只处理父控件（infoCell）的事件 if obj.objectName() == "infoCell": # 鼠标进入父控件 → 改#13f0f3 if event.type() == QEvent.Enter: if hasattr(obj, "value"): # 确保存在value控件 obj.value.setStyleSheet("background: none; color: #13f0f3;") # 鼠标离开父控件 → 恢复默认色 elif event.type() == QEvent.Leave: if hasattr(obj, "value"): obj.value.setStyleSheet("background: none; color: #9fbfd4;") return super().eventFilter(obj, event) def paintEvent(self, event): if not self.bg_pixmap.isNull(): painter = QPainter(self) painter.drawPixmap(self.rect(), self.bg_pixmap) super().paintEvent(event) def _update_ui_from_data(self, data): """根据TCP获取的数据更新界面状态""" # 1、更新顶部独立的“对应管片ID” if "artifact_id" in data: self.id_value_label.setText(str(data["artifact_id"])) # 2、更新网格中的所有字段 for widget in self.statusWidgets: api_field = widget['api_field'] value_label = widget['valueLabel'] if api_field in data: new_value = str(data[api_field]) value_label.setText(new_value) # ------------------- 对外修改接口 ------------------- # row 对应行号(0-6)，从0开始 # -------------------------------------------------- def set_segment_id(self, new_id): """修改上方的 对应的管片ID的值""" if self.id_value_label: self.id_value_label.setText(str(new_id)) def set_row_label(self, row, new_label_text: str): """修改左侧的显示的标签的文本，如: 创建时间、派单时间等""" if 0 <= row < len(self.rows): self.rows[row].label.setText(new_label_text) def set_row_value(self, row, new_value_text: str): """修改右侧的显示的值, 如: 2025年9月9日 9:9:9""" if 0 <= row < len(self.rows): self.rows[row].value.setText(new_value_text) # ------------------ # 按钮点击事件 # ------------------ def on_error_clicked(self): """点击“生产异常”：向服务端发送异常指令""" print("🔘 点击「生产异常」按钮") self._clear_ui_info() self._send_tcp_request("production_error") def on_cancel_clicked(self): """点击“生产取消”：向服务端发送取消指令""" print("🔘 点击「生产取消」按钮") self._clear_ui_info() self._send_tcp_request("cancel_feed") # -------------------- # 清空界面信息的通用方法 # -------------------- def _clear_ui_info(self): """清空管片ID和网格信息""" if self.id_value_label: self.id_value_label.setText("") for widget in self.statusWidgets: widget['valueLabel'].setText("") print("ℹ️ 界面信息已清空") # ------------------ # TCP客户端核心功能 # ------------------ @Slot() def _on_tcp_connected(self): """TCP连接成功回调""" self.is_tcp_connected = True self.has_connected_once = True self.reconnect_timer.stop() # 停止重连定时器 self.reconnect_count = 0 # 重连计数器清零 self.is_running = True print(f"TCP连接成功：{self.tcp_server_host}:{self.tcp_server_port}") # 连接成功后，向服务器发送“请求初始数据”指令 self._send_tcp_request("get_initial_data") # 更新按钮状态：启用“生产异常”“生产取消” self.errorButton.setDisabled(False) self.cancelButton.setDisabled(False) @Slot() def _on_tcp_disconnected(self): """TCP连接断开回调""" self.is_tcp_connected = False self.is_running = False print(f"TCP连接断开：{self.tcp_server_host}:{self.tcp_server_port}") # 启用/禁用按钮 self.errorButton.setDisabled(True) self.cancelButton.setDisabled(True) # 重置状态指示灯为“未连接”状态 for widget in self.statusWidgets: widget['indicator'].setStyleSheet(""" QLabel { background-color: #9E9E9E; border-radius: 10px; border: 2px solid #555555; } """) @Slot() def _on_tcp_data_received(self): """TCP数据接收回调（服务器发送数据时触发）""" tcp_data = self.tcp_socket.readAll().data().decode("utf-8").strip() print(f"TCP数据接收：{tcp_data}") # 解析数据 try: status_data = json.loads(tcp_data) self.latest_json_data = status_data self._update_ui_from_data(status_data) except json.JSONDecodeError as e: print(f"TCP数据解析失败（非JSON格式）：{e}, 原始数据：{tcp_data}") except Exception as e: print(f"TCP数据处理异常：{e}") @Slot(QAbstractSocket.SocketError) def _on_tcp_error(self, error): """TCP错误回调""" if not self.is_tcp_connected: error_str = self.tcp_socket.errorString() print(f"TCP错误：{error_str}") self.is_tcp_connected = False self.is_running = False # 启用/禁用按钮 self.errorButton.setDisabled(True) self.cancelButton.setDisabled(True) # 首次连接失败时，启动重连定时器 if not self.has_connected_once and self.reconnect_count == 0: print(f"将在{RECONNECT_INTERVAL / 1000}秒后启动重连（最多{MAX_RECONNECT}次）") self.reconnect_timer.start() def _send_tcp_request(self, request_cmd="get_status"): """向TCP服务器发送请求指令""" if not self.is_tcp_connected: print("TCP连接未建立，无法发送请求") return print(self.latest_json_data) # 构造请求数据 request_data = json.dumps({ "cmd": request_cmd, "erp_id": self.latest_json_data["erp_id"], "artifact_id":self.latest_json_data["artifact_id"], "timestamp": self.get_current_time(), "client_info": "布料系统客户端" }) + "\n" # 增加换行符作为数据结束标识 # 发送请求数据 self.tcp_socket.write(request_data.encode("utf-8")) print(f"TCP请求发送：{request_data.strip()}") # ------------------ # 时间相关的通用方法 # ------------------ def get_current_time(self): """获取格式化的当前时间""" return datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 测试代码 if __name__ == "__main__": app = QApplication(sys.argv) dialog = DispatchDetailsDialog() dialog.show() sys.exit(app.exec())