876 lines
37 KiB
Python
876 lines
37 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
TEM 下位机模拟器 (GUI 版)
|
||
|
||
依赖: Python 3.8+ 标准库 (tkinter, socket, threading, struct, math, random)
|
||
用法: python tools/mock_device_gui.py [--port 4321]
|
||
"""
|
||
|
||
import tkinter as tk
|
||
from tkinter import ttk, scrolledtext
|
||
import socket
|
||
import struct
|
||
import threading
|
||
import time
|
||
import math
|
||
import random
|
||
import argparse
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
# ── 协议常量 ──────────────────────────────────────────────────────────────────
|
||
|
||
MAGIC = b'\x68\x68\xff\xff'
|
||
FRAME_FLAG = 0xFE
|
||
|
||
class FC:
|
||
SETUP_REQ = 0x01
|
||
CONTINUOUS_REQ = 0x02
|
||
SINGLE_REQ = 0x03
|
||
STOP_REQ = 0x04
|
||
ACTIVE_REQ = 0x05
|
||
SPLITFRAME_REQ = 0x08
|
||
SETUP_ACK = 0x81
|
||
CONTINUOUS_ACK = 0x82
|
||
SINGLE_ACK = 0x83
|
||
STOP_ACK = 0x84
|
||
DATA_ACK = 0x85
|
||
|
||
SEND_FREQ_OPTIONS = [
|
||
(0x00, '0.5 Hz'), (0x01, '1 Hz'), (0x02, '2 Hz'), (0x03, '4 Hz'),
|
||
(0x04, '8 Hz'), (0x05, '12.5 Hz'), (0x06, '16 Hz'), (0x07, '25 Hz'),
|
||
(0x08, '32 Hz'), (0x09, '50 Hz'), (0x0a, '64 Hz'),
|
||
]
|
||
SEND_FREQ_HZ = {c: float(lbl.split()[0]) for c, lbl in SEND_FREQ_OPTIONS}
|
||
AMP_GAIN = [0.125, 0.25, 0.5, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 1.0]
|
||
AMP_LABELS = ['1/8×','1/4×','1/2×','1×','2×','4×','8×','16×','32×','64×','128×','自动']
|
||
SIGNAL_TYPES = ['正弦衰减', '指数衰减', '正弦波', '阶跃响应', '纯噪声', '零信号']
|
||
CH_COLORS = ['#e05050', '#44aa44', '#4466ff', '#dd9900', '#9944dd', '#00aacc']
|
||
|
||
# ── 信号生成 ──────────────────────────────────────────────────────────────────
|
||
|
||
_CH_DECAY = [0.5, 0.8, 1.2, 0.3, 1.8, 0.7]
|
||
_CH_FREQ = [1.0, 2.0, 3.0, 5.0, 7.0, 11.0]
|
||
|
||
def gen_signal(sig_type: int, ch: int, s: int, total: int, phase: float) -> float:
|
||
"""归一化信号值 -1..1"""
|
||
if total <= 0:
|
||
return 0.0
|
||
t = s / total # 0..1 归一化时间
|
||
dk = _CH_DECAY[ch % len(_CH_DECAY)]
|
||
freq = _CH_FREQ [ch % len(_CH_FREQ)]
|
||
|
||
if sig_type == 0: # 正弦衰减
|
||
return math.exp(-t * dk * 8) * math.sin(2 * math.pi * (freq * t + phase * freq))
|
||
elif sig_type == 1: # 指数衰减
|
||
return math.exp(-t * dk * 8) * math.cos(2 * math.pi * phase * 0.3)
|
||
elif sig_type == 2: # 正弦
|
||
return math.sin(2 * math.pi * (freq * t + phase * freq))
|
||
elif sig_type == 3: # 阶跃响应
|
||
rise = 1.0 - math.exp(-t * 20)
|
||
return rise * math.exp(-t * dk * 5)
|
||
elif sig_type == 4: # 纯噪声
|
||
return random.gauss(0, 0.3)
|
||
else: # 零
|
||
return 0.0
|
||
|
||
# ── 帧构建 ──────────────────────────────────────────────────────────────────
|
||
|
||
def _pack_frame(func: int, payload: bytes) -> bytes:
|
||
return MAGIC + struct.pack('<BBI', FRAME_FLAG, func, len(payload)) + payload
|
||
|
||
def build_ack(func_ack: int, ok: bool = True) -> bytes:
|
||
p = bytearray(54)
|
||
p[0] = 0x01 if ok else 0x02
|
||
return _pack_frame(func_ack, bytes(p))
|
||
|
||
def build_data_frame(snap: dict, phase: float) -> bytes:
|
||
ch = snap['channelNum']
|
||
depth = snap['sampleDepth']
|
||
amp = snap['ampRatio']
|
||
acc = max(snap['accNum'], 1)
|
||
gain = AMP_GAIN[amp] if amp < len(AMP_GAIN) else 1.0
|
||
sig = snap['signalType']
|
||
sig_uv = snap['signalUV']
|
||
noise = snap['noiseRatio']
|
||
|
||
meta = bytearray(54)
|
||
meta[0] = 1
|
||
struct.pack_into('<I', meta, 1, int(time.time()))
|
||
struct.pack_into('<d', meta, 5, snap['longitude'])
|
||
struct.pack_into('<d', meta, 13, snap['latitude'])
|
||
struct.pack_into('<f', meta, 21, snap['altitude'])
|
||
struct.pack_into('<f', meta, 25, 1.0)
|
||
meta[29] = (snap['sdStatus'] & 0xF) | ((snap['gpsStatus'] & 0xF) << 4)
|
||
meta[30] = amp & 0xFF
|
||
struct.pack_into('<f', meta, 31, snap['roll'])
|
||
struct.pack_into('<f', meta, 35, snap['pitch'])
|
||
struct.pack_into('<f', meta, 39, snap['yaw'])
|
||
meta[43] = ch & 0xFF
|
||
struct.pack_into('<I', meta, 44, 1000)
|
||
struct.pack_into('<H', meta, 48, snap['tempRaw'])
|
||
struct.pack_into('<H', meta, 50, snap['battRaw'])
|
||
meta[52] = snap['sourceMode'] & 0xFF
|
||
|
||
adc = bytearray(ch * depth * 4)
|
||
for c in range(ch):
|
||
for s in range(depth):
|
||
v = gen_signal(sig, c, s, depth, phase)
|
||
v += random.gauss(0, noise) if noise > 0 else 0
|
||
raw = int(v * sig_uv * gain * acc)
|
||
raw = max(-0x80000000, min(0x7FFFFFFF, raw))
|
||
struct.pack_into('<i', adc, (c * depth + s) * 4, raw)
|
||
|
||
payload = bytes(meta) + bytes(adc)
|
||
return _pack_frame(FC.DATA_ACK, payload)
|
||
|
||
# ── 帧解析器 ──────────────────────────────────────────────────────────────────
|
||
|
||
class FrameParser:
|
||
def __init__(self):
|
||
self.buf = bytearray()
|
||
|
||
def feed(self, data: bytes):
|
||
self.buf += data
|
||
frames = []
|
||
while len(self.buf) >= 10:
|
||
idx = next((i for i in range(len(self.buf) - 3)
|
||
if self.buf[i:i+4] == b'\x68\x68\xff\xff'), -1)
|
||
if idx == -1:
|
||
self.buf = bytearray(self.buf[-3:]) if len(self.buf) >= 3 else bytearray()
|
||
break
|
||
if idx > 0:
|
||
self.buf = self.buf[idx:]
|
||
if len(self.buf) < 10:
|
||
break
|
||
plen = struct.unpack_from('<I', self.buf, 6)[0]
|
||
if plen > 400 * 1024:
|
||
self.buf = self.buf[4:]
|
||
continue
|
||
total = 10 + plen
|
||
if len(self.buf) < total:
|
||
break
|
||
frames.append((self.buf[5], bytes(self.buf[10:total])))
|
||
self.buf = self.buf[total:]
|
||
return frames
|
||
|
||
# ── 模拟配置（主线程写，采集线程读） ────────────────────────────────────────────
|
||
|
||
class SimConfig:
|
||
def __init__(self):
|
||
self.channel_num = 6
|
||
self.send_freq = 0x01
|
||
self.sample_depth = 256
|
||
self.acc_num = 1
|
||
self.amp_ratio = 3
|
||
self.source_mode = 0
|
||
self.signal_type = 0
|
||
self.signal_uv = 5e5
|
||
self.noise_ratio = 0.02
|
||
self.lat = 39.9086
|
||
self.lon = 116.3972
|
||
self.alt = 50.0
|
||
self.gps_walk = False
|
||
self.gps_status = 1
|
||
self.sd_status = 1
|
||
self.roll = 0.0
|
||
self.pitch = 2.5
|
||
self.yaw = 0.0
|
||
self.batt_raw = 3760
|
||
self.temp_raw = 2100
|
||
self.batt_drain = False
|
||
self.frame_count = 0
|
||
self.bytes_sent = 0
|
||
|
||
def snapshot(self) -> dict:
|
||
return {
|
||
'channelNum': self.channel_num, 'sendFreq': self.send_freq,
|
||
'sampleDepth': self.sample_depth, 'accNum': self.acc_num,
|
||
'ampRatio': self.amp_ratio, 'sourceMode': self.source_mode,
|
||
'signalType': self.signal_type, 'signalUV': self.signal_uv,
|
||
'noiseRatio': self.noise_ratio,
|
||
'latitude': self.lat, 'longitude': self.lon, 'altitude': self.alt,
|
||
'gpsStatus': self.gps_status, 'sdStatus': self.sd_status,
|
||
'roll': self.roll, 'pitch': self.pitch, 'yaw': self.yaw,
|
||
'tempRaw': self.temp_raw, 'battRaw': self.batt_raw,
|
||
}
|
||
|
||
# ── 客户端会话 ─────────────────────────────────────────────────────────────────
|
||
|
||
class Session:
|
||
def __init__(self, sock: socket.socket, addr, cfg: SimConfig, on_event):
|
||
self.sock = sock
|
||
self.addr = addr
|
||
self.cfg = cfg
|
||
self.on_event = on_event
|
||
self.parser = FrameParser()
|
||
self.running = False
|
||
self.phase = 0.0
|
||
|
||
def _ev(self, kind: str, data=None):
|
||
self.on_event(kind, data)
|
||
|
||
def _log(self, msg: str, tag: str = 'info'):
|
||
self._ev('log', (tag, f'{self.addr[0]}:{self.addr[1]} | {msg}'))
|
||
|
||
def _send(self, data: bytes) -> bool:
|
||
try:
|
||
self.sock.sendall(data)
|
||
self.cfg.bytes_sent += len(data)
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
# ── 连续发送线程 ──
|
||
|
||
def _data_loop(self):
|
||
self._log('开始连续推送', 'send')
|
||
while self.running:
|
||
t0 = time.monotonic()
|
||
snap = self.cfg.snapshot()
|
||
hz = SEND_FREQ_HZ.get(snap['sendFreq'], 1.0)
|
||
ivl = 1.0 / hz
|
||
|
||
frame = build_data_frame(snap, self.phase)
|
||
if not self._send(frame):
|
||
break
|
||
|
||
self.cfg.frame_count += 1
|
||
self.phase += ivl
|
||
|
||
if self.cfg.gps_walk:
|
||
self.cfg.lon += random.gauss(0, 5e-6)
|
||
self.cfg.lat += random.gauss(0, 3e-6)
|
||
if self.cfg.batt_drain and self.cfg.batt_raw > 2800:
|
||
if self.cfg.frame_count % 20 == 0:
|
||
self.cfg.batt_raw -= 1
|
||
|
||
self._ev('frame', {'ch': snap['channelNum'], 'depth': snap['sampleDepth'],
|
||
'bytes': len(frame) - 10, 'hz': hz})
|
||
|
||
elapsed = time.monotonic() - t0
|
||
if ivl - elapsed > 0.002:
|
||
time.sleep(ivl - elapsed)
|
||
|
||
self._log('连续推送已停止', 'info')
|
||
self._ev('acq_status', '已停止')
|
||
|
||
def start_continuous(self):
|
||
if self.running:
|
||
return
|
||
self.running = True
|
||
threading.Thread(target=self._data_loop, daemon=True).start()
|
||
self._ev('acq_status', '连续推送')
|
||
|
||
def stop_continuous(self):
|
||
self.running = False
|
||
|
||
# ── 帧处理 ──
|
||
|
||
def _parse_setup(self, payload: bytes):
|
||
if len(payload) < 8:
|
||
return
|
||
self.cfg.channel_num = max(1, min(6, payload[0]))
|
||
self.cfg.send_freq = payload[1]
|
||
depth = struct.unpack_from('<H', payload, 3)[0]
|
||
self.cfg.sample_depth = depth if depth > 0 else 256
|
||
acc_flags = struct.unpack_from('<H', payload, 5)[0]
|
||
acc = (acc_flags >> 2) & 0x3FFF
|
||
self.cfg.acc_num = acc if acc > 0 else 1
|
||
self.cfg.amp_ratio = payload[7] if payload[7] < len(AMP_GAIN) else 3
|
||
if len(payload) > 13:
|
||
self.cfg.source_mode = payload[13]
|
||
hz = SEND_FREQ_HZ.get(self.cfg.send_freq, 1.0)
|
||
self._log(f'SETUP ch={self.cfg.channel_num} freq={hz}Hz '
|
||
f'depth={self.cfg.sample_depth} acc={self.cfg.acc_num} '
|
||
f'amp={AMP_LABELS[self.cfg.amp_ratio]}', 'recv')
|
||
self._ev('config_changed', None)
|
||
|
||
def handle(self, func: int, payload: bytes):
|
||
if func == FC.SETUP_REQ:
|
||
self._parse_setup(payload)
|
||
self._send(build_ack(FC.SETUP_ACK))
|
||
self._log('SETUP_ACK →', 'send')
|
||
elif func == FC.CONTINUOUS_REQ:
|
||
self._log('CONTINUOUS_REQ ←', 'recv')
|
||
self._send(build_ack(FC.CONTINUOUS_ACK))
|
||
self._log('CONTINUOUS_ACK →', 'send')
|
||
self.start_continuous()
|
||
elif func == FC.SINGLE_REQ:
|
||
self._log('SINGLE_REQ ←', 'recv')
|
||
self._send(build_ack(FC.SINGLE_ACK))
|
||
frame = build_data_frame(self.cfg.snapshot(), self.phase)
|
||
self._send(frame)
|
||
self.cfg.frame_count += 1
|
||
self.phase += 1.0
|
||
self._log(f'DATA_ACK (单次) {len(frame)-10}B →', 'send')
|
||
self._ev('acq_status', '单次完成')
|
||
elif func == FC.STOP_REQ:
|
||
self._log('STOP_REQ ←', 'recv')
|
||
self.stop_continuous()
|
||
self._send(build_ack(FC.STOP_ACK))
|
||
self._log('STOP_ACK →', 'send')
|
||
elif func == FC.ACTIVE_REQ:
|
||
pass
|
||
elif func == FC.SPLITFRAME_REQ:
|
||
self._log('SPLITFRAME_REQ (忽略) ←', 'recv')
|
||
else:
|
||
self._log(f'未知 0x{func:02X} ←', 'info')
|
||
|
||
def run(self):
|
||
self._log('已连接 ✓', 'info')
|
||
self._ev('connected', f'{self.addr[0]}:{self.addr[1]}')
|
||
try:
|
||
while True:
|
||
data = self.sock.recv(4096)
|
||
if not data:
|
||
break
|
||
for func, payload in self.parser.feed(data):
|
||
self.handle(func, payload)
|
||
except OSError:
|
||
pass
|
||
finally:
|
||
self.stop_continuous()
|
||
self.sock.close()
|
||
self._log('已断开 ✗', 'info')
|
||
self._ev('disconnected', f'{self.addr[0]}:{self.addr[1]}')
|
||
|
||
# ── GUI 主窗口 ─────────────────────────────────────────────────────────────────
|
||
|
||
class MockDeviceApp:
|
||
def __init__(self, root: tk.Tk, port: int = 4321):
|
||
self.root = root
|
||
self.cfg = SimConfig()
|
||
self.session: Optional[Session] = None
|
||
self._srv_sock: Optional[socket.socket] = None
|
||
self._port = port
|
||
|
||
root.title('TEM 下位机模拟器')
|
||
root.minsize(860, 560)
|
||
|
||
style = ttk.Style()
|
||
for theme in ('clam', 'alt', 'default'):
|
||
try:
|
||
style.theme_use(theme)
|
||
break
|
||
except tk.TclError:
|
||
pass
|
||
|
||
self._build_ui()
|
||
self._start_server()
|
||
root.after(600, self._refresh) # first redraw after window is sized
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# UI 构建
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def _build_ui(self):
|
||
r = self.root
|
||
|
||
# ── 顶部状态栏
|
||
top = ttk.Frame(r, padding=(6, 3))
|
||
top.pack(fill='x', side='top')
|
||
|
||
self._sv_var = tk.StringVar(value='启动中…')
|
||
self._conn_var = tk.StringVar(value='未连接')
|
||
self._acq_var = tk.StringVar(value='已停止')
|
||
|
||
ttk.Label(top, text='监听:', foreground='gray').pack(side='left')
|
||
ttk.Label(top, textvariable=self._sv_var, foreground='#666').pack(side='left', padx=(2, 12))
|
||
|
||
ttk.Label(top, text='客户端:', foreground='gray').pack(side='left')
|
||
self._conn_lbl = ttk.Label(top, textvariable=self._conn_var, foreground='#999')
|
||
self._conn_lbl.pack(side='left', padx=(2, 12))
|
||
|
||
ttk.Label(top, text='采集:', foreground='gray').pack(side='left')
|
||
self._acq_lbl = ttk.Label(top, textvariable=self._acq_var, foreground='#999')
|
||
self._acq_lbl.pack(side='left', padx=(2, 0))
|
||
|
||
ttk.Separator(r, orient='horizontal').pack(fill='x')
|
||
|
||
# ── 主体
|
||
body = tk.Frame(r)
|
||
body.pack(fill='both', expand=True, padx=6, pady=4)
|
||
|
||
# Left panel (fixed width)
|
||
left = ttk.Frame(body, width=270)
|
||
left.pack(side='left', fill='y', padx=(0, 6))
|
||
left.pack_propagate(False)
|
||
|
||
# Right panel
|
||
right = ttk.Frame(body)
|
||
right.pack(side='left', fill='both', expand=True)
|
||
|
||
self._build_signal_frame(left)
|
||
self._build_position_frame(left)
|
||
self._build_device_frame(left)
|
||
self._build_right(right)
|
||
|
||
# ── 底部统计栏
|
||
ttk.Separator(r, orient='horizontal').pack(fill='x')
|
||
bot = ttk.Frame(r, padding=(6, 2))
|
||
bot.pack(fill='x', side='bottom')
|
||
self._stat_var = tk.StringVar(value='帧数: 0 | 发送: 0 B')
|
||
ttk.Label(bot, textvariable=self._stat_var, foreground='gray',
|
||
font=('Consolas', 9)).pack(side='left')
|
||
|
||
ttk.Button(bot, text='重置统计', command=self._reset_stats,
|
||
width=8).pack(side='right')
|
||
|
||
def _build_signal_frame(self, parent):
|
||
f = ttk.LabelFrame(parent, text='信号配置', padding=6)
|
||
f.pack(fill='x', padx=2, pady=(0, 4))
|
||
|
||
def row(label, widget_factory, r):
|
||
ttk.Label(f, text=label).grid(row=r, column=0, sticky='w', pady=1)
|
||
w = widget_factory(f)
|
||
w.grid(row=r, column=1, sticky='we', padx=(4, 0), pady=1)
|
||
return w
|
||
|
||
# Signal type
|
||
self._sig_var = tk.StringVar(value=SIGNAL_TYPES[0])
|
||
cb = ttk.Combobox(f, textvariable=self._sig_var, values=SIGNAL_TYPES,
|
||
state='readonly', width=14)
|
||
ttk.Label(f, text='信号类型').grid(row=0, column=0, sticky='w', pady=1)
|
||
cb.grid(row=0, column=1, sticky='we', padx=(4, 0), pady=1)
|
||
cb.bind('<<ComboboxSelected>>', lambda _: self._apply_signal())
|
||
|
||
# Channel count
|
||
self._ch_var = tk.IntVar(value=6)
|
||
sp = ttk.Spinbox(f, from_=1, to=6, textvariable=self._ch_var, width=5,
|
||
command=self._apply_signal)
|
||
ttk.Label(f, text='通道数').grid(row=1, column=0, sticky='w', pady=1)
|
||
sp.grid(row=1, column=1, sticky='w', padx=(4, 0), pady=1)
|
||
|
||
# Send freq
|
||
self._freq_var = tk.StringVar(value='1 Hz')
|
||
cb2 = ttk.Combobox(f, textvariable=self._freq_var,
|
||
values=[lbl for _, lbl in SEND_FREQ_OPTIONS],
|
||
state='readonly', width=10)
|
||
ttk.Label(f, text='发送频率').grid(row=2, column=0, sticky='w', pady=1)
|
||
cb2.grid(row=2, column=1, sticky='w', padx=(4, 0), pady=1)
|
||
cb2.bind('<<ComboboxSelected>>', lambda _: self._apply_signal())
|
||
|
||
# Sample depth
|
||
self._depth_var = tk.IntVar(value=256)
|
||
ttk.Label(f, text='采样深度').grid(row=3, column=0, sticky='w', pady=1)
|
||
ttk.Spinbox(f, from_=16, to=8192, increment=16, textvariable=self._depth_var,
|
||
width=7, command=self._apply_signal).grid(
|
||
row=3, column=1, sticky='w', padx=(4, 0), pady=1)
|
||
|
||
# Amplitude
|
||
self._amp_uv_var = tk.StringVar(value='500000')
|
||
ttk.Label(f, text='幅度 (μV)').grid(row=4, column=0, sticky='w', pady=1)
|
||
ttk.Entry(f, textvariable=self._amp_uv_var, width=10).grid(
|
||
row=4, column=1, sticky='w', padx=(4, 0), pady=1)
|
||
|
||
# Noise %
|
||
self._noise_var = tk.StringVar(value='2')
|
||
ttk.Label(f, text='噪声 (%)').grid(row=5, column=0, sticky='w', pady=1)
|
||
ttk.Entry(f, textvariable=self._noise_var, width=6).grid(
|
||
row=5, column=1, sticky='w', padx=(4, 0), pady=1)
|
||
|
||
# Amp ratio
|
||
self._amp_ratio_var = tk.StringVar(value='1×')
|
||
cb3 = ttk.Combobox(f, textvariable=self._amp_ratio_var, values=AMP_LABELS,
|
||
state='readonly', width=8)
|
||
ttk.Label(f, text='增益倍率').grid(row=6, column=0, sticky='w', pady=1)
|
||
cb3.grid(row=6, column=1, sticky='w', padx=(4, 0), pady=1)
|
||
cb3.bind('<<ComboboxSelected>>', lambda _: self._apply_signal())
|
||
|
||
# Acc num
|
||
self._acc_var = tk.IntVar(value=1)
|
||
ttk.Label(f, text='叠加次数').grid(row=7, column=0, sticky='w', pady=1)
|
||
ttk.Spinbox(f, from_=1, to=512, textvariable=self._acc_var, width=6,
|
||
command=self._apply_signal).grid(
|
||
row=7, column=1, sticky='w', padx=(4, 0), pady=1)
|
||
|
||
ttk.Button(f, text='应用', command=self._apply_signal).grid(
|
||
row=8, column=1, sticky='e', pady=(4, 0))
|
||
f.columnconfigure(1, weight=1)
|
||
|
||
def _build_position_frame(self, parent):
|
||
f = ttk.LabelFrame(parent, text='位置与姿态', padding=6)
|
||
f.pack(fill='x', padx=2, pady=(0, 4))
|
||
|
||
self._lat_var = tk.StringVar(value='39.9086')
|
||
self._lon_var = tk.StringVar(value='116.3972')
|
||
self._alt_var = tk.StringVar(value='50.0')
|
||
self._pitch_var = tk.StringVar(value='2.5')
|
||
self._roll_var = tk.StringVar(value='0.0')
|
||
|
||
fields = [
|
||
('纬度', self._lat_var), ('经度', self._lon_var),
|
||
('海拔 (m)', self._alt_var), ('俯仰角 (°)', self._pitch_var),
|
||
('侧倾角 (°)', self._roll_var),
|
||
]
|
||
for r, (lbl, var) in enumerate(fields):
|
||
ttk.Label(f, text=lbl).grid(row=r, column=0, sticky='w', pady=1)
|
||
ttk.Entry(f, textvariable=var, width=12).grid(
|
||
row=r, column=1, sticky='we', padx=(4, 0), pady=1)
|
||
|
||
self._gps_walk_var = tk.BooleanVar(value=False)
|
||
ttk.Checkbutton(f, text='GPS 走点 (小幅随机漂移)',
|
||
variable=self._gps_walk_var,
|
||
command=self._apply_pos).grid(
|
||
row=len(fields), column=0, columnspan=2, sticky='w', pady=(4, 0))
|
||
|
||
ttk.Button(f, text='应用', command=self._apply_pos).grid(
|
||
row=len(fields) + 1, column=1, sticky='e', pady=(4, 0))
|
||
f.columnconfigure(1, weight=1)
|
||
|
||
def _build_device_frame(self, parent):
|
||
f = ttk.LabelFrame(parent, text='设备状态', padding=6)
|
||
f.pack(fill='x', padx=2, pady=(0, 4))
|
||
|
||
self._batt_var = tk.StringVar(value='3760')
|
||
self._temp_var = tk.StringVar(value='2100')
|
||
self._batt_drain_var = tk.BooleanVar(value=False)
|
||
self._gps_st_var = tk.BooleanVar(value=True)
|
||
self._sd_st_var = tk.BooleanVar(value=True)
|
||
|
||
ttk.Label(f, text='电池 raw').grid(row=0, column=0, sticky='w', pady=1)
|
||
ttk.Entry(f, textvariable=self._batt_var, width=8).grid(
|
||
row=0, column=1, sticky='w', padx=(4, 0))
|
||
ttk.Checkbutton(f, text='模拟耗电', variable=self._batt_drain_var,
|
||
command=self._apply_dev).grid(row=0, column=2, sticky='w', padx=4)
|
||
|
||
ttk.Label(f, text='温度 raw').grid(row=1, column=0, sticky='w', pady=1)
|
||
ttk.Entry(f, textvariable=self._temp_var, width=8).grid(
|
||
row=1, column=1, sticky='w', padx=(4, 0))
|
||
|
||
ttk.Checkbutton(f, text='GPS 有效', variable=self._gps_st_var,
|
||
command=self._apply_dev).grid(row=2, column=0, sticky='w', pady=(4,0))
|
||
ttk.Checkbutton(f, text='SD 就绪', variable=self._sd_st_var,
|
||
command=self._apply_dev).grid(row=2, column=1, sticky='w', pady=(4,0))
|
||
|
||
btn_row = ttk.Frame(f)
|
||
btn_row.grid(row=3, column=0, columnspan=3, sticky='e', pady=(4, 0))
|
||
ttk.Button(btn_row, text='充满电', width=7,
|
||
command=lambda: (self._batt_var.set('4200'),
|
||
setattr(self.cfg, 'batt_raw', 4200))).pack(side='left', padx=2)
|
||
ttk.Button(btn_row, text='应用', width=7,
|
||
command=self._apply_dev).pack(side='left')
|
||
|
||
def _build_right(self, parent):
|
||
# ── 波形预览
|
||
wf = ttk.LabelFrame(parent, text='波形预览', padding=4)
|
||
wf.pack(fill='x', padx=2, pady=(0, 4))
|
||
|
||
self._wc = tk.Canvas(wf, height=140, bg='#0a0a14', highlightthickness=1,
|
||
highlightbackground='#222')
|
||
self._wc.pack(fill='x')
|
||
self._wc.bind('<Configure>', lambda _: self._redraw_waveform())
|
||
|
||
# ── 通信日志
|
||
lf = ttk.LabelFrame(parent, text='通信日志', padding=4)
|
||
lf.pack(fill='both', expand=True, padx=2)
|
||
|
||
self._log = scrolledtext.ScrolledText(
|
||
lf, height=12, font=('Consolas', 9), bg='#080810', fg='#aaaacc',
|
||
insertbackground='white', state='disabled', wrap='none',
|
||
)
|
||
self._log.pack(fill='both', expand=True)
|
||
self._log.tag_config('recv', foreground='#88bbff')
|
||
self._log.tag_config('send', foreground='#88ffaa')
|
||
self._log.tag_config('info', foreground='#dddd88')
|
||
self._log.tag_config('error', foreground='#ff7777')
|
||
|
||
ctrl = ttk.Frame(lf)
|
||
ctrl.pack(fill='x', pady=(2, 0))
|
||
ttk.Button(ctrl, text='清空日志', command=self._clear_log, width=8).pack(side='right')
|
||
ttk.Button(ctrl, text='注入帧', command=self._inject_frame, width=8).pack(side='right', padx=4)
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# 服务器
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def _start_server(self):
|
||
try:
|
||
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||
srv.bind(('0.0.0.0', self._port))
|
||
srv.listen(3)
|
||
self._srv_sock = srv
|
||
except OSError as e:
|
||
self._sv_var.set(f'启动失败: {e}')
|
||
return
|
||
|
||
# Detect LAN IP
|
||
try:
|
||
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||
s.connect(('8.8.8.8', 80))
|
||
lan = s.getsockname()[0]
|
||
s.close()
|
||
except Exception:
|
||
lan = '127.0.0.1'
|
||
|
||
self._sv_var.set(
|
||
f'0.0.0.0:{self._port} 局域网: {lan}:{self._port}'
|
||
f' Android模拟器: 10.0.2.2:{self._port}'
|
||
)
|
||
threading.Thread(target=self._accept_loop, daemon=True).start()
|
||
self._log_line('服务器已启动', 'info')
|
||
|
||
def _accept_loop(self):
|
||
while self._srv_sock:
|
||
try:
|
||
conn, addr = self._srv_sock.accept()
|
||
conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
||
except OSError:
|
||
break
|
||
sess = Session(conn, addr, self.cfg, self._on_event)
|
||
self.session = sess
|
||
threading.Thread(target=sess.run, daemon=True).start()
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# 事件处理（后台线程 → root.after → 主线程）
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def _on_event(self, kind: str, data):
|
||
self.root.after(0, lambda k=kind, d=data: self._handle(k, d))
|
||
|
||
def _handle(self, kind: str, data):
|
||
if kind == 'log':
|
||
tag, msg = data
|
||
self._log_line(msg, tag)
|
||
elif kind == 'connected':
|
||
self._conn_var.set(f'已连接: {data}')
|
||
self._conn_lbl.configure(foreground='#44cc44')
|
||
elif kind == 'disconnected':
|
||
self._conn_var.set('未连接')
|
||
self._conn_lbl.configure(foreground='#999')
|
||
self._acq_var.set('已停止')
|
||
self._acq_lbl.configure(foreground='#999')
|
||
elif kind == 'acq_status':
|
||
colors = {'连续推送': '#44cc44', '已停止': '#999', '单次完成': '#ddaa44'}
|
||
self._acq_var.set(data)
|
||
self._acq_lbl.configure(foreground=colors.get(data, '#999'))
|
||
elif kind == 'config_changed':
|
||
self._sync_ui_from_cfg()
|
||
elif kind == 'frame':
|
||
pass # stats updated via refresh timer
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# 控件回调
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def _apply_signal(self, *_):
|
||
try:
|
||
self.cfg.channel_num = max(1, min(6, self._ch_var.get()))
|
||
self.cfg.sample_depth = max(16, self._depth_var.get())
|
||
self.cfg.acc_num = max(1, self._acc_var.get())
|
||
self.cfg.signal_uv = float(self._amp_uv_var.get())
|
||
self.cfg.noise_ratio = float(self._noise_var.get()) / 100.0
|
||
except (ValueError, tk.TclError):
|
||
pass
|
||
|
||
sig_name = self._sig_var.get()
|
||
if sig_name in SIGNAL_TYPES:
|
||
self.cfg.signal_type = SIGNAL_TYPES.index(sig_name)
|
||
|
||
freq_lbl = self._freq_var.get()
|
||
for code, lbl in SEND_FREQ_OPTIONS:
|
||
if lbl == freq_lbl:
|
||
self.cfg.send_freq = code
|
||
break
|
||
|
||
amp_lbl = self._amp_ratio_var.get()
|
||
if amp_lbl in AMP_LABELS:
|
||
self.cfg.amp_ratio = AMP_LABELS.index(amp_lbl)
|
||
|
||
self._redraw_waveform()
|
||
|
||
def _apply_pos(self, *_):
|
||
try:
|
||
self.cfg.lat = float(self._lat_var.get())
|
||
self.cfg.lon = float(self._lon_var.get())
|
||
self.cfg.alt = float(self._alt_var.get())
|
||
self.cfg.pitch = float(self._pitch_var.get())
|
||
self.cfg.roll = float(self._roll_var.get())
|
||
except ValueError:
|
||
pass
|
||
self.cfg.gps_walk = self._gps_walk_var.get()
|
||
|
||
def _apply_dev(self, *_):
|
||
try:
|
||
self.cfg.batt_raw = int(self._batt_var.get())
|
||
self.cfg.temp_raw = int(self._temp_var.get())
|
||
except ValueError:
|
||
pass
|
||
self.cfg.batt_drain = self._batt_drain_var.get()
|
||
self.cfg.gps_status = 1 if self._gps_st_var.get() else 0
|
||
self.cfg.sd_status = 1 if self._sd_st_var.get() else 0
|
||
|
||
def _inject_frame(self):
|
||
"""手动注入一帧（不管当前是否在连续推送中）"""
|
||
if self.session and self.session.sock:
|
||
frame = build_data_frame(self.cfg.snapshot(), self.session.phase)
|
||
try:
|
||
self.session.sock.sendall(frame)
|
||
self.cfg.frame_count += 1
|
||
self.cfg.bytes_sent += len(frame)
|
||
self.session.phase += 1.0
|
||
self._log_line(f'注入帧 {len(frame)-10}B →', 'send')
|
||
except OSError as e:
|
||
self._log_line(f'注入失败: {e}', 'error')
|
||
else:
|
||
self._log_line('无连接，无法注入帧', 'error')
|
||
|
||
def _sync_ui_from_cfg(self):
|
||
"""SETUP_REQ 后将配置同步回 UI 控件"""
|
||
self._ch_var.set(self.cfg.channel_num)
|
||
self._depth_var.set(self.cfg.sample_depth)
|
||
self._acc_var.set(self.cfg.acc_num)
|
||
if self.cfg.amp_ratio < len(AMP_LABELS):
|
||
self._amp_ratio_var.set(AMP_LABELS[self.cfg.amp_ratio])
|
||
freq_lbl = next((lbl for c, lbl in SEND_FREQ_OPTIONS if c == self.cfg.send_freq), '1 Hz')
|
||
self._freq_var.set(freq_lbl)
|
||
|
||
def _reset_stats(self):
|
||
self.cfg.frame_count = 0
|
||
self.cfg.bytes_sent = 0
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# 日志
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def _log_line(self, msg: str, tag: str = ''):
|
||
ts = datetime.now().strftime('%H:%M:%S.%f')[:-3]
|
||
line = f'[{ts}] {msg}\n'
|
||
self._log.configure(state='normal')
|
||
self._log.insert('end', line, tag or '')
|
||
self._log.see('end')
|
||
# 保留最近 600 行
|
||
n = int(self._log.index('end-1c').split('.')[0])
|
||
if n > 600:
|
||
self._log.delete('1.0', f'{n - 600}.0')
|
||
self._log.configure(state='disabled')
|
||
|
||
def _clear_log(self):
|
||
self._log.configure(state='normal')
|
||
self._log.delete('1.0', 'end')
|
||
self._log.configure(state='disabled')
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# 波形预览
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def _redraw_waveform(self):
|
||
c = self._wc
|
||
c.delete('all')
|
||
W = c.winfo_width()
|
||
H = c.winfo_height()
|
||
if W < 20 or H < 20:
|
||
return
|
||
|
||
PL, PR, PT, PB = 6, 8, 8, 22
|
||
pw = W - PL - PR
|
||
ph = H - PT - PB
|
||
|
||
# Background fill
|
||
c.create_rectangle(PL, PT, PL + pw, PT + ph, fill='#0a0a14', outline='#1a1a28')
|
||
|
||
# Horizontal grid
|
||
for i in range(5):
|
||
y = PT + i * ph // 4
|
||
c.create_line(PL, y, PL + pw, y, fill='#15152a', dash=(3, 4))
|
||
|
||
# Zero line
|
||
y0 = PT + ph // 2
|
||
c.create_line(PL, y0, PL + pw, y0, fill='#22224a')
|
||
|
||
# Draw each channel curve
|
||
N = 300
|
||
n_ch = min(self.cfg.channel_num, 6)
|
||
sig = self.cfg.signal_type
|
||
phase = 0.0
|
||
|
||
for ch in range(n_ch):
|
||
pts = []
|
||
for i in range(N):
|
||
v = gen_signal(sig, ch, i, N, phase)
|
||
v = max(-1.0, min(1.0, v))
|
||
x = PL + i * pw // (N - 1)
|
||
y = PT + ph // 2 - int(v * ph * 0.44)
|
||
y = max(PT, min(PT + ph, y))
|
||
pts.extend([x, y])
|
||
if len(pts) >= 4:
|
||
c.create_line(*pts, fill=CH_COLORS[ch], width=1, smooth=False)
|
||
|
||
# Axes
|
||
c.create_line(PL, PT, PL, PT + ph, fill='#334', width=1)
|
||
c.create_line(PL, PT + ph, PL + pw, PT + ph, fill='#334', width=1)
|
||
|
||
# Channel legend (bottom)
|
||
for ch in range(n_ch):
|
||
lx = PL + ch * 44
|
||
c.create_rectangle(lx, PT + ph + 5, lx + 10, PT + ph + 12,
|
||
fill=CH_COLORS[ch], outline='')
|
||
c.create_text(lx + 13, PT + ph + 9, text=f'CH{ch+1}',
|
||
fill=CH_COLORS[ch], anchor='w', font=('Arial', 7))
|
||
|
||
# Signal type badge
|
||
c.create_text(PL + pw - 2, PT + 3,
|
||
text=SIGNAL_TYPES[sig], fill='#555577',
|
||
anchor='ne', font=('Arial', 8))
|
||
|
||
# Amplitude label
|
||
amp_str = f'{self.cfg.signal_uv/1000:.0f} kμV' if self.cfg.signal_uv >= 1000 else f'{self.cfg.signal_uv:.0f} μV'
|
||
c.create_text(PL + 2, PT + 3, text=amp_str, fill='#444466',
|
||
anchor='nw', font=('Arial', 7))
|
||
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
# 定时刷新（主线程，500ms 周期）
|
||
# ──────────────────────────────────────────────────────────────────────────
|
||
|
||
def _refresh(self):
|
||
# Sync batt / GPS widgets if auto-changing
|
||
if self.cfg.batt_drain:
|
||
self._batt_var.set(str(self.cfg.batt_raw))
|
||
if self.cfg.gps_walk:
|
||
self._lat_var.set(f'{self.cfg.lat:.6f}')
|
||
self._lon_var.set(f'{self.cfg.lon:.6f}')
|
||
|
||
batt_v = self.cfg.batt_raw / 1000.0
|
||
temp_c = (self.cfg.temp_raw - 1820) / 11.0 # rough raw→°C
|
||
hz_str = next((lbl for c, lbl in SEND_FREQ_OPTIONS if c == self.cfg.send_freq), '?')
|
||
self._stat_var.set(
|
||
f'帧数: {self.cfg.frame_count:,} | '
|
||
f'发送: {self.cfg.bytes_sent / 1024:.1f} KB | '
|
||
f'电池: {batt_v:.3f} V | '
|
||
f'温度: {temp_c:.1f} °C | '
|
||
f'频率: {hz_str}'
|
||
)
|
||
|
||
self._redraw_waveform()
|
||
self.root.after(500, self._refresh)
|
||
|
||
|
||
# ── 入口 ──────────────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description='TEM 下位机模拟器 (GUI 版)')
|
||
ap.add_argument('--port', type=int, default=4321, help='监听端口 (默认: 4321)')
|
||
args = ap.parse_args()
|
||
|
||
root = tk.Tk()
|
||
app = MockDeviceApp(root, port=args.port)
|
||
|
||
def _on_close():
|
||
if app._srv_sock:
|
||
try:
|
||
app._srv_sock.close()
|
||
except Exception:
|
||
pass
|
||
if app.session:
|
||
app.session.stop_continuous()
|
||
root.destroy()
|
||
|
||
root.protocol('WM_DELETE_WINDOW', _on_close)
|
||
root.mainloop()
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|