Files
zjsh_ui_sysytem/main.py

588 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
# @Time : 2025/9/10 11:29
# @Author : reenrr
# @File : test.py
'''
import sys
from PySide6.QtWidgets import (
QApplication, QWidget, QVBoxLayout, QHBoxLayout,
QLabel, QFrame, QSizePolicy, QPushButton
)
from PySide6.QtCore import Qt, QTimer, QPoint, QSize, Slot
from PySide6.QtGui import QFont, QPainter, QColor, QPen, QIcon
from datetime import datetime
from PySide6.QtNetwork import QTcpSocket, QAbstractSocket
import json
import os
# -----------
# 参数配置
# -----------
tcp_server_host = "127.0.0.1"
tcp_server_port = 8888
# 数据保存目录
SAVE_DIR = "operation_records"
class StatusMonitor(QWidget):
"""
中交三航精准布料浇筑要料系统 - 主界面类(深色主题)
使用TCP进行数据传输客户端模型与TCP服务器通信
"""
def __init__(self, parent=None):
"""构造函数初始化主界面的UI布局、控件和定时器"""
super().__init__(parent=parent)
self.is_running = False # 系统运行状态标记
self.current_datetime = self.get_current_time() # 当前日期时间
# 缓存服务端发送的最新JSON数据
self.latest_json_data = {}
# ---------------
# 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连接状态标记
# 绑定TCP信号与槽事件驱动
self._bind_tcp_signals()
# 窗口基础设置
self.setWindowTitle("中交三航精准布料浇筑要料系统")
self.setGeometry(100, 100, 850, 500) # 设置窗口位置和大小
self.setStyleSheet("background-color: #121212;") # 窗口背景设为深黑色
# 初始化主布局(垂直布局)
self.mainLayout = QVBoxLayout(self)
self.mainLayout.setContentsMargins(10, 40, 10, 10) # 上边距留空用于显示日期
self.mainLayout.setSpacing(10) # 相邻控件的间距
# 初始化界面组件
self._init_title_label()
self._init_status_container()
self._init_timers()
self._init_save_dir()
# ----------------
# 客户端自动后自动连接服务端
# ----------------
print(f"客户端启动,自动连接服务端{self.tcp_server_host}:{self.tcp_server_port}...")
self.tcp_socket.connectToHost(self.tcp_server_host, self.tcp_server_port)
def _init_save_dir(self):
"""初始化数据保存目录"""
if not os.path.exists(SAVE_DIR):
os.makedirs(SAVE_DIR)
print(f"已创建数据保存目录:{os.path.abspath(SAVE_DIR)}")
else:
print(f"数据保存目录已存在:{os.path.abspath(SAVE_DIR)}")
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_title_label(self):
"""初始化系统标题标签"""
titleLabel = QLabel("中交三航精准布料浇筑要料系统")
titleLabel.setStyleSheet("""
QLabel {
color: #00FF9D;
font-size: 20px;
font-weight: bold;
}
""")
titleLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
titleLabel.setFont(QFont("Microsoft YaHei", 24, QFont.Bold))
self.mainLayout.addWidget(titleLabel)
def _init_status_container(self):
"""初始化核心状态监控容器(包含状态组和操作按钮)"""
self.bigContainer = QFrame()
self.bigContainer.setStyleSheet("""
QFrame {
background-color: #1E1E1E;
border: 2px solid #333333;
border-radius: 8px;
}
""")
self.bigContainer.setSizePolicy(
QSizePolicy.Policy.Expanding,
QSizePolicy.Policy.Expanding
)
containerLayout = QVBoxLayout(self.bigContainer)
containerLayout.setContentsMargins(20, 20, 20, 20)
self._init_status_groups(containerLayout)
self._init_operation_buttons(containerLayout)
self.mainLayout.addWidget(self.bigContainer, 1)
def _init_status_groups(self, parent_layout):
"""初始化左右信息各包含5个状态项"""
statusWidget = QWidget()
statusLayout = QHBoxLayout(statusWidget)
statusLayout.setContentsMargins(0, 0, 0, 0)
statusLayout.setSpacing(30) # 减小中间空白间距原30
leftGroup = QWidget()
leftLayout = QVBoxLayout(leftGroup)
leftLayout.setSpacing(15)
leftLayout.setContentsMargins(30, 0, 0, 0) # 左边组左内边距设为20增加左边留白
rightGroup = QWidget()
rightLayout = QVBoxLayout(rightGroup)
rightLayout.setSpacing(15)
rightLayout.setContentsMargins(0, 0, 30, 0) # 右边组右内边距设为20增加右边留白若需左边也留白可设左内边距
# 左边5个状态项及对应初始值
leftStatusInfo = [
{"name": "任务单号", "value": "20250706-01", "api_field": "task_id"},
{"name": "工程名称", "value": "18号线二期工程", "api_field": "project_name"},
{"name": "区间段", "value": "停车场工作并上行", "api_field": "section"},
{"name": "坍落度", "value": "50~70 mm", "api_field": "slump"},
{"name": "配合比编号", "value": "P2022=001", "api_field": "mix_ratio_id"}
]
# 右边5个状态项及对应初始值
rightStatusInfo = [
{"name": "要料状态", "value": "请求中", "api_field": "request_status"},
{"name": "要料标号", "value": "C50P12", "api_field": "material_grade"},
{"name": "要料方量", "value": "2m³", "api_field": "volume"},
{"name": "要料时间", "value": "2分钟后", "api_field": "request_time"},
{"name": "小车状态", "value": "移动后", "api_field": "car_status"}
]
self.statusWidgets = []
# 处理左边状态项
for info in leftStatusInfo:
statusItem = self._create_status_item(info)
leftLayout.addWidget(statusItem)
# 处理右边状态项
for info in rightStatusInfo:
statusItem = self._create_status_item(info)
rightLayout.addWidget(statusItem)
statusLayout.addWidget(leftGroup)
statusLayout.addStretch(0) # 减小中间空白比例原1
statusLayout.addWidget(rightGroup)
parent_layout.addWidget(statusWidget)
def _create_status_item(self, info):
"""创建单个状态项"""
statusItem = QFrame()
statusItem.setStyleSheet("""
QFrame {
background-color: #2D2D2D;
border: 1px solid #444444;
border-radius: 6px;
padding: 10px;
}
""")
statusItem.setFixedHeight(80)
statusItem.setFixedWidth(320) # 统一加长状态项宽度(原无固定宽度)
itemLayout = QHBoxLayout(statusItem)
itemLayout.setContentsMargins(10, 5, 10, 5)
# 状态指示灯
indicator = QLabel()
indicator.setFixedSize(20, 20)
indicator.setStyleSheet("""
QLabel {
background-color: #9E9E9E;
border-radius: 10px;
border: 2px solid #555555;
}
""")
# 状态名称标签
nameLabel = QLabel(info["name"])
nameLabel.setFixedWidth(100) # 加宽名称标签原90
nameLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
nameLabel.setStyleSheet("font-size: 14px; color: #FFFFFF;")
nameLabel.setFont(QFont("Microsoft YaHei", 12))
# 状态值标签
valueLabel = QLabel(info["value"])
valueLabel.setStyleSheet("""
QLabel {
font-size: 16px;
font-weight: bold;
color: #FFFFFF;
background-color: #2D2D2D;
border: none;
padding: 0px;
}
""")
valueLabel.setAlignment(Qt.AlignmentFlag.AlignCenter)
valueLabel.setMinimumWidth(150) # 加宽值标签原80
itemLayout.addWidget(indicator)
itemLayout.addSpacing(10)
itemLayout.addWidget(nameLabel)
itemLayout.addStretch()
itemLayout.addWidget(valueLabel)
self.statusWidgets.append({
'indicator': indicator,
'nameLabel': nameLabel,
'valueLabel': valueLabel,
'status': False,
'initial_value': info["value"],
'api_field': info["api_field"]
})
return statusItem
def _init_operation_buttons(self, parent_layout):
"""初始化操作按钮(下料完成/生产异常/生产取消)"""
buttonContainer = QWidget()
buttonLayout = QHBoxLayout(buttonContainer)
buttonLayout.setContentsMargins(0, 20, 0, 0)
buttonLayout.setSpacing(30)
# 按钮图标(需替换为实际图标路径)
start_icon_path = "img.png"
down_icon_path = "img.png"
error_icon_path = "img.png"
cancel_icon_path = "img.png"
self.startFeedButton = QPushButton("开始下料")
self.finishButton = QPushButton("下料完成")
self.errorButton = QPushButton("生产异常")
self.cancelButton = QPushButton("生产取消")
# 设置按钮图标
self.startFeedButton.setIcon(QIcon(start_icon_path))
self.finishButton.setIcon(QIcon(down_icon_path))
self.errorButton.setIcon(QIcon(error_icon_path))
self.cancelButton.setIcon(QIcon(cancel_icon_path))
# 设置图标大小
self.startFeedButton.setIconSize(QSize(20, 20))
self.finishButton.setIconSize(QSize(20, 20))
self.errorButton.setIconSize(QSize(20, 20))
self.cancelButton.setIconSize(QSize(20, 20))
button_base_style = """
QPushButton {
font-size: 16px;
font-weight: bold;
padding: 10px 20px;
border-radius: 15px;
border: 1px solid;
min-width: 100px;
}
QPushButton:hover {
opacity: 0.9;
}
QPushButton:pressed {
opacity: 0.8;
}
"""
self.startFeedButton.setStyleSheet(button_base_style + """
QPushButton {
background-color: #2196F3;
color: white;
border-color: #1976D2;
}
""")
self.finishButton.setStyleSheet(button_base_style + """
QPushButton {
background-color: #00796B;
color: white;
border-color: #004D40;
}
""")
self.errorButton.setStyleSheet(button_base_style + """
QPushButton {
background-color: #E65100;
color: white;
border-color: #BF360C;
}
""")
self.cancelButton.setStyleSheet(button_base_style + """
QPushButton {
background-color: #C62828;
color: white;
border-color: #8E0000;
}
""")
button_font = QFont("Microsoft YaHei", 12, QFont.Bold)
self.startFeedButton.setFont(button_font)
self.finishButton.setFont(button_font)
self.errorButton.setFont(button_font)
self.cancelButton.setFont(button_font)
self.startFeedButton.clicked.connect(self.on_start_clicked)
self.finishButton.clicked.connect(self.on_finish_clicked)
self.errorButton.clicked.connect(self.on_error_clicked)
self.cancelButton.clicked.connect(self.on_cancel_clicked)
# 初始禁用“停止”和“异常”按钮(未连接时不可用)
self.startFeedButton.setDisabled(False)
self.finishButton.setDisabled(True)
self.errorButton.setDisabled(True)
self.cancelButton.setDisabled(True)
buttonLayout.addStretch(1)
buttonLayout.addWidget(self.startFeedButton)
buttonLayout.addWidget(self.finishButton)
buttonLayout.addWidget(self.errorButton)
buttonLayout.addWidget(self.cancelButton)
buttonLayout.addStretch(1)
parent_layout.addWidget(buttonContainer)
def _init_timers(self):
"""初始化定时器(时间更新+状态模拟)"""
# 时间更新定时器(每秒更新一次)
self.time_timer = QTimer()
self.time_timer.timeout.connect(self.update_time)
self.time_timer.start(1000)
def get_current_time(self):
"""获取格式化的当前时间"""
return datetime.now().strftime('%Y-%m-%d %H:%M:%S')
def get_timestamp(self):
"""获取当前时间戳(秒级)"""
return datetime.now().strftime('%Y%m%d_%H%M%S_%f')[:-3]
def update_time(self):
"""更新时间显示并触发重绘"""
self.current_datetime = self.get_current_time()
self.update() # 触发paintEvent重绘
def paintEvent(self, event):
"""重写绘画事件,在右上角绘制日期时间文本"""
super().paintEvent(event) # 调用父类方法保持原有绘制
# 创建QPainter对象
painter = QPainter(self)
painter.setRenderHint(QPainter.RenderHint.TextAntialiasing) # 文本抗锯齿
# 设置字体
font = QFont("Arial", 12, QFont.Bold)
painter.setFont(font)
# 设置文本颜色
painter.setPen(QColor("#00FF9D"))
# 计算文本位置(右上角,留出边距)
text = f"🕒 {self.current_datetime}"
text_rect = painter.boundingRect(self.rect(), Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignTop, text)
x = self.width() - text_rect.width() - 15 # 右边距15px
y = 15 # 上边距15px
# 绘制文本
painter.drawText(x, y + text_rect.height(), text)
def _save_data_to_file(self, button_name):
"""
将服务端数据+按钮操作信息保存到JSON文件
参数button_name:点击的按钮名称
"""
# 1、检查是否有服务端数据
if not self.latest_server_data:
print(f"⚠️ 未收到服务端数据,无法保存「{button_name}」操作记录")
return
# 2、构建完整数据服务端数据+按钮操作信息+存档时间)
save_data = {
"opration_button": button_name,
"save_time": self.get_current_time(),
"server_data":self.latest_server_data
}
# 3、生成唯一文件名按时间戳命名
file_name = f"operation_record_{self.get_timestamp()}.json"
file_path = os.path.join(SAVE_DIR, file_name)
# 4、写入JSON文件
try:
with open(file_path, "w", encoding="utf-8") as f:
json.dump(save_data, f, ensure_ascii=False, indent=4)
print(f"💾 保存「{button_name}」操作记录成功:{os.path.abspath(file_path)}")
except Exception as e:
print(f"💾 保存「{button_name}」操作记录失败:{str(e)}")
# ------------------
# TCP客户端核心功能
# ------------------
@Slot()
def _on_tcp_connected(self):
"""TCP连接成功回调"""
self.is_tcp_connected = True
self.is_running = True
print(f"TCP连接成功{self.tcp_server_host}:{self.tcp_server_port}")
# 连接成功后,向服务器发送“请求初始数据”指令
self._send_tcp_request("get_initial_data")
# 更新按钮状态:启用“下料完成”“生产异常”“生产取消”
self.finishButton.setDisabled(False)
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.startFeedButton.setDisabled(False)
self.finishButton.setDisabled(True)
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_server_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错误回调"""
error_str = self.tcp_socket.errorString()
print(f"TCP错误{error_str}")
self.is_tcp_connected = False
self.is_running = False
# 启用/禁用按钮
self.startFeedButton.setDisabled(False)
self.finishButton.setDisabled(True)
self.errorButton.setDisabled(True)
self.cancelButton.setDisabled(True)
def _send_tcp_request(self, request_cmd="get_status"):
"""向TCP服务器发送请求指令"""
if not self.is_tcp_connected:
print("TCP连接未建立无法发送请求")
return
# 构造请求数据
request_data = json.dumps({
"cmd": request_cmd,
"timestamp": self.get_current_time(),
"client_info": "布料系统客户端"
}) + "\n" # 增加换行符作为数据结束标识
# 发送请求数据
self.tcp_socket.write(request_data.encode("utf-8"))
print(f"TCP请求发送{request_data.strip()}")
def _update_ui_from_data(self, data):
"""根据TCP获取的数据更新界面状态"""
for widget in self.statusWidgets:
api_field = widget['api_field']
if api_field in data:
new_value = str(data[api_field])
widget['valueLabel'].setText(new_value)
self.set_indicator_status(widget, new_value)
# ------------------
# 状态指示灯逻辑
# ------------------
def set_indicator_status(self, widget, value):
"""根据值设置状态指示灯颜色"""
if value and value != "未知" and value != "" and value != "None":
# 有效数据:绿色指示灯
widget['indicator'].setStyleSheet("""
QLabel {
background-color: #00E676;
border-radius: 10px;
border: 2px solid #00796B;
}
""")
else:
# 无效数据:红色指示灯
widget['indicator'].setStyleSheet("""
QLabel {
background-color: #FF5252;
border-radius: 10px;
border: 2px solid #C62828;
}
""")
# ------------------
# 按钮点击事件
# ------------------
def on_start_clicked(self):
"""点击“开始下料”:向服务端发送开始指令"""
print("🔘 点击「开始下料」按钮")
if not self.is_tcp_connected:
print("TCP连接未建立尝试重新连接")
self.tcp_socket.connectToHost(self.tcp_server_host, self.tcp_server_port)
QTimer.singleShot(1000, lambda: self._send_tcp_request("start_feed")) # 等待连接成功后再发送请求
else:
print("TCP连接已建立")
self._send_tcp_request("start_feed")
def on_finish_clicked(self):
"""点击“下料完成”:向服务端发送完成指令"""
print("🔘 点击「下料完成」按钮")
self._send_tcp_request("finish_feed")
self._save_data_to_file("finish_feed")
def on_error_clicked(self):
"""点击“生产异常”:向服务端发送异常指令"""
print("🔘 点击「生产异常」按钮")
self._send_tcp_request("production_error")
self._save_data_to_file("production_error")
def on_cancel_clicked(self):
"""点击“生产取消”:向服务端发送取消指令"""
print("🔘 点击「生产取消」按钮")
self._send_tcp_request("cancel_feed")
self._save_data_to_file("cancel_feed")
if __name__ == "__main__":
app = QApplication(sys.argv)
window = StatusMonitor()
window.show()
sys.exit(app.exec())