clint/MqttAccess
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

创建项目框架

Browse Source
This commit is contained in:
clinton 2025-07-14 16:24:02 +08:00
parent d74d46705d
commit 94b7ceedc4
9 changed files with 194 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
src/mqtt_acess/config/settings.py
View File

@ -1,8 +1,42 @@
import json
import os
class Settings:
CONFIG_FILE = os.path.join(os.path.dirname(__file__), 'user_config.json')
MQTT_BROKER = "localhost"
MQTT_PORT = 1883
MQTT_USERNAME = ""
MQTT_PASSWORD = ""
MQTT_TLS = False
MQTT_TOPICS = [("test/topic", 0)]
MQTT_TOPICS = ["test/topic"]
PROTOCOL_TYPE = "json"
RESPONDER_TYPE = "alarm"
@classmethod
def load(cls):
if os.path.exists(cls.CONFIG_FILE):
with open(cls.CONFIG_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
cls.MQTT_BROKER = data.get('MQTT_BROKER', cls.MQTT_BROKER)
cls.MQTT_PORT = data.get('MQTT_PORT', cls.MQTT_PORT)
cls.MQTT_USERNAME = data.get('MQTT_USERNAME', cls.MQTT_USERNAME)
cls.MQTT_PASSWORD = data.get('MQTT_PASSWORD', cls.MQTT_PASSWORD)
cls.MQTT_TLS = data.get('MQTT_TLS', cls.MQTT_TLS)
cls.MQTT_TOPICS = data.get('MQTT_TOPICS', cls.MQTT_TOPICS)
cls.PROTOCOL_TYPE = data.get('PROTOCOL_TYPE', cls.PROTOCOL_TYPE)
cls.RESPONDER_TYPE = data.get('RESPONDER_TYPE', cls.RESPONDER_TYPE)
@classmethod
def save(cls):
data = {
'MQTT_BROKER': cls.MQTT_BROKER,
'MQTT_PORT': cls.MQTT_PORT,
'MQTT_USERNAME': cls.MQTT_USERNAME,
'MQTT_PASSWORD': cls.MQTT_PASSWORD,
'MQTT_TLS': cls.MQTT_TLS,
'MQTT_TOPICS': cls.MQTT_TOPICS,
'PROTOCOL_TYPE': cls.PROTOCOL_TYPE,
'RESPONDER_TYPE': cls.RESPONDER_TYPE
}
with open(cls.CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)

12
src/mqtt_acess/config/user_config.json Normal file
View File

@ -0,0 +1,12 @@
{
"MQTT_BROKER": "localhost",
"MQTT_PORT": 1883,
"MQTT_USERNAME": "",
"MQTT_PASSWORD": "",
"MQTT_TLS": false,
"MQTT_TOPICS": [
"test/topic"
],
"PROTOCOL_TYPE": "json",
"RESPONDER_TYPE": "alarm"
}

6
src/mqtt_acess/mqtt/client.py
View File

@ -29,5 +29,9 @@ class MqttClient:
self.client.on_message = callback
def on_connect(self, client, userdata, flags, rc):
for topic, qos in Settings.MQTT_TOPICS:
for item in Settings.MQTT_TOPICS:
if isinstance(item, (list, tuple)) and len(item) == 2:
topic, qos = item
else:
topic, qos = item, 0
client.subscribe(topic, qos)

12
src/mqtt_acess/parser/custom_parser.py Normal file
View File

@ -0,0 +1,12 @@
from .base import IParser
class CustomParser(IParser):
def parse(self, data: bytes) -> dict:
# 示例：自定义协议，假设用逗号分隔key:value
text = data.decode("utf-8")
result = {}
for item in text.split(","):
if ":" in item:
k, v = item.split(":", 1)
result[k.strip()] = v.strip()
return result

7
src/mqtt_acess/parser/factory.py
View File

@ -1,8 +1,13 @@
from .json_parser import JsonParser
from .xml_parser import XmlParser
class ParserFactory:
def get_parser(self, protocol_type: str):
if protocol_type == "json":
return JsonParser()
# 可扩展更多协议
elif protocol_type == "xml":
return XmlParser()
elif protocol_type == "custom":
from .custom_parser import CustomParser
return CustomParser()
raise ValueError(f"不支持的协议类型: {protocol_type}")

7
src/mqtt_acess/parser/xml_parser.py Normal file
View File

@ -0,0 +1,7 @@
import xml.etree.ElementTree as ET
from .base import IParser
class XmlParser(IParser):
def parse(self, data: bytes) -> dict:
root = ET.fromstring(data.decode("utf-8"))
return {child.tag: child.text for child in root}

7
src/mqtt_acess/responder/auto_reply.py Normal file
View File

@ -0,0 +1,7 @@
from .base import IResponder
class AutoReplyResponder(IResponder):
def respond(self, parsed_data: dict):
# 示例：自动回复，打印回复内容
if 'reply' in parsed_data:
print("自动回复：", parsed_data['reply'])

7
src/mqtt_acess/responder/device_control.py Normal file
View File

@ -0,0 +1,7 @@
from .base import IResponder
class DeviceControlResponder(IResponder):
def respond(self, parsed_data: dict):
# 示例：设备控制，打印控制命令
if 'control' in parsed_data:
print("设备控制命令：", parsed_data['control'])

113
src/mqtt_acess/ui/main_window.py
View File

@ -1,43 +1,134 @@
from PySide6.QtWidgets import QMainWindow, QTextEdit, QPushButton, QVBoxLayout, QWidget
from PySide6.QtWidgets import (QMainWindow, QTextEdit, QPushButton, QVBoxLayout, QWidget,
QLineEdit, QLabel, QComboBox, QHBoxLayout, QMessageBox)
from config.settings import Settings
from responder.alarm import AlarmResponder
from responder.auto_reply import AutoReplyResponder
from responder.device_control import DeviceControlResponder
class MainWindow(QMainWindow):
def __init__(self, mqtt_client, parser_factory, responder):
super().__init__()
self.setWindowTitle("MQTT在线测试平台")
self.resize(600, 400)
self.resize(700, 500)
self.mqtt_client = mqtt_client
self.parser = parser_factory.get_parser(Settings.PROTOCOL_TYPE)
self.parser_factory = parser_factory
self.responder = responder
self.connected = False
self.log = []
# 配置区
self.broker_edit = QLineEdit(Settings.MQTT_BROKER)
self.port_edit = QLineEdit(str(Settings.MQTT_PORT))
self.user_edit = QLineEdit(Settings.MQTT_USERNAME)
self.pass_edit = QLineEdit(Settings.MQTT_PASSWORD)
self.topic_edit = QLineEdit(",".join(Settings.MQTT_TOPICS))
self.protocol_combo = QComboBox()
self.protocol_combo.addItems(["json", "xml", "custom"])
self.protocol_combo.setCurrentText(Settings.PROTOCOL_TYPE)
self.responder_combo = QComboBox()
self.responder_combo.addItems(["alarm", "auto_reply", "device_control"])
self.responder_combo.setCurrentText(Settings.RESPONDER_TYPE)
# 状态与日志
self.status_label = QLabel("未连接")
self.text_edit = QTextEdit()
self.text_edit.setReadOnly(True)
# 按钮
self.btn_connect = QPushButton("连接MQTT")
self.btn_disconnect = QPushButton("断开MQTT")
self.btn_save = QPushButton("保存配置")
self.btn_connect.clicked.connect(self.connect_mqtt)
self.btn_disconnect.clicked.connect(self.disconnect_mqtt)
self.btn_save.clicked.connect(self.save_config)
# 布局
form_layout = QHBoxLayout()
form_layout.addWidget(QLabel("Broker:"))
form_layout.addWidget(self.broker_edit)
form_layout.addWidget(QLabel("端口:"))
form_layout.addWidget(self.port_edit)
form_layout.addWidget(QLabel("用户名:"))
form_layout.addWidget(self.user_edit)
form_layout.addWidget(QLabel("密码:"))
form_layout.addWidget(self.pass_edit)
form_layout.addWidget(QLabel("主题(逗号分隔):"))
form_layout.addWidget(self.topic_edit)
form_layout.addWidget(QLabel("协议类型:"))
form_layout.addWidget(self.protocol_combo)
form_layout.addWidget(QLabel("响应机制:"))
form_layout.addWidget(self.responder_combo)
form_layout.addWidget(self.btn_save)
btn_layout = QHBoxLayout()
btn_layout.addWidget(self.btn_connect)
btn_layout.addWidget(self.btn_disconnect)
btn_layout.addWidget(self.status_label)
layout = QVBoxLayout()
layout.addLayout(form_layout)
layout.addLayout(btn_layout)
layout.addWidget(self.text_edit)
layout.addWidget(self.btn_connect)
layout.addWidget(self.btn_disconnect)
container = QWidget()
container.setLayout(layout)
self.setCentralWidget(container)
def connect_mqtt(self):
self.apply_config()
self.mqtt_client.set_on_message(self.on_message)
self.mqtt_client.connect()
self.text_edit.append("已连接MQTT")
self.connected = True
self.status_label.setText("已连接")
self.append_log("已连接MQTT")
def disconnect_mqtt(self):
self.mqtt_client.disconnect()
self.text_edit.append("已断开MQTT")
self.connected = False
self.status_label.setText("未连接")
self.append_log("已断开MQTT")
def on_message(self, client, userdata, msg):
try:
parsed = self.parser.parse(msg.payload)
self.text_edit.append(f"收到消息: {parsed}")
self.responder.respond(parsed)
parser = self.parser_factory.get_parser(Settings.PROTOCOL_TYPE)
parsed = parser.parse(msg.payload)
self.append_log(f"收到消息: {parsed}")
responder = self.get_responder()
responder.respond(parsed)
except Exception as e:
self.text_edit.append(f"解析错误: {e}")
self.append_log(f"解析错误: {e}")
def append_log(self, text):
self.log.append(text)
self.text_edit.append(text)
def save_config(self):
Settings.MQTT_BROKER = self.broker_edit.text()
Settings.MQTT_PORT = int(self.port_edit.text())
Settings.MQTT_USERNAME = self.user_edit.text()
Settings.MQTT_PASSWORD = self.pass_edit.text()
Settings.MQTT_TOPICS = [t.strip() for t in self.topic_edit.text().split(",") if t.strip()]
Settings.PROTOCOL_TYPE = self.protocol_combo.currentText()
Settings.RESPONDER_TYPE = self.responder_combo.currentText()
Settings.save()
QMessageBox.information(self, "提示", "配置已保存！")
def apply_config(self):
self.save_config()
# 重新设置MQTT订阅主题等
self.mqtt_client.client.unsubscribe("#")
for topic in Settings.MQTT_TOPICS:
self.mqtt_client.subscribe(topic)
def get_responder(self):
t = Settings.RESPONDER_TYPE
if t == "alarm":
return AlarmResponder()
elif t == "auto_reply":
from responder.auto_reply import AutoReplyResponder
return AutoReplyResponder()
elif t == "device_control":
from responder.device_control import DeviceControlResponder
return DeviceControlResponder()
else:
return AlarmResponder()