Начало Каталог Помощ
Регистрация Вход
FWH
/
FranklinWH-Toolbox
2
0
Разклонения 0
Файлове Задачи 0 Заявки за сливане 0 Уики
ИН на ревизия: 0ac8d537b7
Клонове Маркери
FranklinWH-Tool...
/
docs
/
fault-record
/
technical-design.md

technical-design.md 18 KB
История Директен файл

故障录波 MQTT 整体设计

状态：设计草案，2026-09-30；2026-10-16 首版已按本文实现，实现与代码位置见 implementation-status.md；2026-10-14 按 FaultFilter_MQTT协议文档 更新业务协议；2026-10-16 补充文件上传链路、双环境、分层架构、异步任务、存储和取消等用户已确认的决策。功能目标见 requirements.md，页面设计见 ui-design.md，报文依据见 protocol-observations.md 和 protocol-specification.md。

1. 设计目标与原则

建立直接经 MQTT 与设备交互的故障录波能力，覆盖设备信息（实时数据）、录波记录获取、参数读取和参数设置，同时支持生产服和测试服。

  • 只发起、不应答：本工具只发送 377（以及取消用的 cmd=4），其余消息全部监听。380、607、610 由 FWH 平台发出，平台在收到 379(cmd=2) 后自动发起文件上传。
  • 分层清晰：每层只负责自己的事，只依赖下层。MQTT 客户端只提供底层收发，业务判断全部在上层。
  • 复用现有模块：asynq 任务系统、MongoDB、MinIO、Redis、fwh_pro／fwh_test 站点信息接口、uPlot。
  • 只显示原始值：录波数据不做物理量换算（B15）。
  • DataDownload 暂不接入：本功能完整实现并测试通过后，再考虑开放给 DataDownload；DataDownload 继续使用原 FWH API 方案。

2. 分层架构

L5 Route / Controller        HTTP 入参校验、出参转换
L4 Service                   对外用例：创建操作、查询状态与结果、取消、导出、查询设备列表
L3 Task Executor (asynq)     操作执行器：按状态机推进、持久化每一步、重启恢复
L2 Protocol / Parser         纯函数，不访问网络、数据库和文件系统
     ├─ protocol：构造 377 dataArea，解析 378／379／607～610（兼容旧版本 608 把 `id` 误写为第二个 `result`）
     └─ parser：解压、hexData(code=1/2/3)、failuredata、waveformdata、CSV 生成
L1 FWH MQTT Client           按环境实例化；FWH 报文外层（len／crc／信封）+ Request／Publish／Subscribe
L0 utils/mqttClient          paho 封装

存储：MongoDB（操作记录） · MinIO（原始文件） · Redis（网关锁、asynq）
外部：FWH 平台 HTTP（站点信息、hes-upload 下载）
层 建议位置 只负责 不负责
L0 utils/mqttClient 连接、订阅、发布、重连 FWH 报文格式
L1 repositories/mqtt/fwh_mqtt 按环境的连接实例；FWH 信封编解码；三种收发方法；订阅期间保持连接 任何 cmdType／cmd 的业务含义
L2 function/FaultRecordFile（解析器）、function/FaultRecordProtocol（协议编解码） 字节和 JSON 的结构化转换 IO
L3 task/queue/faultRecordQueue 一次操作的完整执行和恢复 HTTP 入参、页面格式
L4 services/faultRecordServices.go 用例编排，调用 Repository 和入队 协议细节
L5 controllers/faultRecordControllers.go、routes/faultRecordRoutes.go HTTP 业务
存储 repositories/mongo/faultRecordRepo.go、repositories/minio/faultRecordRepo.go 数据读写 业务
下载 fwh_pro／fwh_test 中各一个无 token 下载函数 按环境域名下载 解压、解析

L2 是纯函数，用仓库中的样本文件（CloudLog_2026-10-14_104644.csv、2026-10-14_05_46_20.tar.bz2）做单元测试。以后 DataDownload 接入时，复用 L1 和 L2。

3. L1 FWH MQTT 客户端

3.1 对外方法

