电量与增益控制：两套串口协议及 PyQt5 参考界面

学习教程更新于 2026-10-03编辑本文

两套历史方案先选定一种

电池读数、串口传输和模拟增益控制是三个独立环节。原方案同时包含可变长度 CRC 帧与固定八字节 SUM 帧；它们的命令码、字段含义和校验方式不同，不能混用，也不能仅凭文章中的 V1/V2 标题推断已部署固件版本。

项目	扩展 CRC 帧	固定八字节 SUM 帧
外层	`AA 55 CMD LEN DATA CRC16 0D 0A`	`AA CMD D1 D2 00 00 SUM 55`
帧长	`LEN + 8`，LEN 为 0～128	恒为 8
设置增益	`0x10`，参数为档位索引 0～7	`0x02`，参数为倍数值 1～255
完整电池数据	有 19 字节历史响应	无此定义
校验	CRC-16/MODBUS 参数及覆盖范围见下文	CMD 至第二个保留字节之和，取低八位
当前限制	都没有事务序号、采样时间或统一无效值字段	同左

串口约定为 8 个数据位、无校验、1 个停止位，无流控；115200 是原方案默认值。9600、256000 可作为双方另行约定的候选，但适配器和设备未必支持非标准速率。端口、协议与波特率应在连接前明确选定。

扩展帧：长度、端序和命令

CRC 初值为 0xFFFF，反射多项式 0xA001，无最终异或；只覆盖 CMD + LEN + DATA，不含帧头帧尾。CRC 低字节先发。测试字符串 123456789 的结果为 0x4B37。这里沿用原 Python 实现所表达的参数，不用含糊的“CRC16”名称代替完整约定。

数据里的 AA 55 或 0D 0A 是合法载荷，接收端必须依据 LEN 定界。多字节整数均按小端传输；有符号量用二进制补码。固件应逐字节序列化，不能仅靠 C 的 packed 结构体推断跨平台的端序、对齐与内存访问行为。

请求 / 响应	请求载荷	响应载荷
`01 / 81` 基础电量	空	6 字节，见下一表
`02 / 82` 完整电量	空	19 字节，见下一表
`10 / 90` 设置档位	档位一字节，0～7	结果、当前档位、当前倍数，各一字节；结果 0 成功、1 失败、2 无效档位
`12 / 92` 档位列表	空	数量 N 一字节，随后 N 个非零倍数；本方案 N 为 1～8
`F0 / F0` 心跳	空	空；主机发起、设备应答一次，主机不再回声
`F1 / F1` 设备信息	空	28 字节，见下文

普通请求的响应码为请求码或上 0x80；F0/F1 是同码例外，方向要结合事务角色判断。

电池载荷

字段	基础包偏移	完整包偏移	类型与单位
SOC	0	0	u8，0～100%
电压	1～2	1～2	u16，mV
净电流	3～4	3～4	i16，mA
温度	无	5～6	i16，0.1 ℃
剩余容量	无	7～8	u16，mAh
满充容量	无	9～10	u16，mAh
平均功率	无	11～12	i16，mW
循环次数	无	13～14	u16，必须明确系统数据来源
SOH	无	15	u8，0～100%
历史充电状态	5	16	u8，0 未报告充电、1 报告充电
满充标志	无	17	u8，0/1
电池存在标志	无	18	u8，0/1

Python 标准格式分别为 <BHhB 和 <BHhhHHhHBBBB，长度为 6 和 19。原完整包格式 <BHhhhHHhHBBBB 多出一个有符号短整型，变成 21 字节、13 项，与 12 字段定义矛盾；解包前必须检查长度恰好匹配。Python struct 格式说明

BQ27427 不提供原方案假设的 CycleCount(0x20)；该地址实际是 SOH。历史协议的循环次数字段可以保留用于已有设备兼容，但发送端必须有独立可信来源。没有来源时，当前旧协议没有无效标志，不能随意发零冒充已知数值；应停用该完整响应或协商带有效位的新版本。充电状态也不能由 CHG 位直接赋值，0 不足以区分放电、静置和未检测到充电。芯片读数的语义见BQ27427 读数与配置。

增益与设备信息

原八档示例为 1、2、5、10、20、50、100、200 倍，索引 3 对应 10 倍。三个 GPIO 至多提供八种组合，是否对应这八个增益还要由模拟电路、开关真值表和实测决定。设置成功时主机核对 ACK 中的档位后再显示设备确认值；失败时保留上次成功状态，不能先在界面宣布设置完成。

设备信息的字节 0～15 为 ASCII 名称，短名称以零填充；16～19、20～23 分别为硬件和软件版本的主、次、修订、保留字节；24～27 为 u32 序列号。例如 0x12345678 在线路上为 78 56 34 12。名称满 16 字节时可无终止零，接收端仍必须按固定长度读取。

