# -*- coding: utf-8 -*- """ 电表「上x次整点冻结数据」抄读程序（重写版） ============================================================ 在原 sim.py 基础上重写。**组包 / 解包部分原样保留**： DLT645FrameBuilder DL/T 645-2007 组包 DLT645Parser DL/T 645-2007 解包 LocalProtocolBuilder 南网集中器本地通信规约组包 LocalProtocolParser 南网集中器本地通信规约解包 SerialPort 串口收发 重写的抄读流程（核心变更） ------------------------------------------------------------ 1. 外层循环 = 「表」，内层循环 = 「上1次 → 上2次 → ... 上N次冻结数据」； 2. 电表一旦回复**错误帧**（645 应答帧 ERR=1），立即放弃该表后面所有 「上x次冻结数据」的抄读，直接切换到下一只表； 3. 错误帧不重试（电表已经明确应答，重试没有意义）； 只有 超时 / 无应答 / 帧校验失败 才重试； 这两类问题（拥塞/回放导致的偶发）改为「连续 N 次」才放弃本表， 避免一次偶发超时把整只表 250 多项全部误杀（实测踩过）； 4. 被放弃的冻结项以 SKIPPED 状态记账，最终落盘 4 份文本日志。 【关键概念】TaskID ≠ 上x次任务块（务必分清） ------------------------------------------------------------ · 「上x次任务块」= 内层 645 的数据标识 DI（0504FF01..0504FFFE）＝ 真正的业务身份， 它才是归属应答的唯一依据。 · 外层本地规约的 TaskID = 我们下发时填的一个 16 位**流水号**，模块**原样回显**。 实测（COM4 / 260498001817，同一个 DI 换不同 TaskID 下发）： 上1次 配 TaskID=0x002A(42) → 回复外层 TaskID=42 上14次 配 TaskID=0x0001(1) → 回复外层 TaskID=1 上9次 配 TaskID=0x0100(256) → 回复外层 TaskID=256 上9次 配 TaskID=0x0F00(3840) → 回复外层 TaskID=3840 可见 TaskID 与「上x次」编号毫无关系，模块只是照抄。 之前之所以容易看混，是因为旧脚本的流水号恰好从 1 开始、又恰好按 上1、上2… 顺序下发， 于是日志里出现「TaskID=5 ↔ 上5次」这种**巧合**，让人误以为 TaskID 就是任务块编号。 一旦中途改了起始位置（如 --from 9），这个巧合立刻不成立：上9次 的 TaskID 就是 1。 【本次修复】应答归属彻底不用 TaskID，异常应答也按 DI 归属 ------------------------------------------------------------ 现象：运行中大量出现 [跳过不匹配任务] TaskID=1 期望=14 （实测一次运行出现 76 次），并且本项要白等一个完整超时才靠重发拿到数据。 原因：CCO/模块的上行通道是**积压回放**的 —— 下发第 14 项时，模块先把队列里 第 1..9 项（会话内更早下发的任务）的**旧应答**成批吐出来，真正的第 14 项应答 被排在这些回放帧之后。旧代码拿外层 TaskID 做严格相等判断，于是： ① 这批明明带着真实冻结数据的帧被整批丢弃； ② 当前项一直等不到"自己的"帧 → 判超时 → 靠下一次重发才成功（每次白等 timeout 秒）。 修复： · 归属改用 (表号, DI)：只要内层 645 的 DI 就是当前要抄的 DI，就是答案； TaskID 一律不参与归属（只在日志里作为流水号显示）； · ★ 645 异常应答（ERR=1）也解析 DI：数据域 ≥4 字节时前 4 字节就是 DI， 于是异常帧同样按 (表号, DI) 归属 —— 之前拿 TaskID 归属异常帧， 一旦回放/延迟就会把某一块的异常算到另一块头上； · ★ 异常帧若确实不带 DI（部分表数据域只有 1 字节错误码），只在「正在等本项应答」 的窗口内按上下文归属；窗口外（重试前排空缓冲收到的残留帧）一律不归属、不入缓存， 杜绝被后面别的任务块误吃（旧版用 ("ERR",表号,TaskID or 0) 做键，TaskID=0 时键退化成 通配键，会被**下一个**任务块误弹出来，直接把好数据判成错误帧并中止整表）； · 不属于当前任务的帧不再丢弃 → 进「延迟应答缓存」； · 若它属于本表、且是此前超时没取到的项 → 立即回填（**补抄**），不用重发； · 若该 DI 此前已抄回 → 标 [重复帧]（回放重复），只记 raw 日志； · 每次下发前先查缓存：命中直接用（**不重发**，省一次往返）； · 下发前 drain 串口残留帧入缓存，避免跨任务串帧； · 帧切分带残留缓冲（原来的半帧会被丢弃）； · 识别到「本表回放帧挤占窗口」且静默超过 replay_grace 秒 → 提前结束本次等待进入重发，不再干等整个 timeout； · 21 字节的 C=0xC0 模块状态帧等归为「非抄读帧」，不再报"非法帧/上报帧过短"。 【本次修复2】模块三种"非数据"上行报文的识别与处置（载波调试软件逐字段核对） ------------------------------------------------------------ 下发一条抄读任务后，模块除了回 645 数据帧，还可能回下面这些帧；它们都不含数据， 但**每一条都必须收到我们的确认帧**，否则模块会一直重传： ① 受理回执（确认帧） C=0x80 / PRM=0 / AFN=00 确认·否认 680E00800034010001E800009E16 → 只表示"任务已受理"，数据随后另发。只能记 [确认帧] 跳过， 绝不能判成非法帧/错误帧（早期版本会报"上报帧过短"）。 ② 上报任务状态 C=0xC0 / AFN=05 上报信息 / DUID=050505E8 上报任务状态 681500C00505050505E80100171800980426FFB216 解析：TaskID=0001 从节点地址=260498001817 任务状态=FF(其它) → 模块无法执行该抄读任务时不回数据帧，而回这个。 · 任务状态=00(成功)：继续等数据帧； · 任务状态≠00（如 FF 其它）：本项**不会有数据**，立即结束本次等待， 按"无应答"记账并重试；连续 streak_limit 次 → 按策略放弃本表剩余，切下一只表。 这样就不必再为每一项白等一整个 timeout。 ③ 上报任务数据 C=0xE0（带地址域）/ AFN=05 / DUID=010505E8 682F00E0010000074C5808000000000005D0010505E80100146868010000074C586891… （SEQ=D0、TaskID=0001、报文长度 14H、内层 645 应答）→ 正常解析数据。 ★ 确认帧（回给模块）：C=0x00 下行 / PRM=0 / AFN=00 确认·否认 68 0E 00 00 00 SEQ 01 00 01 E8 00 00 BA 16 ← SEQ=0xD0 时的实例 SEQ 必须**与被确认的那条上报报文完全一致**（取上报帧的"帧序列域SEQ"）。 同一 (源地址, SEQ) 在 ack_dedup 秒内只确认一次：回放风暴时同一帧会被成批吐出， 重复确认没有意义。可用 --no-ack 关闭、--ack-dedup 调窗口。 日志文件（全部纯文本 .txt，直接可读；.log/.csv 会被部分工具当二进制） ------------------------------------------------------------ read_freeze.txt 主日志：流程 + 每帧逐字段解析（含内层 645、冻结块） raw_frames.txt 原始报文流水：一行一帧（TX-LOCAL / RX-LOCAL / RX-645 / GIVE-UP） freeze_blocks.txt 冻结数据块明细：一行一块（时间 + 正反向 + 原始字节） freeze_data.txt 业务数据表（逗号分隔，Excel/WPS 可直接打开） 用法 ------------------------------------------------------------ python buddy.py # 按文件内配置抄读 python buddy.py --port COM3 --meter 260498001817 --meter 260498001820 python buddy.py --from 1 --to 24 # 只抄 上1次~上24次 python buddy.py --selftest # 假串口自测（不占用真实串口） """ from __future__ import annotations import csv import os import sys import time import threading from dataclasses import dataclass, field from datetime import datetime, timedelta from enum import Enum from typing import Dict, List, Optional, Sequence, Tuple try: import serial from serial import SerialException SERIAL_AVAILABLE = True except ImportError: SERIAL_AVAILABLE = False SerialException = Exception # ============================================================ # 数据模型 # ============================================================ @dataclass class MeterAddress: """电表地址（DL/T 645 使用 6 字节 BCD 地址，低位在前）""" addr: str description: str = "" def to_bytes(self) -> bytes: if len(self.addr) != 12: raise ValueError(f"电表地址必须为 12 位 BCD 字符串: {self.addr}") return bytes.fromhex(self.addr)[::-1] @dataclass class DataItem: """一个 645 数据项（DI）""" di: str value: bytes = b"" description: str = "" is_write: bool = False def di_bytes(self) -> bytes: if len(self.di) != 8: raise ValueError(f"数据标识必须为 8 位 HEX: {self.di}") return bytes.fromhex(self.di)[::-1] @dataclass class SendTask: """ 一条待发送的任务 = 一个电表 + 一个「上x次冻结数据块」(DI)。 seq = 本地任务流水号，写进本地规约帧的 TaskID 字段，模块**原样回显**。 ★ 它与 item.di 里的「上x次」编号没有任何关系，绝不能当身份用： 归属只认 (表号, DI)。 """ meter: MeterAddress item: DataItem seq: int = 0 @property def freeze_no(self) -> int: """「上x次」编号 —— 来自 DI 的第 4 字节（0504FFxx）""" di = self.item.di.upper() return int(di[6:8], 16) if len(di) == 8 else -1 @property def label(self) -> str: return (f"[流水号{self.seq:03d}] 表号={self.meter.addr} " f"任务块=上{self.freeze_no}次 DI={self.item.di} ({self.item.description})") class TaskStatus(str, Enum): """单条任务的结局""" OK = "OK" # 正常抄回 METER_ERROR = "ERR_FRAME" # 电表回复错误帧（ERR=1）→ 触发放弃本表 NO_RESPONSE = "NO_RESP" # 超时 / 无应答 BAD_FRAME = "BAD_FRAME" # 帧结构 / 校验 / 解析失败 SKIPPED = "SKIPPED" # 因本表被放弃而未抄读 @dataclass class TaskResult: """单条任务的完整结果""" seq: int meter_addr: str di: str di_name: str = "" status: TaskStatus = TaskStatus.OK request: bytes = b"" response: bytes = b"" retries: int = 0 err_code: Optional[int] = None err_text: str = "" note: str = "" freeze_time: str = "" positive: str = "" reverse: str = "" value: str = "" parsed_local: Optional[dict] = None parsed_645: Optional[dict] = None inner_645: bytes = b"" # 内层 645 应答原始报文 freeze_block: bytes = b"" # 冻结数据块原始 16 字节 @property def success(self) -> bool: return self.status is TaskStatus.OK @dataclass class MeterReport: """单只表的抄读报告""" meter: MeterAddress results: List[TaskResult] = field(default_factory=list) aborted: bool = False abort_reason: str = "" @property def ok_count(self) -> int: return sum(1 for r in self.results if r.status is TaskStatus.OK) @property def skipped_count(self) -> int: return sum(1 for r in self.results if r.status is TaskStatus.SKIPPED) # ============================================================ # 抄读策略：什么情况下放弃该表剩余的「上x次冻结数据」 # ============================================================ @dataclass class AbortPolicy: """ 放弃策略。 语义：A 表在读「上x次冻结数据」过程中命中下面任一情况， 则不再抄读该表剩余的所有冻结项，直接切到下一只表。 ★ 修复要点（实测数据支撑）： · METER_ERROR（电表回 ERR=1 错误帧）是**电表明确应答**「没有这个数据」， 是可靠信号 → 立即放弃本表剩余项。 · NO_RESPONSE / BAD_FRAME 往往是链路拥塞、模块回放积压造成的**偶发**现象。 实测：第 14 项第一次下发被回放帧挤占而超时，但第 15~255 项全部能正常抄回； 若按"单次超时就放弃整表"，这一只表 240 多项就被误杀了。 因此这两类改为「连续 streak_limit 次」才放弃（任何一次成功即清零）。 """ on_error_frame: bool = True # 核心需求：电表回错误帧 → 放弃本表剩余 on_no_response: bool = True # 连续无应答 → 放弃本表剩余 on_bad_frame: bool = True # 连续坏帧 → 放弃本表剩余 streak_limit: int = 5 # 「连续 N 次」阈值；<=0 表示永不因该类问题放弃 def should_abort(self, status: TaskStatus, streak: int = 1) -> bool: if status is TaskStatus.METER_ERROR: return self.on_error_frame if status is TaskStatus.NO_RESPONSE: return self.on_no_response and 0 < self.streak_limit <= streak if status is TaskStatus.BAD_FRAME: return self.on_bad_frame and 0 < self.streak_limit <= streak return False def reason(self, status: TaskStatus, streak: int = 1) -> str: return { TaskStatus.METER_ERROR: "电表回复错误帧", TaskStatus.NO_RESPONSE: f"连续 {streak} 次无应答（重试耗尽）", TaskStatus.BAD_FRAME: f"连续 {streak} 次应答帧无效（重试耗尽）", }.get(status, str(status)) # ============================================================ # 类 1：DL/T 645 协议组包类 （原样保留） # ============================================================ class DLT645FrameBuilder: """ DL/T 645-2007 协议组包类 帧格式： 68H | A0..A5 | 68H | C | L | DATA | CS | 16H """ START = 0x68 END = 0x16 CTRL_READ = 0x11 CTRL_WRITE = 0x14 def __init__(self, plus33: bool = True): self.plus33 = plus33 @staticmethod def _checksum(data: bytes) -> int: return sum(data) & 0xFF def _encode_data_field(self, data: bytes) -> bytes: if not self.plus33: return data return bytes((b + 0x33) & 0xFF for b in data) def build_read(self, meter: MeterAddress, item: DataItem) -> bytes: return self._build(meter, item, self.CTRL_READ) def build_write(self, meter: MeterAddress, item: DataItem) -> bytes: if not item.value: raise ValueError("写操作必须提供 value") return self._build(meter, item, self.CTRL_WRITE) def build(self, meter: MeterAddress, item: DataItem) -> bytes: return self.build_write(meter, item) if item.is_write else self.build_read(meter, item) def _build(self, meter: MeterAddress, item: DataItem, ctrl: int) -> bytes: addr_bytes = meter.to_bytes() di_bytes = item.di_bytes() data_field = di_bytes + (item.value if ctrl == self.CTRL_WRITE else b"") data_encoded = self._encode_data_field(data_field) length = len(data_encoded) head = bytes([self.START]) + addr_bytes + bytes([self.START]) mid = bytes([ctrl, length]) + data_encoded cs = self._checksum(head + mid) return head + mid + bytes([cs, self.END]) # ============================================================ # 类 2：DL/T 645 协议解析类 （原样保留） # ============================================================ class DLT645Parser: """ DL/T 645-2007 应答报文解析类 帧： 68H | A0..A5 | 68H | C | L | DATA | CS | 16H 正常应答帧（ERR 位=0）：解析 DI + 数据。 异常应答帧（ERR 位=1）：数据域同样是「DI(4B) + 数据」，只要长度够就把 DI 解析出来 （字段 has_di=True），供上层按「表号 + DI」归属；只有数据域短于 4 字节时才没有 DI。 """ START = 0x68 END = 0x16 DI_NAMES = { "00000000": "组合有功总电能", "00010000": "当前正向有功总电能", "00020000": "当前反向有功总电能", "00030000": "当前正向无功总电能", "00040000": "当前反向无功总电能", "0504FF01": "上一次整点冻结", } # 645 应答异常码（ERR=1 时，数据域第一字节为异常码） ERR_CODES = { 0x01: "其他错误", 0x02: "无请求数据", 0x03: "密码错/未授权", 0x04: "通信速率不能更改", 0x05: "年时区数超", 0x06: "日时段数超", 0x07: "费率数超", } def parse(self, frame: bytes, plus33: bool = True) -> dict: # ---------- 1. 基本结构校验 ---------- if len(frame) < 12: raise ValueError("645 帧过短") if frame[0] != self.START or frame[7] != self.START: raise ValueError("645 起始字节错误") if frame[-1] != self.END: raise ValueError("645 结束字节错误") addr = frame[1:7] ctrl = frame[8] length = frame[9] data_encoded = frame[10:10 + length] cs = frame[10 + length] end = frame[11 + length] if plus33: data = bytes((b - 0x33) & 0xFF for b in data_encoded) else: data = data_encoded cs_calc = sum(frame[0:10 + length]) & 0xFF dir_bit = (ctrl >> 7) & 1 err_bit = (ctrl >> 6) & 1 follow = (ctrl >> 5) & 1 func = ctrl & 0x1F # ---------- 2. 公共字段 ---------- result = { "raw": frame.hex().upper(), "addr_bytes": addr.hex().upper(), "addr": addr[::-1].hex().upper(), "C": ctrl, "DIR": dir_bit, "ERR": err_bit, "FOLLOW": follow, "FUNC": func, "L": length, "DataHex": data.hex().upper(), "PayloadHex": "", "DI": "", "DI_name": "", "has_di": False, "Value": "", "ValueBCD": "", "kind": "raw", "abnormal": False, "err_code": None, "err_text": "", "CS": cs, "CS_calc": cs_calc, "CS_ok": cs == cs_calc, "END": end, } # ---------- 3. 异常应答帧 ---------- # ★ 异常应答的数据域结构与正常应答一样，都是「数据标识 DI(4B) + 数据」： # · L >= 4 → 前 4 字节（低字节在前）就是 DI，可据此准确归属到「上x次任务块」； # 若 L >= 5，紧随其后的 1 字节为异常码。 # · L < 4 → 帧里确实没有 DI（少数表只回 1 字节异常码），只能由上层按上下文归属。 # 旧版这里直接 return，把 DI 丢掉了 —— 这正是「异常帧归属只能靠 TaskID」的由来， # 也是 TaskID 与 上x次任务块 被看混的根源之一。 if err_bit == 1: result["abnormal"] = True if len(data) >= 4: di = data[0:4][::-1].hex().upper() result["DI"] = di result["DI_name"] = self.DI_NAMES.get(di, self._freeze_di_name(di)) result["has_di"] = True if len(data) >= 5: code = data[4] result["err_code"] = code result["err_text"] = self.ERR_CODES.get(code, f"未知错误码 0x{code:02X}") return result result["has_di"] = False # 异常应答时数据域通常 1 字节，为错误码 if len(data) >= 1: code = data[0] result["err_code"] = code result["err_text"] = self.ERR_CODES.get(code, f"未知错误码 0x{code:02X}") return result # ---------- 4. 正常应答帧 ---------- # 数据域至少要 4 字节 DI if len(data) < 4: result["abnormal"] = True result["err_text"] = f"数据域过短（{len(data)} 字节），无法取 DI" return result di = data[0:4][::-1].hex().upper() payload = data[4:] result["DI"] = di result["DI_name"] = self.DI_NAMES.get(di, self._freeze_di_name(di)) result["PayloadHex"] = payload.hex().upper() extra = self._parse_payload(di, payload) result.update(extra) return result # ======================================================== # 内部：按 DI 分发 # ======================================================== def _parse_payload(self, di: str, payload: bytes) -> dict: if not payload: return {"kind": "raw"} if di.startswith("0504FF"): return self._parse_freeze(payload) if di.endswith("0000"): value = self._bcd_to_str(payload) return { "kind": "energy", "ValueBCD": payload[::-1].hex().upper(), "Value": self._format_energy(value, 2), } return { "kind": "raw", "ValueBCD": payload[::-1].hex().upper(), "Value": payload.hex().upper(), } # ======================================================== # 内部：冻结块 # ======================================================== def _parse_freeze(self, payload: bytes) -> dict: """ 一次整点冻结数据块（16 字节，低字节在前存储）： [0:6] 冻结时间，末尾 AA 分隔 [6:11] 正向有功总电能，末尾 AA 分隔 [11:16] 反向有功总电能，末尾 AA 分隔 """ if len(payload) < 16: return {"kind": "freeze", "error": "冻结块长度不足"} t = payload[0:6][::-1] # [AA, 年, 月, 日, 时, 分] year = self._bcd(t[1]) month = self._bcd(t[2]) day = self._bcd(t[3]) hour = self._bcd(t[4]) minute = self._bcd(t[5]) freeze_time = f"{year}年{month:0>2}月{day:0>2}日{hour:0>2}时{minute:0>2}分" e1 = payload[6:11][::-1] positive = self._format_energy(self._bcd_to_str(e1[1:]), 2) e2 = payload[11:16][::-1] reverse = self._format_energy(self._bcd_to_str(e2[1:]), 2) return { "kind": "freeze", "freeze_time": freeze_time, "positive": positive, "reverse": reverse, # 便于调用方用统一的 Value 字段 "Value": f"正:{positive} 反:{reverse}", } # ======================================================== # 内部工具 # ======================================================== @staticmethod def _bcd_to_str(data: bytes) -> str: s = "" for b in data: hi = (b >> 4) & 0x0F lo = b & 0x0F if hi > 9 or lo > 9: return data.hex().upper() s += f"{hi}{lo}" return s @staticmethod def _bcd(b: int) -> int: hi = (b >> 4) & 0x0F lo = b & 0x0F if hi > 9 or lo > 9: return -1 return hi * 10 + lo @classmethod def _format_energy(cls, val_str: str, decimals: int) -> str: if not val_str.isdigit(): return val_str if decimals <= 0: return val_str val_str = val_str.zfill(decimals + 1) return val_str[:-decimals] + "." + val_str[-decimals:] @staticmethod def _freeze_di_name(di: str) -> str: if len(di) == 8 and di.startswith("0504FF"): try: n = int(di[6:8], 16) return f"上{n}次整点冻结数据块" except ValueError: pass return "" # ============================================================ # 类 3：集中器本地通信规约组包类 （原样保留） # ============================================================ class LocalProtocolBuilder: """ 南网集中器本地通信规约组包类（下行） 帧： 68H | L(2B, 小端) | C(1B) | 用户数据区 | CS(1B) | 16H L = 用户数据区长度 + 6 CS = (C + 用户数据区) & 0xFF 用户数据区（下行）： ASRC(6B) 源地址（CCO 地址），原样写入 ADST(6B) 目的地址（STA/电表地址），低字节在前 AFN(1B) + SEQ(1B) DUID(4B) TaskID(2B, LE) PrioResp(1B) Timeout(2B, LE) InnerLen(1B) InnerFrame(NB) """ START = 0x68 END = 0x16 CTRL_DOWN_ADD_TASK = 0x60 AFN_MANAGE = 0x02 DUID_ADD_TASK = bytes.fromhex("010202E8") ASRC_LEN = 6 ADST_LEN = 6 def __init__(self, asrc_addr: str, adst_addr: str, **kwargs): self.asrc = self._addr_raw(asrc_addr, self.ASRC_LEN) self.adst = self._addr_le(adst_addr, self.ADST_LEN) @staticmethod def _addr_raw(addr: str, nbytes: int) -> bytes: need = nbytes * 2 if len(addr) > need: raise ValueError(f"地址 {addr} 超出 {nbytes} 字节") return bytes.fromhex(addr.zfill(need)) @staticmethod def _addr_le(addr: str, nbytes: int) -> bytes: need = nbytes * 2 if len(addr) > need: raise ValueError(f"地址 {addr} 超出 {nbytes} 字节") return bytes.fromhex(addr.zfill(need))[::-1] def build_add_task(self, target_meter, inner_645_frame: bytes, seq: int = 0x09, task_id: int = 0x0001, prio_resp: int = 0x83, timeout: int = 0x000A) -> bytes: user_data = ( self.asrc + self.adst + bytes([self.AFN_MANAGE, seq]) + self.DUID_ADD_TASK + task_id.to_bytes(2, "little") + bytes([prio_resp]) + timeout.to_bytes(2, "little") + bytes([len(inner_645_frame)]) + inner_645_frame ) return self._wrap(user_data, ctrl=self.CTRL_DOWN_ADD_TASK) def build_forward_645(self, target_meter, inner_645_frame, **kwargs) -> bytes: return self.build_add_task(target_meter, inner_645_frame, **kwargs) def _wrap(self, user_data: bytes, ctrl: int) -> bytes: L = len(user_data) + 6 head = bytes([self.START]) + L.to_bytes(2, "little") + bytes([ctrl]) + user_data cs = (ctrl + sum(user_data)) & 0xFF return head + bytes([cs, self.END]) # ============================================================ # 类 4：集中器本地通信规约解析类 （原样保留） # ============================================================ class LocalProtocolParser: """ 南网集中器本地通信规约解析类 支持两类上行帧： 1) 确认帧：C 的 PRM=0（如 0x80），DUID=01000E8?? 无 InnerLen/Inner —— 解析时仅取基本字段 2) 上报帧：C=0xE0（PRM=1），AFN=0x05，带 InnerLen + Inner645 """ START = 0x68 END = 0x16 AFN_REPORT = 0x05 def parse(self, frame: bytes) -> dict: if len(frame) < 6: raise ValueError("帧过短") if frame[0] != self.START: raise ValueError(f"起始字节错误: {frame[0]:02X}") if frame[-1] != self.END: raise ValueError(f"结束字节错误: {frame[-1]:02X}") L = int.from_bytes(frame[1:3], "little") ctrl = frame[3] DIR = (ctrl >> 7) & 1 PRM = (ctrl >> 6) & 1 ADD = (ctrl >> 5) & 1 VER = ctrl & 0x0F # 确认帧（PRM=0）结构不同，简单返回 if PRM == 0: return { "raw": frame.hex().upper(), "L": L, "C": ctrl, "DIR": DIR, "PRM": PRM, "ADD": ADD, "VER": VER, "is_ack": True, "InnerLen": 0, "Inner": "", "InnerBytes": b"", "CS_ok": None, "L_ok": None, } if len(frame) < 26: raise ValueError("上报帧过短") asrc = frame[4:10] adst = frame[10:16] afn = frame[16] seq = frame[17] duid = frame[18:22] task_id = int.from_bytes(frame[22:24], "little") inner_len = frame[24] inner_start = 25 inner_end = inner_start + inner_len inner = frame[inner_start:inner_end] cs = frame[inner_end] end = frame[inner_end + 1] user_data = frame[4:inner_end] cs_calc = (ctrl + sum(user_data)) & 0xFF L_calc = len(user_data) + 6 return { "raw": frame.hex().upper(), "L": L, "L_calc": L_calc, "L_ok": L == L_calc, "C": ctrl, "DIR": DIR, "PRM": PRM, "ADD": ADD, "VER": VER, "is_ack": False, "ASRC_bytes": asrc.hex().upper(), "ASRC": asrc[::-1].hex().upper(), "ADST_bytes": adst.hex().upper(), "ADST": adst[::-1].hex().upper(), "AFN": afn, "SEQ": seq, "DUID": duid.hex().upper(), "TaskID": task_id, "InnerLen": inner_len, "Inner": inner.hex().upper(), "InnerBytes": inner, "CS": cs, "CS_calc": cs_calc, "CS_ok": cs == cs_calc, "END": end, } # ============================================================ # 类 5：串口通信类 （原样保留） # ============================================================ # ---- 上报任务状态帧（C=0xC0 无地址域，AFN=05H 上报信息）---- # 实测样例（载波调试软件帧解析）： # 68 15 00 C0 05 05 05 05 05 E8 01 00 17 18 00 98 04 26 FF B2 16 # AFN=05(上报信息) SEQ=05 DUID=050505E8(上报任务状态) # TaskID=0001 从节点地址=260498001817 任务状态=FF(其它) # 下发抄读任务后，模块若无法执行会回这种帧（而不是 645 数据帧）； # 另有确认帧（C=0x80, PRM=0, AFN=00 确认/否认）表示任务已受理。 TASK_STATUS_CODES = {0x00: "成功", 0xFF: "其它"} def parse_task_status_frame(frame: bytes) -> Optional[dict]: """ 识别并解析「上报任务状态」帧（C=0xC0：上行走廊、PRM=1、无地址域）。 布局（总长 21 字节）： 68 L(2) C AFN(1) SEQ(1) DUID(4) TaskID(2,LE) 从节点地址(6,BCD倒序) 任务状态(1) CS 16 不是该类帧返回 None。 """ if len(frame) != 21 or frame[0] != 0x68 or frame[-1] != 0x16: return None ctrl = frame[3] if (ctrl & 0xC0) != 0xC0 or ((ctrl >> 5) & 1): # 需 PRM=1 且无地址域 return None afn, seq = frame[4], frame[5] duid = frame[6:10] if afn != 0x05 or duid != b"\x05\x05\x05\xE8": # DUID=050505E8 上报任务状态 return None task_id = int.from_bytes(frame[10:12], "little") addr = frame[12:18][::-1].hex().upper() code = frame[18] cs, cs_calc = frame[19], (ctrl + sum(frame[4:19])) & 0xFF return { "kind": "task_status", "C": ctrl, "AFN": afn, "SEQ": seq, "DUID": duid.hex().upper(), "task_id": task_id, "addr": addr, "status_code": code, "status_text": TASK_STATUS_CODES.get(code, f"未知(0x{code:02X})"), "CS": cs, "CS_calc": cs_calc, "CS_ok": cs == cs_calc, } def build_report_ack(seq: int) -> bytes: """ 对模块「上行报文」的确认帧（我们回给模块）。 实测样例：68 0E 00 00 00 D0 01 00 01 E8 00 00 BA 16 C=0x00 下行报文 / PRM=0 来自从动站 / 无地址域 应用数据域 = AFN(00 确认·否认) + SEQ + DUID(01 00 01 E8 确认) + 等待时间(00 00) 要点：**SEQ 必须与被确认的那条上行报文完全一致**，否则模块不认这条确认。 """ ud = bytes([0x00, seq & 0xFF]) + bytes.fromhex("010001E8") + b"\x00\x00" ctrl = 0x00 head = bytes([0x68]) + (len(ud) + 6).to_bytes(2, "little") + bytes([ctrl]) + ud return head + bytes([(ctrl + sum(ud)) & 0xFF, 0x16]) class SerialPortError(Exception): """串口操作异常""" pass class SerialPort: """ 串口通信封装。 默认参数：115200 / 8 数据位 / 偶校验 (E) / 1 停止位 """ def __init__(self, port: str, baudrate: int = 115200, bytesize: int = 8, parity: str = "E", stopbits: int = 1, timeout: float = 1.0, write_timeout: float = 1.0, verbose: bool = True): if not SERIAL_AVAILABLE: raise SerialPortError("未安装 pyserial，请执行: pip install pyserial") self.port = port self.baudrate = baudrate self.bytesize = bytesize self.parity = parity.upper() self.stopbits = stopbits self.timeout = timeout self.write_timeout = write_timeout self.verbose = verbose self._ser: Optional[serial.Serial] = None self._lock = threading.RLock() # ---------- 打开 / 关闭 ---------- def open(self) -> None: with self._lock: if self._ser and self._ser.is_open: return try: self._ser = serial.Serial( port=self.port, baudrate=self.baudrate, bytesize=self.bytesize, parity=self._parity_const(), stopbits=self.stopbits, timeout=self.timeout, write_timeout=self.write_timeout ) self._log(f"[串口] 已打开 {self.port} " f"@{self.baudrate},{self.bytesize}{self.parity}{self.stopbits}") except SerialException as e: raise SerialPortError(f"打开串口失败 {self.port}: {e}") def close(self) -> None: with self._lock: if self._ser and self._ser.is_open: self._ser.close() self._log(f"[串口] 已关闭 {self.port}") def is_open(self) -> bool: return self._ser is not None and self._ser.is_open def __enter__(self): self.open() return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() # ---------- 收发 ---------- def send(self, data: bytes) -> int: if not self.is_open(): raise SerialPortError("串口未打开") with self._lock: try: n = self._ser.write(data) self._ser.flush() self._log(f"[串口][TX] {data.hex().upper()}") return n except SerialException as e: raise SerialPortError(f"发送失败: {e}") def recv(self, expected_len: int = 0, end_flag: Optional[bytes] = None, max_wait: Optional[float] = None) -> bytes: if not self.is_open(): raise SerialPortError("串口未打开") deadline = time.time() + (max_wait if max_wait is not None else self.timeout) buf = bytearray() with self._lock: old_timeout = self._ser.timeout try: self._ser.timeout = 0.05 while time.time() < deadline: chunk = self._ser.read(256) if chunk: buf.extend(chunk) if end_flag and buf.endswith(end_flag): break if expected_len and len(buf) >= expected_len: break else: if not expected_len and not end_flag and buf: break finally: self._ser.timeout = old_timeout data = bytes(buf) # if data: # self._log(f"[串口][RX] {data.hex().upper()}") return data def query(self, request: bytes, expected_len: int = 0, end_flag: Optional[bytes] = b"\x16", max_wait: Optional[float] = None, retries: int = 0, retry_interval: float = 0.15) -> Optional[bytes]: for attempt in range(retries + 1): if attempt > 0: self._log(f"[串口] 第 {attempt} 次重试...") time.sleep(retry_interval) self.send(request) resp = self.recv(expected_len=expected_len, end_flag=end_flag, max_wait=max_wait) if resp: self._log(f"[串口][RX] {resp.hex().upper()}") return resp return None def clear_input(self) -> None: if not self.is_open(): return with self._lock: try: self._ser.reset_input_buffer() except Exception: pass def drain(self, max_wait: float = 0.03) -> bytes: """ 把串口输入缓冲区里**已经到达**的字节全部取走并返回（非阻塞，不等待新数据）。 用途：每次下发任务前先排空上一次交互残留的帧 —— 它们会被当作"归属未知"处理 （正常应答进缓存/补抄；不带 DI 的异常应答直接丢弃），而不是被下一个任务块误吃。 """ if not self.is_open(): return b"" with self._lock: try: n = self._ser.in_waiting if n <= 0: return b"" return self._ser.read(n) except Exception: return b"" # ---------- 内部工具 ---------- def _parity_const(self): mapping = { "N": serial.PARITY_NONE, "E": serial.PARITY_EVEN, "O": serial.PARITY_ODD, "M": serial.PARITY_MARK, "S": serial.PARITY_SPACE, } if self.parity not in mapping: raise SerialPortError(f"不支持的校验方式: {self.parity}") return mapping[self.parity] def _log(self, msg: str): if self.verbose: print(msg) # ============================================================ # 报文日志工具：把原始帧逐字段拆成可读文本 # ============================================================ _P645 = DLT645Parser() def fmt_hex(data: bytes, sep: str = " ") -> str: """字节 → 空格分隔的 HEX 串""" return sep.join(f"{b:02X}" for b in data) def fmt_ts() -> str: """毫秒级时间戳，用于原始报文日志""" return datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] def annotate_645_frame(frame: bytes, plus33: bool = True, indent: str = " ") -> List[str]: """把一帧 645 报文（上行/下行）拆成逐字段说明""" out: List[str] = [] if len(frame) < 12: return [f"{indent}<645 帧过短: {fmt_hex(frame)}>"] addr = frame[1:7] ctrl = frame[8] length = frame[9] data_enc = frame[10:10 + length] cs = frame[10 + length] cs_calc = sum(frame[0:10 + length]) & 0xFF data = bytes((b - 0x33) & 0xFF for b in data_enc) if plus33 else data_enc direction = "上行(表→CCO)" if (ctrl >> 7) & 1 else "下行(CCO→表)" func = ctrl & 0x1F func_name = {0x11: "读数据", 0x14: "写数据"}.get(func, "?") out.append(f"{indent}┌ 645 报文 [{direction}]: {fmt_hex(frame)}") out.append(f"{indent}│ 起始符 68 : {frame[0]:02X}") out.append(f"{indent}│ 表地址 A0-A5: {fmt_hex(addr)} → {addr[::-1].hex().upper()}") out.append(f"{indent}│ 起始符 68 : {frame[7]:02X}") out.append(f"{indent}│ 控制码 C : {ctrl:02X} (DIR={(ctrl >> 7) & 1} " f"ERR={(ctrl >> 6) & 1} FOLLOW={(ctrl >> 5) & 1} FUNC=0x{func:02X} {func_name})") out.append(f"{indent}│ 数据长度 L : {length:02X} ({length} 字节)") out.append(f"{indent}│ 数据域 +33 : {fmt_hex(data_enc)}") out.append(f"{indent}│ 数据域 -33 : {fmt_hex(data)}") if (ctrl >> 6) & 1: # ERR=1 异常应答 code = data[0] if data else None if code is None: out.append(f"{indent}│ ★ 异常应答 : 无错误码") else: txt = DLT645Parser.ERR_CODES.get(code, f"未知错误码 0x{code:02X}") out.append(f"{indent}│ ★ 异常应答 : 错误码 0x{code:02X} {txt}") elif len(data) >= 4: di = data[0:4][::-1].hex().upper() payload = data[4:] name = DLT645Parser.DI_NAMES.get(di) or DLT645Parser._freeze_di_name(di) out.append(f"{indent}│ 数据标识 DI : {di}" + (f" ({name})" if name else "")) if payload: out.append(f"{indent}│ 数据块原始 : {fmt_hex(payload)} ({len(payload)} 字节)") out.extend(_annotate_645_payload(di, payload, indent + "│ ")) elif ((ctrl >> 7) & 1) == 0: out.append(f"{indent}│ 数据块 : 无（下行读请求，仅带 DI）") out.append(f"{indent}│ 校验 CS : {cs:02X} (计算 {cs_calc:02X}) {'OK' if cs == cs_calc else '★MISMATCH★'}") out.append(f"{indent}└ 结束符 16 : {frame[-1]:02X} 帧总长 {len(frame)} 字节") return out def _annotate_645_payload(di: str, payload: bytes, ind: str) -> List[str]: """按 DI 拆解数据块（冻结块逐字段展开）""" lines: List[str] = [] if not payload: return lines if di.startswith("0504FF"): if len(payload) < 16: lines.append(f"{ind}★ 冻结块长度不足：{len(payload)} 字节（应为 16）") return lines p = _P645._parse_freeze(payload) lines.append(f"{ind}冻结块 16B 布局 = 时间[0:6] + 正向有功[6:11] + 反向有功[11:16]") lines.append(f"{ind} 时间 6B : {fmt_hex(payload[0:6])} → {p.get('freeze_time')}") lines.append(f"{ind} 正向 5B : {fmt_hex(payload[6:11])} → {p.get('positive')} kWh") lines.append(f"{ind} 反向 5B : {fmt_hex(payload[11:16])} → {p.get('reverse')} kWh") return lines if di.endswith("0000"): d = _P645._parse_payload(di, payload) lines.append(f"{ind}电能值 : {fmt_hex(payload)} → {d.get('Value')} kWh") return lines lines.append(f"{ind}原始值 : {fmt_hex(payload)}") return lines def annotate_local_frame(frame: bytes, indent: str = " ", with_inner: bool = True) -> List[str]: """ 把一帧南网集中器本地规约报文拆成逐字段说明： 下行请求帧(DIR=0,PRM=1)：ASRC ADST AFN SEQ DUID TaskID Prio Timeout InnerLen Inner 上行业务帧(DIR=1,PRM=1)：ASRC ADST AFN SEQ DUID TaskID InnerLen Inner 确认帧 (PRM=0) ：仅基本字段 """ out: List[str] = [] if len(frame) < 6: return [f"{indent}<本地规约帧过短: {fmt_hex(frame)}>"] L = int.from_bytes(frame[1:3], "little") ctrl = frame[3] DIR, PRM, ADD, VER = (ctrl >> 7) & 1, (ctrl >> 6) & 1, (ctrl >> 5) & 1, ctrl & 0x0F kind = "确认帧" if PRM == 0 else ("下行请求帧" if DIR == 0 else "上行业务帧") out.append(f"{indent}┌ 本地规约·{kind}: {fmt_hex(frame)}") out.append(f"{indent}│ 起始符 68 : {frame[0]:02X}") out.append(f"{indent}│ 长度 L : {fmt_hex(frame[1:3])} (小端) → {L} " f"帧长 {len(frame)} {'OK' if L == len(frame) else '★MISMATCH★'}") out.append(f"{indent}│ 控制码 C : {ctrl:02X} (DIR={DIR} PRM={PRM} ADD={ADD} VER={VER})") # 确认帧：结构不同，只做通用拆解 if PRM == 0: out.append(f"{indent}│ 用户数据 : {fmt_hex(frame[4:-2])}") out.append(f"{indent}│ 校验 CS : {frame[-2]:02X} (计算 {sum(frame[:-2]) & 0xFF:02X})") out.append(f"{indent}└ 结束符 16 : {frame[-1]:02X}") return out if len(frame) < 26: out.append(f"{indent}└ <帧过短: {len(frame)} 字节>") return out down = (DIR == 0) asrc, adst = frame[4:10], frame[10:16] afn, seq, duid = frame[16], frame[17], frame[18:22] task_id = int.from_bytes(frame[22:24], "little") if down: prio, timeout = frame[24], int.from_bytes(frame[25:27], "little") inner_len = frame[27] inner = frame[28:28 + inner_len] cs = frame[28 + inner_len] end = frame[29 + inner_len] user_data = frame[4:28 + inner_len] else: inner_len = frame[24] inner = frame[25:25 + inner_len] cs = frame[25 + inner_len] end = frame[26 + inner_len] user_data = frame[4:25 + inner_len] cs_calc = (ctrl + sum(user_data)) & 0xFF out.append(f"{indent}│ ASRC 源 : {fmt_hex(asrc)} → {asrc[::-1].hex().upper()}") out.append(f"{indent}│ ADST 目 : {fmt_hex(adst)} → {adst[::-1].hex().upper()}") out.append(f"{indent}│ AFN / SEQ : {afn:02X} / {seq:02X}") out.append(f"{indent}│ DUID : {duid.hex().upper()}") out.append(f"{indent}│ TaskID : 0x{task_id:04X} ({task_id}) —— 流水号，模块原样回显，" f"与「上x次」编号无关") if down: out.append(f"{indent}│ 优先级/超时: PrioResp={prio:02X} Timeout={timeout}") out.append(f"{indent}│ 内层长度 : {inner_len}") out.append(f"{indent}│ 校验 CS : {cs:02X} (计算 {cs_calc:02X}) {'OK' if cs == cs_calc else '★MISMATCH★'}") out.append(f"{indent}└ 结束符 16 : {end:02X} 帧总长 {len(frame)} 字节") if with_inner: if inner[:1] == b"\x68" and inner[-1:] == b"\x16": out.append(f"{indent} ▼ 内层 645 报文（查表归属身份取这里的 DI）") out.extend(annotate_645_frame(inner, indent=indent + " ")) elif inner: out.append(f"{indent} ▼ 内层非 645 数据: {fmt_hex(inner)}") return out def _di_x(di: str) -> str: """0504FFxx → 第几次（字符串，仅日志用）""" try: return str(int(di[6:8], 16)) except Exception: return "?" def _peek_di(outer: bytes) -> str: """从下行本地规约帧里窥探内层 645 的 DI（仅用于日志）""" try: inner_len = outer[27] inner = outer[28:28 + inner_len] di_enc = inner[10:14] return bytes((b - 0x33) & 0xFF for b in di_enc)[::-1].hex().upper() except Exception: return "?" def _peek_di_up(inner_645: bytes) -> str: """从上行 645 帧里窥探 DI（错误帧会返回 ?，仅用于日志）""" try: if inner_645[8] & 0x40: code = (inner_645[10] - 0x33) & 0xFF return f"<异常应答 0x{code:02X}>" length = inner_645[9] data = bytes((b - 0x33) & 0xFF for b in inner_645[10:10 + length]) return data[0:4][::-1].hex().upper() except Exception: return "?" # ============================================================ # 类 6：抄读会话（重写核心） # ============================================================ class FreezeReadSession: """ 「上x次整点冻结数据」抄读会话。 for meter in meters: # 外层：表 for item in freeze_items: # 内层：上1次 → 上N次 read(item) 若命中 AbortPolicy → 放弃该表剩余全部冻结项，break 到下一只表 """ CSV_HEADER = ["表号", "表名", "序号", "DI", "数据项", "状态", "冻结时间", "正向有功(kWh)", "反向有功(kWh)", "数值", "重试", "错误码", "错误信息", "645应答报文", "冻结块原始数据(16B)"] def __init__(self, asrc_addr: str = "000000000000", transport=None, policy: Optional[AbortPolicy] = None, timeout: float = 10.0, max_retries: int = 2, retry_interval: float = 0.3, retry_on_error_frame: bool = False, freeze_start: int = 1, freeze_end: int = 255, # 注意：全部用 .txt 扩展名，纯文本可直接打开/预览； # .log 会被某些读取工具当成二进制而无法查看。 csv_path: Optional[str] = "freeze_data.csv", log_path: Optional[str] = "read_freeze.txt", raw_log_path: Optional[str] = "raw_frames.txt", freeze_log_path: Optional[str] = "freeze_blocks.txt", dump_frames: bool = True, dump_console: bool = False, raw_console: bool = False, replay_grace: float = 1.0, ack_reports: bool = True, ack_dedup: float = 0.0, verbose: bool = True): self.asrc_addr = asrc_addr self.transport = transport self.policy = policy or AbortPolicy() self.timeout = timeout self.max_retries = max_retries self.retry_interval = retry_interval self.retry_on_error_frame = retry_on_error_frame self.replay_grace = replay_grace # 回放帧挤占窗口后，静默多久就提前重发（<=0 关闭） self.ack_reports = ack_reports # ★ 每条上行报文都回确认帧（SEQ 与上报帧一致） self.ack_dedup = ack_dedup # 同一 (源地址,SEQ) 去重窗口，0=每次上报都回（默认） self.freeze_start = freeze_start self.freeze_end = freeze_end # ---- 日志目标 ---- self.csv_path = csv_path # 业务数据 CSV self.log_path = log_path # 主日志（含逐字段报文解析） self.raw_log_path = raw_log_path # 原始报文流水（一行一帧） self.freeze_log_path = freeze_log_path # 冻结数据块（一行一块） self.dump_frames = dump_frames # 是否把逐字段解析写日志 self.dump_console = dump_console # 逐字段解析是否也打控制台 self.raw_console = raw_console # 原始报文是否也打控制台 self.verbose = verbose # 控制台总开关 self.dlt645 = DLT645FrameBuilder(plus33=True) self.dlt645_parser = DLT645Parser() self.local_parser = LocalProtocolParser() self._seq = 0x09 # 本地规约 SEQ self._task_no = 0 # 全局任务序号（0x0001 起） # ---- 应答归属 / 缓存（本次修复核心）---- self._rx_resid = bytearray() # 帧切分残留缓冲（不丢半帧） self._cache: Dict[Tuple[str, str], dict] = {} # (表号, DI) → 延迟应答 self._got: Dict[Tuple[str, str], TaskResult] = {} # 已成功抄回的项（判重） self._wait_fill: Dict[Tuple[str, str], TaskResult] = {} # 尚未取到、等待补抄的项 self._late_fill = 0 # 补抄成功次数 self._dup_frames = 0 # 回放重复帧次数 self._status_fail: Optional[dict] = None # 最近一次「任务状态=失败」上报帧 self._acked: Dict[Tuple[str, int], float] = {} # (源地址, SEQ) → 上次回确认的时间 self._ack_frames = 0 # 已回确认帧数 self._cache_hits = 0 # 缓存命中（省掉一次重发）次数 self._fp = None self._raw_fp = None self._freeze_fp = None self._lock = threading.RLock() # ---------------- 日志 / 资源 ---------------- @staticmethod def _append(path: Optional[str], line: str) -> None: if not path: return with open(path, "a", encoding="utf-8") as f: f.write(line + "\n") def _log(self, msg: str = "") -> None: """主日志（人可读流程）""" if self.verbose: print(msg) if self.log_path: with self._lock: if self._fp is None: self._fp = open(self.log_path, "a", encoding="utf-8") self._fp.write(msg + "\n") self._fp.flush() def _dump(self, title: str, lines: Sequence[str]) -> None: """逐字段报文解析块 → 主日志""" if not self.dump_frames: return block = [f"{fmt_ts()} | {title}"] + list(lines) if self.log_path: with self._lock: if self._fp is None: self._fp = open(self.log_path, "a", encoding="utf-8") self._fp.write("\n".join(block) + "\n") self._fp.flush() if self.dump_console: print("\n".join(block)) def _raw(self, tag: str, frame: bytes, note: str = "") -> None: """原始报文流水日志：一行一帧，便于和抓包工具比对""" line = (f"{fmt_ts()} | {tag:<10} | len={len(frame):<3} | " f"{note:<44} | {fmt_hex(frame)}") if self.raw_log_path: with self._lock: if self._raw_fp is None: self._raw_fp = open(self.raw_log_path, "a", encoding="utf-8") self._raw_fp.write(line + "\n") self._raw_fp.flush() if self.raw_console: print(line) def _freeze(self, r: "TaskResult", tag: str = "") -> None: """冻结数据块明细日志：一行一块（时间 + 正反向 + 原始字节） tag 非空时在状态后追加标记，例如「补抄」——表示该行是迟到应答回填后的最终结果。 """ if not self.freeze_log_path: return blk = r.freeze_block seg = "" if len(blk) >= 16: seg = (f"时间[{fmt_hex(blk[0:6])}] 正[{fmt_hex(blk[6:11])}] 反[{fmt_hex(blk[11:16])}]") x = r.di[6:8] status = r.status.value + (f"({tag})" if tag else "") line = (f"{fmt_ts()} | 表={r.meter_addr} | 上{int(x, 16) if x else '?'}次 | DI={r.di} | " f"状态={status} | {r.freeze_time or '-'} | " f"正向={r.positive or '-'}kWh | 反向={r.reverse or '-'}kWh | {seg or '-'}") self._append(self.freeze_log_path, line) def close(self) -> None: with self._lock: for name in ("_fp", "_raw_fp", "_freeze_fp"): fp = getattr(self, name) if fp is not None: fp.close() setattr(self, name, None) def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() # ---------------- 会话头 ---------------- def log_banner(self, meters: Sequence[MeterAddress], items: Sequence[DataItem]) -> None: port = getattr(self.transport, "port", None) self._log("=" * 78) self._log(f"会话开始 {fmt_ts()}") self._log(f" 通信方式 : {'串口 ' + port if port else type(self.transport).__name__}") self._log(f" CCO 地址 : {self.asrc_addr}") self._log(f" 电表数量 : {len(meters)} {[m.addr for m in meters]}") self._log(f" 冻结项 : 上{self.freeze_start}次 ~ 上{self.freeze_end}次，共 {len(items)} 项") self._log(f" 超时/重试 : timeout={self.timeout}s, max_retries={self.max_retries}, " f"retry_interval={self.retry_interval}s, replay_grace={self.replay_grace}s") self._log(f" 错误帧重试 : {self.retry_on_error_frame}") self._log(f" 放弃策略 : 错误帧={self.policy.on_error_frame}(立即) " f"无应答={self.policy.on_no_response}(连续{self.policy.streak_limit}次) " f"坏帧={self.policy.on_bad_frame}(连续{self.policy.streak_limit}次)") self._log(f" 应答归属 : 只按「表号 + DI(上x次任务块)」判定；外层 TaskID 是模块" f"原样回显的流水号，不参与归属") self._log(f" （迟到应答入缓存→补抄；重复回放帧忽略；不带 DI 的异常帧只认" f"当前等待窗口，窗口外丢弃）") self._log(f" 上报确认 : {'启用' if self.ack_reports else '关闭'}" f"（模块每条上报都回确认帧，SEQ 与上报帧一致" f"{'；同 SEQ %gs 内去重' % self.ack_dedup if self.ack_dedup > 0 else '，不去重'}）") self._log(f" 任务状态帧 : AFN=05/DUID=E805050 上报任务状态 → 非 0 状态立即判本项失败" f"（不白等超时），连续 {self.policy.streak_limit} 次按策略放弃本表") self._log(f" 日志文件 : 主日志={self.log_path} 原始报文={self.raw_log_path} " f"冻结块={self.freeze_log_path} CSV={self.csv_path}") self._log("=" * 78) # ---------------- 冻结数据项清单 ---------------- def freeze_items(self) -> List[DataItem]: """上1次 .. 上N次整点冻结数据（0504FF01 .. 0504FFxx，按顺序抄）""" return [ DataItem(di=f"0504FF{i:02X}", description=f"上{i}次整点冻结数据块") for i in range(self.freeze_start, self.freeze_end + 1) ] # ---------------- 组包（复用原组包类） ---------------- def build_frames(self, task: SendTask) -> Dict[str, bytes]: inner = self.dlt645.build(task.meter, task.item) builder = LocalProtocolBuilder( asrc_addr=self.asrc_addr, adst_addr=task.meter.addr, ) outer = builder.build_add_task( target_meter=task.meter, inner_645_frame=inner, seq=self._seq & 0xFF, task_id=task.seq & 0xFFFF, ) return {"645": inner, "local": outer} # ---------------- 帧切分（带残留缓冲，不再丢半帧） ---------------- def _feed(self, chunk: bytes) -> List[bytes]: """ 把新到达的字节追加进残留缓冲区，切出所有完整 68...16 帧（按 L 字段定长）。 尾部不完整的帧保留在缓冲区里等下一批字节（旧版直接丢弃，会丢帧/错帧）。 """ if chunk: self._rx_resid.extend(chunk) buf = self._rx_resid frames: List[bytes] = [] i, n = 0, len(buf) while i < n: if buf[i] != 0x68: i += 1 continue if i + 3 > n: break L = int.from_bytes(buf[i + 1:i + 3], "little") if L < 6: i += 1 continue if i + L > n: break frames.append(bytes(buf[i:i + L])) i += L del buf[:i] if len(buf) > 4096: # 防御：长期无有效帧时截断，避免缓冲无限增长 self._raw("RX-BAD", bytes(buf[:64]), f"残留缓冲异常增长({len(buf)}B)，已截断") del buf[:len(buf) - 512] return frames # ---------------- 归属工具 ---------------- @staticmethod def _norm_addr(addr: str) -> str: return (addr or "").strip().upper().zfill(12) def _task_key(self, task: SendTask) -> Tuple[str, str]: return (self._norm_addr(task.meter.addr), task.item.di.upper()) # ---------------- 上行帧解析 / 归属路由 ---------------- def _parse_uplink(self, frame: bytes) -> Optional[dict]: """ 解析一条上行本地规约帧 → {表号, DI, 两级解析结果, TaskID, 是否异常应答}。 确认帧 / 非抄读帧 / 非法帧返回 None（并各自归类记日志）。 """ if len(frame) < 4: self._raw("RX-BAD", frame, "帧过短") return None # 1) 确认帧（PRM=0）→ 跳过（AFN=00 = 确认/否认，表示任务已受理，数据随后另发） if ((frame[3] >> 6) & 1) == 0: afn = frame[4] if len(frame) > 4 else None afn_txt = {0x00: "AFN=00 确认/否认"}.get(afn, f"AFN={afn:02X}" if afn is not None else "") self._log(f" [确认帧] {frame.hex().upper()}（{afn_txt}，任务已受理，等待数据帧）") self._raw("RX-ACK", frame, f"确认帧(PRM=0 {afn_txt})，跳过") if self.dump_frames: self._dump("· 确认帧", annotate_local_frame(frame, with_inner=False)) return None # 1.5) 上报任务状态帧（C=0xC0 / AFN=05 / DUID=050505E8） # 模块无法执行抄读任务时不回 645 数据帧，而是回这种帧（如 任务状态=FF 其它） ts = parse_task_status_frame(frame) if ts is not None: if not ts["CS_ok"]: self._log(f" [任务状态帧CS错] {frame.hex().upper()}") self._raw("RX-BAD", frame, f"任务状态帧 CS={ts['CS']:02X} 应为 {ts['CS_calc']:02X}") return None self._ack_report(ts["SEQ"], ts["addr"]) # ★ 先确认，再处置 return {"status_report": True, **ts} # 2) 上报帧 → 解析 try: p = self.local_parser.parse(frame) except Exception as e: # 21 字节的 C=0xC0 帧（模块状态/回执，非抄读数据）单独归类，不算"非法帧" if len(frame) < 26: self._log(f" [非抄读帧] {frame.hex().upper()} ({e})") self._raw("RX-MISC", frame, f"非抄读帧: {e}") else: self._log(f" [跳过非法帧] {frame.hex().upper()} ({e})") self._raw("RX-BAD", frame, f"非法帧: {e}") return None # ★ 本地规约合法的上行报文 → 立即回确认帧（SEQ 取自本帧，模块据此停止重传） # 校验错（CS/L）的帧不回确认，让模块重传，避免"确认了一条坏帧" if p.get("CS_ok") is not False: self._ack_report(p.get("SEQ") or 0, p.get("ASRC") or "") else: self._log(f" [不上报确认] 本地规约 CS 校验失败，等待模块重传") inner = p.get("InnerBytes", b"") if not inner or inner[:1] != b"\x68" or inner[-1:] != b"\x16": self._log(f" [非抄读帧] {frame.hex().upper()}（无 645 内层）") self._raw("RX-MISC", frame, "无 645 内层，非抄读帧") return None try: d = self.dlt645_parser.parse(inner) except Exception as e: self._log(f" [跳过非法645帧] {inner.hex().upper()} ({e})") self._raw("RX-BAD", inner, f"645 非法帧: {e}") return None return { "local": p, "645": d, "inner": inner, "frame": frame, "addr": self._norm_addr(d.get("addr", "")), "di": (d.get("DI") or "").upper(), "has_di": bool(d.get("has_di")), "task_id": p.get("TaskID") or 0, "abnormal": bool(d.get("abnormal")) or d.get("ERR", 0) == 1, } def _route(self, frame: bytes, task: Optional[SendTask]) -> Tuple[Optional[bytes], bool]: """ ★ 核心：上行帧归属路由。身份只有一个 —— (表号, DI)=「上x次任务块」。 规则（TaskID 完全不参与归属）： · 正常应答 / 带 DI 的异常应答：归属键 = (表号, DI)。 只要内层 DI 就是当前在抄的 DI，就是答案 —— 模块回放/延迟会让外层 TaskID 滞后， 拿它判断会误杀真实数据。 · 不是当前任务块的帧 → 进缓存，而不是丢弃： - 属于本表、且是此前超时未取到的项 → 立即回填（补抄）； - 该 DI 此前已抄回 → 记 [重复帧]（回放重复）。 · 不带 DI 的异常应答（数据域只有错误码）：只在「正在等当前项」时按上下文归属； 窗口外收到的一律不归属、不入缓存（否则会被后续别的任务块误吃）。 返回 (命中当前任务的原始帧 或 None, 该帧是否属于当前表)。 """ rep = self._parse_uplink(frame) if rep is None: return None, False same_meter = task is not None and rep.get("addr") == self._norm_addr(task.meter.addr) # ---- 上报任务状态帧（模块受理/执行结果汇报，不含 645 数据）---- if rep.get("status_report"): tid, addr = rep["task_id"], rep["addr"] desc = f"任务状态={rep['status_text']}(0x{rep['status_code']:02X})" if rep["status_code"] == 0x00: # 成功 → 数据帧随后另发，继续等即可 self._log(f" [任务状态] TaskID=0x{tid:04X} 表={addr} {desc}" f"（执行成功，等待数据帧）") self._raw("RX-STATUS", frame, f"任务状态 {desc}（等待数据帧）") return None, same_meter if (task is not None and tid == (task.seq & 0xFFFF) and addr == self._norm_addr(task.meter.addr)): # ★ 当前任务块执行失败（如 FF-其它）→ 不会再有数据帧，提前结束等待 self._status_fail = rep self._log(f" [任务状态] TaskID=0x{tid:04X} 表={addr} {desc}" f" → 当前任务块 上{_di_x(task.item.di)}次 不会返回数据，提前结束等待") self._raw("RX-STATUS", frame, f"任务状态 {desc}（当前任务块失败，提前结束等待）") return None, True self._log(f" [任务状态] TaskID=0x{tid:04X} 表={addr} {desc}" f"（非当前任务块，忽略）") self._raw("RX-STATUS", frame, f"任务状态 {desc}（非当前任务块，忽略）") return None, False # ---- 645 异常应答（ERR=1）---- if rep["abnormal"]: tid = rep["task_id"] di = rep["di"] code = rep["645"].get("err_code") text = rep["645"].get("err_text", "") desc = f"{f'0x{code:02X} ' if code is not None else ''}{text}".strip() # (a) 异常帧里带 DI → 和正常应答一样按 (表号, DI) 归属，与 TaskID 无关 if di: blk = f"上{_di_x(di)}次(DI={di})" if task is not None and (rep["addr"], di) == self._task_key(task): self._log(f" [命中] 645 异常应答 {desc} → 任务块 {blk}" f"（外层TaskID=0x{tid:04X} 仅为回显，不参与归属）") return frame, True ekey = ("ERR", rep["addr"], di) if ekey in self._cache: self._dup_frames += 1 self._log(f" [重复帧] 645 异常应答 {desc} 任务块 {blk} 已在缓存，忽略") self._raw("RX-DUP", frame, f"重复异常应答 {blk}") return None, same_meter self._cache[ekey] = rep cur = self._task_key(task)[1] if task is not None else "-" self._log(f" [缓存延迟异常应答] 任务块 {blk} {desc} → 入缓存" f"（当前任务块 DI={cur}），轮到该项时再取用") self._raw("RX-CACHE", frame, f"延迟异常应答 {blk} {desc} 入缓存") return None, same_meter # (b) 异常帧不带 DI → 只能按上下文归属 if task is not None: self._log(f" [命中] 645 异常应答 {desc}（帧内无 DI，按上下文归属到当前任务块" f" 上{_di_x(task.item.di)}次 DI={task.item.di}）") return frame, same_meter self._log(f" [丢弃] 645 异常应答 {desc}：帧内无 DI，且当前没有在等的任务块 ——" f"不作归属，避免被后面的任务块误吃") self._raw("RX-DROP", frame, f"无DI异常应答 {desc}，无上下文可归属，丢弃") return None, False # ---- 正常应答：按 (表号, DI)=任务块 归属 ---- key = (rep["addr"], rep["di"]) if task is not None and key == self._task_key(task): self._log(f" [命中] 任务块 上{_di_x(rep['di'])}次(DI={rep['di']}) 表={rep['addr']}" f"（外层TaskID=0x{rep['task_id']:04X} 仅为流水号回显）") return frame, True if key in self._got: self._dup_frames += 1 self._log(f" [重复帧] 任务块 上{_di_x(rep['di'])}次(DI={rep['di']}) 此前已抄回，忽略") self._raw("RX-DUP", frame, f"重复帧 DI={rep['di']}（已抄回，忽略）") return None, same_meter if key in self._cache: self._dup_frames += 1 self._log(f" [重复帧] 任务块 上{_di_x(rep['di'])}次(DI={rep['di']}) 已在缓存中，忽略") self._raw("RX-DUP", frame, f"重复帧 DI={rep['di']}（已在缓存，忽略）") return None, same_meter # 属于本表、且是此前超时没取到的项 → 立即补抄回填（不入缓存） tgt = self._wait_fill.get(key) if tgt is not None: self._patch_late(tgt, rep) return None, same_meter self._cache[key] = rep cur = self._task_key(task)[1] if task is not None else "-" self._log(f" [缓存延迟应答] DI={rep['di']}(上{_di_x(rep['di'])}次) 与当前任务" f"(DI={cur})不符 → 入缓存，继续等当前项") self._raw("RX-CACHE", frame, f"延迟应答 DI={rep['di']} 入缓存（当前任务 DI={cur}）") return None, same_meter def _take_cache(self, task: SendTask) -> Optional[dict]: """ 取属于该「任务块(表号+DI)」的延迟应答（有则用，没有则 None）。 ★ 只用 (表号, DI) 做键。旧版还有一个 ("ERR", 表号, TaskID) 的键 + 一个 ("ERR", 表号, 0) 的通配兜底 —— 那正是「拿 TaskID 当任务块身份」的残留： 异常帧的 TaskID 来自别的会话/别的项，永远匹配不上，却又会被通配键 弹出来喂给**下一个**任务块，把正常数据判成错误帧并中止整表。 """ addr, di = self._task_key(task) rep = self._cache.pop((addr, di), None) if rep is not None: return rep return self._cache.pop(("ERR", addr, di), None) # ---------------- 对模块上报报文回确认帧 ---------------- def _ack_report(self, seq: int, src_addr: str = "") -> None: """ 模块每上报一条报文（上报任务数据 / 上报任务状态），都要回一条确认帧， 且确认帧的 SEQ 必须与该上报报文一致（否则模块会一直重传）。 去重：同一 (源地址, SEQ) 在 ack_dedup 秒内只确认一次 —— 回放风暴时 同一帧会被成批吐出，重复确认没有意义，只会占用总线时间。 """ if not self.ack_reports or self.transport is None: return now = time.time() key = (self._norm_addr(src_addr) if src_addr else "", seq & 0xFF) prev = self._acked.get(key) if prev is not None and now - prev < self.ack_dedup: self._raw("TX-ACK-DUP", b"", f"SEQ=0x{seq:02X} 已在 {self.ack_dedup:g}s 内确认过，跳过重复确认") return self._acked[key] = now ack = build_report_ack(seq) try: self.transport.send(ack) except Exception as e: # 确认帧发失败不该影响抄读主流程 self._log(f" · 确认帧发送失败（{e}），继续等待数据帧") self._raw("TX-ACK-ERR", ack, f"确认帧发送失败: {e}") return self._ack_frames += 1 self._log(f" → 回确认帧（SEQ=0x{seq:02X} 与上报报文一致）: {ack.hex().upper()}") self._raw("TX-ACK", ack, f"确认模块上报报文 SEQ=0x{seq:02X}") def _drain_to_cache(self, max_wait: float = 0.03) -> None: """把串口缓冲区里已到达的残留帧全部收进缓存（而不是丢弃）""" if self.transport is None: return drain = getattr(self.transport, "drain", None) if drain is None: return try: chunk = drain(max_wait) except TypeError: chunk = drain() except Exception: return if not chunk: return for f in self._feed(chunk): self._route(f, None) def _fill_from_rep(self, r: TaskResult, rep: dict) -> None: """用一条应答填充 TaskResult（正常应答路径）""" d = rep["645"] r.status = TaskStatus.OK r.response = rep["frame"] r.parsed_local = rep["local"] r.parsed_645 = d r.inner_645 = rep["inner"] r.err_code = None r.err_text = "" r.note = "补抄：应答迟到到达后回填" try: r.freeze_block = bytes.fromhex(d["PayloadHex"]) if d.get("PayloadHex") else b"" except ValueError: r.freeze_block = b"" r.freeze_time = d.get("freeze_time", "") r.positive = d.get("positive", "") r.reverse = d.get("reverse", "") r.value = d.get("Value", "") def _patch_late(self, r: TaskResult, rep: dict) -> None: """★ 补抄：把迟到到达的应答回填到此前判失败的条目上（不用重发）""" old = r.status.value self._fill_from_rep(r, rep) self._late_fill += 1 key = (self._norm_addr(r.meter_addr), r.di.upper()) self._wait_fill.pop(key, None) self._got[key] = r self._log(f" ↩ 补抄成功：上{_di_x(r.di)}次(DI={r.di}) 的应答迟到到达，" f"已回填（原状态 {old}，无需重发）") self._log(f" 冻结时间: {r.freeze_time} 正向: {r.positive} kWh 反向: {r.reverse} kWh") self._raw("RX-FILL", rep["frame"], f"补抄命中 DI={r.di}（原状态 {old}）") self._freeze(r, tag="补抄") # ---------------- 单次交互：发一帧，等到属于本任务的应答 ---------------- def _exchange(self, task: SendTask, outer: bytes, timeout: float, task_label: str = "", attempt: int = 0) -> Optional[bytes]: """ 返回：属于「本表 + 本 DI」的上报帧；超时返回 None。 · 下发前先查缓存 → 命中直接用（不重发） · 等待中收到的其它 DI 帧 → 入缓存/补抄，不丢弃 · 若本表回放帧挤占窗口且静默超过 replay_grace → 提前返回，交给外层重发 """ if self.transport is None: raise RuntimeError("未注入 transport") # ---- 0) 缓存里已有本任务的应答 → 直接取用，省一次往返 ---- rep = self._take_cache(task) if rep is not None and rep.get("abnormal"): # 缓存里的异常应答可能是上一次会话/回放的陈旧结论 —— 中止整表代价太大， # 所以不直接采信，重新下发一次让它再确认。 self._log(f" · 缓存里有一条该任务块的 645 异常应答，但仍重新下发一次确认" f"（避免用陈旧的异常结论中止整表）") self._raw("RX-CACHE-ERR", rep["frame"], "缓存异常应答 → 重新下发确认") rep = None if rep is not None: self._cache_hits += 1 kind = "645 异常应答（ERR=1）" if rep.get("abnormal") else "应答" self._log(f" ✔ 命中延迟应答缓存：任务块 上{_di_x(task.item.di)}次" f"(DI={task.item.di}) 的{kind}此前已到达，本次不重发") self._raw("RX-CACHE-HIT", rep["frame"], f"{task_label} 命中缓存，未重发") return rep["frame"] # ---- 下行：本地规约帧（内含 645 请求）---- self._raw("TX-LOCAL", outer, f"{task_label} 第{attempt + 1}次发送（TaskID 流水号=0x{task.seq & 0xFFFF:04X}）") if self.dump_frames: self._dump(f">>> 下行报文（{task_label}）", annotate_local_frame(outer, with_inner=True)) self._raw("TX-645", outer[28:28 + outer[27]], f"{task_label} 内层645请求 DI={_peek_di(outer)}") self.transport.send(outer) self._status_fail = None # 本次等待窗口内收到的「任务状态=失败」帧 t0 = time.time() deadline = t0 + timeout last_frame_at = None # 最近一次收到本表数据帧的时间 saw_replay = False # 本窗口内是否被本表回放帧挤占过 while True: now = time.time() if now >= deadline: self._log(f" · 等待超时({timeout:g}s)，未收到 DI={task.item.di} 的上报帧") return None # 回放帧挤占窗口：静默一小段时间后提前重发，不再干等整个 timeout if (saw_replay and self.replay_grace > 0 and self.max_retries > attempt and last_frame_at is not None and now - last_frame_at >= self.replay_grace): self._log(f" · 本表回放帧挤占窗口且已静默 {now - last_frame_at:.1f}s，" f"提前结束本次等待（进入重发）") return None resp = self.transport.recv(end_flag=b"\x16", max_wait=min(0.5, deadline - now)) if not resp: continue self._raw("RX-RAW", resp, f"{task_label} 串口收到 {len(resp)} 字节") hit: Optional[bytes] = None for f in self._feed(resp): got, same_meter = self._route(f, task) if got is not None: if hit is None: hit = got elif same_meter: # 收到了本表的帧（延迟/重复）→ 说明链路在动，记录时间 saw_replay = True last_frame_at = time.time() if hit is not None: self._raw("RX-LOCAL", hit, f"{task_label} 命中上报帧（本表+DI={task.item.di}）") self._raw("RX-645", hit[25:25 + hit[24]], f"{task_label} 内层645应答 DI={_peek_di_up(hit[25:25 + hit[24]])}") return hit # 模块上报「任务状态=失败」→ 本项不会有数据帧，立即结束等待（进入重试/记账） if self._status_fail is not None: return None # ---------------- 应答解析：本地规约 + 内层 645 ---------------- def _parse_reply(self, task: SendTask, raw: bytes, attempt: int) -> TaskResult: r = TaskResult(seq=task.seq, meter_addr=task.meter.addr, di=task.item.di, di_name=task.item.description, response=raw, retries=attempt) try: pl = self.local_parser.parse(raw) except Exception as e: r.status = TaskStatus.BAD_FRAME r.err_text = f"本地规约解析失败: {e}" return r r.parsed_local = pl # 本地规约自身的长度/校验 if pl.get("L_ok") is False: r.status = TaskStatus.BAD_FRAME r.err_text = f"本地规约 L 校验失败(L={pl['L']}, 应为 {pl['L_calc']})" return r if pl.get("CS_ok") is False: r.status = TaskStatus.BAD_FRAME r.err_text = f"本地规约 CS 校验失败(CS={pl['CS']:02X}, 应为 {pl['CS_calc']:02X})" return r inner = pl.get("InnerBytes", b"") if not inner: r.status = TaskStatus.BAD_FRAME r.err_text = "上报帧不含 645 内层数据" return r r.inner_645 = inner if self.dump_frames: self._dump(f"<<< 上行报文逐字段解析（{task.meter.addr} {task.item.description}）", annotate_local_frame(raw, with_inner=True)) try: d = self.dlt645_parser.parse(inner) except Exception as e: r.status = TaskStatus.BAD_FRAME r.err_text = f"645 解析失败: {e}" return r r.parsed_645 = d if d.get("PayloadHex"): try: r.freeze_block = bytes.fromhex(d["PayloadHex"]) except ValueError: r.freeze_block = b"" # 645 帧校验 if not d.get("CS_ok", True): r.status = TaskStatus.BAD_FRAME r.err_text = f"645 CS 校验失败(CS={d['CS']:02X}, 应为 {d['CS_calc']:02X})" return r # ★ 错误帧：ERR=1 → 由调用方决定「放弃本表剩余冻结项」 if d.get("ERR", 0) == 1: r.status = TaskStatus.METER_ERROR r.err_code = d.get("err_code") r.err_text = d.get("err_text") or "645 异常应答" return r if d.get("abnormal"): r.status = TaskStatus.BAD_FRAME r.err_text = d.get("err_text") or "645 应答数据异常" return r # 正常应答 r.status = TaskStatus.OK r.freeze_time = d.get("freeze_time", "") r.positive = d.get("positive", "") r.reverse = d.get("reverse", "") r.value = d.get("Value", "") if not r.di_name: r.di_name = d.get("DI_name", "") return r # ---------------- 抄读单条任务 ---------------- def read_one(self, task: SendTask) -> TaskResult: outer = self.build_frames(task)["local"] label = f"表={task.meter.addr} {task.item.description}({task.item.di})" key = self._task_key(task) last: Optional[TaskResult] = None # ★ 下发前先排空串口残留帧： # 残留帧按「归属未知」处理（正常应答→缓存/补抄；不带 DI 的异常应答→丢弃）， # 避免上一轮的回放/延迟帧被本任务块误吃。 self._drain_to_cache() for attempt in range(self.max_retries + 1): if attempt > 0: time.sleep(self.retry_interval) self._log(f" [重试 {attempt}/{self.max_retries}]") self._drain_to_cache() # 先收下这段时间到达的帧（可能已含本项） raw = self._exchange(task, outer, timeout=self.timeout, task_label=label, attempt=attempt) if raw is None: if self._status_fail is not None: sf = self._status_fail last = TaskResult(seq=task.seq, meter_addr=task.meter.addr, di=task.item.di, di_name=task.item.description, status=TaskStatus.NO_RESPONSE, request=outer, retries=attempt, err_text=f"模块上报任务状态 {sf['status_text']}" f"(0x{sf['status_code']:02X})，无 645 数据应答") else: last = TaskResult(seq=task.seq, meter_addr=task.meter.addr, di=task.item.di, di_name=task.item.description, status=TaskStatus.NO_RESPONSE, request=outer, retries=attempt, err_text="超时/未收到本项上报帧") continue r = self._parse_reply(task, raw, attempt) r.request = outer last = r if r.status is TaskStatus.OK: break # ★ 错误帧：默认不重试，立即退出，交由外层终止本表 if r.status is TaskStatus.METER_ERROR and not self.retry_on_error_frame: break assert last is not None # 记账：成功的登记为「已抄回」（用于回放判重）；失败的登记为「等待补抄」 if last.status is TaskStatus.OK: self._got[key] = last self._wait_fill.pop(key, None) else: self._wait_fill[key] = last self._freeze(last) # 冻结块明细落盘（成功/失败都记） return last # ---------------- 抄读单只表 ---------------- def read_meter(self, meter: MeterAddress, items: Optional[Sequence[DataItem]] = None) -> MeterReport: items = list(items) if items is not None else self.freeze_items() report = MeterReport(meter=meter) self._log(f"\n===== 开始抄读 [{meter.description or ''} {meter.addr}] " f"共 {len(items)} 项冻结数据 =====") self._log(f" 表地址 ADST(低字节在前) = {fmt_hex(meter.to_bytes())}") streak = 0 for idx, item in enumerate(items): self._seq = (self._seq + 1) & 0xFF self._task_no = (self._task_no + 1) & 0xFFFF if self._task_no == 0: self._task_no = 1 task = SendTask(meter=meter, item=item, seq=self._task_no) self._log(f"--- {task.label} ---") r = self.read_one(task) report.results.append(r) self._log_result(r) # 连续失败计数：任何一次成功即清零（单次偶发超时不再误杀整表） if r.status is TaskStatus.OK: streak = 0 else: streak += 1 # ★★ 核心策略：命中放弃条件 → 该表剩余「上x次冻结数据」全部不抄，切下一只表 if self.policy.should_abort(r.status, streak): rest = items[idx + 1:] reason = self.policy.reason(r.status, streak) detail = f"({r.err_text})" if r.err_text else "" self._log(f" ✘ 放弃本表剩余 {len(rest)} 项「上x次冻结数据」" f" —— 原因：{reason}{detail}，切换到下一只表") if rest: self._log(f" 放弃区间：DI {rest[0].di}(上{_di_x(rest[0].di)}) " f"~ DI {rest[-1].di}(上{_di_x(rest[-1].di)})，共 {len(rest)} 项") self._raw("GIVE-UP", b"", f"表={meter.addr} 放弃 DI {rest[0].di}~{rest[-1].di} 共 {len(rest)} 项") for it in rest: report.results.append(TaskResult( seq=task.seq, meter_addr=meter.addr, di=it.di, di_name=it.description, status=TaskStatus.SKIPPED, note=f"{reason}，放弃抄读", )) report.aborted = True report.abort_reason = f"{reason}{detail} @ 上{idx + 1}次(DI={item.di})" break self._write_csv(report) # ---------------- 收尾：捞最后一批迟到应答（自动补抄） ---------------- self._drain_to_cache(max_wait=0.3) self._log(f"===== 抄读结束 [{meter.addr}]：成功 {report.ok_count}" f"{'，已放弃剩余 ' + str(report.skipped_count) + ' 项' if report.aborted else ''}" f" | 补抄回填 {self._late_fill} 项 | 命中缓存 {self._cache_hits} 次" f" | 忽略回放重复帧 {self._dup_frames} 帧 =====") return report def _log_result(self, r: TaskResult) -> None: blk = r.freeze_block if r.status is TaskStatus.OK: self._log(f" ✔ 成功 (重试 {r.retries} 次)") self._log(f" 表号: {r.meter_addr}") self._log(f" DI: {r.di} ({r.di_name})") if blk and len(blk) >= 16: self._log(f" 冻结时间: {r.freeze_time}") self._log(f" 正向总电能: {r.positive} kWh") self._log(f" 反向总电能: {r.reverse} kWh") self._log(f" 冻结块原始 16B : {fmt_hex(blk)}") self._log(f" 时间[0:6]={fmt_hex(blk[0:6])} " f"正向[6:11]={fmt_hex(blk[6:11])} 反向[11:16]={fmt_hex(blk[11:16])}") else: self._log(f" 数据: {r.value}") if r.inner_645: self._log(f" 645 应答报文: {fmt_hex(r.inner_645)}") else: self._log(f" ✘ {r.status.value} (重试 {r.retries} 次)") if r.err_code is not None: self._log(f" 错误码: 0x{r.err_code:02X} {r.err_text}") elif r.err_text: self._log(f" 错误: {r.err_text}") if blk: self._log(f" 冻结块原始 16B : {fmt_hex(blk)}") if r.inner_645: self._log(f" 645 应答报文: {fmt_hex(r.inner_645)}") self._log("") # ---------------- 抄读多只表 ---------------- def run(self, meters: Sequence[MeterAddress], items: Optional[Sequence[DataItem]] = None) -> List[MeterReport]: items = list(items) if items is not None else self.freeze_items() self.log_banner(meters, items) reports: List[MeterReport] = [] for i, m in enumerate(meters, 1): self._log(f"\n>>>>>>>>>> [{i}/{len(meters)}] {m.description or ''} {m.addr} <<<<<<<<<<") try: reports.append(self.read_meter(m, items)) except Exception as e: # 单表异常不影响其它表 self._log(f" !! 该表抄读过程中出现异常: {e}，跳过并继续下一只表") rep = MeterReport(meter=m, aborted=True, abort_reason=f"异常: {e}") reports.append(rep) self._print_summary(reports) return reports def _print_summary(self, reports: Sequence[MeterReport]) -> None: self._log("\n" + "=" * 70) self._log("================ 全部抄读结果汇总 ================") total_ok = total_skip = 0 for i, rep in enumerate(reports, 1): ok = rep.ok_count err = sum(1 for r in rep.results if r.status is TaskStatus.METER_ERROR) bad = sum(1 for r in rep.results if r.status is TaskStatus.BAD_FRAME) nor = sum(1 for r in rep.results if r.status is TaskStatus.NO_RESPONSE) skip = rep.skipped_count total_ok += ok total_skip += skip line = f" {i}. {rep.meter.addr} {rep.meter.description}：成功 {ok}" if err: line += f"，错误帧 {err}" if bad: line += f"，坏帧 {bad}" if nor: line += f"，无应答 {nor}" if skip: line += f"，放弃 {skip}" self._log(line) if rep.aborted: self._log(f" └─ 已放弃本表剩余冻结项：{rep.abort_reason}") self._log(f" --------------------------------------------------") self._log(f" 合计：成功 {total_ok} 项，因放弃未抄读 {total_skip} 项") self._log(f" 应答归属：补抄回填 {self._late_fill} 项，" f"命中延迟应答缓存 {self._cache_hits} 次，忽略回放重复帧 {self._dup_frames} 帧") self._log(f" 确认帧 ：已对模块上报回确认 {self._ack_frames} 条" f"（SEQ 与上报帧一致{'，同 SEQ 去重 %gs' % self.ack_dedup if self.ack_dedup > 0 else '，每次都回'}）") self._log(f" 日志落盘：") self._log(f" 主日志(含逐字段报文) : {self.log_path}") self._log(f" 原始报文流水(TX/RX) : {self.raw_log_path}") self._log(f" 冻结数据块明细 : {self.freeze_log_path}") self._log(f" 业务数据表(逗号分隔) : {self.csv_path}") self._log("=" * 70 + "\n") # ---------------- CSV ---------------- def _write_csv(self, report: MeterReport) -> None: if not self.csv_path: return new_file = not os.path.exists(self.csv_path) with open(self.csv_path, "a", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) if new_file: w.writerow(self.CSV_HEADER) for r in report.results: w.writerow([ r.meter_addr, report.meter.description, r.seq, r.di, r.di_name, r.status.value, r.freeze_time, r.positive, r.reverse, r.value if r.status is TaskStatus.OK else "", r.retries, f"0x{r.err_code:02X}" if r.err_code is not None else "", r.err_text or r.note, fmt_hex(r.inner_645), fmt_hex(r.freeze_block), ]) # ============================================================ # 自测：假串口复现「回放帧挤占 / 延迟应答 / 补抄 / 错误帧中止」 # ============================================================ def _di_bytes(di: str) -> bytes: return bytes.fromhex(di)[::-1] def _bcd_bytes(digits: str) -> bytes: """'00071973' → b'\\x00\\x07\\x19\\x73'（每两位十进制数字压成 1 字节 BCD）""" return bytes(int(digits[i:i + 2], 16) for i in range(0, len(digits), 2)) def _freeze_payload(year, month, day, hour, minute, pos_digits, rev_digits) -> bytes: """按 DLT645Parser._parse_freeze 的字段布局组一个 16 字节冻结块（低字节在前）""" head = bytes([int(f"{minute:02d}", 16), int(f"{hour:02d}", 16), int(f"{day:02d}", 16), int(f"{month:02d}", 16), int(f"{year % 100:02d}", 16), 0xAA]) return (head + _bcd_bytes(pos_digits)[::-1] + b"\xAA" + _bcd_bytes(rev_digits)[::-1] + b"\xAA") def _build_645_resp(addr: str, ctrl: int, data_field: bytes) -> bytes: """组一条 645 应答帧（数据域 +33H 编码）""" addr_bytes = bytes.fromhex(addr)[::-1] enc = bytes((b + 0x33) & 0xFF for b in data_field) head = bytes([0x68]) + addr_bytes + bytes([0x68, ctrl, len(enc)]) + enc return head + bytes([sum(head) & 0xFF, 0x16]) def _build_local_report(inner: bytes, asrc: str, meter: str, seq: int = 0x05, task_id: int = 1) -> bytes: """把 645 帧包成本地规约上报帧（上行：C=0xE0 / AFN=0x05，无 PrioResp/Timeout）""" asrc_b = bytes.fromhex(asrc.zfill(12)) adst_b = bytes.fromhex(meter.zfill(12))[::-1] user_data = (asrc_b + adst_b + bytes([0x05, seq]) + bytes.fromhex("010505E8") + task_id.to_bytes(2, "little") + bytes([len(inner)]) + inner) ctrl = 0xE0 L = len(user_data) + 6 head = bytes([0x68]) + L.to_bytes(2, "little") + bytes([ctrl]) + user_data return head + bytes([(ctrl + sum(user_data)) & 0xFF, 0x16]) def _build_ack(seq: int = 0x0A) -> bytes: """确认帧（PRM=0）—— 模块对我们下行任务的受理回执（C=0x80）""" ud = bytes([0x00, seq, 0x01, 0x00, 0x01, 0xE8, 0x00, 0x00]) ctrl = 0x80 head = bytes([0x68]) + (len(ud) + 6).to_bytes(2, "little") + bytes([ctrl]) + ud return head + bytes([(ctrl + sum(ud)) & 0xFF, 0x16]) def _build_status(meter: str, task_id: int, code: int, seq: int = 0x05) -> bytes: """ 上报任务状态帧（C=0xC0）：模块汇报任务执行结果。 实测样例：68 15 00 C0 05 05 05 05 05 E8 01 00 17 18 00 98 04 26 FF B2 16 """ ud = (bytes([0x05, seq]) + bytes.fromhex("050505E8") + task_id.to_bytes(2, "little") + bytes.fromhex(meter)[::-1] + bytes([code])) ctrl = 0xC0 head = bytes([0x68]) + (len(ud) + 6).to_bytes(2, "little") + bytes([ctrl]) + ud return head + bytes([(ctrl + sum(ud)) & 0xFF, 0x16]) class _ScriptedTransport: """自测用假串口：behavior(表号, DI, TaskID, 第几次下发) → 要投递的帧列表""" def __init__(self, behavior, on_ack=None): self.port = "FAKE" self.behavior = behavior self.on_ack = on_ack # 收到我们回的确认帧时的额外行为（可选） self.sent: List[bytes] = [] self.acks: List[bytes] = [] # 我们回给模块的确认帧 self._buf = bytearray() self._open = True def is_open(self) -> bool: return self._open def open(self) -> None: self._open = True def close(self) -> None: self._open = False def send(self, data: bytes) -> int: # 我们回给模块的确认帧（C=0x00，14 字节）→ 单独记账，不触发任务行为脚本 if len(data) == 14 and data[3] == 0x00: self.acks.append(bytes(data)) if self.on_ack is not None: for f in self.on_ack(bytes(data)): self._buf.extend(f) return len(data) self.sent.append(bytes(data)) addr = data[10:16][::-1].hex().upper() inner_len = data[27] # 下行比上行多 PrioResp+Timeout inner = data[28:28 + inner_len] di = bytes((b - 0x33) & 0xFF for b in inner[10:14])[::-1].hex().upper() tid = int.from_bytes(data[22:24], "little") for f in self.behavior(addr, di, tid, len(self.sent)): self._buf.extend(f) return len(data) def drain(self, max_wait: float = 0.0) -> bytes: out = bytes(self._buf) self._buf.clear() return out def recv(self, expected_len: int = 0, end_flag=None, max_wait=None) -> bytes: if not self._buf: time.sleep(0.005) # 模拟空读，避免忙等 return b"" out = bytes(self._buf) self._buf.clear() return out def selftest_replay() -> None: """用假串口复现真实抓包里观察到的全部异常模式，验证本次修复""" import tempfile addr_a, addr_b = "260498001817", "260498001820" items = [DataItem(f"0504FF{i:02X}", description=f"上{i}次整点冻结数据块") for i in range(1, 11)] DI = {i: f"0504FF{i:02X}" for i in range(1, 11)} sends: Dict[str, int] = {} def report(a, d, tid, hh): digits = f"{int(hh * 100):08d}" inner = _build_645_resp(a, 0x91, _di_bytes(d) + _freeze_payload(26, 10, 16, hh % 24, 0, digits, "00000000")) return _build_local_report(inner, "000000000000", a, task_id=tid) def behavior(addr, d, tid, n): sends[(addr, d)] = sends.get((addr, d), 0) + 1 att = sends[(addr, d)] if addr == addr_b: # B 表：一切正常 return [_build_ack(), report(addr_b, d, tid, 1)] # ---------- A 表：复现真实抓包里的各种异常 ---------- if d == DI[1]: return [_build_ack(), report(addr_a, DI[1], tid, 17)] if d == DI[2]: # ① 回放上1次的旧帧（重复）+ ② 上3次的答案提前到达 + ③ 本项答案 return [report(addr_a, DI[1], 1, 17), report(addr_a, DI[3], 3, 15), report(addr_a, DI[2], tid, 16)] if d == DI[3]: # 本轮不应被下发（上3次的答案已在上一窗口入缓存）——若被下发说明缓存没生效 return [report(addr_a, DI[3], tid, 15)] if d == DI[4] and att == 1: # ★ 回放风暴：旧帧挤占窗口，本项答案这一轮不来（实测就是这种情形） return [report(addr_a, DI[1], 1, 17), report(addr_a, DI[2], 2, 16), report(addr_a, DI[3], 3, 15), bytes.fromhex("681500C0055A050505E82A00171800980426FF3016")] # 模块非抄读帧 if d == DI[4]: return [report(addr_a, DI[4], tid, 13)] if d == DI[5]: return [] # 整轮无应答 → 依赖后续补抄 if d == DI[6]: # 上5次的迟到应答（补抄）+ 本项答案 return [report(addr_a, DI[5], 5, 12), report(addr_a, DI[6], tid, 11)] if d == DI[7]: # 错误帧 ERR=1 / 0x02 无请求数据 → 触发放弃本表剩余。 # ★ 故意把外层 TaskID 写成一个与本项流水号无关的值，验证异常帧只按 DI 归属。 err = _build_645_resp(addr_a, 0xD1, _di_bytes(DI[7]) + bytes([0x02])) return [_build_local_report(err, "000000000000", addr_a, task_id=0xABCD)] return [] tmp = tempfile.mkdtemp(prefix="buddy_selftest_") paths = dict(csv_path=os.path.join(tmp, "data.csv"), log_path=os.path.join(tmp, "read.txt"), raw_log_path=os.path.join(tmp, "raw.txt"), freeze_log_path=os.path.join(tmp, "freeze.txt")) transport = _ScriptedTransport(behavior) session = FreezeReadSession( asrc_addr="000000000000", transport=transport, policy=AbortPolicy(streak_limit=5), timeout=0.4, # 自测用短超时 max_retries=1, retry_interval=0.01, replay_grace=0.15, freeze_start=1, freeze_end=10, dump_frames=False, verbose=True, **paths, ) print("=" * 78) print("[自测] 假串口复现：回放帧挤占 / 延迟应答 / 补抄 / 错误帧中止") print("=" * 78) reports = session.run([MeterAddress(addr_a, "A 表"), MeterAddress(addr_b, "B 表")], items) session.close() a, b = reports[0], reports[1] log_text = open(paths["log_path"], encoding="utf-8").read() # ---- 断言 ---- print("\n" + "=" * 78) print("[自测结果]") assert [r.di for r in a.results[:7]] == [DI[i] for i in range(1, 8)], \ f"结果顺序须与 DI 顺序一致，实际 {[r.di for r in a.results[:7]]}" assert a.results[0].status is TaskStatus.OK, "上1次应成功" assert a.results[1].status is TaskStatus.OK, "上2次应成功（同批里先到的旧帧不能串位）" assert sends.get((addr_a, DI[3]), 0) == 0, "上3次的答案已在缓存里，本轮不应再下发（未命中缓存）" assert a.results[2].status is TaskStatus.OK and a.results[2].retries == 0, "上3次应由缓存命中" assert a.results[3].status is TaskStatus.OK and a.results[3].retries == 1, \ "回放风暴后应重发成功" assert a.results[4].status is TaskStatus.OK and "补抄" in a.results[4].note, \ "上5次应由迟到应答补抄回填" assert a.results[4].freeze_time == "26年10月16日12时00分", \ f"补抄回填的冻结时间不对: {a.results[4].freeze_time}" assert a.results[6].status is TaskStatus.METER_ERROR and a.results[6].err_code == 0x02, \ "上7次的错误帧应识别为 METER_ERROR" assert a.aborted and all(r.status is TaskStatus.SKIPPED for r in a.results[7:]), \ "错误帧后应放弃本表剩余项" assert b.ok_count == 10, f"A 表被放弃不应影响 B 表，B 成功 {b.ok_count}/10" assert session._late_fill >= 1, "应有补抄回填" assert session._cache_hits >= 1, "应有缓存命中" assert session._dup_frames >= 1, "应识别出回放重复帧" assert "跳过不匹配任务" not in log_text, "旧的 [跳过不匹配任务] 提示不应再出现" assert "补抄成功" in log_text and "重复帧" in log_text and "提前结束本次等待" in log_text print(f" A 表: 成功 {a.ok_count} | 补抄 {session._late_fill} | 缓存命中 {session._cache_hits} " f"| 重复帧 {session._dup_frames} | 放弃 {a.skipped_count}") print(f" B 表: 成功 {b.ok_count}/10（未受 A 表放弃影响）") print(f" 日志中已无「跳过不匹配任务」：{'跳过不匹配任务' not in log_text}") print(" CSV 内容：") for line in open(paths["csv_path"], encoding="utf-8-sig").read().strip().splitlines()[:9]: print(" " + line) print("\n OK：回放/延迟/补抄/中止全部按预期工作") def selftest_taskid_semantics() -> None: """ 专项自测：TaskID 与「上x次任务块」绝不能互相混淆。 ① 模块所有上报帧统一用 TaskID=0xABCD（既不是下发的流水号，也不是 x 编号） → 仍应全部按 DI 正确抄回（证明归属完全不依赖 TaskID）。 ② 不带 DI 的异常应答：等待窗口内 → 归属当前任务块； 窗口外（task=None，例如下发前排空串口时的残留帧）→ 丢弃且不入缓存， 绝不能被**下一个**任务块误吃（旧版 ("ERR",表号,TaskID or 0) 键 + 通配兜底就会）。 ③ 带 DI 的异常应答：属于别的任务块时只能算到那个块头上，不能算到当前块。 """ import tempfile addr = "260498001817" items = [DataItem(f"0504FF{i:02X}", description=f"上{i}次整点冻结数据块") for i in range(1, 6)] DI = {i: f"0504FF{i:02X}" for i in range(1, 6)} WEIRD = 0xABCD # 刻意用一个与「上x次」编号、与流水号都无关的值 def report(d, hh, tid=WEIRD): inner = _build_645_resp(addr, 0x91, _di_bytes(d) + _freeze_payload(26, 10, 16, hh % 24, 0, f"{int(hh * 100):08d}", "00000000")) return _build_local_report(inner, "000000000000", addr, task_id=tid) def err_report(data_field, tid=WEIRD): inner = _build_645_resp(addr, 0xD1, data_field) # 0xD1 = 读应答 + ERR=1 return _build_local_report(inner, "000000000000", addr, task_id=tid) def behavior(a, d, tid, n): return [report(d, int(d[6:8], 16))] # 全部用 WEIRD 回 print("=" * 78) print("[自测·TaskID 语义] 归属只认 (表号, DI)，TaskID 只是回显的流水号") print("=" * 78) tmp = tempfile.mkdtemp(prefix="buddy_tidsem_") paths = dict(csv_path=os.path.join(tmp, "data.csv"), log_path=os.path.join(tmp, "read.txt"), raw_log_path=os.path.join(tmp, "raw.txt"), freeze_log_path=os.path.join(tmp, "freeze.txt")) session = FreezeReadSession( asrc_addr="000000000000", transport=_ScriptedTransport(behavior), policy=AbortPolicy(), timeout=0.3, max_retries=0, retry_interval=0.01, replay_grace=0.0, freeze_start=1, freeze_end=5, dump_frames=False, verbose=False, **paths, ) rep = session.read_meter(MeterAddress(addr, "TID 表"), items) assert rep.ok_count == len(items), \ f"模块用 TaskID=0x{WEIRD:04X} 回帧时仍应按 DI 全部抄回，实际成功 {rep.ok_count}/{len(items)}" assert not rep.aborted, "TaskID 是无关字段，不应触发任何放弃" print(f" ① 模块统一回 TaskID=0x{WEIRD:04X}（非流水号、非 x 编号）" f" → 仍然 {rep.ok_count}/{len(items)} 全部抄回，顺序正确") # ---- 645 解析：异常应答帧也要能取出 DI ---- p645 = DLT645Parser() pd = p645.parse(_build_645_resp(addr, 0xD1, _di_bytes(DI[4]) + bytes([0x02]))) assert pd["abnormal"] and pd["has_di"] and pd["DI"] == DI[4] and pd["err_code"] == 0x02, \ f"异常应答应能解析出 DI 与错误码，实际 {pd.get('DI')}/{pd.get('err_code')}" print(f" ② 异常应答帧解析：DI={pd['DI']}(上{int(pd['DI'][6:8], 16)}次) " f"错误码=0x{pd['err_code']:02X} {pd['err_text']} ← 旧版这里 DI 是空的") task2 = SendTask(meter=MeterAddress(addr, "t"), item=DataItem(DI[2], description="上2次"), seq=0x000A) task3 = SendTask(meter=MeterAddress(addr, "t"), item=DataItem(DI[3], description="上3次"), seq=0x000B) # ---- ② 不带 DI 的异常帧：窗口内归属当前块；窗口外丢弃且不入缓存 ---- f_in = err_report(bytes([0x02])) # 数据域只有错误码，没有 DI hit, same = session._route(f_in, task2) assert hit is f_in, "窗口内收到的无 DI 异常帧应归属当前任务块" f_out = err_report(bytes([0x02])) hit2, same2 = session._route(f_out, None) # 模拟下发前排空时收到的残留帧 assert hit2 is None and same2 is False, "窗口外的无 DI 异常帧不得归属任何块" assert session._take_cache(task3) is None, \ "窗口外的无 DI 异常帧不得留在缓存里被后面的任务块误吃（旧版会张冠李戴）" print(" ③ 无 DI 异常帧：窗口内 → 归属当前块；窗口外 → 丢弃且不入缓存" "（≫ 旧版会喂给下一个块，把好数据判成错误帧并中止整表）") # ---- ③ 带 DI 的异常帧：只算到它自己的块头上 ---- f_other = err_report(_di_bytes(DI[3]) + bytes([0x02])) hit3, _ = session._route(f_other, task2) assert hit3 is None, "属于上3次的异常帧不能算到上2次头上" assert session._take_cache(task2) is None, "上2次不该被别的块的异常牵连" cached3 = session._take_cache(task3) assert cached3 is not None and cached3["di"] == DI[3], "上3次的异常应答应能被上3次取用" f_self = err_report(_di_bytes(DI[2]) + bytes([0x02])) hit4, _ = session._route(f_self, task2) assert hit4 is f_self, "属于当前块的异常帧必须命中当前块（与 TaskID 无关）" print(" ④ 带 DI 异常帧：只算到它自己的任务块头上，不牵连当前块") session.close() print("\n OK：TaskID 与 上x次任务块 已彻底解耦，异常帧一律按 DI 归属") # ============================================================ # 主程序（实际使用：串口直连抄读） # ============================================================ def selftest_status_and_ack() -> None: """ 专项自测（依据载波调试软件帧解析出的真实报文）： ① 模块「受理回执」确认帧 680E00800034010001E800009E16（C=0x80 / PRM=0 / AFN=00 确认） → 只代表任务已受理，绝不能判成非法帧/错误帧。 ② 模块「上报任务状态」帧 68 15 00 C0 05 05 05 05 05 E8 01 00 17 18 00 98 04 26 FF B2 16 （C=0xC0 / AFN=05 上报信息 / DUID=050505E8 / 任务状态=FF 其它） → 立刻判定当前任务块本次不会返回数据（不再白等满 timeout）； 连续 streak_limit 次 → 按策略放弃本表剩余，切下一只表。 ③ 任务状态=00（成功）→ 不算失败，继续等数据帧。 ④ 每条上行报文都要回确认帧，且 SEQ 与上报报文一致 （680E000000 SEQ 010001E8 0000 CS 16）；同一 SEQ 短时间内不重复确认。 """ import tempfile addr_a, addr_b = "260498001817", "260498001820" items = [DataItem(f"0504FF{i:02X}", description=f"上{i}次整点冻结数据块") for i in range(1, 6)] DI = {i: f"0504FF{i:02X}" for i in range(1, 6)} TIMEOUT = 2.0 # 若 FF 不快速失败，每项要白等 2s # ---- 真实报文回归 ---- real_ack = bytes.fromhex("680E00800034010001E800009E16") # 模块受理回执 real_status = bytes.fromhex("681500C00505050505E80100171800980426FFB216") # 任务状态=FF 其它 ts = parse_task_status_frame(real_status) assert ts is not None and ts["CS_ok"], "真实任务状态帧应能被识别且校验通过" assert (ts["task_id"], ts["addr"], ts["status_code"], ts["status_text"]) == \ (1, "260498001817", 0xFF, "其它"), f"任务状态帧解析有误: {ts}" assert build_report_ack(0xD0).hex().upper() == "680E000000D0010001E80000BA16", \ f"确认帧组帧与实测不符: {build_report_ack(0xD0).hex().upper()}" print("=" * 78) print("[自测·任务状态与确认帧] 真实报文回归 + FF 快速失败 + 每条上报都回确认帧") print("=" * 78) print(f" ① 任务状态帧解析：TaskID={ts['task_id']} 表={ts['addr']} " f"状态={ts['status_text']}(0x{ts['status_code']:02X}) CS={'OK' if ts['CS_ok'] else 'BAD'}") print(" ② 确认帧组帧：SEQ=0xD0 → 680E000000D0010001E80000BA16（与实测逐字节一致）") report_seqs = [] # 模块上报过的所有 SEQ（用于核对我们逐条都回了确认） def status(a, tid, code, seq): report_seqs.append(seq) return _build_status(a, tid, code, seq=seq) def report(a, d, tid, hh, seq): inner = _build_645_resp(a, 0x91, _di_bytes(d) + _freeze_payload(26, 10, 16, hh % 24, 0, f"{int(hh * 100):08d}", "00000000")) report_seqs.append(seq) return _build_local_report(inner, "000000000000", a, seq=seq, task_id=tid) sends: Dict[Tuple[str, str], int] = {} def behavior(addr, d, tid, n): sends[(addr, d)] = sends.get((addr, d), 0) + 1 att = sends[(addr, d)] idx = int(d[6:8], 16) if addr == addr_b: # B 表：一切正常 return [_build_ack(), report(addr_b, d, tid, idx, 0x30 + idx)] if idx == 1: # ③ 受理回执 + 任务状态=成功 + 数据帧 全在一批里 → 本项必须成功 return [real_ack, status(addr_a, tid, 0x00, 0xD1), report(addr_a, d, tid, 17, 0xD0)] if idx == 2: # ② 任务状态=FF 其它 → 本项不会有数据，必须快速失败（不白等 timeout） return [_build_ack(), status(addr_a, tid, 0xFF, 0xE0 + att)] if idx == 3: return [_build_ack(), status(addr_a, tid, 0xFF, 0xF0 + att)] return [] tmp = tempfile.mkdtemp(prefix="buddy_status_") paths = dict(csv_path=os.path.join(tmp, "data.csv"), log_path=os.path.join(tmp, "read.txt"), raw_log_path=os.path.join(tmp, "raw.txt"), freeze_log_path=os.path.join(tmp, "freeze.txt")) transport = _ScriptedTransport(behavior) session = FreezeReadSession( asrc_addr="000000000000", transport=transport, policy=AbortPolicy(streak_limit=2), timeout=TIMEOUT, max_retries=0, retry_interval=0.01, replay_grace=0.0, freeze_start=1, freeze_end=5, dump_frames=False, verbose=False, **paths) t0 = time.time() reports = session.run([MeterAddress(addr_a, "A 表"), MeterAddress(addr_b, "B 表")], items) elapsed = time.time() - t0 session.close() a, b = reports[0], reports[1] log_text = open(paths["log_path"], encoding="utf-8").read() assert a.results[0].status is TaskStatus.OK, "任务状态=成功 + 数据帧 → 本项应成功" assert a.results[1].status is TaskStatus.NO_RESPONSE and "任务状态" in a.results[1].err_text, \ f"任务状态=FF 应快速失败并注明原因，实际 {a.results[1].status} / {a.results[1].err_text}" assert a.results[1].err_code is None, "这不是 645 错误帧，不该记错误码" assert a.aborted and all(r.status is TaskStatus.SKIPPED for r in a.results[3:]), \ "连续 2 次 FF → 按策略放弃本表剩余" assert elapsed < TIMEOUT, \ f"FF 必须快速失败：本次耗时 {elapsed:.2f}s，若白等超时应≥ {TIMEOUT}s×2 项" assert b.ok_count == len(items), f"B 表不应受 A 表影响，实际 {b.ok_count}/{len(items)}" ack_seqs = sorted(ack[5] for ack in transport.acks) assert transport.acks, "必须对模块每条上报回确认帧" assert ack_seqs == sorted(report_seqs), \ f"模块每条上报都应有同 SEQ 的确认帧：上报 {sorted(report_seqs)} vs 实回 {ack_seqs}" assert "TX-ACK-DUP" not in open(paths["raw_log_path"], encoding="utf-8").read(), \ "默认不去重：不应出现 TX-ACK-DUP" assert "[确认帧]" in log_text and "任务状态" in log_text, "日志应含确认帧与任务状态记录" assert "[跳过非法帧]" not in log_text and "[非法帧]" not in log_text, \ "受理回执/任务状态帧都不能被判成非法帧" print(f" ③ A 表：上1次成功（受理回执+状态成功+数据帧同批）；" f"上2/上3次 收到 任务状态=FF → 立即失败并放弃剩余 {a.skipped_count} 项") print(f" 耗时 {elapsed:.2f}s（若白等超时需 ≥ {TIMEOUT:g}s×2 = {TIMEOUT * 2:g}s）") print(f" ④ 模块上报 {len(report_seqs)} 条 → 我们回确认帧 {len(transport.acks)} 条，" f"SEQ 逐条对齐：{' '.join(f'0x{s:02X}' for s in ack_seqs)}") print(f" ⑤ B 表 {b.ok_count}/{len(items)} 全部成功，未受 A 表影响") print("\n OK：任务状态帧识别 + FF 快速失败 + 逐条上报回确认帧（SEQ 对齐）全部生效") def parse_args(): import argparse p = argparse.ArgumentParser( description="电表「上x次整点冻结数据」抄读：电表回错误帧即放弃该表后续冻结项，切下一只表") p.add_argument("--port", default="COM4", help="串口名，如 COM4 / /dev/ttyUSB0") p.add_argument("--meter", action="append", default=None, help="电表地址(12位HEX，如 260498001817)，可重复传入多只表按顺序抄") p.add_argument("--from", dest="x_from", type=int, default=1, help="从 上x次 开始（默认 1）") p.add_argument("--to", dest="x_to", type=int, default=255, help="到 上y次 结束（默认 255）") p.add_argument("--timeout", type=float, default=5.0, help="单条等待应答超时(秒)，默认 5（实测应答 0.13~0.35s 到达）") p.add_argument("--retries", type=int, default=2, help="超时/坏帧重试次数，默认 2") p.add_argument("--replay-grace", type=float, default=1.0, help="被本表回放帧挤占窗口后，静默该秒数即提前重发，默认 1.0（0=关闭）") p.add_argument("--abort-streak", type=int, default=5, help="连续 N 次无应答/坏帧才放弃本表剩余，默认 5（0=永不放弃）") p.add_argument("--keep-on-timeout", action="store_true", help="兼容旧参数：等价于 --abort-streak 0（无应答不放弃本表）") p.add_argument("--dump-console", action="store_true", help="把逐字段报文解析也打印到控制台（默认只写日志文件）") p.add_argument("--no-ack", action="store_true", help="不对模块的上报报文回确认帧（默认每条上报都回，SEQ 与上报帧一致）") p.add_argument("--ack-dedup", type=float, default=0.0, help="同一 (源地址,SEQ) 去重窗口秒数；默认 0 = 每次上报都回确认帧") p.add_argument("--selftest", action="store_true", help="用假串口跑内建自测（验证回放/延迟应答/补抄/异常中止/任务状态/确认帧），" "不占用真实串口") return p.parse_args() def main(): args = parse_args() if args.selftest: selftest_replay() selftest_taskid_semantics() selftest_status_and_ack() return # ==================== 用户配置 ==================== CCO_ADDR = "000000000000" # ASRC：CCO/模块地址，原样写入 SERIAL_PORT = args.port # 串口名，也可用 --port 覆盖 BAUDRATE = 9600 # 115200 8E1 PARITY = "E" SEND_PROBE = False # 是否先发一条探测帧（原脚本里那条固定帧） PROBE_FRAME = "681500C00503050505E80300171800980426FFB216" # 电表清单：按顺序抄；命令行给了 --meter 则以其为准 METERS = [ MeterAddress("260498001817", "1# 表"), # MeterAddress("584c07000006", "2# 表"), ] if args.meter: METERS = [MeterAddress(a.lower(), f"{i}# 表") for i, a in enumerate(args.meter, 1)] # 日志文件（全部纯文本 .txt；置 None 即关闭该类输出） LOG = dict( log_path = "read_freeze.txt", # 主日志：流程 + 每帧逐字段解析 raw_log_path = "raw_frames.txt", # 原始报文流水：一行一帧（TX/RX/GIVE-UP） freeze_log_path = "freeze_blocks.txt", # 冻结数据块明细：一行一块 csv_path = "freeze_data.txt", # 业务数据表（逗号分隔，Excel/WPS 可打开） ) # 放弃策略：命中即不再抄该表剩余「上x次冻结数据」，直接切下一只表 streak = 0 if args.keep_on_timeout else args.abort_streak policy = AbortPolicy( on_error_frame = True, # ★ 电表回错误帧 → 立即放弃本表剩余 on_no_response = True, # 连续无应答 → 放弃本表剩余 on_bad_frame = True, # 连续坏帧 → 放弃本表剩余 streak_limit = streak, # 连续 N 次阈值（0 = 永不因超时/坏帧放弃） ) # ================================================== try: transport = SerialPort(port=SERIAL_PORT, baudrate=BAUDRATE, parity=PARITY, timeout=args.timeout + 1.0, verbose=True) transport.open() except SerialPortError as e: print(f"[错误] 串口打开失败：{e}") print(" 请检查串口号；若提示未安装 pyserial，执行：pip install pyserial") return session = FreezeReadSession( asrc_addr=CCO_ADDR, transport=transport, policy=policy, timeout=args.timeout, max_retries=args.retries, retry_interval=0.3, retry_on_error_frame=False, freeze_start=args.x_from, freeze_end=args.x_to, replay_grace=args.replay_grace, ack_reports=not args.no_ack, ack_dedup=args.ack_dedup, dump_frames=True, dump_console=args.dump_console, raw_console=False, verbose=True, **LOG, ) try: if SEND_PROBE: transport.send(bytes.fromhex(PROBE_FRAME)) session.run(METERS, session.freeze_items()) except KeyboardInterrupt: print("\n[中断] 已手动终止，此前抄回的数据已写入日志与数据表。") finally: session.close() transport.close() if __name__ == "__main__": main()