# -*- coding: utf-8 -*- """ 电表「上x次整点冻结数据」抄读程序（v2：类型化 + 拆类） ============================================================ 演进路线：buddy.py（重写版）→ freeze_reader.py（精简：删自测 + 一轮优化） → 本文件（在 freeze_reader.py 基础上做结构重构，行为不变） 【v2 做了什么】 ------------------------------------------------------------ 1) 解析结果类型化，干掉「嵌套 dict」。 旧写法调用方要写 rep["645"]["err_code"]、d.get("PayloadHex")， 键名写错要等运行时才炸，也看不出一个应答到底有哪些字段。 现在全部是 frozen dataclass，两层封顶且字段有名字、有类型： DLT645Frame 一帧 645 应答（.kind / .di / .payload / .freeze / .error_code） FreezeBlock 16B 冻结块（.freeze_time / .positive / .reverse / .raw） LocalFrame 一帧本地规约（.task_id / .seq / .inner / .checksum_ok） TaskStatusReport 上报任务状态帧 Uplink 上行帧的分类结果（UplinkKind 判别式 + 该分支的字段） 于是 frame.freeze.freeze_time 就是最深一层，不再是 session._cache[(addr, di)]["645"]["freeze_time"] 这种钻法。 2) 拆类。v1 的 FreezeReadSession 有 900 行，同时兼任协议解析器、日志文件 管理器、CSV 写入器、缓存管理器。v2 按职责拆开： DLT645Codec / LocalProtocolCodec 帧解析（纯解析，不碰 IO） UplinkClassifier 上行帧分类（bytes → Uplink，含日志） FrameSplitter 字节流切帧（带残留缓冲） UplinkPump 收发 + 分类 + 回确认 + 交归属 FrameRouter 应答归属 / 延迟缓存 / 补抄 / 放弃 AckResponder 确认帧与去重 LogSink / CsvWriter 四份文本日志 / 业务数据表 SessionStats 会话计数 FreezeReadSession 只剩流程编排、重试与放弃策略 3) 顺手修掉一个继承自 v1 的显示缺陷：日志注解里，ERR=1 且数据域为 「DI(4B) + 错误码」时，旧代码把 DI 的首字节当成错误码打印 （实测：0x02 无请求数据 被显示成 0x07 费率数超）。现在注解复用类型化 解析器，不再自己切一遍字节 —— 同一语义只有一份实现。 协议语义、归属规则、放弃策略、日志文本全部与 v1 保持一致：v1 的三套自测 **原样** 跑在 v2 上全部通过。（v1 那轮精简的改动见文末。） 组包部分依旧原样保留： DLT645FrameBuilder DL/T 645-2007 组包 DLT645Codec DL/T 645-2007 解包（类型化，替代 DLT645Parser） LocalProtocolBuilder 南网集中器本地通信规约组包 LocalProtocolCodec 南网集中器本地通信规约解包（类型化，替代 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_v2.py # 按文件内配置抄读 python freeze_reader_v2.py --port COM3 --baud 9600 # 覆盖串口号 / 波特率 python freeze_reader_v2.py --meter 260498001817 --meter 260498001820 python freeze_reader_v2.py --from 1 --to 24 # 只抄 上1次~上24次 python freeze_reader_v2.py --keep-on-timeout # 超时/坏帧永不放弃本表 python freeze_reader_v2.py --no-ack --dump-console # 不回确认帧 + 解析打屏 # 当库用（组合式装配，测试可用假串口替掉 Transport）： logs = LogSink(log_path="read.txt", raw_log_path="raw.txt") session = FreezeReadSession(transport=SerialPort("COM4"), logs=logs, csv=CsvWriter("data.txt")) session.run([MeterAddress("260498001817", "1# 表")], make_freeze_items(1, 24)) freeze_reader.py（v1）相对 buddy.py 的改动（v2 继承） ------------------------------------------------------------ 【删除】 · 移除 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, Protocol, 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: """ 单条任务的完整结果。 v1 在这里挂了两个无类型 dict（parsed_local / parsed_645），调用方要 r.parsed_645["err_code"] 这样往下钻；v2 直接挂类型化对象， 冻结时间/正反向电能则做成属性，避免同一份数据存两遍。 """ 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 = "" value: str = "" inner_645: bytes = b"" # 内层 645 应答原始报文 local_frame: Optional["LocalFrame"] = None # 本地规约解析结果 dlt645_frame: Optional["DLT645Frame"] = None # 内层 645 解析结果 freeze: Optional["FreezeBlock"] = None # 冻结块（时间 + 正反向电能） payload_raw: bytes = b"" # 数据域原始字节（含不足 16B 的短冻结块） @property def success(self) -> bool: return self.status is TaskStatus.OK @property def freeze_time(self) -> str: return self.freeze.freeze_time if self.freeze is not None else "" @property def positive(self) -> str: return self.freeze.positive if self.freeze is not None else "" @property def reverse(self) -> str: return self.freeze.reverse if self.freeze is not None else "" @property def freeze_block(self) -> bytes: """冻结块 16B 原始字节；非冻结项/短块时退化为数据域原始字节（与 v1 一致）""" return self.freeze.raw if self.freeze is not None else self.payload_raw @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 协议解析（类型化） # ------------------------------------------------------------ # v1 的解析结果是「嵌套 dict」：d["DI"] / d["PayloadHex"] / d["freeze_time"]， # 路由里到处是 rep["645"]["err_code"] 这种字符串键往下钻 —— 键名写错要等运行时 # 才炸，也看不出一个应答到底有哪些字段。v2 换成 frozen dataclass： # frame.kind / frame.di / frame.freeze.freeze_time （两层封顶，全部有名字） # ============================================================ class FrameError(ValueError): """帧结构错误（起止符/长度/字段越界）。与 v1 的 ValueError 同族，便于 except""" class FrameKind(str, Enum): """一帧 645 应答的解析结论（v1 用 kind + abnormal 两个松散标记表示，容易漏判）""" ENERGY = "energy" # 电能类数据项 FREEZE = "freeze" # 整点冻结数据块 RAW = "raw" # 其它数据项，原样保留 ERROR_RESPONSE = "error_response" # ERR=1 异常应答（电表明确应答：没有这个数据） MALFORMED = "malformed" # 结构/长度/校验问题 # ---- BCD / 电能值纯函数（解析与日志注解共用同一份实现，避免两处各写一遍） ---- def bcd_to_str(data: bytes) -> str: """BCD 字节 → 十进制字符串；任一半字节 >9 则整体退化为 HEX（与 v1 行为一致）""" 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 def bcd_to_int(b: int) -> int: """单字节 BCD → 整数；非法 BCD 返回 -1""" hi = (b >> 4) & 0x0F lo = b & 0x0F if hi > 9 or lo > 9: return -1 return hi * 10 + lo def format_energy(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:] def decode_645_data(encoded: bytes, plus33: bool = True) -> bytes: """数据域 +33H 解码""" if not plus33: return encoded return bytes((b - 0x33) & 0xFF for b in encoded) @dataclass(frozen=True) class FreezeBlock: """ 一次整点冻结数据块（16 字节，低字节在前存）： [0:6] 冻结时间，末尾 AA 分隔 [6:11] 正向有功总电能，末尾 AA 分隔 [11:16] 反向有功总电能，末尾 AA 分隔 """ raw: bytes time_bytes: bytes positive_bytes: bytes reverse_bytes: bytes freeze_time: str = "" positive: str = "" reverse: str = "" MIN_LEN = 16 @classmethod def parse(cls, payload: bytes) -> Optional["FreezeBlock"]: """长度不足（<16B）返回 None —— 调用方据此判 MALFORMED""" if len(payload) < cls.MIN_LEN: return None time_bytes = payload[0:6] positive_bytes = payload[6:11] reverse_bytes = payload[11:16] t = time_bytes[::-1] # [AA, 年, 月, 日, 时, 分] year, month, day = bcd_to_int(t[1]), bcd_to_int(t[2]), bcd_to_int(t[3]) hour, minute = bcd_to_int(t[4]), bcd_to_int(t[5]) freeze_time = f"{year}年{month:0>2}月{day:0>2}日{hour:0>2}时{minute:0>2}分" # 每段首字节是 0xAA 分隔符，跳过 positive = format_energy(bcd_to_str(positive_bytes[::-1][1:]), 2) reverse = format_energy(bcd_to_str(reverse_bytes[::-1][1:]), 2) return cls(raw=payload[:cls.MIN_LEN], time_bytes=time_bytes, positive_bytes=positive_bytes, reverse_bytes=reverse_bytes, freeze_time=freeze_time, positive=positive, reverse=reverse) @property def display(self) -> str: return f"正:{self.positive} 反:{self.reverse}" @dataclass(frozen=True) class DLT645Frame: """一帧 DL/T 645-2007 应答（字段全部有名字，不再是字符串键）""" raw: bytes addr_bytes: bytes addr: str control: int length: int data: bytes # 已 -33H 的数据域 checksum: int checksum_calc: int end: int kind: FrameKind di: str = "" di_name: str = "" payload: bytes = b"" value: str = "" value_bcd: str = "" freeze: Optional[FreezeBlock] = None error_code: Optional[int] = None error_text: str = "" @property def direction(self) -> int: return (self.control >> 7) & 1 @property def is_error_response(self) -> bool: """ERR 位=1：电表明确回复「我这边没有这个数据」""" return ((self.control >> 6) & 1) == 1 @property def follow(self) -> int: return (self.control >> 5) & 1 @property def func(self) -> int: return self.control & 0x1F @property def checksum_ok(self) -> bool: return self.checksum == self.checksum_calc @property def has_di(self) -> bool: return bool(self.di) @property def is_conclusive_error(self) -> bool: """ 「这条应答的结论是坏的」—— 等价于 v1 的 abnormal 标记： ERR=1 异常应答，或（正常应答但结构坏掉、取不出 DI）。 """ return self.kind in (FrameKind.ERROR_RESPONSE, FrameKind.MALFORMED) class DLT645Codec: """ DL/T 645-2007 解析器：bytes → DLT645Frame；结构问题抛 FrameError。 ★ 异常应答也解析 DI：正常应答与异常应答的数据域结构一样，都是「DI(4B) + 数据」。 只要长度够就把 DI 取出来（kind=ERROR_RESPONSE 且 di 非空），供上层按 「表号 + DI」归属；只有数据域短于 4 字节时才没有 DI。早期版本直接把 DI 丢掉， 那正是「异常帧只能靠 TaskID 归属」的根源。 """ 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 __init__(self, plus33: bool = True): self.plus33 = plus33 # ---------------- 解析 ---------------- def parse(self, frame: bytes) -> DLT645Frame: if len(frame) < 12: raise FrameError("645 帧过短") if frame[0] != self.START or frame[7] != self.START: raise FrameError("645 起始字节错误") if frame[-1] != self.END: raise FrameError("645 结束字节错误") addr = frame[1:7] ctrl = frame[8] length = frame[9] data = decode_645_data(frame[10:10 + length], self.plus33) checksum = frame[10 + length] end = frame[11 + length] base = dict(raw=frame, addr_bytes=addr, addr=addr[::-1].hex().upper(), control=ctrl, length=length, data=data, checksum=checksum, checksum_calc=sum(frame[0:10 + length]) & 0xFF, end=end) # ---------- 异常应答（ERR=1）---------- if ((ctrl >> 6) & 1) == 1: if len(data) >= 4: di = data[0:4][::-1].hex().upper() code = data[4] if len(data) >= 5 else None return DLT645Frame(**base, kind=FrameKind.ERROR_RESPONSE, di=di, di_name=self.describe_di(di), error_code=code, error_text=self.error_text(code)) code = data[0] if data else None return DLT645Frame(**base, kind=FrameKind.ERROR_RESPONSE, error_code=code, error_text=self.error_text(code)) # ---------- 正常应答 ---------- if len(data) < 4: return DLT645Frame(**base, kind=FrameKind.MALFORMED, error_text=f"数据域过短（{len(data)} 字节），无法取 DI") di = data[0:4][::-1].hex().upper() payload = data[4:] if not payload: return DLT645Frame(**base, kind=FrameKind.RAW, di=di, di_name=self.describe_di(di)) if di.startswith("0504FF"): block = FreezeBlock.parse(payload) if block is None: return DLT645Frame(**base, kind=FrameKind.MALFORMED, di=di, di_name=self.describe_di(di), payload=payload, error_text=f"冻结块长度不足（{len(payload)} 字节）") return DLT645Frame(**base, kind=FrameKind.FREEZE, di=di, di_name=self.describe_di(di), payload=payload, freeze=block, value=block.display) if di.endswith("0000"): return DLT645Frame(**base, kind=FrameKind.ENERGY, di=di, di_name=self.describe_di(di), payload=payload, value_bcd=payload[::-1].hex().upper(), value=format_energy(bcd_to_str(payload), 2)) return DLT645Frame(**base, kind=FrameKind.RAW, di=di, di_name=self.describe_di(di), payload=payload, value_bcd=payload[::-1].hex().upper(), value=payload.hex().upper()) # ---------------- 名字 / 错误码 ---------------- def describe_di(self, di: str) -> str: """DI → 中文名；冻结类 DI 自动生成「上n次整点冻结数据块」""" name = self.DI_NAMES.get(di) if name: return name if len(di) == 8 and di.startswith("0504FF"): try: return f"上{int(di[6:8], 16)}次整点冻结数据块" except ValueError: return "" return "" def error_text(self, code: Optional[int]) -> str: if code is None: return "" return self.ERR_CODES.get(code, f"未知错误码 0x{code:02X}") # ============================================================ # 类 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：南网集中器本地通信规约解析（类型化） # ============================================================ # ---- 上报任务状态帧（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: "其它"} @dataclass(frozen=True) class LocalFrame: """ 一帧南网集中器本地通信规约报文。 下行请求帧(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) ：仅基本字段（length_ok / checksum_ok 为 None，与 v1 一致） """ raw: bytes length: int control: int is_ack: bool = False asrc: str = "" adst: str = "" asrc_bytes: bytes = b"" adst_bytes: bytes = b"" afn: int = 0 seq: int = 0 duid: bytes = b"" task_id: int = 0 inner: bytes = b"" length_calc: Optional[int] = None checksum: Optional[int] = None checksum_calc: Optional[int] = None end: int = 0 @property def direction(self) -> int: return (self.control >> 7) & 1 @property def prm(self) -> int: return (self.control >> 6) & 1 @property def add(self) -> int: return (self.control >> 5) & 1 @property def ver(self) -> int: return self.control & 0x0F @property def length_ok(self) -> Optional[bool]: return None if self.length_calc is None else self.length == self.length_calc @property def checksum_ok(self) -> Optional[bool]: return None if self.checksum is None else self.checksum == self.checksum_calc @property def inner_645(self) -> Optional[bytes]: """内层是不是一整帧 645（68 ... 16）""" if self.inner[:1] == b"\x68" and self.inner[-1:] == b"\x16": return self.inner return None @dataclass(frozen=True) class TaskStatusReport: """ 模块「上报任务状态」帧（C=0xC0：上行走廊、PRM=1、无地址域，总长 21 字节）： 68 L(2) C AFN(1) SEQ(1) DUID(4) TaskID(2,LE) 从节点地址(6,BCD倒序) 任务状态(1) CS 16 """ raw: bytes seq: int duid: bytes task_id: int addr: str status_code: int checksum: int checksum_calc: int @property def status_text(self) -> str: return TASK_STATUS_CODES.get(self.status_code, f"未知(0x{self.status_code:02X})") @property def checksum_ok(self) -> bool: return self.checksum == self.checksum_calc @property def ok(self) -> bool: """00=成功（数据帧随后另发）；非 00 表示本项不会有数据""" return self.status_code == 0x00 class LocalProtocolCodec: """南网集中器本地通信规约解析器：bytes → LocalFrame；结构问题抛 FrameError""" START = 0x68 END = 0x16 AFN_REPORT = 0x05 DUID_TASK_STATUS = b"\x05\x05\x05\xE8" def parse(self, frame: bytes) -> LocalFrame: if len(frame) < 6: raise FrameError("帧过短") if frame[0] != self.START: raise FrameError(f"起始字节错误: {frame[0]:02X}") if frame[-1] != self.END: raise FrameError(f"结束字节错误: {frame[-1]:02X}") length = int.from_bytes(frame[1:3], "little") ctrl = frame[3] if ((ctrl >> 6) & 1) == 0: # 确认帧（PRM=0）：结构不同 return LocalFrame(raw=frame, length=length, control=ctrl, is_ack=True) if len(frame) < 26: raise FrameError("上报帧过短") asrc, adst = frame[4:10], frame[10:16] inner_len = frame[24] inner = frame[25:25 + inner_len] user_data = frame[4:25 + inner_len] return LocalFrame( raw=frame, length=length, control=ctrl, is_ack=False, asrc_bytes=asrc, asrc=asrc[::-1].hex().upper(), adst_bytes=adst, adst=adst[::-1].hex().upper(), afn=frame[16], seq=frame[17], duid=frame[18:22], task_id=int.from_bytes(frame[22:24], "little"), inner=inner, length_calc=len(user_data) + 6, checksum=frame[25 + inner_len], checksum_calc=(ctrl + sum(user_data)) & 0xFF, end=frame[26 + inner_len], ) def parse_task_status(self, frame: bytes) -> Optional[TaskStatusReport]: """ 识别并解析「上报任务状态」帧；不是该类帧返回 None。 （总长 21 字节，需 PRM=1 且无地址域，AFN=05 且 DUID=050505E8） """ if len(frame) != 21 or frame[0] != self.START or frame[-1] != self.END: return None ctrl = frame[3] if (ctrl & 0xC0) != 0xC0 or ((ctrl >> 5) & 1): return None if frame[4] != self.AFN_REPORT or frame[6:10] != self.DUID_TASK_STATUS: return None return TaskStatusReport( raw=frame, seq=frame[5], duid=frame[6:10], task_id=int.from_bytes(frame[10:12], "little"), addr=frame[12:18][::-1].hex().upper(), status_code=frame[18], checksum=frame[19], checksum_calc=(ctrl + sum(frame[4:19])) & 0xFF, ) 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]) # ============================================================ # 上行帧分类：一条「字节帧」→ 可判别类型 # ------------------------------------------------------------ # v1 用 Optional[dict] 表示分类结果，调用方靠 rep.get("status_report")、 # rep["645"]、rep.get("abnormal") 猜结构。v2 换成 UplinkKind 判别式 + Uplink， # 每个分支该有哪些字段，类型上就写明了。 # ============================================================ class UplinkKind(str, Enum): CONFIRM = "confirm" # 受理回执：任务已受理，数据随后另发 TASK_STATUS = "task_status" # 上报任务状态：模块对该任务的执行结论 DATA = "data" # 上报任务数据（带内层 645） NON_DATA = "non_data" # 本地规约合法，但不含 645 内层 ILLEGAL = "illegal" # 结构/长度/校验失败，无法解析 @dataclass(frozen=True) class Uplink: """ 一条上行报文的分类结果。归属只用到 inner（表号 + DI）， 再也不需要 rep["645"]["err_code"] 这种按字符串键往下钻的写法。 """ kind: UplinkKind frame: bytes local: Optional[LocalFrame] = None inner: Optional[DLT645Frame] = None status: Optional[TaskStatusReport] = None note: str = "" should_ack: bool = False ack_seq: int = 0 ack_src: str = "" @property def addr(self) -> str: return self.inner.addr if self.inner is not None else "" @property def di(self) -> str: return self.inner.di if self.inner is not None else "" @property def is_conclusive_error(self) -> bool: """这条应答的结论是坏的（v1 的 abnormal 语义，用于缓存命中时的重新确认判断）""" return self.inner is not None and self.inner.is_conclusive_error class UplinkClassifier: """ 上行帧分类器：bytes → Uplink。分类的同时写日志与原始报文流水 （这些日志本身就是排障依据，按 v1 的原文保留）。 """ def __init__(self, logs: LogSink, local: Optional[LocalProtocolCodec] = None, dlt645: Optional[DLT645Codec] = None): self.logs = logs self.local = local or LocalProtocolCodec() self.dlt645 = dlt645 or DLT645Codec() def classify(self, frame: bytes) -> Uplink: if len(frame) < 4: self.logs.raw("RX-BAD", frame, "帧过短") return Uplink(kind=UplinkKind.ILLEGAL, frame=frame, note="帧过短") # 1) 确认帧（PRM=0）→ 只表示任务已受理，数据随后另发；不再对它回确认 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.logs.flow(f" [确认帧] {frame.hex().upper()}" f"（{afn_txt}，任务已受理，等待数据帧）") self.logs.raw("RX-ACK", frame, f"确认帧(PRM=0 {afn_txt})，跳过") self.logs.block("· 确认帧", annotate_local_frame(frame, with_inner=False)) return Uplink(kind=UplinkKind.CONFIRM, frame=frame, note=afn_txt) # 2) 上报任务状态帧（C=0xC0 / AFN=05 / DUID=050505E8） status = self.local.parse_task_status(frame) if status is not None: if not status.checksum_ok: self.logs.flow(f" [任务状态帧CS错] {frame.hex().upper()}") self.logs.raw("RX-BAD", frame, f"任务状态帧 CS={status.checksum:02X} " f"应为 {status.checksum_calc:02X}") return Uplink(kind=UplinkKind.ILLEGAL, frame=frame, status=status, note="任务状态帧 CS 错") return Uplink(kind=UplinkKind.TASK_STATUS, frame=frame, status=status, should_ack=True, ack_seq=status.seq, ack_src=status.addr) # 3) 上报帧 → 解析本地规约 try: local = self.local.parse(frame) except Exception as e: # 21 字节的 C=0xC0 帧（模块状态/回执，非抄读数据）单独归类，不算"非法帧" if len(frame) < 26: self.logs.flow(f" [非抄读帧] {frame.hex().upper()} ({e})") self.logs.raw("RX-MISC", frame, f"非抄读帧: {e}") return Uplink(kind=UplinkKind.NON_DATA, frame=frame, note=f"非抄读帧: {e}") self.logs.flow(f" [跳过非法帧] {frame.hex().upper()} ({e})") self.logs.raw("RX-BAD", frame, f"非法帧: {e}") return Uplink(kind=UplinkKind.ILLEGAL, frame=frame, note=f"非法帧: {e}") # ★ 本地规约合法 → 立即回确认帧（SEQ 取自本帧，模块据此停止重传） # 校验错（CS/L）的帧不回确认，让模块重传，避免"确认了一条坏帧" should_ack = local.checksum_ok is not False if not should_ack: self.logs.flow(" [不上报确认] 本地规约 CS 校验失败，等待模块重传") ack = dict(should_ack=should_ack, ack_seq=local.seq, ack_src=local.asrc) if not local.inner or local.inner_645 is None: self.logs.flow(f" [非抄读帧] {frame.hex().upper()}（无 645 内层）") self.logs.raw("RX-MISC", frame, "无 645 内层，非抄读帧") return Uplink(kind=UplinkKind.NON_DATA, frame=frame, local=local, note="无 645 内层", **ack) try: inner = self.dlt645.parse(local.inner) except Exception as e: self.logs.flow(f" [跳过非法645帧] {local.inner.hex().upper()} ({e})") self.logs.raw("RX-BAD", local.inner, f"645 非法帧: {e}") return Uplink(kind=UplinkKind.ILLEGAL, frame=frame, local=local, note=f"645 非法帧: {e}", **ack) return Uplink(kind=UplinkKind.DATA, frame=frame, local=local, inner=inner, **ack) # ============================================================ # 类 5：串口通信类 （原样保留） # ============================================================ 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) # ============================================================ # 报文日志工具：把原始帧逐字段拆成可读文本 # ============================================================ _645_CODEC = DLT645Codec() _645_CODEC_NO_PLUS33 = DLT645Codec(plus33=False) 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)}") # ★ 语义部分交给类型化解析器，不再在这里自己切一遍字节。 # 旧写法（v1 同样如此）把异常应答的数据域当「1 字节错误码」处理， # 于是数据域是「DI(4B) + 错误码」时，会把 DI 的第一个字节当成错误码打出来 # （实测：0x02 无请求数据 会显示成 0x07 费率数超）。 codec = _645_CODEC if plus33 else _645_CODEC_NO_PLUS33 try: parsed = codec.parse(frame) except Exception: parsed = None if parsed is not None and parsed.is_error_response: if parsed.di: code_txt = (f"错误码 0x{parsed.error_code:02X} {parsed.error_text}" if parsed.error_code is not None else "无错误码") name = f" ({parsed.di_name})" if parsed.di_name else "" out.append(f"{indent}│ ★ 异常应答 : DI={parsed.di}{name} {code_txt}") elif parsed.error_code is not None: out.append(f"{indent}│ ★ 异常应答 : 错误码 0x{parsed.error_code:02X} " f"{parsed.error_text}") else: out.append(f"{indent}│ ★ 异常应答 : 无错误码") elif parsed is not None and parsed.has_di: out.append(f"{indent}│ 数据标识 DI : {parsed.di}" + (f" ({parsed.di_name})" if parsed.di_name else "")) if parsed.payload: out.append(f"{indent}│ 数据块原始 : {fmt_hex(parsed.payload)} " f"({len(parsed.payload)} 字节)") out.extend(_annotate_645_payload(parsed.di, parsed.payload, indent + "│ ")) elif parsed.direction == 0: out.append(f"{indent}│ 数据块 : 无（下行读请求，仅带 DI）") else: out.append(f"{indent}│ 数据域 : {fmt_hex(data)}") if parsed is not None and parsed.error_text: out.append(f"{indent}│ 说明 : {parsed.error_text}") 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 block = FreezeBlock.parse(payload) if block is None: lines.append(f"{ind}★ 冻结块长度不足：{len(payload)} 字节（应为 16）") return lines lines.append(f"{ind}冻结块 16B 布局 = 时间[0:6] + 正向有功[6:11] + 反向有功[11:16]") lines.append(f"{ind} 时间 6B : {fmt_hex(payload[0:6])} → {block.freeze_time}") lines.append(f"{ind} 正向 5B : {fmt_hex(payload[6:11])} → {block.positive} kWh") lines.append(f"{ind} 反向 5B : {fmt_hex(payload[11:16])} → {block.reverse} kWh") return lines if di.endswith("0000"): lines.append(f"{ind}电能值 : {fmt_hex(payload)} → " f"{format_energy(bcd_to_str(payload), 2)} 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：日志落盘（LogSink / CsvWriter / SessionStats） # ------------------------------------------------------------ # v1 把「流程日志 / 报文注解 / 原始报文 / 冻结块明细 / CSV」五种落盘全塞在 # FreezeReadSession 里（_log / _dump / _raw / _freeze / _write_csv）， # 会话类因此同时兼任协议注解器和文件管理器。v2 拆成只管落盘的两个对象。 # ============================================================ @dataclass class LogSink: """ 四份文本日志的出口（按路径复用句柄，不再每行 open/close）。 flow() 主日志：流程 + 逐字段报文解析（文件里逐行带毫秒时间戳，控制台不加前缀） block() 逐字段报文解析块（块头自带完整时间戳） raw() 原始报文流水：一行一帧，便于和抓包工具比对 freeze() 冻结数据块明细：一行一块 """ log_path: Optional[str] = None raw_log_path: Optional[str] = None freeze_log_path: Optional[str] = None dump_frames: bool = True # 是否把逐字段解析写进主日志 dump_console: bool = False # 逐字段解析是否也打控制台 raw_console: bool = False # 原始报文是否也打控制台 verbose: bool = True # 控制台总开关 _handles: Dict[str, TextIO] = field(default_factory=dict, repr=False) _lock: threading.RLock = field(default_factory=threading.RLock, repr=False) def _handle(self, path: str) -> TextIO: 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 flow(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 block(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])}] " f"反[{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 '?'}次 | " f"DI={r.di} | 状态={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() class CsvWriter: """ 业务数据表（逗号分隔）。首次创建时写表头；utf-8-sig 便于 Excel/WPS 直接打开。 注意：**只在会话收尾时写**（run() 里，等最后一轮补抄回填完）， 这样「日志写着补抄成功、CSV 里还是 NO_RESP」的矛盾不会出现。 """ HEADER = ["表号", "表名", "序号", "DI", "数据项", "状态", "冻结时间", "正向有功(kWh)", "反向有功(kWh)", "数值", "重试", "错误码", "错误信息", "645应答报文", "冻结块原始数据(16B)"] def __init__(self, path: Optional[str] = None): self.path = path def write(self, report: MeterReport) -> None: if not self.path: return new_file = not os.path.exists(self.path) with open(self.path, "a", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) if new_file: w.writerow(self.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), ]) @dataclass class SessionStats: """会话级计数（v1 是散落在会话类里的 5 个 int/布尔属性）""" late_fill: int = 0 # 补抄回填成功的项数 dup_frames: int = 0 # 忽略掉的重放重复帧 cache_hits: int = 0 # 命中延迟应答缓存（省掉一次重发） ack_frames: int = 0 # 已回给模块的确认帧数 interrupted: bool = False # 是否被 Ctrl-C 中断 @dataclass(frozen=True) class SplitResult: """一次分帧的结果；overflow 非空表示残留缓冲异常增长被截断（供记日志）""" frames: List[bytes] overflow: Optional[bytes] = None overflow_len: int = 0 @dataclass class FrameSplitter: """ 把串口字节流切成本地规约帧（68H | L(2) | ... | 16H，L = 整帧长度）。 尾部不完整的帧留在缓冲里等下一批字节（v1 早期版本直接丢弃，会丢帧/错帧）。 """ residue: bytearray = field(default_factory=bytearray) MAX_RESIDUE: int = 4096 def feed(self, chunk: bytes) -> SplitResult: if chunk: self.residue.extend(chunk) buf = self.residue frames: List[bytes] = [] i, n = 0, len(buf) while i < n: if buf[i] != 0x68: i += 1 continue if i + 3 > n: break length = int.from_bytes(buf[i + 1:i + 3], "little") if length < 6: i += 1 continue if i + length > n: break frames.append(bytes(buf[i:i + length])) i += length del buf[:i] if len(buf) > self.MAX_RESIDUE: # 防御：长期无有效帧时截断，避免无限增长 overflow = bytes(buf[:64]) overflow_len = len(buf) del buf[:len(buf) - 512] return SplitResult(frames=frames, overflow=overflow, overflow_len=overflow_len) return SplitResult(frames=frames) class AckResponder: """ 对模块的上报报文回确认帧：SEQ 必须与该上报报文完全一致，否则模块会一直重传。 去重：同一 (源地址, SEQ) 在 dedup_window 秒内只确认一次 —— 回放风暴时同一帧 会被成批吐出，重复确认没有意义，只会占用总线时间（默认 0 = 每次都回）。 """ def __init__(self, transport: Optional[Transport], stats: SessionStats, logs: LogSink, enabled: bool = True, dedup_window: float = 0.0): self.transport = transport self.stats = stats self.logs = logs self.enabled = enabled self.dedup_window = dedup_window self._sent: Dict[Tuple[str, int], float] = {} def ack(self, seq: int, src_addr: str = "") -> None: if not self.enabled or self.transport is None: return now = time.time() key = (norm_addr(src_addr) if src_addr else "", seq & 0xFF) prev = self._sent.get(key) if prev is not None and now - prev < self.dedup_window: self.logs.raw("TX-ACK-DUP", b"", f"SEQ=0x{seq:02X} 已在 {self.dedup_window:g}s 内确认过，跳过重复确认") return self._sent[key] = now ack = build_report_ack(seq) try: self.transport.send(ack) except Exception as e: # 确认帧发失败不该影响抄读主流程 self.logs.flow(f" · 确认帧发送失败（{e}），继续等待数据帧") self.logs.raw("TX-ACK-ERR", ack, f"确认帧发送失败: {e}") return self.stats.ack_frames += 1 self.logs.flow(f" → 回确认帧（SEQ=0x{seq:02X} 与上报报文一致）: {ack.hex().upper()}") self.logs.raw("TX-ACK", ack, f"确认模块上报报文 SEQ=0x{seq:02X}") # ============================================================ # 类 7：应答归属（FrameRouter） # ============================================================ TaskKey = Tuple[str, str] # (表号, DI) —— 归属的唯一身份 def norm_addr(addr: str) -> str: """表号规范化：去空白、转大写、左补零到 12 位（v1 是 FreezeReadSession 的静态方法）""" return (addr or "").strip().upper().zfill(12) class Transport(Protocol): """ 串口/假串口的统一接口。v1 完全靠鸭子类型，注入假串口时看不出到底要用哪几个方法， 这里把它写成一个显式协议（SerialPort 与测试用假串口都天然满足）。 """ def send(self, data: bytes) -> int: ... def recv(self, expected_len: int = 0, end_flag: Optional[bytes] = None, max_wait: Optional[float] = None) -> bytes: ... def drain(self, max_wait: float = 0.0) -> bytes: ... def is_open(self) -> bool: ... def close(self) -> None: ... @dataclass(frozen=True) class RouteOutcome: """一次归属路由的结论（v1 是裸的 (frame, same_meter) 二元组）""" frame: Optional[bytes] = None # 命中当前任务的原始帧 same_meter: bool = False # 该帧是否属于当前正在抄的表 status_failed: bool = False # 模块上报「任务状态=失败」→ 本项不会有数据 class FrameRouter: """ ★ 核心：上行帧归属。身份只有一个 —— (表号, DI) =「上x次任务块」。 规则（TaskID 完全不参与归属）： · 正常应答 / 带 DI 的异常应答：归属键 = (表号, DI)。 只要内层 DI 就是当前在抄的 DI，就是答案 —— 模块回放/延迟会让外层 TaskID 滞后， 拿它判断会误杀真实数据。 · 不是当前任务块的帧 → 进缓存，而不是丢弃： - 属于本表、且是此前超时没取到的项 → 立即回填（补抄）； - 该 DI 此前已抄回 → 记 [重复帧]（回放重复）。 · 不带 DI 的异常应答（数据域只有错误码）：只在「正在等当前项」时按上下文归属； 窗口外收到的一律不归属、不入缓存（否则会被后续别的任务块误吃）。 · 已放弃任务块的应答：直接丢弃，不入缓存（否则缓存条目永远没人取）。 """ def __init__(self, logs: LogSink, stats: SessionStats): self.logs = logs self.stats = stats self.cache: Dict[TaskKey, Uplink] = {} # 延迟应答 self.got: Dict[TaskKey, TaskResult] = {} # 已成功抄回的项（判重） self.wait_fill: Dict[TaskKey, TaskResult] = {} # 尚未取到、等待补抄的项 self.abandoned: Set[TaskKey] = set() # 已随本表放弃的任务块 self.status_fail: Optional[TaskStatusReport] = None # ---------------- 归属 ---------------- @staticmethod def task_key(task: SendTask) -> TaskKey: return (norm_addr(task.meter.addr), task.item.di.upper()) def route(self, uplink: Uplink, task: Optional[SendTask]) -> RouteOutcome: if uplink.kind in (UplinkKind.CONFIRM, UplinkKind.NON_DATA, UplinkKind.ILLEGAL): return RouteOutcome() same_meter = bool(task is not None and uplink.addr and norm_addr(uplink.addr) == norm_addr(task.meter.addr)) # ---- 上报任务状态帧（模块受理/执行结果汇报，不含 645 数据）---- if uplink.kind is UplinkKind.TASK_STATUS: assert uplink.status is not None status = uplink.status desc = f"任务状态={status.status_text}(0x{status.status_code:02X})" if status.ok: # 成功 → 数据帧随后另发，继续等即可 self.logs.flow(f" [任务状态] TaskID=0x{status.task_id:04X} 表={status.addr} " f"{desc}（执行成功，等待数据帧）") self.logs.raw("RX-STATUS", uplink.frame, f"任务状态 {desc}（等待数据帧）") return RouteOutcome(same_meter=same_meter) if (task is not None and status.task_id == (task.seq & 0xFFFF) and status.addr == norm_addr(task.meter.addr)): # ★ 当前任务块执行失败（如 FF-其它）→ 不会再有数据帧，提前结束等待 self.status_fail = status self.logs.flow(f" [任务状态] TaskID=0x{status.task_id:04X} 表={status.addr} " f"{desc} → 当前任务块 上{_di_x(task.item.di)}次 不会返回数据，" f"提前结束等待") self.logs.raw("RX-STATUS", uplink.frame, f"任务状态 {desc}（当前任务块失败，提前结束等待）") return RouteOutcome(same_meter=True, status_failed=True) self.logs.flow(f" [任务状态] TaskID=0x{status.task_id:04X} 表={status.addr} " f"{desc}（非当前任务块，忽略）") self.logs.raw("RX-STATUS", uplink.frame, f"任务状态 {desc}（非当前任务块，忽略）") return RouteOutcome() # ---- 上报任务数据 ---- assert uplink.inner is not None inner = uplink.inner addr = norm_addr(uplink.addr) key = (addr, inner.di.upper()) tid = uplink.local.task_id if uplink.local is not None else 0 # 已随本表放弃的任务块 → 丢弃，避免「永远没人取的缓存条目」 if inner.di and key in self.abandoned: self.logs.flow(f" [已放弃] 任务块 上{_di_x(inner.di)}次(DI={inner.di}) 属于已放弃的 " f"{addr}，应答直接丢弃（不入缓存）") self.logs.raw("RX-SKIP", uplink.frame, f"已放弃任务块的应答 DI={inner.di}，丢弃") return RouteOutcome() # ---- 645 异常应答（ERR=1）与结构坏掉的应答 ---- if inner.is_conclusive_error: code, text = inner.error_code, inner.error_text desc = f"{f'0x{code:02X} ' if code is not None else ''}{text}".strip() # (a) 帧里带 DI → 和正常应答一样按 (表号, DI) 归属，与 TaskID 无关 if inner.di: block = f"上{_di_x(inner.di)}次(DI={inner.di})" if task is not None and (addr, inner.di) == self.task_key(task): self.logs.flow(f" [命中] 645 异常应答 {desc} → 任务块 {block}" f"（外层TaskID=0x{tid:04X} 仅为回显，不参与归属）") return RouteOutcome(frame=uplink.frame, same_meter=True) err_key = ("ERR", addr, inner.di) if err_key in self.cache: self.stats.dup_frames += 1 self.logs.flow(f" [重复帧] 645 异常应答 {desc} 任务块 {block} 已在缓存，忽略") self.logs.raw("RX-DUP", uplink.frame, f"重复异常应答 {block}") return RouteOutcome(same_meter=same_meter) self.cache[err_key] = uplink cur = self.task_key(task)[1] if task is not None else "-" self.logs.flow(f" [缓存延迟异常应答] 任务块 {block} {desc} → 入缓存" f"（当前任务块 DI={cur}），轮到该项时再取用") self.logs.raw("RX-CACHE", uplink.frame, f"延迟异常应答 {block} {desc} 入缓存") return RouteOutcome(same_meter=same_meter) # (b) 帧里不带 DI → 只能按上下文归属 if task is not None: self.logs.flow(f" [命中] 645 异常应答 {desc}（帧内无 DI，按上下文归属到当前任务块" f" 上{_di_x(task.item.di)}次 DI={task.item.di}）") return RouteOutcome(frame=uplink.frame, same_meter=same_meter) self.logs.flow(f" [丢弃] 645 异常应答 {desc}：帧内无 DI，且当前没有在等的任务块 ——" f"不作归属，避免被后面的任务块误吃") self.logs.raw("RX-DROP", uplink.frame, f"无DI异常应答 {desc}，无上下文可归属，丢弃") return RouteOutcome() # ---- 正常应答：按 (表号, DI) = 任务块 归属 ---- if task is not None and key == self.task_key(task): self.logs.flow(f" [命中] 任务块 上{_di_x(inner.di)}次(DI={inner.di}) 表={addr}" f"（外层TaskID=0x{tid:04X} 仅为流水号回显）") return RouteOutcome(frame=uplink.frame, same_meter=True) if key in self.got: self.stats.dup_frames += 1 self.logs.flow(f" [重复帧] 任务块 上{_di_x(inner.di)}次(DI={inner.di}) 此前已抄回，忽略") self.logs.raw("RX-DUP", uplink.frame, f"重复帧 DI={inner.di}（已抄回，忽略）") return RouteOutcome(same_meter=same_meter) if key in self.cache: self.stats.dup_frames += 1 self.logs.flow(f" [重复帧] 任务块 上{_di_x(inner.di)}次(DI={inner.di}) 已在缓存中，忽略") self.logs.raw("RX-DUP", uplink.frame, f"重复帧 DI={inner.di}（已在缓存，忽略）") return RouteOutcome(same_meter=same_meter) # 属于本表、且是此前超时没取到的项 → 立即补抄回填（不入缓存） target = self.wait_fill.get(key) if target is not None: self.fill_late(target, uplink) return RouteOutcome(same_meter=same_meter) self.cache[key] = uplink cur = self.task_key(task)[1] if task is not None else "-" self.logs.flow(f" [缓存延迟应答] DI={inner.di}(上{_di_x(inner.di)}次) 与当前任务" f"(DI={cur})不符 → 入缓存，继续等当前项") self.logs.raw("RX-CACHE", uplink.frame, f"延迟应答 DI={inner.di} 入缓存（当前任务 DI={cur}）") return RouteOutcome(same_meter=same_meter) def take_cached(self, task: SendTask) -> Optional[Uplink]: """ 取属于该「任务块(表号+DI)」的延迟应答（有则用，没有则 None）。 ★ 只用 (表号, DI) 做键。v1 早期还有一个 ("ERR", 表号, TaskID) 的键 + 一个 ("ERR", 表号, 0) 的通配兜底 —— 那正是「拿 TaskID 当任务块身份」的残留： 异常帧的 TaskID 来自别的会话/别的项，永远匹配不上，却又会被通配键弹出来 喂给**下一个**任务块，把正常数据判成错误帧并中止整表。 """ addr, di = self.task_key(task) cached = self.cache.pop((addr, di), None) if cached is not None: return cached return self.cache.pop(("ERR", addr, di), None) # ---------------- 记账 ---------------- def remember(self, result: TaskResult) -> None: """成功的登记「已抄回」（用于回放判重）；失败的登记「等待补抄」""" key = (norm_addr(result.meter_addr), result.di.upper()) if result.status is TaskStatus.OK: self.got[key] = result self.wait_fill.pop(key, None) else: self.wait_fill[key] = result def abandon(self, meter_addr: str, items: Sequence[DataItem]) -> None: """本表放弃这些任务块：后续回放应答直接丢弃，并清掉已进缓存的条目""" addr = norm_addr(meter_addr) for item in items: key = (addr, item.di.upper()) self.abandoned.add(key) self.cache.pop(key, None) self.cache.pop(("ERR",) + key, None) # ---------------- 用应答填充结果 ---------------- def apply_reply(self, r: TaskResult, uplink: Uplink) -> None: """用一条应答填充 TaskResult（正常应答路径 / 补抄回填）""" inner = uplink.inner assert inner is not None r.status = TaskStatus.OK r.response = uplink.frame r.local_frame = uplink.local r.dlt645_frame = inner r.inner_645 = inner.raw r.freeze = inner.freeze r.value = inner.value r.err_code = None r.err_text = "" r.note = "补抄：应答迟到到达后回填" def fill_late(self, r: TaskResult, uplink: Uplink) -> None: """★ 补抄：把迟到到达的应答回填到此前判失败的条目上（不用重发）""" old = r.status.value self.apply_reply(r, uplink) self.stats.late_fill += 1 key = (norm_addr(r.meter_addr), r.di.upper()) self.wait_fill.pop(key, None) self.got[key] = r self.logs.flow(f" ↩ 补抄成功：上{_di_x(r.di)}次(DI={r.di}) 的应答迟到到达，" f"已回填（原状态 {old}，无需重发）") self.logs.flow(f" 冻结时间: {r.freeze_time} 正向: {r.positive} kWh " f"反向: {r.reverse} kWh") self.logs.raw("RX-FILL", uplink.frame, f"补抄命中 DI={r.di}（原状态 {old}）") self.logs.freeze(r, tag="补抄") # ============================================================ # 类 8：收发与分帧（UplinkPump） # ============================================================ @dataclass class PumpResult: """一次收帧的结果：命中当前任务的帧 / 本表有没有帧（判断回放挤占）/ 任务状态失败""" hit: Optional[bytes] = None saw_same_meter: bool = False status_failed: bool = False class UplinkPump: """ 串口 ↔ 帧流：收字节 → 分帧 → 分类 → 回确认帧 → 交给 FrameRouter 归属。 （v1 这些逻辑散在 FreezeReadSession 的 _feed / _drain_to_cache / _exchange 里， 并且直接读写 self._rx_resid / self._acked 等会话内部状态。） """ def __init__(self, transport: Optional[Transport], classifier: UplinkClassifier, router: FrameRouter, acks: AckResponder, logs: LogSink, splitter: Optional[FrameSplitter] = None): self.transport = transport self.classifier = classifier self.router = router self.acks = acks self.logs = logs self.splitter = splitter or FrameSplitter() # ---------------- 一帧：分类 → 确认 → 归属 ---------------- def handle(self, frame: bytes, task: Optional[SendTask]) -> RouteOutcome: uplink = self.classifier.classify(frame) if uplink.should_ack: self.acks.ack(uplink.ack_seq, uplink.ack_src) return self.router.route(uplink, task) def feed(self, chunk: bytes, task: Optional[SendTask]) -> PumpResult: """处理一批刚收到的字节（可能含多帧粘包），返回本批的结论""" result = PumpResult() split = self.splitter.feed(chunk) if split.overflow is not None: self.logs.raw("RX-BAD", split.overflow, f"残留缓冲异常增长({split.overflow_len}B)，已截断") for frame in split.frames: outcome = self.handle(frame, task) if outcome.frame is not None: if result.hit is None: result.hit = outcome.frame elif outcome.same_meter: # 收到本表的帧（延迟/重复）→ 说明链路在动 result.saw_same_meter = True if outcome.status_failed: result.status_failed = True return result def pump(self, task: Optional[SendTask], label: str = "", max_wait: float = 0.5) -> PumpResult: """收一次串口（最多等 max_wait 秒）并处理""" if self.transport is None: raise RuntimeError("未注入 transport") chunk = self.transport.recv(end_flag=b"\x16", max_wait=max_wait) if not chunk: return PumpResult() self.logs.raw("RX-RAW", chunk, f"{label} 串口收到 {len(chunk)} 字节") return self.feed(chunk, task) def drain(self, max_wait: float = 0.03) -> None: """ 把串口输入缓冲区里**已经到达**的字节全部取走，按「归属未知」（task=None）处理。 用途：每次下发任务前先排空上一次交互残留的帧 —— 正常应答进缓存/补抄； 不带 DI 的异常应答直接丢弃。避免被下一个任务块误吃。 """ 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 chunk: self.feed(chunk, None) def make_freeze_items(start: int = 1, end: int = 255) -> List[DataItem]: """上1次 .. 上N次整点冻结数据（0504FF01 .. 0504FFxx，按顺序抄）""" return [DataItem(di=f"0504FF{i:02X}", description=f"上{i}次整点冻结数据块") for i in range(start, end + 1)] # ============================================================ # 类 9：抄读会话（只做编排与策略） # ============================================================ class FreezeReadSession: """ 「上x次整点冻结数据」抄读会话。 for meter in meters: # 外层：表 for item in freeze_items: # 内层：上1次 → 上N次 read(item) 若命中 AbortPolicy → 放弃该表剩余全部冻结项，break 到下一只表 职责划分（v1 是一个 900 行、什么都干的类）： 帧解析 → DLT645Codec / LocalProtocolCodec 帧分类 → UplinkClassifier 收发/分帧 → UplinkPump + FrameSplitter 归属缓存 → FrameRouter 落盘 → LogSink（四份文本日志）+ CsvWriter 本类 → 流程编排、重试与放弃策略、记账 """ def __init__(self, asrc_addr: str = "000000000000", transport: Optional[Transport] = None, logs: Optional[LogSink] = None, csv: Optional[CsvWriter] = 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, 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 self.ack_dedup = ack_dedup self.freeze_start = freeze_start self.freeze_end = freeze_end self.verbose = verbose # ---- 协作对象（谁干什么一目了然，不再是同一个类里的 20 个私有状态）---- self.logs = logs or LogSink(verbose=verbose) self.csv = csv or CsvWriter(None) self.stats = SessionStats() self.router = FrameRouter(logs=self.logs, stats=self.stats) self.acks = AckResponder(transport, stats=self.stats, logs=self.logs, enabled=ack_reports, dedup_window=ack_dedup) self.pump = UplinkPump(transport=transport, classifier=UplinkClassifier(self.logs), router=self.router, acks=self.acks, logs=self.logs) self.builder = DLT645FrameBuilder(plus33=True) self.dlt645 = DLT645Codec() self.local_codec = LocalProtocolCodec() # ---- 流水号 ---- self._seq = 0x09 # 本地规约 SEQ（帧序列域） self._task_no = 0 # 全局任务流水号（0x0001 起，写进 TaskID，模块原样回显） self._builders: Dict[str, LocalProtocolBuilder] = {} # 表号 → 组包器 # ---------------- 会话头 ---------------- def log_banner(self, meters: Sequence[MeterAddress], items: Sequence[DataItem]) -> None: port = getattr(self.transport, "port", None) p = self.policy self.logs.flow("=" * 78) self.logs.flow(f"会话开始 {fmt_ts()}") self.logs.flow(f" 通信方式 : {'串口 ' + port if port else type(self.transport).__name__}") self.logs.flow(f" CCO 地址 : {self.asrc_addr}") self.logs.flow(f" 电表数量 : {len(meters)} {[m.addr for m in meters]}") self.logs.flow(f" 冻结项 : 上{self.freeze_start}次 ~ 上{self.freeze_end}次，" f"共 {len(items)} 项") self.logs.flow(f" 超时/重试 : timeout={self.timeout}s, max_retries={self.max_retries}, " f"retry_interval={self.retry_interval}s, replay_grace={self.replay_grace}s") self.logs.flow(f" 错误帧重试 : {self.retry_on_error_frame}") self.logs.flow(f" 放弃策略 : 错误帧={p.on_error_frame}(立即) " f"无应答={p.on_no_response}(连续{p.streak_limit}次) " f"坏帧={p.on_bad_frame}(连续{p.streak_limit}次)") self.logs.flow(" 应答归属 : 只按「表号 + DI(上x次任务块)」判定；外层 TaskID 是模块" "原样回显的流水号，不参与归属") self.logs.flow(" （迟到应答入缓存→补抄；重复回放帧忽略；不带 DI 的异常帧" "只认当前等待窗口，窗口外丢弃）") self.logs.flow(f" 上报确认 : {'启用' if self.ack_reports else '关闭'}" f"（模块每条上报都回确认帧，SEQ 与上报帧一致" f"{'；同 SEQ %gs 内去重' % self.ack_dedup if self.ack_dedup > 0 else '，不去重'}）") self.logs.flow(f" 任务状态帧 : AFN=05/DUID=E805050 上报任务状态 → 非 0 状态立即判本项失败" f"（不白等超时），连续 {p.streak_limit} 次按策略放弃本表") self.logs.flow(f" 日志文件 : 主日志={self.logs.log_path} 原始报文={self.logs.raw_log_path} " f"冻结块={self.logs.freeze_log_path} CSV={self.csv.path}") self.logs.flow("=" * 78) # ---------------- 组包 ---------------- 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.builder.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 _exchange(self, task: SendTask, outer: bytes, timeout: float, label: str = "", attempt: int = 0) -> Optional[bytes]: """ 返回：属于「本表 + 本 DI」的上报帧；超时返回 None。 · 下发前先查缓存 → 命中直接用（不重发） · 等待中收到的其它 DI 帧 → 入缓存/补抄，不丢弃 · 若本表回放帧挤占窗口且静默超过 replay_grace → 提前返回，交给外层重发 """ if self.transport is None: raise RuntimeError("未注入 transport") # ---- 0) 缓存里已有本任务的应答 → 直接取用，省一次往返 ---- cached = self.router.take_cached(task) if cached is not None and cached.is_conclusive_error: # 缓存里的异常应答可能是上一次会话/回放的陈旧结论 —— 中止整表代价太大， # 所以不直接采信，重新下发一次让它再确认。 self.logs.flow(" · 缓存里有一条该任务块的 645 异常应答，但仍重新下发一次确认" "（避免用陈旧的异常结论中止整表）") self.logs.raw("RX-CACHE-ERR", cached.frame, "缓存异常应答 → 重新下发确认") cached = None if cached is not None: self.stats.cache_hits += 1 is_err = cached.inner is not None and cached.inner.is_error_response kind = "645 异常应答（ERR=1）" if is_err else "应答" self.logs.flow(f" ✔ 命中延迟应答缓存：任务块 上{_di_x(task.item.di)}次" f"(DI={task.item.di}) 的{kind}此前已到达，本次不重发") self.logs.raw("RX-CACHE-HIT", cached.frame, f"{label} 命中缓存，未重发") return cached.frame # ---- 下行：本地规约帧（内含 645 请求）---- self.logs.raw("TX-LOCAL", outer, f"{label} 第{attempt + 1}次发送（TaskID 流水号=0x{task.seq & 0xFFFF:04X}）") self.logs.block(f">>> 下行报文（{label}）", annotate_local_frame(outer, with_inner=True)) self.logs.raw("TX-645", outer[28:28 + outer[27]], f"{label} 内层645请求 DI={_peek_di(outer)}") self.transport.send(outer) self.router.status_fail = None # 本次等待窗口内收到的「任务状态=失败」帧 deadline = time.time() + timeout last_frame_at: Optional[float] = None # 最近一次收到本表数据帧的时间 saw_replay = False # 本窗口内是否被本表回放帧挤占过 while True: now = time.time() if now >= deadline: self.logs.flow(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.logs.flow(f" · 本表回放帧挤占窗口且已静默 {now - last_frame_at:.1f}s，" f"提前结束本次等待（进入重发）") return None outcome = self.pump.pump(task, label=label, max_wait=min(0.5, deadline - now)) if outcome.hit is not None: hit = outcome.hit self.logs.raw("RX-LOCAL", hit, f"{label} 命中上报帧（本表+DI={task.item.di}）") self.logs.raw("RX-645", hit[25:25 + hit[24]], f"{label} 内层645应答 DI={_peek_di_up(hit[25:25 + hit[24]])}") return hit if outcome.saw_same_meter: saw_replay = True last_frame_at = time.time() # 模块上报「任务状态=失败」→ 本项不会有数据帧，立即结束等待（进入重试/记账） if outcome.status_failed: 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: local = self.local_codec.parse(raw) except Exception as e: r.status = TaskStatus.BAD_FRAME r.err_text = f"本地规约解析失败: {e}" return r r.local_frame = local if local.length_ok is False: r.status = TaskStatus.BAD_FRAME r.err_text = f"本地规约 L 校验失败(L={local.length}, 应为 {local.length_calc})" return r if local.checksum_ok is False: r.status = TaskStatus.BAD_FRAME r.err_text = (f"本地规约 CS 校验失败(CS={local.checksum:02X}, " f"应为 {local.checksum_calc:02X})") return r if not local.inner: r.status = TaskStatus.BAD_FRAME r.err_text = "上报帧不含 645 内层数据" return r r.inner_645 = local.inner self.logs.block(f"<<< 上行报文逐字段解析（{task.meter.addr} {task.item.description}）", annotate_local_frame(raw, with_inner=True)) try: inner = self.dlt645.parse(local.inner) except Exception as e: r.status = TaskStatus.BAD_FRAME r.err_text = f"645 解析失败: {e}" return r r.dlt645_frame = inner r.freeze = inner.freeze r.payload_raw = inner.payload if not inner.checksum_ok: r.status = TaskStatus.BAD_FRAME r.err_text = (f"645 CS 校验失败(CS={inner.checksum:02X}, " f"应为 {inner.checksum_calc:02X})") return r # ★ 错误帧：ERR=1 → 由调用方决定「放弃本表剩余冻结项」 if inner.is_error_response: r.status = TaskStatus.METER_ERROR r.err_code = inner.error_code r.err_text = inner.error_text or "645 异常应答" return r if inner.kind is FrameKind.MALFORMED: r.status = TaskStatus.BAD_FRAME r.err_text = inner.error_text or "645 应答数据异常" return r r.status = TaskStatus.OK r.value = inner.value if not r.di_name: r.di_name = inner.di_name return r def _no_response(self, task: SendTask, outer: bytes, attempt: int) -> TaskResult: """两种「没有数据」的记账：模块明确报了任务状态失败 / 纯超时无应答""" failed = self.router.status_fail if failed is not None: err_text = (f"模块上报任务状态 {failed.status_text}" f"(0x{failed.status_code:02X})，无 645 数据应答") else: err_text = "超时/未收到本项上报帧" return 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=err_text) # ---------------- 抄读单条任务 ---------------- def read_one(self, task: SendTask) -> TaskResult: outer = self.build_frames(task)["local"] label = f"表={task.meter.addr} {task.item.description}({task.item.di})" # ★ 下发前先排空串口残留帧：按「归属未知」处理（正常应答→缓存/补抄； # 不带 DI 的异常应答→丢弃），避免上一轮的回放/延迟帧被本任务块误吃。 self.pump.drain() last: Optional[TaskResult] = None for attempt in range(self.max_retries + 1): if attempt > 0: time.sleep(self.retry_interval) self.logs.flow(f" [重试 {attempt}/{self.max_retries}]") self.pump.drain() # 先收下这段时间到达的帧（可能已含本项） raw = self._exchange(task, outer, timeout=self.timeout, label=label, attempt=attempt) if raw is None: last = self._no_response(task, outer, attempt) continue last = self.parse_reply(task, raw, attempt) last.request = outer if last.status is TaskStatus.OK: break # ★ 错误帧：默认不重试，立即退出，交由外层终止本表 if last.status is TaskStatus.METER_ERROR and not self.retry_on_error_frame: break assert last is not None self.router.remember(last) self.logs.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 make_freeze_items( self.freeze_start, self.freeze_end) report = MeterReport(meter=meter) self.logs.flow(f"\n===== 开始抄读 [{meter.description or ''} {meter.addr}] " f"共 {len(items)} 项冻结数据 =====") self.logs.flow(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.logs.flow(f"--- {task.label} ---") r = self.read_one(task) report.results.append(r) self.log_result(r) # 连续失败计数：任何一次成功即清零（单次偶发超时不再误杀整表） streak = 0 if r.status is TaskStatus.OK else streak + 1 # ★★ 核心策略：命中放弃条件 → 该表剩余「上x次冻结数据」全部不抄，切下一只表 if not self.policy.should_abort(r.status, streak): continue rest = items[idx + 1:] reason = self.policy.reason(r.status, streak) detail = f"({r.err_text})" if r.err_text else "" self.logs.flow(f" ✘ 放弃本表剩余 {len(rest)} 项「上x次冻结数据」" f" —— 原因：{reason}{detail}，切换到下一只表") if rest: self.logs.flow(f" 放弃区间：DI {rest[0].di}(上{_di_x(rest[0].di)}) " f"~ DI {rest[-1].di}(上{_di_x(rest[-1].di)})，共 {len(rest)} 项") self.logs.raw("GIVE-UP", b"", f"表={meter.addr} 放弃 DI {rest[0].di}~{rest[-1].di} " f"共 {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}，放弃抄读")) # 这些任务块的应答以后直接丢弃（同时清掉可能已进缓存的条目） self.router.abandon(meter.addr, rest) report.aborted = True report.abort_reason = f"{reason}{detail} @ 上{idx + 1}次(DI={item.di})" break except KeyboardInterrupt: # ★ 原来 Ctrl-C 直接穿出去：当前表的 CSV 不写、未抄的项没有任何记录。 # 现在把没来得及抄的项按 SKIPPED 记账，收尾照常落盘、照常打汇总。 self.stats.interrupted = True done = len(report.results) rest = items[done:] for it in rest: 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）")) self.router.abandon(meter.addr, rest) report.aborted = True report.abort_reason = f"手动中断（Ctrl-C）@ 已处理 {done}/{len(items)} 项" self.logs.flow(f" ⏹ 收到 Ctrl-C：本表剩余 {len(items) - done} 项以 SKIPPED 记账，" f"停止后续电表并收尾落盘") # ---------------- 收尾：捞最后一批迟到应答（自动补抄） ---------------- self.pump.drain(max_wait=0.3) self.logs.flow(f"===== 抄读结束 [{meter.addr}]：成功 {report.ok_count}" f"{'，已放弃剩余 ' + str(report.skipped_count) + ' 项' if report.aborted else ''}" f" | 补抄回填 {self.stats.late_fill} 项" f" | 命中缓存 {self.stats.cache_hits} 次" f" | 忽略回放重复帧 {self.stats.dup_frames} 帧 =====") return report def log_result(self, r: TaskResult) -> None: blk = r.freeze_block if r.status is TaskStatus.OK: self.logs.flow(f" ✔ 成功 (重试 {r.retries} 次)") self.logs.flow(f" 表号: {r.meter_addr}") self.logs.flow(f" DI: {r.di} ({r.di_name})") if blk and len(blk) >= 16: self.logs.flow(f" 冻结时间: {r.freeze_time}") self.logs.flow(f" 正向总电能: {r.positive} kWh") self.logs.flow(f" 反向总电能: {r.reverse} kWh") self.logs.flow(f" 冻结块原始 16B : {fmt_hex(blk)}") self.logs.flow(f" 时间[0:6]={fmt_hex(blk[0:6])} " f"正向[6:11]={fmt_hex(blk[6:11])} " f"反向[11:16]={fmt_hex(blk[11:16])}") else: self.logs.flow(f" 数据: {r.value}") if r.inner_645: self.logs.flow(f" 645 应答报文: {fmt_hex(r.inner_645)}") else: self.logs.flow(f" ✘ {r.status.value} (重试 {r.retries} 次)") if r.err_code is not None: self.logs.flow(f" 错误码: 0x{r.err_code:02X} {r.err_text}") elif r.err_text: self.logs.flow(f" 错误: {r.err_text}") if blk: self.logs.flow(f" 冻结块原始 16B : {fmt_hex(blk)}") if r.inner_645: self.logs.flow(f" 645 应答报文: {fmt_hex(r.inner_645)}") self.logs.flow("") # ---------------- 抄读多只表 ---------------- def run(self, meters: Sequence[MeterAddress], items: Optional[Sequence[DataItem]] = None) -> List[MeterReport]: """ 按顺序抄多只表；单表异常/放弃都不影响后续电表。 四份日志在抄读过程中逐行 flush；CSV 在本方法末尾统一落盘（含补抄回填后的最终结果）。 """ items = list(items) if items is not None else make_freeze_items( self.freeze_start, self.freeze_end) self.log_banner(meters, items) reports: List[MeterReport] = [] try: for i, meter in enumerate(meters, 1): self.logs.flow(f"\n>>>>>>>>>> [{i}/{len(meters)}] " f"{meter.description or ''} {meter.addr} <<<<<<<<<<") try: reports.append(self.read_meter(meter, items)) except Exception as e: # 单表异常不影响其它表 self.logs.flow(f" !! 该表抄读过程中出现异常: {e}，跳过并继续下一只表") reports.append(MeterReport(meter=meter, aborted=True, abort_reason=f"异常: {e}")) if self.stats.interrupted: self.logs.flow(" ⏹ 已手动中断：不再抄读后续电表") break finally: # ★ 落盘推迟到这里：上面的收尾 drain 会把迟到应答回填进各表结果对象， # 先回填完再写 CSV，日志与 CSV 才不会互相矛盾（原来在 read_meter 里就写了， # 补抄回来的项在 CSV 里仍是 NO_RESP）。 self.pump.drain(max_wait=0.3) for report in reports: self.csv.write(report) self.print_summary(reports) return reports def print_summary(self, reports: Sequence[MeterReport]) -> None: self.logs.flow("\n" + "=" * 70) self.logs.flow("================ 全部抄读结果汇总 ================") total_ok = total_skip = 0 for i, rep in enumerate(reports, 1): 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 += rep.ok_count total_skip += skip line = f" {i}. {rep.meter.addr} {rep.meter.description}：成功 {rep.ok_count}" if err: line += f"，错误帧 {err}" if bad: line += f"，坏帧 {bad}" if nor: line += f"，无应答 {nor}" if skip: line += f"，放弃 {skip}" self.logs.flow(line) if rep.aborted: self.logs.flow(f" └─ 已放弃本表剩余冻结项：{rep.abort_reason}") self.logs.flow(" --------------------------------------------------") self.logs.flow(f" 合计：成功 {total_ok} 项，因放弃未抄读 {total_skip} 项") if self.stats.interrupted: self.logs.flow(" ⚠ 本次运行被手动中断（Ctrl-C）：已抄回的数据与未抄项均已落盘") self.logs.flow(f" 应答归属：补抄回填 {self.stats.late_fill} 项，" f"命中延迟应答缓存 {self.stats.cache_hits} 次，" f"忽略回放重复帧 {self.stats.dup_frames} 帧") self.logs.flow(f" 确认帧 ：已对模块上报回确认 {self.stats.ack_frames} 条" f"（SEQ 与上报帧一致" f"{'，同 SEQ 去重 %gs' % self.ack_dedup if self.ack_dedup > 0 else '，每次都回'}）") self.logs.flow(" 日志落盘：") self.logs.flow(f" 主日志(含逐字段报文) : {self.logs.log_path}") self.logs.flow(f" 原始报文流水(TX/RX) : {self.logs.raw_log_path}") self.logs.flow(f" 冻结数据块明细 : {self.logs.freeze_log_path}") self.logs.flow(f" 业务数据表(逗号分隔) : {self.csv.path}") self.logs.flow("=" * 70 + "\n") def close(self) -> None: """关闭日志句柄（内容已逐行 flush）""" self.logs.close() # ============================================================ # 默认配置（全部可用命令行参数覆盖） # ============================================================ 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 # 落盘对象与「会话」分开构造：日志走 LogSink，业务表走 CsvWriter logs = LogSink(log_path=LOG_FILES["log_path"], raw_log_path=LOG_FILES["raw_log_path"], freeze_log_path=LOG_FILES["freeze_log_path"], dump_frames=True, dump_console=args.dump_console, raw_console=False, verbose=True) csv_writer = CsvWriter(LOG_FILES["csv_path"]) session = FreezeReadSession( asrc_addr=CCO_ADDR, transport=transport, logs=logs, csv=csv_writer, 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, verbose=True, ) try: if args.probe: transport.send(bytes.fromhex(PROBE_FRAME)) session.run(meters, make_freeze_items(args.x_from, args.x_to)) except KeyboardInterrupt: print("\n[中断] 已手动终止：未抄项按 SKIPPED 记账，结果已写入日志与数据表。") finally: session.close() # 关闭四份日志的句柄（内容已逐行 flush） transport.close() return 0 if __name__ == "__main__": sys.exit(main())