下列命令在原方案只有名称或声明，尚缺少完整载荷、失败语义或执行实现。参考程序保留其设计意图，但不假装已经支持。

请求 / 响应	原设计用途	尚需定义
`03/83、04/84、05/85、06/86`	单项 SOC、电压、电流、温度	完整请求/响应及错误约定
`07/87`	健康信息	数据来源、字段与有效性
`11/91`	查询当前档位	载荷及状态语义
`20/A0、21/A1、22/A2`	读参数、保存参数、恢复默认	参数表、持久化时机、权限、幂等性和失败响应
`F2/F2`	系统复位	ACK 与复位的先后、超时和重试语义

可复现的十六进制示例

以下数据是协议编码样例，不是电池实测。75%、2000/3000 mAh、800 mW 等数值只用于检查字段位置；不能据此推导物理一致性或 SOC 精度。每一行均包括已计算的 CRC，可在确认设备采用本协议后用于串口调试。

基础请求：AA 55 01 00 00 20 0D 0A
基础响应：AA 55 81 06 4B 68 10 C8 00 01 C5 2B 0D 0A
完整请求：AA 55 02 00 00 D0 0D 0A
完整响应：AA 55 82 13 4B 68 10 C8 00 E6 00 D0 07 B8 0B 20 03 0A 00 5A 01 00 01 2D 1F 0D 0A
设置档3： AA 55 10 01 03 31 94 0D 0A
成功响应：AA 55 90 03 00 03 0A 64 AE 0D 0A
列表请求：AA 55 12 00 0D 10 0D 0A
列表响应：AA 55 92 09 08 01 02 05 0A 14 32 64 C8 B2 80 0D 0A
心跳双方：AA 55 F0 00 45 B0 0D 0A
设备请求：AA 55 F1 00 44 20 0D 0A

基础响应表示 75%、4200 mV、+200 mA、充电状态 1。原示例载荷 4B 10 68 00 C8 01 把多字节数值写成大端；按本文小端契约会被解成 26640 mV、−14336 mA，不能照抄后再将其解释为 4.2 V、200 mA。完整响应依次包含 23.0 ℃、2000 mAh、3000 mAh、800 mW、示例循环计数 10、SOH 90%、状态 1/0/1。

固定八字节方案

SUM 为字节 1～5 之和的低八位。两个保留字节必须为零，校验通过后还要检查命令方向和字段范围。该校验不能发现所有多字节错误，不能与 CRC 的错误检测能力等同。

操作	请求	响应
查询 SOC	`01 00 00`	`81 SOC STATE`，SOC 为 0～100，STATE 为 0/1
设置倍数	`02 GAIN 00`	`82 GAIN RESULT`
查询倍数	`03 00 00`	`83 GAIN 00`，GAIN 非零
心跳	`F0 00 00`	`F0 00 00`，设备只应答一次

表中仅列 CMD、D1、D2。GAIN 线编码范围为 1～255，不等于硬件能够产生任意 1～255 倍；原八档硬件仍需做值到 GPIO 组合的映射并拒绝不支持值。RESULT 的原约定为 00 成功、01 参数错误、02 硬件错误、FF 未知错误。

查询电量：AA 01 00 00 00 00 01 55
75%响应：AA 81 4B 01 00 00 CD 55
设置10倍：AA 02 0A 00 00 00 0C 55
成功确认：AA 82 0A 00 00 00 8C 55
设置20倍：AA 02 14 00 00 00 16 55
成功确认：AA 82 14 00 00 00 96 55
设置50倍：AA 02 32 00 00 00 34 55
设置100倍：AA 02 64 00 00 00 66 55
读取倍数：AA 03 00 00 00 00 03 55
10倍响应：AA 83 0A 00 00 00 8D 55
心跳双方：AA F0 00 00 00 00 F0 55
无效零值：AA 02 00 00 00 00 02 55
错误确认：AA 82 00 01 00 00 83 55

最后一行保留了原错误示例：失败 ACK 的 D1 为请求的零值，因此不能在失败时将 D1 当成当前有效增益。成功 ACK 才可用于更新显示。另一个原例 AA 01 4B 01 00 00 4D 55 虽然求和正确，命令仍是请求 01，不能作为电量响应；上表改成响应 81 并重算为 CD。

流式解析与事务边界

一次串口读取可能只有半帧，也可能含多帧。解析器保留尾部可能的 AA，限制 CRC 的 LEN，在坏帧处逐字节重新寻找帧头；不能根据损坏的长度把后面的正常帧一起丢掉。未完成候选帧若长时间停滞，允许按主机观察到的间隔丢弃并重同步。该超时反映软件收到分块的时间，不能当作线路真实字节间隔；慢速链路或调度暂停时需要调整。

