活动介绍
file-type

QT与西门子PLC通讯的SNAP7协议模板程序

ZIP文件

QTC++
4星 · 超过85%的资源 | 下载需积分: 5 | 3.8MB | 更新于2025-02-03 | 135 浏览量 | 189 下载量 举报 10 收藏
download 立即下载
根据所提供的文件信息，我们可以详细阐述如下知识点： ### QT SNAP7协议 demo 知识点 #### 1. QT 技术概述 QT 是一个跨平台的应用程序和用户界面框架，由挪威公司 Trolltech 开发，后被诺基亚收购。QT 支持多种操作系统，包括 Windows、Mac OS X、Linux、嵌入式 Linux、Android、iOS 等。它使用 C++ 语言编写，同时提供了一套丰富的库集合（Qt Library），能够用于开发图形用户界面应用程序，也可以开发非GUI程序，例如命令行工具和服务器。 #### 2. C++ 语言特点 C++ 是一种静态类型、编译式、通用的编程语言。它支持过程化编程、面向对象编程和泛型编程。C++ 是 C 语言的一个超集，它在继承了 C 语言强大功能的同时，增加了面向对象的特性，如类、继承和多态等。C++广泛应用于系统软件、游戏开发、桌面应用、实时物理模拟等领域。 #### 3. SNAP7 协议介绍 SNAP7 是一个开源的通讯协议，它遵循 Siemens S7 协议规范，可用于与西门子 PLC（可编程逻辑控制器）进行通讯。SNAP7 通过 TCP/IP 或者 ISO-on-TCP（ISO 8073/30003）协议建立连接，能够实现与西门子 PLC 的读写操作。SNAP7 与商业软件相比，最大的优势在于其完全开源并且功能全面，它支持对西门子 PLC 的各种数据类型进行读写操作。 #### 4. PLC 与计算机通讯 PLC（Programmable Logic Controller）是可编程逻辑控制器，它广泛应用于工业自动化控制。PLC 与计算机通讯是指通过特定的通讯协议或接口，实现计算机与 PLC 之间的数据交换。通讯方式可以是串口通讯、以太网通讯等。在以太网通讯中，常使用 Modbus TCP、Profinet、Profibus、Ethernet/IP 等协议。 #### 5. 与西门子 PLC 通讯的数据类型 在与西门子 PLC 进行通讯时，可以操作的数据类型非常丰富，包括但不限于： - **bit**：二进制位，是 PLC 中最小的数据单位。 - **byte**：字节，包含 8 位，可以表示 0 到 255 的数值。 - **word**：字，通常由 2 个字节组成，可以表示 0 到 65535 的数值。 - **Dword**：双字，由 4 个字节组成，可以表示更大范围的数值。 - **Uint**：无符号整型，用于表示大于 Word 范围的整数。 - **float**：浮点数，用于表示小数或者实数。 #### 6. 线程读取的概念 线程读取是指在计算机程序中，同时进行的多个操作或者说是多个执行流程。在多核 CPU 时代，线程读取可以充分地利用 CPU 的多核资源，提高程序的执行效率。线程可以理解为程序执行流的最小单元，它可以被操作系统调度和执行。在读取操作中，可以利用多线程实现多个数据块的并行读取，从而减少等待时间和提高效率。 #### 7. 西门子 PLC 的编程和维护 西门子 PLC 系列产品广泛应用于工业控制领域。其编程软件，如 STEP 7、TIA Portal，提供了丰富的编程工具和环境，使得 PLC 的编程、调试、维护变得更为便捷。在实际应用中，PLC 的编程和维护工作包括：程序的编写、上载、调试、修改、故障诊断、数据记录等。 #### 8. 模板程序的作用和使用 模板程序是一种预先设计好的程序结构或者代码框架，它提供了一种通用的解决方案或者解决方法的示例。在本例中，QT SNAP7 协议 demo 就作为一个模板程序，它展示了如何使用 QT 技术结合 SNAP7 协议实现与西门子 PLC 的通讯。开发者可以利用这个模板程序作为基础，根据自己的需求进行修改和扩展，从而快速地搭建出满足特定功能需求的应用程序。 #### 9. 文件压缩包的文件名称列表说明 从提供的压缩包名称 "QtSnap7-master" 可以推断，该压缩包是 QT 技术与 SNAP7 协议结合使用的一个项目或者库的源代码。"master" 通常表示主分支，意味着这个压缩包包含了该开源项目主分支的最新代码。开发者可以通过解压此压缩包获取项目的所有代码和资源文件，并对其进行编辑和使用。 总结以上知识点，我们可以发现本文件提供的信息涉及到了跨平台的 QT 应用程序开发、C++ 编程语言的使用、与西门子 PLC 的通讯协议和数据类型操作、多线程编程的基本概念、以及如何利用模板程序快速搭建应用程序等内容。开发者利用这个 QT SNAP7协议 demo，可以更快速地实现与西门子 PLC 的通讯功能，并且通过扩展模板程序达到符合自身业务需求的应用软件开发。

相关推荐

filetype

Qt下基于snap7西门子PLC通信.zip

Qt基于snap7第三方库西门子PLC通信示例，附件中包括snap7.dll、snap7.lib、Qt示例源码和配置文档
filetype

QtSnap7:Qt snap7 S7-1200循环读取

QtSnap7 [wiki] Qt5.9 snap7 S7-1200循环读取 快照7 Qt5
filetype

Qt调用Snap7.zip

Qt5调用Snap7与西门子PLC通信, 读取DB块
filetype

Qt之PLC snap7开发入门demo及源码

Qt之PLC snap7开发入门demo及源码
filetype

qt与西门子PLC通讯snap7协议自动重连

我们正在使用Qt与西门子PLC通信，基于Snap7协议。用户询问如何实现自动重连功能。 根据引用内容，我们知道Snap7是一个用于与西门子PLC通信的库，在Qt中可以通过封装Snap7库来实现通信功能。 自动重连功能通常包括...
filetype

SNAP7协议说明书

snap7协议是西门子PLC通讯的一种协议，它具有跨平台、开源的特性，可以让开发者通过C++等编程语言实现与西门子PLC的通信。snap7库支持S7-300、S7-400、S7-1200和S7-1500系列的西门子PLC设备。 snap7协议说明书提供...
filetype