方法 用途 行为
Request(ctx, req) (*Msg, error) 一问一答（保留原方式） 订阅应答 topic → 发送 → 等到第一条满足 Match 的消息或超时 → 取消订阅
Publish(ctx, topic, msg) error 只请求 发送后立即返回
Subscribe(topic, handler) (cancel func(), error) 持续监听 同一 topic 可有多个订阅者，各自收到每条消息；cancel 只移除自己
  • handler 收到 {Topic, Envelope, Raw []byte, ReceivedAt}。Envelope 为解析后的外层，Raw 为原始字节，供上层按需重新解析（如旧版本 608 的重复 result 键）。
  • Request 的匹配条件由调用方传入（如 cmdType=378 && dataArea.cmd=1），客户端不内置任何业务过滤。
  • 信封构造（len、crc、timeStamp、snno）属于 L1 的通用能力，以 NewEnvelope(cmdType, equipNo, dataArea) 提供。
  • 现有 EnableXlog、FaultRecordParamSetting、ChargeChannel 中的业务内容移到上层，L1 只保留通用部分。

3.2 连接与订阅管理

  • 内部维护 topic → 订阅者集合。第一个订阅者加入时向 broker 订阅；最后一个退出时向 broker 取消订阅。
  • 有活动订阅时不做空闲断开；没有订阅且空闲 5 分钟后才断开（保持现有省连接的行为）。
  • 断线自动重连后恢复全部 broker 订阅（L0 已有 ResumeSubscribe）。L1 需要记录断线区间，供上层判断“这段时间可能漏了消息”。
  • 发布仍经过一个 worker 串行发送；订阅回调不阻塞 paho 线程，按订阅者各自的缓冲 channel 分发，满了就丢弃并记日志。

4. 操作模型

4.1 操作类型

操作 cmd 输入 结果
device_info 0 codeList ⊆ {1,2,3} hexList 原文 + 解析结果
record_fetch 2 一个设备、一条或多条记录序号（最多 32） 原始文件（MinIO）+ 解析结果
param_read 3 idList（功能操作页只用于 4、7） idInfo 原文 + 解析结果
param_write 1 paraList setResultList

每次操作只针对一个网关下的一个设备（aGate 或某个 pe）。

cmd=3（查询参数）和 cmd=4（退出）只在 FaultFilter 文档中定义，旧描述、现有代码和实报只有 0/1/2。用户判断外层可能支持，先按支持实现，实测后确认（B01）。如不支持：功能操作页的 4、7 只能设置不能读取，取消只停止本工具的等待。参数设置页（5、6）读取和复读走 cmd=0，不受影响。record_fetch 可一次获取多条记录（2026-10-19 用户确认设备支持）：文件按压缩包中的 index_<n> 目录区分，不会混淆；379 中的多组 4/5 结果按“每条记录一组、顺序与请求一致”对应（待实测）。解析后核对压缩包是否包含全部请求的序号，缺少时在结果中提示。等待时间按记录数延长：379 每多一条加 1 分钟，609 每多一条加 30 秒。

下载时机：收到 609 成功后推进到 uploaded，收到 610 后 15 秒开始下载，失败后每 15 秒重试，一直到总超时（从 379 起算 5 分钟，每多一条记录加 30 秒）；仍失败记为“下载失败”并注明尝试次数（用户要求，2026-10-19）。连续相同的失败只保留最后一条下载记录，并记录累计失败次数和首次失败时间。610 没有文件名，按“收到 609 之后的下一条 610”对应，其 result 只记录不判断。未收到 609 或 610 时只在超时时按规则直接下载一次；服务重启恢复时先尝试下载一次。不做收到 379 后定期试探下载的兜底（用户决定，2026-10-19）。

手动重试下载：上传已完成（收到 609）但下载失败或超时的操作，页面提供“重试下载”按钮：操作重新置为“上传完成”，立即下载，失败后每 15 秒重试，最多尝试 3 次（用户要求，2026-10-19）；每次触发都在原始报文中追加一条 action 记录（操作人、第几次重试、重试前状态与错误）；每次重试使用新的任务 ID，服务重启后按当前重试序号恢复。

文件名：不同固件的 609.fileName 有的带 .tar.bz2、有的不带。每次下载（含自动和手动重试）都由批次名重新拼接 <批次>.tar.bz2 生成下载链接，不依赖已保存的文件名（10-19 测试服实测缺后缀导致下载为空）；MinIO 对象键和原始包导出文件名同样使用补齐后缀的名称。

前端轮询：开始后 10 秒内每 2 秒查询一次，之后每 5 秒一次；同一操作只保留一条轮询。

文件下载是 HTTP 请求，不是 MQTT 报文；每次下载作为一条 http 记录写入操作的原始报文（路径、结果、大小或错误、耗时）。操作结束后再记录约 2 秒内到达的报文（如 609 之后的 610）。

4.2 异步执行