一个连接仅允许一个未完成请求，收到匹配命令且内容有效的响应才完成事务。两套格式都没有事务序号，因此串行请求、清空缓存或等待隔离期仍不能证明迟到的同码响应属于新请求。 对增益等有副作用操作，应以设备侧序号、幂等操作和读回机制完善新版协议；当前参考界面在响应超时后停止连接，要求检查后重新连接，不自动重复写入。

原策略中的基础电量每秒、完整数据每十秒、心跳每五秒是主机刷新建议，不保证芯片产生新样本的频率。“最多重发三次”与“连续三次失败”也不是同一计数方式：前者若不包含初次发送共四次尝试。若采用约 1000 ms 响应期限和 500 ms 重试间隔，需要明确计数与哪些只读命令可以重试；保存参数、恢复默认、复位不能直接套用同一重试策略。

编解码与有界解析器

保存为 battery_protocol.py。该模块不打开串口，分离外层校验和主机响应语义检查；未知或仅保留名称的命令不会自动变成已支持功能。

"""Two historical battery/gain envelopes. No serial port is opened here."""
from dataclasses import dataclass
import struct


class ProtocolError(ValueError):
    pass


@dataclass(frozen=True)
class Frame:
    cmd: int
    payload: bytes


def crc16_modbus(data):
    crc = 0xFFFF
    for value in data:
        crc ^= value
        for _ in range(8):
            crc = (crc >> 1) ^ (0xA001 if crc & 1 else 0)
    return crc


def byte(value):
    if not isinstance(value, int) or not 0 <= value <= 255:
        raise ProtocolError('byte out of range')
    return value


def encode_crc(cmd, payload=b''):
    payload = bytes(payload)
    if len(payload) > 128:
        raise ProtocolError('payload exceeds 128 bytes')
    protected = bytes((byte(cmd), len(payload))) + payload
    return b'\xAA\x55' + protected + struct.pack('<H', crc16_modbus(protected)) + b'\r\n'


def decode_crc(packet):
    packet = bytes(packet)
    if len(packet) < 8 or packet[:2] != b'\xAA\x55':
        raise ProtocolError('bad CRC-frame header or length')
    length = packet[3]
    if length > 128 or len(packet) != length + 8:
        raise ProtocolError('declared CRC-frame length does not match')
    if packet[-2:] != b'\r\n':
        raise ProtocolError('bad CRC-frame tail')
    received = struct.unpack_from('<H', packet, 4 + length)[0]
    if received != crc16_modbus(packet[2:4 + length]):
        raise ProtocolError('CRC mismatch')
    return Frame(packet[2], packet[4:4 + length])


def encode_sum(cmd, data1=0, data2=0):
    protected = bytes((byte(cmd), byte(data1), byte(data2), 0, 0))
    return b'\xAA' + protected + bytes((sum(protected) & 0xFF, 0x55))


def decode_sum(packet):
    packet = bytes(packet)
    if len(packet) != 8 or packet[0] != 0xAA or packet[7] != 0x55:
        raise ProtocolError('bad SUM-frame envelope')
    if packet[4:6] != b'\x00\x00':
        raise ProtocolError('nonzero reserved bytes')
    if packet[6] != sum(packet[1:6]) & 0xFF:
        raise ProtocolError('checksum mismatch')
    return Frame(packet[1], packet[2:4])


class StreamParser:
    """Bounded carry buffer, length-based framing and sliding resynchronization.

    gap_s is a host-observed inter-read gap, not a measurement of wire timing.
    Call feed(b'', now) while idle so incomplete frames can expire.
    """
    def __init__(self, mode='crc', gap_s=0.25):
        if mode not in ('crc', 'sum') or gap_s <= 0:
            raise ValueError('invalid mode or gap')
        self.mode, self.gap_s = mode, gap_s
        self.buffer = bytearray()
        self.last_input = None
        self.last_clock = None
        self.rejected = 0

    def _drain(self):
        frames = []
        prefix = b'\xAA\x55' if self.mode == 'crc' else b'\xAA'
        while self.buffer:
            pos = self.buffer.find(prefix)
            if pos < 0:
                keep = self.mode == 'crc' and self.buffer[-1] == 0xAA
                self.buffer[:] = b'\xAA' if keep else b''
                break
            if pos:
                del self.buffer[:pos]
            if self.mode == 'crc':
                if len(self.buffer) < 4:
                    break
                if self.buffer[3] > 128:
                    del self.buffer[0]
                    self.rejected += 1
                    continue
                length = self.buffer[3] + 8
            else:
                length = 8
            if len(self.buffer) < length:
                break
            try:
                decode = decode_crc if self.mode == 'crc' else decode_sum
                frame = decode(self.buffer[:length])
            except ProtocolError:
                # A corrupt length may overlap a later valid frame.
                del self.buffer[0]
                self.rejected += 1
            else:
                del self.buffer[:length]
                frames.append(frame)
        return frames

    def feed(self, data, now):
        if self.last_clock is not None and now < self.last_clock:
            raise ValueError('use a monotonic clock')
        self.last_clock = now
        frames = []
        if self.buffer and self.last_input is not None and now - self.last_input >= self.gap_s:
            # Discard the expired candidate's first byte, recover any complete
            # later frames, then discard its remaining incomplete suffix.
            del self.buffer[0]
            self.rejected += 1
            frames.extend(self._drain())
            self.buffer.clear()
        for value in data:
            self.buffer.append(value)
            frames.extend(self._drain())
        if data:
            self.last_input = now
        return frames