snap7-SDK包

"python-snap7demo"标签暗示了压缩包内有一个Python演示程序，即"PLCdemo.py"，这个文件很可能是用来展示如何在Python中使用Snap7库进行基本的PLC操作，如读取和写入数据。 在"PLCdemo.py"中，我们可能看到以下关键...
filetype

PLC通信c++ 库Snap7

Snap7是一个开源的C++库，专门用于PC与Siemens S7系列PLC之间的通信。本文将深入探讨Snap7库及其在C++中的应用。 1. **Snap7概述** Snap7库为C++开发者提供了一套完整的工具，使得他们能够轻松地与西门子S7系列的...
filetype

qucsdk:Qt编写的自定义控件插件的sdk集合，包括了各个操作系统的动态库文件以及控件的头文件和sdk使用demo。心中有坐标，万物皆painter，欢迎各位咨询、购买、定制控件，QQ：517216493 微信：feiyangqingyun

Qt编写的自定义控件插件的sdk集合，包括了各个操作系统的动态库文件以及控件的头文件和sdk使用demo。 体验地址 体验地址： 提取密码：877p 文件名称：bin_quc.zip 尊贵提示：可付费购买控件完整源码，每个控件都有...
filetype

Qt GUI 界面通用系统模板 源码

7. **other、snap、api、form、temp** 这些可能是项目中的其他目录，分别存储了辅助文件、截图、API接口定义、表单文件以及临时文件。"other"可能包含辅助资源，如图标、图片等；"snap"可能存放应用的截图；"api...
filetype

Qt与snap7的PLC开发示例教程及完整源码

资源摘要信息:"Qt之PLC snap7开发入门demo及源码" 知识点概述： 1. Qt框架应用 2. PLC与snap7的集成开发 3. 源码结构与组件功能解析 4. 实时监控与控制的实现方法 5. 开发环境的搭建与配置 1. Qt框架应用 Qt是一个...
filetype

Ubuntu下Qt与SQLite数据库操作的Demo教程

7. Qt数据库模块（Qt SQL） Qt SQL模块是Qt框架的一部分，提供了与各种SQL数据库交互的能力。它包括数据库驱动程序、SQL语句的执行、查询结果的检索等功能。通过Qt SQL模块，开发者可以更加便捷地在Qt应用程序中添加...
filetype

C++利用snap7实现西门子PLC整型与小数读写操作教程

这样的demo对于学习和理解如何将snap7集成到Qt应用程序中，并通过图形界面与PLC进行交互是非常有帮助的。 从文件名称列表中，我们可以看到这个demo程序的文件构成。'QtGuiApplicationSanp7.sdf'可能是一个项目文件...
filetype

Qt系统通用界面模板源码分享，自定义换肤功能

- **项目文件（uidemo25.pro）**：Qt特有的项目文件，描述了项目源代码的位置、编译选项和依赖关系等。 - **资源文件夹（other、snap）**：包含程序运行时所需的图片、图标和其他资源文件。 - **API接口（api）**：...
filetype

用Python实现与西门子S7-300/S7-400 PLC的TCP/IP通信

一个著名的库叫做python-snap7，这是一个开源的Python库，它基于snap7项目，用于实现与西门子PLC的通信。python-snap7实现了与PLC的多种通信模式，包括直接连接、监听和被动连接等，而且支持数据读写、块读取、定时...
filetype

VS中Qt项目部署指南：应用发布与分发的专业技巧

