# -*- coding: utf-8 -*- """ 电表「上x次整点冻结数据」抄读程序（精简优化版） ============================================================ 由 buddy.py 精简而来：**删去全部自测代码**（假串口复现、TaskID 语义自测、 任务状态/确认帧自测，以及它们用的造帧辅助函数），只保留生产路径； 另做了一轮优化，逐条列在文末「本版相对 buddy.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 freeze_reader.py # 按文件内配置抄读 python freeze_reader.py --port COM3 --baud 9600 # 覆盖串口号 / 波特率 python freeze_reader.py --meter 260498001817 --meter 260498001820 python freeze_reader.py --from 1 --to 24 # 只抄 上1次~上24次 python freeze_reader.py --keep-on-timeout # 超时/坏帧永不放弃本表 python freeze_reader.py --no-ack --dump-console # 不回确认帧 + 报文解析打屏 本版相对 buddy.py 的改动 ------------------------------------------------------------ 【删除】 · 移除 selftest_replay / selftest_taskid_semantics / selftest_status_and_ack 三套自测，以及 _ScriptedTransport、_di_bytes、_bcd_bytes、_freeze_payload、 _build_645_resp、_build_local_report、_build_ack、_build_status、--selftest。 · 移除从未被赋值的死属性 _freeze_fp、未使用的 timedelta 导入、 以及 recv() 里注释掉的一行日志。 【修掉的真问题】 · CSV 落盘时机：原来在「本表收尾补抄」之前就写，迟到应答回填的那一项 会与 CSV 里记的 NO_RESP 不一致。现在统一推迟到 run() 末尾（全表抄完、 最后一轮补抄收干之后）一次性落盘，日志与 CSV 必然一致。 · 放弃本表后，该表被放弃的 DI 若还有应答上来，原来会进「延迟应答缓存」 永远没人取（纯垃圾，还会让日志出现像模像样的 [缓存延迟应答]）。 现在按 (表号, DI) 记入 _abandoned，直接记 [已放弃] 丢弃。 · 串口打开后不清输入缓冲：上一次会话残留的字节会被当成第一只表的应答。 现在 open() 里 reset_input_buffer()。 · Ctrl-C 原来会直接丢掉当前表的全部结果（CSV 未写、剩余项无记录）。 现在按 SKIPPED 记账、停止后续电表、照常落盘并打印汇总。 【效率】 · 日志句柄复用：原来 freeze_blocks.txt 每写一行就 open/close 一次文件 （255 项 × N 只表，纯属浪费），现在四份日志各复用一个句柄。 · SerialPort.recv 用 in_waiting 优先读已在缓冲区的字节，不再每次白等 一个读切片。 · LocalProtocolBuilder 按表地址缓存，不再每项重建。 · 主日志每行加毫秒时间戳（控制台保持原样不加），便于与 raw_frames.txt 对齐。 【易用】 · --baud 参数化（默认 115200，与原 SerialPort 默认值及本地通信接口一致）； buddy.py 里 main() 写死 9600 而注释写着「115200 8E1」，三处互相矛盾， 现统一为 115200 且可覆盖。若你的现场确实是 9600，加 --baud 9600 即可。 · --from/--to 与电表地址在启动时校验，参数错立刻报错退出，而不是抄到一半才炸。 · main() 返回退出码：0 正常、1 串口打不开、2 参数错误。 """ from __future__ import annotations import csv import os import sys import time import threading from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Dict, List, Optional, Sequence, Set, TextIO, Tuple try: import serial from serial import SerialException SERIAL_AVAILABLE = True except ImportError: SERIAL_AVAILABLE = False SerialException = Exception # 串口单次读切片（秒）：没数据时最多阻塞这么久，避免忙等烧 CPU _READ_SLICE = 0.05 # ============================================================ # 数据模型 # ============================================================ @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._ser.reset_input_buffer() self._ser.reset_output_buffer() 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: """ 取一次已到达的字节（可能含多帧粘包，交给上层按 L 字段切分）。 优先读「已经在驱动缓冲区里的字节」（in_waiting）——有数据立刻返回， 不必白等一个读切片；没数据才做一次最多 _READ_SLICE 秒的阻塞读。 """ 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 = _READ_SLICE while time.time() < deadline: pending = self._ser.in_waiting chunk = self._ser.read(pending) if pending else self._ser.read(1) if not chunk: if buf and not expected_len and not end_flag: break continue buf.extend(chunk) if end_flag and buf.endswith(end_flag): break if expected_len and len(buf) >= expected_len: break finally: self._ser.timeout = old_timeout return bytes(buf) 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 fmt_clock() -> str: """时:分:秒.毫秒，用于主日志逐行前缀（便于和 raw_frames.txt 对齐时间）""" return datetime.now().strftime("%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.txt", 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._abandoned: Set[Tuple[str, str]] = set() # 已放弃的任务块 (表号, DI) self._builders: Dict[str, LocalProtocolBuilder] = {} # 表号 → 组包器（按表缓存） self._interrupted = False # 是否被 Ctrl-C 中断（用于优雅收尾） self._handles: Dict[str, TextIO] = {} # 日志句柄，按路径复用 self._lock = threading.RLock() # ---------------- 日志 / 资源 ---------------- def _handle(self, path: str) -> TextIO: """ 按路径复用文件句柄。 ★ 原来每写一行就 open(path, "a") 再关掉：冻结块日志一只表就有 255 行， 多只表纯属白扔 syscall。现在各日志共用一个句柄表，close() 时统一关闭； 把两个日志指到同一个文件也没问题（同一路径复用同一句柄）。 """ fp = self._handles.get(path) if fp is None: fp = open(path, "a", encoding="utf-8") self._handles[path] = fp return fp def _append(self, path: Optional[str], line: str) -> None: if not path: return with self._lock: fp = self._handle(path) fp.write(line + "\n") fp.flush() def _log(self, msg: str = "") -> None: """主日志（人可读流程）。日志文件逐行带毫秒时间戳，控制台保持原样。""" if self.verbose: print(msg) if self.log_path: self._append(self.log_path, f"[{fmt_clock()}] {msg}" if msg else "") 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: self._append(self.log_path, "\n".join(block)) 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: self._append(self.raw_log_path, line) 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 fp in self._handles.values(): try: fp.close() except Exception: pass self._handles.clear() 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 _local_builder(self, addr: str) -> LocalProtocolBuilder: """按表地址缓存组包器（地址字节只算一次，不再每项重建）""" b = self._builders.get(addr) if b is None: b = LocalProtocolBuilder(asrc_addr=self.asrc_addr, adst_addr=addr) self._builders[addr] = b return b def build_frames(self, task: SendTask) -> Dict[str, bytes]: inner = self.dlt645.build(task.meter, task.item) outer = self._local_builder(task.meter.addr).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) # ---- 已经随本表放弃的任务块：任何应答都不再处理 ---- # ★ 原来这些帧会进「延迟应答缓存」，但该 DI 永远不会再被下发， # 缓存条目没人取、只会在日志里装成正常的 [缓存延迟应答]。 if rep.get("di"): fkey = (rep["addr"], rep["di"].upper()) if fkey in self._abandoned: self._log(f" [已放弃] 任务块 上{_di_x(rep['di'])}次(DI={rep['di']}) " f"属于已放弃的 {rep['addr']}，应答直接丢弃（不入缓存）") self._raw("RX-SKIP", frame, f"已放弃任务块的应答 DI={rep['di']}，丢弃") return None, False # ---- 上报任务状态帧（模块受理/执行结果汇报，不含 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: """ 抄一只表的全部冻结项，返回该表报告。 注意：CSV 由 run() 在全部表抄完、最后一轮补抄收干之后统一落盘 （这样补抄回填的结果才会体现在 CSV 里）；直接调用本方法不写 CSV。 """ 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 try: 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}，放弃抄读", )) # 该 DI 本表已放弃：后续万一有回放应答上来，直接丢弃即可 akey = (self._norm_addr(meter.addr), it.di.upper()) self._abandoned.add(akey) self._cache.pop(akey, None) self._cache.pop(("ERR",) + akey, None) report.aborted = True report.abort_reason = f"{reason}{detail} @ 上{idx + 1}次(DI={item.di})" break except KeyboardInterrupt: # ★ 原来 Ctrl-C 直接穿出去：当前表的 CSV 不写、未抄的项没有任何记录。 # 现在把没来得及抄的项按 SKIPPED 记账，收尾照常落盘、照常打汇总。 self._interrupted = True done = len(report.results) for it in items[done:]: report.results.append(TaskResult( seq=self._task_no & 0xFFFF, meter_addr=meter.addr, di=it.di, di_name=it.description, status=TaskStatus.SKIPPED, note="手动中断（Ctrl-C）", )) report.aborted = True report.abort_reason = f"手动中断（Ctrl-C）@ 已处理 {done}/{len(items)} 项" self._log(f" ⏹ 收到 Ctrl-C：本表剩余 {len(items) - done} 项以 SKIPPED 记账，" f"停止后续电表并收尾落盘") # ---------------- 收尾：捞最后一批迟到应答（自动补抄） ---------------- 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]: """ 按顺序抄多只表；单表异常/放弃都不影响后续电表。 四份日志在抄读过程中逐行 flush，CSV 在本方法末尾统一落盘（含补抄回填后的最终结果）。 """ items = list(items) if items is not None else self.freeze_items() self.log_banner(meters, items) reports: List[MeterReport] = [] try: for i, m in enumerate(meters, 1): self._log(f"\n>>>>>>>>>> [{i}/{len(meters)}] " f"{m.description or ''} {m.addr} <<<<<<<<<<") try: reports.append(self.read_meter(m, items)) except Exception as e: # 单表异常不影响其它表 self._log(f" !! 该表抄读过程中出现异常: {e}，跳过并继续下一只表") reports.append(MeterReport(meter=m, aborted=True, abort_reason=f"异常: {e}")) if self._interrupted: self._log(" ⏹ 已手动中断：不再抄读后续电表") break finally: # ★ 落盘推迟到这里：read_meter 收尾和本次收尾都会把迟到应答回填进 # 各表的结果对象，先回填完再写 CSV，日志与 CSV 才不会互相矛盾 # （原来在 read_meter 里就写了，补抄回来的项在 CSV 里仍是 NO_RESP）。 self._drain_to_cache(max_wait=0.3) for rep in reports: self._write_csv(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} 项") if self._interrupted: self._log(" ⚠ 本次运行被手动中断（Ctrl-C）：已抄回的数据与未抄项均已落盘") 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), ]) # ============================================================ # 默认配置（全部可用命令行参数覆盖） # ============================================================ DEFAULT_PORT = "COM4" # 串口名：COM4 / /dev/ttyUSB0 DEFAULT_BAUDRATE = 115200 # 本地通信接口：115200 8E1 DEFAULT_PARITY = "E" CCO_ADDR = "000000000000" # ASRC：CCO/模块地址，原样写入 DEFAULT_METERS = [ MeterAddress("260498001817", "1# 表"), # MeterAddress("584c07000006", "2# 表"), ] # 日志文件（全部纯文本 .txt；置 None 即关闭该类输出） # 注意 .txt 而不是 .log：纯文本可直接打开/预览，.log 会被某些工具当二进制 LOG_FILES = 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 可打开） ) # 可选：下发抄读前先发一条固定探测帧（观测链路用，--probe 打开） PROBE_FRAME = "681500C00503050505E80300171800980426FFB216" def _validate_meter_addr(addr: str) -> str: """电表地址必须是 12 位十六进制 BCD（如 260498001817），错了立刻报出来""" a = (addr or "").strip().lower() if len(a) != 12 or any(c not in "0123456789abcdef" for c in a): raise ValueError(f"电表地址必须是 12 位十六进制（如 260498001817），实际: {addr!r}") return a def parse_args(): import argparse p = argparse.ArgumentParser( description="电表「上x次整点冻结数据」抄读：电表回错误帧即放弃该表后续冻结项，切下一只表", formatter_class=argparse.ArgumentDefaultsHelpFormatter) p.add_argument("--port", default=DEFAULT_PORT, help="串口名，如 COM4 / /dev/ttyUSB0") p.add_argument("--baud", type=int, default=DEFAULT_BAUDRATE, help="波特率（本地通信接口通常 115200；现场若为 9600 请显式指定）") 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次 开始") p.add_argument("--to", dest="x_to", type=int, default=255, help="到 上y次 结束") p.add_argument("--timeout", type=float, default=5.0, help="单条等待应答超时(秒)（实测应答 0.13~0.35s 到达）") p.add_argument("--retries", type=int, default=2, help="超时/坏帧重试次数") p.add_argument("--replay-grace", type=float, default=1.0, help="被本表回放帧挤占窗口后，静默该秒数即提前重发（0=关闭）") p.add_argument("--abort-streak", type=int, default=5, help="连续 N 次无应答/坏帧才放弃本表剩余（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("--probe", action="store_true", help=f"抄读前先发一条固定探测帧 {PROBE_FRAME}") return p.parse_args() def main() -> int: args = parse_args() # ---- 参数校验：错了立刻退出，而不是抄到一半才炸 ---- if not (1 <= args.x_from <= args.x_to <= 255): print(f"[错误] --from/--to 需满足 1 <= from <= to <= 255，" f"实际: --from {args.x_from} --to {args.x_to}") return 2 if args.meter: try: meters = [MeterAddress(_validate_meter_addr(a), f"{i}# 表") for i, a in enumerate(args.meter, 1)] except ValueError as e: print(f"[错误] {e}") return 2 else: meters = list(DEFAULT_METERS) # 放弃策略：命中即不再抄该表剩余「上x次冻结数据」，直接切下一只表 policy = AbortPolicy( on_error_frame = True, # ★ 电表回错误帧 → 立即放弃本表剩余 on_no_response = True, # 连续无应答 → 放弃本表剩余 on_bad_frame = True, # 连续坏帧 → 放弃本表剩余 streak_limit = 0 if args.keep_on_timeout else args.abort_streak, ) transport = SerialPort(port=args.port, baudrate=args.baud, parity=DEFAULT_PARITY, timeout=args.timeout + 1.0, verbose=True) try: transport.open() except SerialPortError as e: print(f"[错误] 串口打开失败：{e}") print(" 请检查串口号 / 波特率；若提示未安装 pyserial，执行：pip install pyserial") return 1 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_FILES, ) try: if args.probe: transport.send(bytes.fromhex(PROBE_FRAME)) session.run(meters, session.freeze_items()) except KeyboardInterrupt: print("\n[中断] 已手动终止：未抄项按 SKIPPED 记账，结果已写入日志与数据表。") finally: session.close() # 关闭四份日志的句柄（内容已逐行 flush） transport.close() return 0 if __name__ == "__main__": sys.exit(main())