def decode_response(frame, mode):
    """Application validation for host reception. Reserved commands are rejected."""
    cmd, data = frame.cmd, frame.payload
    if mode == 'sum':
        if len(data) != 2:
            raise ProtocolError('SUM data must contain exactly two bytes')
        a, b = data
        if cmd == 0x81 and a <= 100 and b in (0, 1):
            return {'kind': 'battery', 'soc': a, 'charging_legacy': b}
        if cmd == 0x82 and b in (0, 1, 2, 255) and (b != 0 or a > 0):
            return {'kind': 'gain_ack', 'value': a, 'result': b}
        if cmd == 0x83 and a > 0 and b == 0:
            return {'kind': 'gain_value', 'value': a}
        if cmd == 0xF0 and data == b'\x00\x00':
            return {'kind': 'heartbeat'}
        raise ProtocolError('unknown SUM response or invalid fields')
    if mode != 'crc':
        raise ProtocolError('unknown profile')
    if cmd == 0x81 and len(data) == 6:
        soc, voltage, current, charging = struct.unpack('<BHhB', data)
        if soc > 100 or charging not in (0, 1):
            raise ProtocolError('invalid basic battery fields')
        return {'kind': 'battery', 'soc': soc, 'voltage_mv': voltage,
                'current_ma': current, 'charging_legacy': charging}
    if cmd == 0x82 and len(data) == 19:
        values = struct.unpack('<BHhhHHhHBBBB', data)
        keys = ('soc', 'voltage_mv', 'current_ma', 'temperature_deci_c',
                'remaining_mah', 'full_mah', 'power_mw', 'cycle_count',
                'soh', 'charging_legacy', 'full', 'detected')
        result = dict(zip(keys, values))
        if result['soc'] > 100 or result['soh'] > 100 or any(result[k] not in (0, 1) for k in keys[-3:]):
            raise ProtocolError('invalid full battery fields')
        result['kind'] = 'battery_full'
        return result
    if cmd == 0x90 and len(data) == 3:
        result, level, value = data
        if result not in (0, 1, 2) or level > 7 or value == 0:
            raise ProtocolError('invalid gain acknowledgement')
        return {'kind': 'gain_ack', 'result': result, 'level': level, 'value': value}
    if cmd == 0x92 and data and 1 <= data[0] <= 8 and len(data) == data[0] + 1 and all(data[1:]):
        return {'kind': 'gain_list', 'values': list(data[1:])}
    if cmd == 0xF1 and len(data) == 28:
        return {'kind': 'device', 'name': data[:16].split(b'\0', 1)[0].decode('ascii', 'replace'),
                'hardware': list(data[16:20]), 'software': list(data[20:24]),
                'serial': struct.unpack_from('<I', data, 24)[0]}
    if cmd == 0xF0 and not data:
        return {'kind': 'heartbeat'}
    raise ProtocolError('unknown CRC response or invalid payload length')


def build_request(mode, name, gain=None):
    """Only commands whose request contract is present in the original article."""
    if mode == 'crc':
        commands = {'basic': 0x01, 'full': 0x02, 'gain_list': 0x12,
                    'heartbeat': 0xF0, 'device': 0xF1}
        if name == 'set_gain' and isinstance(gain, int) and 0 <= gain <= 7:
            return encode_crc(0x10, bytes((gain,))), 0x90
        if name in commands:
            cmd = commands[name]
            return encode_crc(cmd), cmd if cmd >= 0xF0 else cmd | 0x80
    elif mode == 'sum':
        commands = {'basic': 0x01, 'read_gain': 0x03, 'heartbeat': 0xF0}
        if name == 'set_gain' and isinstance(gain, int) and 1 <= gain <= 255:
            return encode_sum(0x02, gain), 0x82
        if name in commands:
            cmd = commands[name]
            return encode_sum(cmd), cmd if cmd == 0xF0 else cmd | 0x80
    raise ProtocolError('unsupported request or out-of-range gain')