前端 ──POST 创建操作──▶ Service ──写 Mongo(pending) + 入队 asynq──▶ 返回 operationId
前端 ──GET 轮询 operationId──▶ Service ──读 Mongo──▶ 状态、阶段时间、结果
asynq worker ──▶ Task Executor ──▶ 执行并逐步写回 Mongo
  • 队列：新建 faultRecordQueue，在 task/queue/taskQueue.go 注册，沿用 dataDownloadQueue 的写法（asynq.TaskID 保证幂等，Retention 保留记录）。
  • 不依赖 asynq 重试做业务重试：重复执行由执行器根据 Mongo 状态自行判断，避免重复发送 377(cmd=2) 清空设备文件。
  • 任务超时 15 分钟；各阶段等待：378 为 30 秒，379 为 2 分钟（record_fetch 4 分钟），379 到 609 为 5 分钟（暂定，实测后调整，见 faultRecordQueue/deps.go）。
  • MaxRetry 设为 3 只用于进程崩溃后重新投递；业务失败由执行器写入状态并正常返回，不触发 asynq 重试。（asynq 对崩溃任务按重试计数处理，MaxRetry(0) 会直接归档、无法恢复。）
  • 收到 609 并下载完成后操作即结束，之后到达的 610 不再记录。

4.3 执行器流程（以 record_fetch 为例）

  1. 取 Redis 网关锁 fault-record:lock:<env>:<gatewaySN>（带 TTL 并续租）。取不到就结束操作，状态为 gateway_locked（本工具内已有操作在进行）。
  2. 订阅 fwh/c2s/mng/<sn> 和 fwh/s2c/mng/<sn>（378/379/380）；record_fetch 另订阅 fwh/s2c/debug/<sn> 和 fwh/c2s/debug/<sn>（607/610 与 608/609，用户确认，2026-10-19）。debug topic 与 Xlog 共用，客户端支持同一 topic 多个订阅者，互不影响。
  3. 用 Publish 发 377，写 sent。

  4. 收到报文就交给 L2 解析，推进状态，并把原始报文追加到记录：

    pending → sent → accepted(378) → ready(379) → uploading(607) → uploaded(609) → downloaded → done

  5. 收到 609.result=0 后推进到 uploaded；收到 610 后 15 秒开始下载（失败每 15 秒重试至总超时），存入 MinIO，解析，写 done。

  6. 释放锁，取消订阅。

终止状态分开记录：busy（设备回复 busy，明确提示，不重试）、rejected（378 失败）、item_failed（379 中 resultOne 不为 ok）、timeout_<阶段>、upload_failed（609 失败）、download_failed、parse_failed、cancelled、interrupted。

其他操作按同一框架执行，只到 ready(379) 为止，再解析结果写 done。

4.4 关联规则

  • 377 → 378／379：环境 + equipNo + cmdType + dataArea.cmd。网关锁保证本工具内同一网关同一时间只有一个操作。
  • 379 → 607／608／609：equipNo + 文件名（609.fileName = 379.filename + ".tar.bz2"）。
  • 608 在设备上行方向：订阅不到 607 时，收到文件名匹配的 608 也推进到“平台上传中”，并从 608.id 取得平台记录 ID。
  • 发送的 377 也会出现在 s2c topic 上，按 cmdType 过滤掉。

4.5 取消（cmd=4）

cmd=4 是 FaultFilter 协议的“退出”命令：通知设备结束当前处理，data 为 {}，没有业务回复。它是设备忙时唯一会被接受的命令。

取消时所处阶段 处理
pending（还没发 377） 直接取消，不发任何消息
sent／accepted（设备正在处理我们的命令） 发 377(cmd=4)，标记 cancelled，释放锁
ready 及之后（设备已处理完，平台在上传） 不发 cmd=4（设备已无处理中的命令，且无法终止平台上传），只停止等待，标记 cancelled
状态为 busy 等终止态 不可取消

限制：

  • cmd=4 作用于设备，不区分发起方。只有在本操作的命令确实被设备接受（没有收到 busy）时才发送，避免中断别人发起的处理。
  • 设备是否真的停止、外层是否有回复，待实测；页面显示“已发送取消”，不承诺设备一定停止。

4.6 重启恢复

asynq 会把服务重启时未完成的任务重新投递。执行器先读 Mongo 状态，从断点继续：

重启时状态 恢复方式
pending 正常执行
sent／accepted 不能确定设备是否已回复，且重发会清空设备文件，标记 interrupted，由用户手动重试
ready／uploading（已知文件名） 直接按下载地址尝试下载；成功则继续解析，失败则重新订阅等待 609，直到超时
uploaded／downloaded 继续下载／解析

