feat: 实现 sensor_report 主动上报功能

vd960DBN:
- 新增 g_report_active 全局标志控制传感器数据主动上报
- handle_report_config: 解析 active_report JSON 字段,设置/清除标志
- tcp_json_push_sensor: 检查 g_report_active 开关,仅在启用时推送

DBNetClient:
- tcp_json_client.py: 新增 report_config(active_report) 方法
- main.py: 线圈标签页添加"启用主动上报"复选框
- main.py: 注册 sensor_report push 处理器,实时显示推送数据
This commit is contained in:
wangfq
2026-07-02 13:42:59 +08:00
parent 615b369690
commit 8526023e06
3 changed files with 96 additions and 20 deletions

View File

@@ -22,6 +22,7 @@ class DBNetApp:
self.root.minsize(800, 600)
self.client = DBNetClient(log_callback=self.log)
self.client.on_push("sensor_report", self.on_sensor_report)
self.client.on_push("loop_data", self.on_loop_data)
self.client.on_push("event_report", self.on_event_report)
@@ -196,31 +197,46 @@ class DBNetApp:
ttk.Separator(f, orient=tk.HORIZONTAL).grid(row=2, column=0, columnspan=2,
sticky=tk.EW, pady=8)
# Row 3: 灵敏度
ttk.Label(f, text="线圈灵敏度 (0x8A):", font=("", 10, "bold")).grid(row=3, column=0, sticky=tk.W)
sens_row = ttk.Frame(f)
sens_row.grid(row=4, column=0, columnspan=2, sticky=tk.W, pady=2)
ttk.Button(sens_row, text="读取灵敏度", command=self._do_loop_sens_read).pack(side=tk.LEFT, padx=2)
# Row 2.5: 主动上报
ttk.Label(f, text="主动上报 (sensor_report):", font=("", 10, "bold")).grid(row=3, column=0, sticky=tk.W)
report_row = ttk.Frame(f)
report_row.grid(row=4, column=0, columnspan=2, sticky=tk.W, pady=2)
self.report_var = tk.BooleanVar(value=False)
self.report_cb = ttk.Checkbutton(report_row, text="启用主动上报",
variable=self.report_var,
command=self._do_report_config)
self.report_cb.pack(side=tk.LEFT, padx=2)
self.report_status = ttk.Label(report_row, text="", foreground="gray")
self.report_status.pack(side=tk.LEFT, padx=5)
ttk.Separator(f, orient=tk.HORIZONTAL).grid(row=5, column=0, columnspan=2,
sticky=tk.EW, pady=8)
# Row 6: 参数
ttk.Label(f, text="线圈参数 (0x63/0x64):", font=("", 10, "bold")).grid(row=6, column=0, sticky=tk.W)
param_row = ttk.Frame(f)
param_row.grid(row=7, column=0, columnspan=2, sticky=tk.W, pady=2)
ttk.Button(param_row, text="查询参数", command=self._do_loop_param_query).pack(side=tk.LEFT, padx=2)
ttk.Label(param_row, text="参数设置通过 Raw JSON 标签页发送 loop_param_set。",
foreground="gray").pack(side=tk.LEFT, padx=10)
# Row 6: 灵敏度
ttk.Label(f, text="线圈灵敏度 (0x8A):", font=("", 10, "bold")).grid(row=6, column=0, sticky=tk.W)
sens_row = ttk.Frame(f)
sens_row.grid(row=7, column=0, columnspan=2, sticky=tk.W, pady=2)
ttk.Button(sens_row, text="读取灵敏度", command=self._do_loop_sens_read).pack(side=tk.LEFT, padx=2)
ttk.Separator(f, orient=tk.HORIZONTAL).grid(row=8, column=0, columnspan=2,
sticky=tk.EW, pady=8)
# Row 9: 参数
ttk.Label(f, text="线圈参数 (0x63/0x64):", font=("", 10, "bold")).grid(row=9, column=0, sticky=tk.W)
param_row = ttk.Frame(f)
param_row.grid(row=10, column=0, columnspan=2, sticky=tk.W, pady=2)
ttk.Button(param_row, text="查询参数", command=self._do_loop_param_query).pack(side=tk.LEFT, padx=2)
ttk.Label(param_row, text="参数设置通过 Raw JSON 标签页发送 loop_param_set。",
foreground="gray").pack(side=tk.LEFT, padx=10)
ttk.Separator(f, orient=tk.HORIZONTAL).grid(row=11, column=0, columnspan=2,
sticky=tk.EW, pady=8)
# Response display
self.loop_text = scrolledtext.ScrolledText(f, height=14, wrap=tk.WORD,
self.loop_text = scrolledtext.ScrolledText(f, height=12, wrap=tk.WORD,
font=("Consolas", 9))
self.loop_text.grid(row=9, column=0, columnspan=2, sticky="nsew")
f.rowconfigure(9, weight=1)
self.loop_text.grid(row=12, column=0, columnspan=2, sticky="nsew")
f.rowconfigure(12, weight=1)
f.columnconfigure(0, weight=1)
return f
@@ -516,12 +532,49 @@ class DBNetApp:
self.raw_resp.insert(tk.END, f"Parse error: {e}\n")
# Push handlers
def on_sensor_report(self, cmd, data):
"""接收设备主动上报的传感器数据"""
import json as _json
text = _json.dumps(data, indent=2, ensure_ascii=False)
self.log(f"<<< PUSH sensor_report")
# 显示在线圈参数标签页
self.root.after(0, self._show_sensor_push, text)
def on_loop_data(self, cmd, data):
self.log(f"<<< PUSH loop_data: {data}")
def on_event_report(self, cmd, data):
self.log(f"<<< PUSH event_report: {data}")
def _show_sensor_push(self, text):
"""在线圈标签页追加传感器推送数据"""
import time
self.loop_text.insert(tk.END,
f"--- PUSH {time.strftime('%H:%M:%S')} ---\n{text}\n\n")
self.loop_text.see(tk.END)
def _do_report_config(self):
"""上报开关切换"""
enabled = self.report_var.get()
self._bg_run(
lambda: self.client.report_config(enabled),
lambda r: self._show_report_config_result(r, enabled))
def _show_report_config_result(self, resp, enabled):
if isinstance(resp, Exception):
self.report_status.config(text=str(resp), foreground="red")
self.report_var.set(not enabled) # revert
else:
code = resp.get("code", -1)
if code == 0:
self.report_status.config(
text="✓ 已启用" if enabled else "✓ 已关闭",
foreground="green")
else:
self.report_status.config(
text=f"{resp.get('msg','')}", foreground="red")
self.report_var.set(not enabled) # revert
def main():
root = tk.Tk()