调用者使用单调时钟向 feed 传入数据，空闲时也调用 feed(b'', now) 处理残帧期限。处理完成后的保留缓冲区上限为 CRC 135 字节、SUM 7 字节；上位机另将每次读取限制为最多 512 字节。解码拒绝 SOC/SOH 大于 100、非法布尔状态、错误长度和非零保留位，不把损坏数值强行截断到合理范围。

可运行的 PyQt5 参考界面

把下面代码保存为同目录的 battery_monitor.py。验证环境为 Python 3.11、PyQt5 5.15.11 和 pyserial 3.5；可在独立环境安装这两个包后运行。代码保留串口选择、基础/完整电量、增益控制、状态颜色、周期刷新、心跳和日志。协议必须先与设备约定，不能通过试发有副作用的命令来猜。

python -m pip install PyQt5==5.15.11 pyserial==3.5
python battery_monitor.py

串口创建、读取、写入和关闭都由同一个工作线程负责；主线程只提交请求和更新控件。工作线程解码成功后仍保留请求所有权，直到 GUI 消费排队结果并确认，才允许下一请求。短读交给流式解析器，部分写入继续发送且有总期限。pySerial 的读超时可能返回不足请求长度的数据，写操作返回写入字节数，不能把一次 read/write 当作完整帧收发。pySerial API

"""PyQt5 reference monitor. Select the agreed wire profile before connecting."""
import queue
import sys
import threading
import time
import serial
import serial.tools.list_ports
from PyQt5.QtCore import QThread, QTimer, pyqtSignal
from PyQt5.QtWidgets import (QApplication, QCheckBox, QComboBox, QHBoxLayout,
    QLabel, QMainWindow, QPlainTextEdit, QProgressBar, QPushButton, QSpinBox, QVBoxLayout, QWidget)
from battery_protocol import StreamParser, ProtocolError, build_request, decode_response


class SerialWorker(QThread):
    opened = pyqtSignal()
    received = pyqtSignal(object)
    failed = pyqtSignal(str)

    def __init__(self, port, baud, mode, parent=None, serial_factory=serial.Serial):
        super().__init__(parent)
        self.port, self.baud, self.mode = port, baud, mode
        self.serial_factory = serial_factory
        self.stop_event = threading.Event()
        self.requests = queue.Queue(maxsize=1)
        self.state_lock = threading.Lock()
        self.busy = False
        self.result_waiting = False

    def request(self, name, gain=None):
        packet, expected = build_request(self.mode, name, gain)
        with self.state_lock:
            if self.busy or self.stop_event.is_set():
                return False
            self.busy = True
            self.requests.put_nowait((packet, expected, name, gain))
        return True

    def acknowledge_result(self):
        # Keep ownership until the GUI has consumed the queued result signal.
        with self.state_lock:
            if self.stop_event.is_set() or not self.result_waiting:
                return False
            self.result_waiting = False
            self.busy = False
        return True

    def stop(self):
        self.stop_event.set()

    def run(self):
        connection = None
        pending = None
        parser = StreamParser(self.mode)
        try:
            # Open, read, write and close all belong to this worker thread.
            connection = self.serial_factory(self.port, self.baud,
                bytesize=8, parity='N', stopbits=1, timeout=0.05, write_timeout=0.2)
            connection.reset_input_buffer()
            self.opened.emit()
            while not self.stop_event.is_set():
                if pending is None:
                    try:
                        packet, expected, name, gain = self.requests.get_nowait()
                    except queue.Empty:
                        pass
                    else:
                        sent, send_deadline = 0, time.monotonic() + 0.5
                        while sent < len(packet):
                            if self.stop_event.is_set():
                                return
                            if time.monotonic() >= send_deadline:
                                raise TimeoutError('发送未在限定时间内完成')
                            count = connection.write(packet[sent:])
                            if not isinstance(count, int) or not 0 < count <= len(packet) - sent:
                                raise OSError('串口没有完整接受发送数据')
                            sent += count
                        pending = (expected, name, gain, time.monotonic() + 1.0)
                if pending and time.monotonic() >= pending[3]:
                    raise TimeoutError('响应超时；连接已停止，请检查设备后重新连接')
                incoming = connection.read(min(max(connection.in_waiting, 1), 512))
                for frame in parser.feed(incoming, time.monotonic()):
                    if pending is None or frame.cmd != pending[0]:
                        continue  # No echo for heartbeat replies or unsolicited frames.
                    try:
                        result = decode_response(frame, self.mode)
                    except ProtocolError:
                        continue  # Keep waiting until the finite transaction deadline.
                    if pending[1] == 'set_gain' and result['result'] == 0:
                        echoed = result['level'] if self.mode == 'crc' else result['value']
                        if echoed != pending[2]:
                            raise ProtocolError('增益确认与当前请求不一致')
                    pending = None
                    with self.state_lock:
                        self.result_waiting = True
                    self.received.emit(result)
        except Exception as error:
            if not self.stop_event.is_set():
                self.failed.emit(str(error))
        finally:
            self.stop_event.set()
            if connection is not None:
                try:
                    connection.close()
                except Exception:
                    pass