服务启动时扫描 Mongo 中未终止、但 asynq 中已不存在的操作（例如 Redis 数据丢失），按上表处理。

5. 设备与站点信息

  • 用户输入网关 SN，Service 按环境调用 fwh_pro.QuerySiteInfoList 或 fwh_test.QuerySiteInfoList 获取站点信息。
  • 可选设备为：aGate（网关 SN）+ fhpSn 中逗号分隔的每个 aPower SN。aPower SN 就是 pe SN（用户确认，2026-10-16）。
  • 生产服和测试服是两个不同的接口（fwh_pro 与 fwh_test），必须按操作所属环境调用，不能用生产服站点信息去操作测试服设备，反之亦然。
  • 只调用 Repository 层接口，不复用 services.GetSiteInfoData：后者会把结果写入站点信息表，测试服数据不能写入。

6. 测试服与生产服

每次操作绑定一个环境，MQTT、站点信息和文件下载都取自同一环境，不能混用。

项目 生产服 测试服
MQTT broker／证书 FWH_MQTT_BROKER、FWH_MQTT_CA/CERT/KEY FWH_MQTT_TEST_BROKER、FWH_MQTT_TEST_CA/CERT/KEY
FWH 平台域名 FWH_ENERGY（默认 energy.franklinwh.com） FWH_TEST_ENERGY（如 test.franklinwh.com）
站点信息 fwh_pro fwh_test
下载凭据 不需要 不需要

文件下载地址（用户确认，2026-10-16）：

https://<环境域名>/hes-upload/v1/file/download/logger/<equipNo>/<609.fileName>
  • <equipNo> 取 609 外层的 equipNo；pe 场景下是否仍为网关 SN 待实测。
  • 域名取自环境配置，不从 607.url 推导（607.url 是上传地址，样本为 http://）。
  • 文件扩展名是 .tar.bz2，实际是 gzip tar；解压按文件头判断，同时兼容 gzip 和 bzip2。

7. 存储

7.1 MongoDB：操作记录

选择 MongoDB：每次操作的报文条数和结果结构不固定，适合文档存储；与 DataDownload 的任务记录方式一致，已有连接与 Repository 写法。不使用 PostgreSQL：波形约 49KB，查看和导出时现场解析即可，无需结构化建表。

集合 fault_record_operations（字段草案）：

字段 说明
_id operationId
env prod／test
gatewaySn、deviceType、deviceSn 目标
kind、cmd、request 操作类型与请求参数
status、stage、stageTimes 当前状态和各阶段时间
operator 操作人（取自登录用户），历史共享可见
messages[] 原始报文：方向、topic、cmdType、接收时间、原文
result 解析后的结果（目录、设备信息、参数等）
fileName、faultRecordId、minioObject record_fetch 专用
error 失败原因
createdAt、updatedAt

索引：env + gatewaySn + createdAt、status、createdAt。暂不实现数据清理。

7.2 MinIO：原始文件

  • 对象键：fault-record/<env>/<gatewaySn>/<fileName>，保留下载到的原始字节和原始文件名。
  • 对象键不经 HTTP 暴露，由 Service 按 operationId 读取。

7.3 导出

格式 内容
原始压缩包 直接从 MinIO 读出返回
CSV（zip） waveform.csv：每行一个采样点，列为点序号、相对故障时刻毫秒、各通道原始值，表头为通道序号和含义代码；summary.csv：基本信息、版本、触发源、快照通道、采样通道、可配通道

CSV 现场生成，不缓存。

8. 时间口径

目录、failuredata、对时功能中的时间均按设备时间原样显示，不做时区换算（暂定，实测有问题再改）。操作记录中的时间（创建、阶段时间）用服务器时间存储，前端按本地时间显示。

9. 待实测前提

  1. 本工具直接发出 377(cmd=2) 后，平台是否同样自动发 607 并上传。这是“只监听”方案成立的前提。
  2. 订阅 s2c 方向：测试服已实测可订阅（收到 380）；607～610 的 topic 已确认（debug），待实测确认能收到。
  3. cmd=0/1/3 的 380 是否也由平台回复。
  4. pe 场景：外层 equipNo 与 info[].sn 的关系、下载路径中的 <equipNo>。
  5. cmd=1/3 经外层的实际报文和失败样本；busy 经外层如何体现。
  6. cmd=4 的外层表现和实际效果。

协议遗留问题见 protocol-specification.md B12、B14。