[VS中Qt项目部署指南：应用发布与分发的专业技巧](https://opengraph.githubassets.com/93b859acae131f024bdcc0bc183b4c9d9385943abc952fc4e16a60659340476e/qtproject/qt-apps-multiscreen-demo) # 摘要 本文旨在...
filetype

import copy import sys import threading import atexit from PyQt5.QtWidgets import * from PyQt5.QtCore import Qt, QRect, QPoint, QSize, pyqtSignal, QTimer, QEvent from PyQt5.QtGui import QFont, QColor from datetime import datetime from multiprocessing import Process, Queue, freeze_support import weakref import psutil import time class MT5ProcessWorker: """ 独立进程工作器（替代原MT5Worker） 通过队列与主进程通信 """ def __init__(self, exe_path, symbol): self.exe_path = exe_path self.symbol = symbol self.queue = Queue() # 每个worker独立队列 self._process = None self._shutdown_flag = False def start(self): self._process = Process( target=self._run, args=(self.exe_path, self.symbol, self.queue) ) self._process.daemon = True # 主进程退出时自动终止 self._process.start() atexit.register(self.stop) # 注册退出清理 def _run(self, exe_path, symbol, queue): """子进程运行方法""" try: import MetaTrader5 as mt5_local # 每个进程独立导入 import time # 初始化独立MT5实例 if not mt5_local.initialize(path=exe_path, timeout=5): queue.put(('error', f"初始化失败: {mt5_local.last_error()}")) return # 获取并发送精度信息 symbol_info = mt5_local.symbol_info(symbol) if not symbol_info: queue.put(('error', f"无效品种: {symbol}")) return # 获取服务器信息 account_info = mt5_local.account_info() if account_info and hasattr(account_info, 'server'): server_name = account_info.server else: server_name = "Unknown Server" print("警告: 无法获取服务器名称") # 添加调试输出 digits = symbol_info.digits queue.put(('connected', (server_name, digits))) # 返回服务器名和精度 print(f"发送连接成功消息: {server_name}") # 添加调试输出 # 报价循环 while not self._shutdown_flag: try: # 增加心跳检测 if not mt5_local.terminal_info().connected: break # 检查队列中的关闭命令 if not queue.empty(): msg_type, _ = queue.get_nowait() if msg_type == 'shutdown': break # 在循环开始处检查队列状态 if queue.qsize() > 10: # 防止队列堆积 queue.queue.clear() terminal = mt5_local.terminal_info() if terminal is None or not hasattr(terminal, 'connected'): queue.put(('disconnected', "MT5连接已断开")) break if not terminal.connected: queue.put(('disconnected', "连接已主动断开")) tick = mt5_local.symbol_info_tick(symbol) if tick and tick.time_msc > 0: queue.put(('price', (symbol, tick.bid, tick.ask))) time.sleep(0.1) # 减少CPU使用率 except AttributeError as ae: queue.put(('error', f"终端状态获取失败: {str(ae)}")) break except Exception as e: print(f"报价循环异常: {e}") except Exception as e: queue.put(('error', str(e))) finally: mt5_local.shutdown() queue.put(('disconnected', "MT5连接关闭")) # 确保关闭时发送通知 def stop(self): if self._process: try: # 发送关闭指令 self.queue.put(('shutdown', None)) self._process.join(2) # 增加等待时间 # 双重终止保障 if self._process.is_alive(): self._process.terminate() self._process.join(1) # 安全关闭队列 self.queue.close() self.queue.cancel_join_thread() except Exception as e: print(f"进程终止失败: {str(e)}") class PriceCard(QWidget): delete_requested = pyqtSignal(object) symbol_changed = pyqtSignal(str) set_as_reference = pyqtSignal(str, float, float) status_updated = pyqtSignal(str) # 新增状态信号 # 添加一个信号用于线程安全地更新UI update_platform_info = pyqtSignal(str, int) update_prices_signal = pyqtSignal(str, float, float) def __init__(self, parent=None, currency_list=None): super().__init__(parent) self.mt5_worker = None self.data_timer = QTimer() # 初始设置货币对列表 if currency_list is not None: self.currency_list = copy.deepcopy(currency_list) else: self.currency_list = ["EURUSD", "GBPUSD", "USDJPY", "XAUUSD"] self.current_bid = 0.0 self.current_ask = 0.0 self._is_connected = False # 新增连接状态标志 self._has_valid_prices = False # 是否接收过有效价格 self.is_reference = False self._valid = True # 新增有效性标志 self.abnormal_count = 0 self.setup_ui() self.setup_style() self.setup_connections() # 初始化MT5工作器 self.start_connection() # 连接信号到UI更新方法 self.update_platform_info.connect(self._safe_update_platform_info) self.update_prices_signal.connect(self._safe_update_prices) def _safe_update_platform_info(self, server_name, digits): """在线程安全的情况下更新平台信息""" self.platform_input.setText(f"已连接 · {server_name}") self.platform_input.setStyleSheet("color: #009900;") self._digits = digits self._is_connected = True def _safe_update_prices(self, symbol, bid, ask): """线程安全地更新价格显示""" if not hasattr(self, '_digits'): self._digits = 5 # 默认精度 format_str = f".{self._digits}f" try: if bid > 0 and ask > 0: self._has_valid_prices = True self.current_bid = bid self.current_ask = ask self.sell_label.setText(f"{bid:{format_str}}") self.buy_label.setText(f"{ask:{format_str}}") # 持续更新reference_label if self.is_reference: self.status_updated.emit(f"参照平台:{symbol} {bid}/{ask} [{datetime.now().strftime("%H:%M:%S")}]") else: self._has_valid_prices = False except Exception as e: print(f"更新价格异常: {e}") # 安全地断开信号连接 @staticmethod def safe_disconnect(signal: pyqtSignal): try: # 尝试断开所有连接 signal.disconnect() except TypeError: # 没有连接或已断开，忽略 pass def _process_queue(self): """非阻塞方式处理队列""" if not self.mt5_worker or not hasattr(self.mt5_worker, 'queue'): return try: max_messages = 10 for _ in range(max_messages): if self.mt5_worker.queue.empty(): break try: msg_type, data = self.mt5_worker.queue.get_nowait() self._handle_queue_message(msg_type, data) except Exception as e: print(f"处理队列消息异常: {e}") break except Exception as e: print(f"队列处理主循环异常: {e}") finally: # 继续轮询 QTimer.singleShot(50, self._process_queue) def setup_ui(self): self.setFixedSize(320, 200) layout = QVBoxLayout(self) layout.setContentsMargins(12, 8, 12, 8) layout.setSpacing(8) # 标题栏 title_layout = QHBoxLayout() title_layout.setContentsMargins(0, 0, 0, 0) title_layout.setSpacing(6) self.platform_input = QLineEdit("等待连接...") self.platform_input.setFont(QFont("微软雅黑", 9)) self.platform_input.setReadOnly(True) # 参考按钮 self.ref_btn = QPushButton("★") self.ref_btn.setFixedHeight(24) self.ref_btn.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) # 删除按钮 self.delete_btn = QPushButton("×") self.delete_btn.setFixedHeight(24) self.delete_btn.setSizePolicy(QSizePolicy.Preferred, QSizePolicy.Fixed) title_layout.addWidget(self.platform_input, stretch=1) title_layout.addWidget(self.ref_btn) title_layout.addWidget(self.delete_btn) layout.addLayout(title_layout) # 交易品种 self.symbol_combo = QComboBox() self.symbol_combo.addItems(self.currency_list) layout.addWidget(self.symbol_combo) # 价格显示 price_layout = QHBoxLayout() price_layout.setContentsMargins(0, 0, 0, 0) price_layout.setSpacing(10) self.sell_label = QLabel("--") self.buy_label = QLabel("--") for label in [self.sell_label, self.buy_label]: label.setFixedHeight(40) label.setAlignment(Qt.AlignCenter) label.setFont(QFont("Arial", 12, QFont.Bold)) price_layout.addWidget(label) layout.addLayout(price_layout) # MT5路径选择 path_layout = QHBoxLayout() self.path_input = QLineEdit() self.path_input.setPlaceholderText("MT5可执行文件路径") self.browse_btn = QPushButton("浏览") self.browse_btn.setMinimumWidth(100) self.browse_btn.setFixedHeight(24) path_layout.addWidget(self.path_input) path_layout.addWidget(self.browse_btn) layout.addLayout(path_layout) self.browse_btn.clicked.connect(self.browse_mt5_exe) self.delete_btn.clicked.connect(self.request_delete) self.symbol_combo.currentTextChanged.connect(self.symbol_changed.emit) self.ref_btn.clicked.connect(self.emit_reference_signal) def setup_style(self): self.base_style = """ QWidget { background: white; border-radius: 8px; border: 2px solid #e0e0e0; } QLineEdit, QComboBox { border: 1px solid #d0d0d0; border-radius: 4px; padding: 4px; } QStatusBar { background: #f8f8f8; border-top: 1px solid #e0e0e0; } """ self.normal_style = "" self.abnormal_style = """ QWidget { border: 2px solid #ff0000; background: #fff0f0; } """ self.ref_style = """ QWidget { border: 2px solid #00aa00; background: #f8fff8; } """ self.ref_btn.setStyleSheet(""" QPushButton { background: transparent; color: #888888; font-size: 12px; border-radius: 4px; border: 1px solid #e0e0e0; } QPushButton:checked { color: #FFFFFF; background: #00aa00; border-radius: 4px; } """) self.delete_btn.setStyleSheet(""" QPushButton { background: transparent; color: #888888; font-size: 12px; border-radius: 4px; border: 1px solid #e0e0e0; } QPushButton:hover { color: #FF0000; border-color: #FF0000; } """) self.sell_label.setStyleSheet(""" QLabel { background-color: #8B0000; color: white; border-radius: 4px; padding: 6px; min-width: 100px; } """) self.buy_label.setStyleSheet(""" QLabel { background-color: #000080; color: white; border-radius: 4px; padding: 6px; min-width: 100px; } """) # 为浏览按钮添加样式 self.browse_btn.setStyleSheet(""" QPushButton { color: black; /* 设置文字颜色为黑色，可按需调整 */ background-color: #f0f0f0; /* 设置背景色，可按需调整 */ border: 1px solid #ccc; /* 设置边框，可按需调整 */ padding: 4px 8px; } """) self.setStyleSheet(self.base_style) def isValid(self): return self._valid def deleteLater(self): if not self._valid: # 防止重复调用 return self._valid = False # 停止定时器 if hasattr(self, 'data_timer') and self.data_timer.isActive(): self.data_timer.stop() # 停止工作进程 if hasattr(self, 'mt5_worker') and self.mt5_worker: try: self.mt5_worker.stop() except Exception as e: print(f"停止MT5工作进程失败: {e}") self.safe_disconnect(self.delete_requested) self.safe_disconnect(self.set_as_reference) self.safe_disconnect(self.symbol_changed) self.safe_disconnect(self.status_updated) self.safe_disconnect(self.update_platform_info) self.safe_disconnect(self.update_prices_signal) super().deleteLater() def setup_connections(self): self.symbol_changed.connect(self.restart_connection) def emit_reference_signal(self): if self.ref_btn.isChecked(): self.set_as_reference.emit( self.symbol_combo.currentText(), self.current_bid, self.current_ask ) self.set_style(self.ref_style) self.is_reference = True else: self.clear_reference_style() self.set_as_reference.emit("", 0.0, 0.0) self.is_reference = False def clear_reference_style(self): self.setFixedSize(320, 200) self.setStyleSheet(f"{self.base_style}") if self.parent(): self.parent().updateGeometry() def set_style(self, style_sheet): self.setStyleSheet(style_sheet) self.update() def update_prices(self, symbol, bid, ask): # 使用信号更新UI，确保线程安全 self.update_prices_signal.emit(symbol, bid, ask) def update_abnormal_status(self, is_abnormal): if self.is_reference: return if is_abnormal: self.abnormal_count += 1 self.set_style(self.abnormal_style) else: self.abnormal_count = 0 self.set_style(self.normal_style) def browse_mt5_exe(self): path, _ = QFileDialog.getOpenFileName( self, "选择MT5可执行文件", filter="Executable Files (*.exe)" ) if path: self.path_input.setText(path) self.start_connection() def start_connection(self): try: if not self.path_input.text().strip(): return if hasattr(self, 'mt5_worker') and self.mt5_worker: self.mt5_worker.stop() self.mt5_worker = MT5ProcessWorker( self.path_input.text().strip(), self.symbol_combo.currentText() ) self.mt5_worker.start() # 启动数据接收定时器 if not self.data_timer.isActive(): self.data_timer.timeout.connect(self._process_queue) self.data_timer.start(50) # 添加临时状态显示，用于调试 self.platform_input.setText("正在连接...") self.platform_input.setStyleSheet("color: #FF8800;") except Exception as e: print(f"启动连接失败: {e}") self.platform_input.setText("连接失败") self.platform_input.setStyleSheet("color: #FF0000;") def _handle_queue_message(self, msg_type, data): try: if msg_type == 'connected': server_name, digits = data print(f"收到连接成功消息: {server_name}, 精度: {digits}") # 使用信号更新UI，确保在主线程中执行 self.update_platform_info.emit(server_name, digits) elif msg_type == 'disconnected': disconnect_msg = data self.on_disconnected(disconnect_msg) elif msg_type == 'price': symbol, bid, ask = data self.update_prices(symbol, bid, ask) elif msg_type == 'error': error_msg = data self.show_error(error_msg) except Exception as e: print(f"处理队列消息异常: {e}") def on_disconnected(self, message): self.platform_input.setText("等待连接...") self.platform_input.setStyleSheet("color: #666666;") self._is_connected = False self._has_valid_prices = False self.sell_label.setText("--") self.buy_label.setText("--") # 自动重连逻辑 if not self._valid: # 卡片已标记为无效时不重连 return QTimer.singleShot(3000, self.start_connection) def request_delete(self): # 立即从父布局中移除自己 if self.parent() and self.parent().layout(): layout = self.parent().layout() layout.removeWidget(self) # 发送删除请求 self.delete_requested.emit(weakref.ref(self)) # 停止数据接收 if hasattr(self, 'data_timer') and self.data_timer.isActive(): self.data_timer.stop() # 终止工作进程 if hasattr(self, 'mt5_worker') and self.mt5_worker: self.mt5_worker.stop() self.mt5_worker = None self.safe_disconnect(self.delete_requested) self.safe_disconnect(self.set_as_reference) self.safe_disconnect(self.symbol_changed) self.safe_disconnect(self.status_updated) self.safe_disconnect(self.update_platform_info) self.safe_disconnect(self.update_prices_signal) self._valid = False self.hide() # 立即隐藏 self.deleteLater() def restart_connection(self, new_symbol): if self.path_input.text().strip(): self.start_connection() def on_connected(self, server_name, digits): print(f"更新连接状态: {server_name}") # 添加调试输出 self.platform_input.setText(f"已连接 · {server_name}") self.platform_input.setStyleSheet("color: #009900;") self._digits = digits self._is_connected = True def show_error(self, message): QMessageBox.critical(self, "连接错误", f"{message}\n请检查：\n1. MT5客户端是否已登录\n2. 路径是否正确\n3. 品种是否有效") self.platform_input.setText("连接失败") self.path_input.clear() self._is_connected = False def update_currencies(self, new_currency=None): """更新交易品种下拉菜单""" # 检查品种是否已存在 if new_currency and new_currency not in self.currency_list: self.currency_list.append(new_currency) if hasattr(self, 'symbol_combo'): self.symbol_combo.clear() self.symbol_combo.addItems(self.currency_list) class CardScrollArea(QScrollArea): """自定义滚动区域，确保卡片完整显示""" def __init__(self, parent=None): super().__init__(parent) self.setWidgetResizable(True) self._block_signal = False # 滚动动画定时器 self.scroll_timer = QTimer(self) self.scroll_timer.timeout.connect(self._smooth_scroll_step) self.target_value = 0 self.start_value = 0 self.scroll_steps = 0 self.current_step = 0 # 安装事件过滤器以捕获滚动事件 self.verticalScrollBar().installEventFilter(self) def eventFilter(self, obj, event): """拦截滚动事件，实现卡片对齐""" # 兼容不同PyQt5版本的事件类型检查 slider_release_type = getattr(QEvent, 'SliderRelease', None) if slider_release_type is None: # 尝试使用Type枚举 slider_release_type = getattr(QEvent.Type, 'SliderRelease', None) if obj == self.verticalScrollBar() and event.type() == slider_release_type: if not self._block_signal: self._snap_to_card() return True return super().eventFilter(obj, event) def _snap_to_card(self): """对齐到最近的完整卡片""" if not self.widget(): return scroll_bar = self.verticalScrollBar() current_value = scroll_bar.value() viewport_height = self.viewport().height() layout = self.widget().layout() valid_items = [layout.itemAt(i).widget() for i in range(layout.count()) if layout.itemAt(i).widget()] positions = [] for widget in valid_items: y = widget.y() height = widget.height() top = y bottom = y + height if top <= current_value + viewport_height and bottom >= current_value: positions.extend([top, bottom]) if positions: closest_pos = min(positions, key=lambda x: abs(x - current_value)) self._start_smooth_scroll(closest_pos) def _start_smooth_scroll(self, target): """开始平滑滚动到目标位置""" scroll_bar = self.verticalScrollBar() self.start_value = scroll_bar.value() self.target_value = target self.current_step = 0 self.scroll_steps = 10 # 动画步数 self.scroll_timer.start(20) # 每20ms更新一次 def _smooth_scroll_step(self): """平滑滚动的每一步""" if self.current_step >= self.scroll_steps: self.scroll_timer.stop() return t = self.current_step / self.scroll_steps ease_t = t * t * (3 - 2 * t) # 缓入缓出函数 scroll_bar = self.verticalScrollBar() value = int(self.start_value + (self.target_value - self.start_value) * ease_t) self._block_signal = True scroll_bar.setValue(value) self._block_signal = False self.current_step += 1 def resizeEvent(self, event): """处理窗口大小变化事件，确保布局更新""" super().resizeEvent(event) # 使用单次定时器延迟布局更新，避免频繁计算 QTimer.singleShot(100, self._safe_layout_update) def _safe_layout_update(self): layout = self.widget().layout() if layout: layout.invalidate() self.updateGeometry() class FlowLayout(QLayout): def __init__(self, parent=None): super().__init__(parent) self._items = [] self._hspace = 10 # 水平间距 self._vspace = 10 # 垂直间距 self._is_layouting = False self._layout_cache = None self._last_width = -1 self._update_pending = False # 用于延迟布局更新 def addItem(self, item): self._items.append(item) self._layout_cache = None self.invalidate() self._schedule_layout_update() def _schedule_layout_update(self): """安排延迟布局更新，避免频繁计算""" if not self._update_pending: self._update_pending = True QTimer.singleShot(50, self._safe_layout_update) def _safe_layout_update(self): """安全地更新布局""" try: if self.parent(): self._do_layout(self.parent().rect()) finally: self._update_pending = False def count(self): return len(self._items) def itemAt(self, index): if 0 <= index < len(self._items): return self._items[index] return None def takeAt(self, index): if 0 <= index < len(self._items): self._layout_cache = None self._schedule_layout_update() return self._items.pop(index) return None def setGeometry(self, rect): super().setGeometry(rect) if self._last_width != rect.width(): self._layout_cache = None self._last_width = rect.width() self._do_layout(rect) def _do_layout(self, rect): if self._is_layouting: return self._is_layouting = True try: x = rect.x() y = rect.y() line_height = 0 available_width = rect.width() valid_items = [item for item in self._items if item.widget() and item.widget().isVisible()] layout_cache = [] for i, item in enumerate(valid_items): widget = item.widget() if not widget: continue # 获取内容边距 margins = widget.contentsMargins() total_margin_width = margins.left() + margins.right() total_margin_height = margins.top() + margins.bottom() # 获取组件的实际大小 size_hint = widget.sizeHint() widget_width = size_hint.width() + total_margin_width widget_height = size_hint.height() + total_margin_height # 计算水平间距 if i > 0: x += self._hspace # 判断是否需要换行 if x + widget_width > available_width and x != rect.x(): x = rect.x() y += line_height + self._vspace line_height = 0 # 设置固定大小并放置组件 widget.setFixedSize(widget_width, widget_height) geom = QRect(QPoint(x, y), QSize(widget_width, widget_height)) widget.setGeometry(geom) layout_cache.append((widget, geom)) # 更新当前行的宽度和最大高度 x += widget_width line_height = max(line_height, widget_height) # 保存布局缓存 self._layout_cache = layout_cache # 强制更新布局和渲染 self.update() if self.parent(): self.parent().update() self.parent().repaint() finally: self._is_layouting = False def sizeHint(self): return self.minimumSize() def minimumSize(self): width = 0 height = 0 line_width = 0 line_height = 0 for item in self._items: widget = item.widget() if not widget or not widget.isVisible(): continue # 考虑内容边距 margins = widget.contentsMargins() total_margin_width = margins.left() + margins.right() total_margin_height = margins.top() + margins.bottom() size = widget.sizeHint() widget_width = size.width() + total_margin_width widget_height = size.height() + total_margin_height # 计算水平间距 if line_width > 0: line_width += self._hspace # 判断是否需要换行 if line_width + widget_width > self.parent().width(): width = max(width, line_width) height += line_height + self._vspace line_width = widget_width line_height = widget_height else: line_width += widget_width line_height = max(line_height, widget_height) # 计算最后一行 width = max(width, line_width) height += line_height return QSize(width, height) def horizontalSpacing(self): return self._hspace def verticalSpacing(self): return self._vspace def setSpacing(self, spacing): self._hspace = spacing self._vspace = spacing self._layout_cache = None self.invalidate() def spacing(self): return self.horizontalSpacing() class MainWindow(QMainWindow): currency_added = pyqtSignal(str) def __init__(self): super().__init__() self._global_currencies = ["EURUSD", "GBPUSD", "USDJPY", "XAUUSD"] self._currency_lock = threading.Lock() # 添加锁保护 self._layout_lock = False self._pending_deletions = set() self.cards = [] self.reference_card = None self.threshold = 20 self.setup_ui() self.setWindowTitle("多平台报价监控系统 版本:V1.0 作者:wbb") self.resize(1280, 720) self.setup_monitoring() self._queue_lock = False self._layout_timer = QTimer() self._layout_timer.setSingleShot(True) self._layout_timer.timeout.connect(self._safe_layout_update) self.start_status_monitor() def setup_ui(self): main_widget = QWidget() self.setCentralWidget(main_widget) main_layout = QHBoxLayout(main_widget) # 整体水平布局 main_layout.setContentsMargins(10, 10, 10, 10) # 左侧垂直布局 (控制栏 + 卡片区) left_vertical_layout = QVBoxLayout() # 控制栏 control_bar = QHBoxLayout() self.threshold_input = QLineEdit("20") self.threshold_input.setFixedWidth(80) self.monitor_switch = QCheckBox("实时监控") self.monitor_switch.setChecked(True) control_bar.addWidget(QLabel("异常阈值(点):")) control_bar.addWidget(self.threshold_input) control_bar.addWidget(self.monitor_switch) # 状态栏 self.status_bar = QStatusBar() self.status_bar.setSizeGripEnabled(False) status_widget = QWidget() status_layout = QHBoxLayout(status_widget) status_layout.setContentsMargins(0, 0, 0, 0) self.reference_info_label = QLabel("未设置参照平台") self.reference_info_label.setStyleSheet("color: #666666; padding: 0 10px;") self.reference_info_label.setFixedWidth(320) status_layout.addWidget(self.reference_info_label) self.performance_info_label = QLabel("内存占用: -- MB | 存活卡片: --") self.performance_info_label.setStyleSheet("color: #666666; padding: 0 10px;") self.performance_info_label.setFixedWidth(280) status_layout.addWidget(self.performance_info_label) self.status_bar.addWidget(status_widget) control_bar.addWidget(self.status_bar) left_vertical_layout.addLayout(control_bar) # 左侧卡片区域 left_card_widget = QWidget() left_card_layout = QVBoxLayout(left_card_widget) left_card_layout.setContentsMargins(0, 0, 0, 0) scroll = CardScrollArea() scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAsNeeded) scroll.setVerticalScrollBarPolicy(Qt.ScrollBarAsNeeded) scroll.setStyleSheet(""" QScrollArea { border: 1px solid #e0e0e0; border-radius: 4px; } """) self.scroll_content = QWidget() self.flow_layout = FlowLayout(self.scroll_content) self.scroll_content.setLayout(self.flow_layout) self.scroll_content.setSizePolicy( QSizePolicy.Expanding, QSizePolicy.MinimumExpanding ) scroll.setWidget(self.scroll_content) left_card_layout.addWidget(scroll) left_vertical_layout.addWidget(left_card_widget, 1) # 设置伸缩因子为1，占据剩余空间 # 右侧垂直布局 (按钮 + 日志区) right_vertical_layout = QVBoxLayout() # 添加按钮水平布局 button_layout = QHBoxLayout() button_layout.setAlignment(Qt.AlignLeft) # 按钮靠左排列 # 添加左侧弹簧 button_layout.addItem(QSpacerItem(0, 0, QSizePolicy.Expanding, QSizePolicy.Minimum)) self.add_card_btn = self.create_tool_button("＋ 添加卡片", "#0078D4") self.add_currency_btn = self.create_tool_button("＋ 添加品种", "#009966") self.clear_display_btn = self.create_tool_button("- 清除显示", "#D40000") # 调整按钮间距和边距 button_layout.setSpacing(10) button_layout.setContentsMargins(0, 0, 0, 10) # 底部留出间距 button_layout.addWidget(self.add_card_btn) button_layout.addWidget(self.add_currency_btn) button_layout.addWidget(self.clear_display_btn) # 添加右侧弹簧 button_layout.addItem(QSpacerItem(0, 0, QSizePolicy.Expanding, QSizePolicy.Minimum)) right_vertical_layout.addLayout(button_layout) # 右侧日志区域 right_log_widget = QWidget() right_log_layout = QVBoxLayout(right_log_widget) right_log_layout.setContentsMargins(0, 0, 0, 0) # 异常信息区域 abnormal_title_widget = QWidget() abnormal_title_layout = QHBoxLayout(abnormal_title_widget) abnormal_title_layout.setContentsMargins(0, 0, 0, 0) abnormal_label = QLabel("异常平台") abnormal_label.setStyleSheet(""" font-weight: bold; padding: 4px 12px; background-color: #f0f0f0; border-radius: 4px; border-left: 1px dashed #cccccc; border-right: 1px dashed #cccccc; """) abnormal_label.setAlignment(Qt.AlignCenter) abnormal_title_layout.addWidget(abnormal_label) abnormal_title_layout.addStretch(1) right_log_layout.addWidget(abnormal_title_widget) self.abnormal_area = QTextEdit() self.abnormal_area.setReadOnly(True) self.abnormal_area.setStyleSheet(""" QTextEdit { background: #fff0f0; border: 1px solid #e0e0e0; border-radius: 4px; padding: 8px; font-family: Consolas; font-size: 12px; } """) right_log_layout.addWidget(self.abnormal_area, 2) # 异常信息区占2份 # 系统日志区域 log_title_widget = QWidget() log_title_layout = QHBoxLayout(log_title_widget) log_title_layout.setContentsMargins(0, 0, 0, 0) log_label = QLabel("系统日志") log_label.setStyleSheet(""" font-weight: bold; padding: 4px 12px; background-color: #f0f0f0; border-radius: 4px; border-left: 1px dashed #cccccc; border-right: 1px dashed #cccccc; """) log_label.setAlignment(Qt.AlignCenter) log_title_layout.addWidget(log_label) log_title_layout.addStretch(1) right_log_layout.addWidget(log_title_widget) self.log_area = QTextEdit() self.log_area.setReadOnly(True) self.log_area.setStyleSheet(""" QTextEdit { background: #f8f8f8; border: 1px solid #e0e0e0; border-radius: 4px; padding: 8px; font-family: Consolas, monospace; font-size: 12px; color: #333333; line-height: 1.4; } """) right_log_layout.addWidget(self.log_area, 1) # 系统日志区占1份 right_vertical_layout.addWidget(right_log_widget, 1) # 设置伸缩因子为1，占据剩余空间 # 将左右两个垂直布局添加到整体水平布局 main_layout.addLayout(left_vertical_layout, 2) # 左侧占2份 main_layout.addLayout(right_vertical_layout, 1) # 右侧占1份 # 连接信号 self.add_card_btn.clicked.connect(self.add_card) self.add_currency_btn.clicked.connect(self.show_add_currency_dialog) self.threshold_input.textChanged.connect(self.update_threshold) self.currency_added.connect(self.add_global_currency) self.clear_display_btn.clicked.connect(self.clear_logs) def clear_logs(self): """清空异常平台和系统日志的显示""" self.abnormal_area.clear() self.log_area.clear() def create_tool_button(self, text, color): btn = QPushButton(text) btn.setStyleSheet(f""" QPushButton {{ background: {color}; color: white; border-radius: 4px; padding: 8px 12px; min-width: 100px; }} QPushButton:pressed {{ color: #D0D0D0; background: {QColor(color).darker(150).name()}; }} """) return btn def setup_monitoring(self): self.check_timer = QTimer() self.check_timer.timeout.connect(self.check_all_platforms) self.check_timer.start(1000) def check_perf(self): try: # 获取内存占用 process = psutil.Process() mem = process.memory_info().rss / 1024 / 1024 # 计算存活卡片数量 valid_cards = len([ref for ref in self.cards if ref() is not None]) # 更新状态栏性能信息 self.performance_info_label.setText(f"内存占用: {mem:.2f} MB | 存活卡片: {valid_cards}") except Exception as e: print(f"更新性能信息错误: {e}") # 出错时保持原显示 valid_cards = len([ref for ref in self.cards if ref() is not None]) self.performance_info_label.setText(f"内存占用: -- MB | 存活卡片: {valid_cards}") def update_threshold(self): try: self.threshold = int(self.threshold_input.text()) self.log(f"异常阈值更新为: {self.threshold}点") except: self.threshold = 20 self.threshold_input.setText("20") def handle_reference(self, symbol, bid, ask): # 清除旧参考样式 if self.reference_card and self.reference_card() and self.reference_card().is_reference: self.reference_card().ref_btn.setChecked(False) self.reference_card().clear_reference_style() # 设置新参考卡片 if symbol: current_card = self.sender() if current_card: current_card.ref_btn.setChecked(True) current_card.set_style(current_card.ref_style) current_card.is_reference = True self.reference_card = weakref.ref(current_card) # 更新参考平台信息 timestamp = datetime.now().strftime("%H:%M:%S") self.reference_info_label.setText(f"参照平台: {symbol} {bid:.5f}/{ask:.5f} [{timestamp}]") self.log(f"设置参考平台: {symbol}") else: self.reference_card = None self.reference_info_label.setText("未设置参照平台") self.log("取消参考平台设置") def add_card(self): """添加新的价格卡片""" card = PriceCard(currency_list=self._global_currencies) card.delete_requested.connect(self.remove_card) card.set_as_reference.connect(self.handle_reference) card.status_updated.connect(self.update_status) card.symbol_changed.connect(self.update_currencies) self.flow_layout.addWidget(card) self.cards.append(weakref.ref(card)) # 延迟更新布局，避免频繁计算 self._schedule_layout_update() # 记录日志 self.log(f"添加新卡片，当前总数: {len(self.cards)}") def remove_card(self, card_ref): """安全移除卡片""" card = card_ref() if card: # 从布局中移除 if self.flow_layout.indexOf(card) >= 0: self.flow_layout.removeWidget(card) card.hide() # 如果是参考卡片，清除参考状态 if self.reference_card and self.reference_card() == card: self.handle_reference("", 0, 0) # 延迟删除，确保布局更新完成 QTimer.singleShot(500, card.deleteLater) # 从卡片列表中移除 self.cards = [ref for ref in self.cards if ref() is not None] # 记录日志 self.log(f"移除卡片，当前总数: {len(self.cards)}") # 延迟更新布局 self._schedule_layout_update() def show_add_currency_dialog(self): """显示添加交易品种对话框""" dialog = QDialog(self) dialog.setWindowTitle("添加交易品种") dialog.setFixedSize(300, 150) layout = QVBoxLayout(dialog) input_label = QLabel("请输入新的交易品种:") input_field = QLineEdit() input_field.setPlaceholderText("例如: EURUSD") button_layout = QHBoxLayout() ok_button = QPushButton("确定") cancel_button = QPushButton("取消") button_layout.addWidget(ok_button) button_layout.addWidget(cancel_button) layout.addWidget(input_label) layout.addWidget(input_field) layout.addLayout(button_layout) ok_button.clicked.connect(lambda: self._handle_add_currency(input_field.text(), dialog)) cancel_button.clicked.connect(dialog.reject) dialog.exec_() def _handle_add_currency(self, currency, dialog): """处理添加交易品种""" if currency.strip(): currency = currency.strip().upper() if currency not in self._global_currencies: self.currency_added.emit(currency) self.log(f"添加新交易品种: {currency}") else: self.log(f"交易品种 '{currency}' 已存在") dialog.accept() def add_global_currency(self, currency): """添加全局交易品种""" with self._currency_lock: if currency not in self._global_currencies: self._global_currencies.append(currency) # 更新所有卡片的品种列表 for ref in self.cards: card = ref() if card: card.update_currencies(currency) def update_currencies(self, symbol): """更新全局交易品种列表""" if symbol and symbol not in self._global_currencies: with self._currency_lock: self._global_currencies.append(symbol) # 更新其他卡片的品种列表 for ref in self.cards: card = ref() if card and card.symbol_combo.currentText() != symbol: card.update_currencies(symbol) def check_all_platforms(self): """检查所有平台价格差异""" if not self.monitor_switch.isChecked() or not self.reference_card or not self.reference_card(): return ref_card = self.reference_card() ref_symbol = ref_card.symbol_combo.currentText() ref_bid = ref_card.current_bid ref_ask = ref_card.current_ask if ref_bid <= 0 or ref_ask <= 0: return self.abnormal_area.clear() abnormal_count = 0 for ref in self.cards: card = ref() if card and card != ref_card and card.isValid() and card._has_valid_prices: # 确保检查相同的交易品种 if card.symbol_combo.currentText() == ref_symbol: card_bid = card.current_bid card_ask = card.current_ask # 计算点差差异 (假设5位小数的品种) bid_diff = (card_bid - ref_bid) * 10000 ask_diff = (card_ask - ref_ask) * 10000 # 判断是否异常 is_abnormal = abs(bid_diff) > self.threshold or abs(ask_diff) > self.threshold card.update_abnormal_status(is_abnormal) if is_abnormal: abnormal_count += 1 platform = card.platform_input.text().split("·")[1].strip() self.abnormal_area.append( f"平台: {platform}\n" f"参考价格: {ref_bid:.5f} / {ref_ask:.5f}\n" f"当前价格: {card_bid:.5f} / {card_ask:.5f}\n" f"差异: {bid_diff:+.2f} / {ask_diff:+.2f} 点\n" f"{'=' * 30}\n" ) # 更新状态栏 if abnormal_count > 0: self.statusBar().showMessage(f"发现 {abnormal_count} 个异常平台", 3000) def log(self, message): """记录系统日志""" timestamp = datetime.now().strftime("%H:%M:%S") log_entry = f"[{timestamp}] {message}\n" self.log_area.insertPlainText(log_entry) # 自动滚动到底部 self.log_area.moveCursor(self.log_area.textCursor().End) def update_status(self, message): """更新状态栏消息""" self.statusBar().showMessage(message, 3000) def _schedule_layout_update(self): """安排布局更新，避免频繁计算""" if not self._layout_timer.isActive(): self._layout_timer.start(100) def _safe_layout_update(self): """安全地更新布局""" if self.flow_layout: self.flow_layout.invalidate() self.flow_layout._layout_cache = None # 清除缓存 self.scroll_content.adjustSize() self.flow_layout._do_layout(self.scroll_content.rect()) def start_status_monitor(self): """启动状态监控定时器""" self.performance_timer = QTimer() self.performance_timer.timeout.connect(self.check_perf) self.performance_timer.start(2000) # 每2秒更新一次 def closeEvent(self, event): """关闭应用程序时的清理工作""" # 停止所有定时器 if hasattr(self, 'check_timer') and self.check_timer.isActive(): self.check_timer.stop() if hasattr(self, 'performance_timer') and self.performance_timer.isActive(): self.performance_timer.stop() if hasattr(self, '_layout_timer') and self._layout_timer.isActive(): self._layout_timer.stop() # 停止所有MT5工作进程 for ref in self.cards: card = ref() if card and card.mt5_worker: card.mt5_worker.stop() # 确认所有进程已停止 time.sleep(0.5) event.accept() if __name__ == "__main__": freeze_support() # 支持多进程打包 app = QApplication(sys.argv) # 设置应用程序样式 app.setStyle("Fusion") # 设置全局字体 font = QFont("微软雅黑", 9) app.setFont(font) # 设置全局样式 app.setStyleSheet(""" QMainWindow, QWidget { background-color: #f8f9fa; } QLabel { color: #333333; } QPushButton { border-radius: 4px; padding: 6px 12px; background-color: #007bff; color: white; border: none; } QPushButton:hover { background-color: #0069d9; } QPushButton:pressed { background-color: #0056b3; } QScrollBar:vertical { width: 12px; background: #f0f0f0; } QScrollBar::handle:vertical { background: #c0c0c0; border-radius: 6px; } QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical { height: 0px; } """) window = MainWindow() window.show() sys.exit(app.exec_()) 添加卡片后选择mt5路径后，mt5连接成功并开始报价更新，但是platform_input中显示没有更新，还是显示正在连接

server="Demo", password="your_password" ) ``` **常见问题排查点**： - 确保`platform_input`对象在窗口初始化后正确创建 - 检查信号连接是否在窗口构造函数中完成 - 使用`print(mt5.last_error())`输出连接...
filetype

西门子利用QT搭载客户端与西门子通信_QT_西门子利用QT搭载客户端与西门子通信_

西门子
filetype

QT用SNAP7协议和西门子PLC 通讯实例源码.zip

【达摩老生出品，必属精品，亲测校正，质量保证】 资源名：QT用SNAP7协议和西门子PLC 通讯实例源码.zip 资源类型：程序源代码 源码说明：该程序是 QT用SNAP7写的 和西门子PLC 通讯的例子 ，可编辑。可打开。支持 对西门子PLC 任意区域 读写 bit,byte,word/Dword，Uint,float 等任意数据类型 。线程读取。是通讯的模板程序 适合人群：新手及有一定经验的开发人员
filetype

史上最强的YUV转换RenderScript，支持转换成RGBA,BGRA，并且同时支持旋转，翻转，YUV数据格式支持NV21, YV12, YUV420_888, 分别对应Android Camer

史上最强的YUV转换RenderScript，支持转换成RGBA,BGRA，并且同时支持旋转，翻转，YUV数据格式支持NV21, YV12, YUV420_888, 分别对应Android Camera1API 以及Camera2 API 的YUV处理 ​ 这段时间在做一些Android Camera数据处理的相关工作，涉及到YUV转RGBA的事情，由于对性能要求极高我先后采用了多种方案，最开...
filetype

byte-buddy-0.5.2.jar中文-英文对照文档.zip

1、压缩文件中包含： 中文-英文对照文档、jar包下载地址、Maven依赖、Gradle依赖、源代码......
嘻嘻666
  • 粉丝: 15
上传资源 快速赚钱

最新资源