class Monitor(QMainWindow):
    def __init__(self, worker_factory=SerialWorker):
        super().__init__()
        self.worker_factory, self.worker = worker_factory, None
        self.closing = False
        self.handshake = False
        self.phase = 'disconnected'
        self.last_reply = None
        self.battery_received_at = None
        self.gain_buttons = []
        self.setWindowTitle('电量与增益监控 · 参考实现')
        root = QWidget()
        layout = QVBoxLayout(root)
        settings = QHBoxLayout()
        self.ports, self.baud, self.profile = QComboBox(), QComboBox(), QComboBox()
        self.baud.addItems(['9600', '115200', '256000'])
        self.baud.setCurrentText('115200')
        self.profile.addItem('扩展 CRC 帧', 'crc')
        self.profile.addItem('固定 8 字节 SUM 帧', 'sum')
        self.refresh_ports = QPushButton('刷新端口')
        self.connect_button = QPushButton('连接')
        for widget in (self.ports, self.baud, self.profile, self.refresh_ports, self.connect_button):
            settings.addWidget(widget)
        layout.addLayout(settings)
        self.connection_label = QLabel('未连接')
        self.battery = QLabel('电量数据：--')
        self.soc_bar = QProgressBar()
        self.soc_bar.setRange(0, 100)
        self.soc_bar.setValue(0)
        self.soc_bar.setFormat('未收到有效电量')
        self.sample_age = QLabel('尚无电量数据接收时间')
        self.gain_label = QLabel('当前增益：--')
        for label in (self.connection_label, self.battery, self.gain_label):
            label.setWordWrap(True)
            layout.addWidget(label)
        layout.addWidget(self.soc_bar)
        layout.addWidget(self.sample_age)
        commands = QHBoxLayout()
        self.basic_button, self.full_button = QPushButton('读取基础数据'), QPushButton('读取完整数据')
        self.auto_refresh = QCheckBox('每秒尝试刷新基础数据')
        for widget in (self.basic_button, self.full_button, self.auto_refresh):
            commands.addWidget(widget)
        layout.addLayout(commands)
        self.gain_layout = QHBoxLayout()
        self.gain_value = QSpinBox()
        self.gain_value.setRange(1, 255)
        self.set_gain_button = QPushButton('设置倍数（SUM）')
        self.gain_layout.addWidget(self.gain_value)
        self.gain_layout.addWidget(self.set_gain_button)
        layout.addLayout(self.gain_layout)
        self.log_view = QPlainTextEdit()
        self.log_view.setReadOnly(True)
        self.log_view.document().setMaximumBlockCount(200)
        layout.addWidget(self.log_view)
        self.setCentralWidget(root)
        self.refresh_ports.clicked.connect(self.load_ports)
        self.connect_button.clicked.connect(self.toggle_connection)
        self.basic_button.clicked.connect(lambda: self.send('basic'))
        self.full_button.clicked.connect(lambda: self.send('full'))
        self.set_gain_button.clicked.connect(lambda: self.send('set_gain', self.gain_value.value()))
        self.tick = QTimer(self)
        self.tick.timeout.connect(self.refresh)
        self.tick.start(1000)
        self.heartbeat = QTimer(self)
        self.heartbeat.timeout.connect(lambda: self.send('heartbeat', quiet=True))
        self.heartbeat.start(5000)
        self.set_connected(False)
        self.load_ports()

    def log(self, message):
        self.log_view.appendPlainText(time.strftime('%H:%M:%S ') + message)

    def load_ports(self):
        self.ports.clear()
        for port in serial.tools.list_ports.comports():
            self.ports.addItem(f'{port.device} — {port.description}', port.device)

    def set_connected(self, enabled):
        mode = self.profile.currentData()
        self.basic_button.setEnabled(enabled)
        self.full_button.setEnabled(enabled and mode == 'crc')
        self.auto_refresh.setEnabled(enabled)
        self.gain_value.setEnabled(enabled and mode == 'sum')
        self.set_gain_button.setEnabled(enabled and mode == 'sum')
        for button in self.gain_buttons:
            button.setEnabled(enabled)
        for widget in (self.ports, self.baud, self.profile, self.refresh_ports):
            widget.setEnabled(self.worker is None)

    def toggle_connection(self):
        if self.worker is not None:
            self.phase = 'stopping'
            self.handshake = False
            self.connection_label.setText('正在断开…')
            self.connect_button.setEnabled(False)
            self.set_connected(False)
            self.worker.stop()
            return
        port = self.ports.currentData()
        if not port:
            self.log('请选择实际串口')
            return
        for button in self.gain_buttons:
            self.gain_layout.removeWidget(button)
            button.deleteLater()
        self.gain_buttons.clear()
        self.handshake, self.last_reply = False, None
        self.phase = 'opening'
        self.battery_received_at = None
        self.battery.setText('电量数据：--')
        self.soc_bar.setValue(0)
        self.soc_bar.setFormat('未收到有效电量')
        self.sample_age.setText('尚无电量数据接收时间')
        self.gain_label.setText('当前增益：--')
        self.worker = self.worker_factory(port, int(self.baud.currentText()), self.profile.currentData(), self)
        self.worker.opened.connect(self.on_open)
        self.worker.received.connect(self.on_result)
        self.worker.failed.connect(self.on_error)
        self.worker.finished.connect(self.on_finished)
        self.connect_button.setText('断开')
        self.connection_label.setText('正在打开端口…')
        self.set_connected(False)
        self.worker.start()

    def on_open(self):
        if self.phase != 'opening' or self.worker is None or self.worker.stop_event.is_set():
            return
        self.connection_label.setText('端口已打开，等待心跳响应…')
        self.startup_request('heartbeat')

    def startup_request(self, name):
        if self.worker is None or self.phase == 'stopping':
            return
        self.phase = name
        if not self.worker.request(name):
            self.on_error('初始化请求未能取得通信所有权')
            self.worker.stop()

    def send(self, name, gain=None, quiet=False):
        if self.worker is None or self.phase != 'ready':
            return False
        try:
            accepted = self.worker.request(name, gain)
        except ProtocolError as error:
            self.log(str(error))
            return False
        if not accepted and not quiet:
            self.log('上一请求尚未完成，请稍后重试')
        return accepted

    def on_result(self, data):
        worker = self.worker
        if worker is None or self.phase in ('disconnected', 'stopping') or worker.stop_event.is_set():
            return  # A queued reply cannot restart initialization after stop.
        if not worker.acknowledge_result():
            return
        phase = self.phase
        expected = {'heartbeat': 'heartbeat', 'device': 'device',
                    'gain_list': 'gain_list', 'read_gain': 'gain_value', 'basic': 'battery'}
        if phase != 'ready' and data['kind'] != expected.get(phase):
            self.on_error('初始化响应与当前阶段不一致')
            worker.stop()
            return
        self.last_reply = time.monotonic()
        self.connection_label.setText('已收到有效响应' if phase == 'ready' else '正在读取设备初始状态…')
        kind, mode = data['kind'], self.profile.currentData()
        if kind == 'device':
            self.log(f"设备 {data['name']}，硬件 {data['hardware']}，软件 {data['software']}，序号 {data['serial']:08X}")
        elif kind == 'gain_list':
            for button in self.gain_buttons:
                self.gain_layout.removeWidget(button)
                button.deleteLater()
            self.gain_buttons.clear()
            for index, value in enumerate(data['values']):
                button = QPushButton(f'{value}×（档位 {index}）')
                button.setEnabled(phase == 'ready')
                button.clicked.connect(lambda checked=False, level=index: self.send('set_gain', level))
                self.gain_layout.addWidget(button)
                self.gain_buttons.append(button)
        elif kind in ('gain_ack', 'gain_value'):
            if data.get('result', 0) == 0:
                self.gain_label.setText(f"当前增益：{data['value']}×（设备确认）")
            else:
                self.log(f"设置失败，错误码 {data['result']}；保留上一次确认值")
        elif kind in ('battery', 'battery_full'):
            self.battery_received_at = time.monotonic()
            self.sample_age.setText('电量数据于 0 秒前接收；该时间不是芯片采样时间')
            self.soc_bar.setValue(data['soc'])
            self.soc_bar.setFormat('%p%')
            color = '#c62828' if data['soc'] <= 20 else '#a66b00' if data['soc'] <= 50 else '#2e7d32'
            self.soc_bar.setStyleSheet('QProgressBar::chunk { background: ' + color + '; }')
            state = '设备报告充电' if data['charging_legacy'] else '设备未报告充电'
            lines = [f"电量 {data['soc']}% · {state}"]
            if 'voltage_mv' in data:
                lines.append(f"电压 {data['voltage_mv']} mV · 净电流 {data['current_ma']} mA")
            if kind == 'battery_full':
                lines.extend([f"温度 {data['temperature_deci_c']/10:.1f} ℃ · 容量 {data['remaining_mah']}/{data['full_mah']} mAh",
                    f"功率 {data['power_mw']} mW · SOH {data['soh']}% · 循环 {data['cycle_count']}（设备报告，来源需确认）",
                    f"满充标志 {data['full']} · 电池存在标志 {data['detected']}"])
            self.battery.setText('\n'.join(lines))
        self.log('收到 ' + kind)
        if phase == 'heartbeat':
            self.startup_request('device' if mode == 'crc' else 'read_gain')
        elif phase == 'device':
            self.startup_request('gain_list')
        elif phase in ('gain_list', 'read_gain'):
            self.startup_request('basic')
        elif phase == 'basic':
            self.phase = 'ready'
            self.handshake = True
            self.connection_label.setText('初始化完成，已收到有效电量')
            self.set_connected(True)

    def refresh(self):
        if self.worker and self.phase == 'ready' and self.auto_refresh.isChecked():
            self.send('basic', quiet=True)
        if self.last_reply is not None and time.monotonic() - self.last_reply > 3:
            self.connection_label.setText('最近响应已超过 3 秒；当前显示可能过期')
        if self.battery_received_at is not None:
            age = time.monotonic() - self.battery_received_at
            self.sample_age.setText(f'电量数据于 {age:.0f} 秒前接收；该时间不是芯片采样时间')

    def on_error(self, message):
        self.phase = 'stopping'
        self.handshake = False
        self.log(message)
        self.connection_label.setText('通信失败；当前显示不是新读数')
        self.set_connected(False)

    def on_finished(self):
        worker, self.worker = self.worker, None
        if worker:
            worker.deleteLater()
        self.auto_refresh.setChecked(False)
        self.handshake = False
        self.phase = 'disconnected'
        self.last_reply = None
        self.set_connected(False)
        self.connect_button.setEnabled(True)
        self.connect_button.setText('连接')
        self.connection_label.setText('已断开；保留的读数可能过期')
        if self.closing:
            self.close()

    def closeEvent(self, event):
        if self.worker is not None and self.worker.isRunning():
            self.closing = True
            self.phase = 'stopping'
            self.handshake = False
            self.set_connected(False)
            self.worker.stop()
            event.ignore()  # finished() closes the window after the worker exits.
        else:
            event.accept()


if __name__ == '__main__':
    application = QApplication(sys.argv)
    window = Monitor()
    window.resize(950, 560)
    window.show()
    sys.exit(application.exec_())

扩展格式的启动顺序为心跳、设备信息、档位列表、基础电量；SUM 格式为心跳、当前倍数、基础电量。只有基础电量响应被 GUI 消费后才进入 ready，此前普通按钮、自动刷新与周期心跳均不能插入请求。断开或关闭窗口立即进入 stopping，已排队的旧响应不会恢复控件或继续启动链。

界面用电池响应自身的接收时间计算数据年龄，心跳不能让旧电池数据看起来刚刷新。收到有效响应仅证明本机解码与事务匹配成功，不能证明该响应携带最新芯片样本。

线程停止通过事件和有限 I/O 等待完成，关闭窗口不会在界面线程无限等待。QThread 对象本身仍由创建线程管理，重写 run 并不会把其所有方法自动搬到工作线程；因此 request 使用有界队列与锁，stop 使用线程事件。这个线程归属原则与 Qt 的说明一致；本程序另在实际 PyQt5 事件循环中检查了串口 I/O 的线程归属。Qt QThread 线程归属说明

验证、联调与后续扩展

已运行九组纯协议测试和九组 PyQt5 界面/工作线程测试，包括全部切分位置、连续帧、载荷内伪帧头、错误 CRC/长度、残帧超时重同步、随机噪声下的缓冲区上界、字段范围、部分写入、错误 ACK、事务超时、打开失败、关闭时退出、日志上限和数据年龄。另在实际事件循环中，让用户操作与周期请求在每个初始化结果消费前竞争，并在结果排队时断开或关闭窗口，验证启动链和停止状态。界面测试使用内存模拟串口，没有连接硬件；只能证明参考实现的这些行为，不能把原文“20/50 ms 响应时间”或“通信量减少 80%”作为测量结论。

实际联调可按以下顺序进行：先核对完整十六进制往返，再测试分包/粘包和故意错误校验，随后断开设备验证超时与旧数据提示，最后测量增益 GPIO/模拟输出及电池读数。电量计侧的更新频率、有效标志和配置过程与串口事务独立管理；固件应缓存最近一份有效读数，而不是每次界面请求都立刻重复访问全部寄存器。相关任务划分、F103 内存预算与单独的历史设计见STM32F103 电量与增益控制，不能把两份历史文档中的状态扩展默认视为同一固件契约。

若新增电压、温度、参数保存、多电池类型、自动校准或 OTA，需要定义新协议版本和可测试的执行流程。SUM 的 D2 已承载充电状态或设置结果，直接塞入电压或拼成 16 位增益会破坏现有含义。新版本宜明确能力协商、事务序号、采样时间、有效位、错误响应、增益映射和持久化规则；这些仍属于扩展设计，未包含在上述参考实现中。