// Package faultRecordProtocol 故障录波 MQTT 业务协议编解码(纯函数,不涉及网络、数据库和文件)。 // // 协议依据:docs/fault-record/FaultFilter_MQTT协议文档.md、docs/fault-record/protocol-specification.md package faultRecordProtocol import ( "bytes" "encoding/json" "fmt" "strings" ) // 外层 cmdType const ( CmdTypeRequest uint = 377 // 本工具 → 设备:故障录波操作指令 CmdTypeAck uint = 378 // 设备 → 本工具:指令应答 CmdTypeReply uint = 379 // 设备 → 平台:信息上传(业务回复) CmdTypeReplyAck uint = 380 // 平台 → 设备:信息上传应答 CmdTypeUploadCall uint = 607 // 平台 → 设备:录波文件上传指令 CmdTypeUploadAck uint = 608 // 设备 → 平台:上传指令应答 CmdTypeUploadDone uint = 609 // 设备 → 平台:文件上传结果 CmdTypeUploadDoneA uint = 610 // 平台 → 设备:上传结果应答 ) // 业务层 cmd const ( CmdRealtime = 0 // 获取实时数据 CmdSetParam = 1 // 设置参数 CmdGetRecord = 2 // 获取记录数据 CmdQueryParam = 3 // 查询参数 CmdExit = 4 // 退出 CmdTestSuccess = 90 CmdTestFail = 99 ) // 业务层设备类型 const ( DeviceTypePE = "pe" DeviceTypeAGate = "aGate" ) // 整体结果 const ( ResultOk = "ok" ResultError = "error" ResultBusy = "busy" ) // NormalizeDeviceType 将页面/HTTP 中的设备类型转换为协议值,不区分大小写 func NormalizeDeviceType(t string) (string, error) { switch strings.ToLower(t) { case "pe": return DeviceTypePE, nil case "agate": return DeviceTypeAGate, nil default: return "", fmt.Errorf("unknown device type: %s", t) } } /************** 377 请求 **************/ type ParaItem struct { Id int `json:"id"` SetVal string `json:"setVal"` } type RequestInfo struct { Sn string `json:"sn"` CodeList []int `json:"codeList,omitempty"` // cmd=0 ParaList []ParaItem `json:"paraList,omitempty"` // cmd=1 Index []int `json:"index,omitempty"` // cmd=2 IdList []int `json:"idList,omitempty"` // cmd=3 } type RequestGroup struct { Type string `json:"type"` Info []RequestInfo `json:"info"` } type RequestData struct { List []RequestGroup `json:"list"` } // Request 377 的 dataArea type Request struct { Cmd int `json:"cmd"` Data any `json:"data"` } func newDeviceRequest(cmd int, deviceType, sn string, info RequestInfo) (Request, error) { t, err := NormalizeDeviceType(deviceType) if err != nil { return Request{}, err } if sn == "" { return Request{}, fmt.Errorf("device sn is empty") } info.Sn = sn return Request{ Cmd: cmd, Data: RequestData{List: []RequestGroup{{Type: t, Info: []RequestInfo{info}}}}, }, nil } // NewRealtimeRequest cmd=0,codeList 取值 1~3 func NewRealtimeRequest(deviceType, sn string, codeList []int) (Request, error) { if len(codeList) == 0 { return Request{}, fmt.Errorf("codeList is empty") } for _, c := range codeList { if c < 1 || c > 3 { return Request{}, fmt.Errorf("invalid code %d, cmd=0 only supports 1~3", c) } } return newDeviceRequest(CmdRealtime, deviceType, sn, RequestInfo{CodeList: codeList}) } // NewSetParamRequest cmd=1,setVal 按功能码校验 func NewSetParamRequest(deviceType, sn string, paraList []ParaItem) (Request, error) { if len(paraList) == 0 { return Request{}, fmt.Errorf("paraList is empty") } for i, p := range paraList { if err := ValidateSetVal(p.Id, p.SetVal); err != nil { return Request{}, fmt.Errorf("paraList[%d]: %w", i, err) } paraList[i].SetVal = strings.ToUpper(p.SetVal) } return newDeviceRequest(CmdSetParam, deviceType, sn, RequestInfo{ParaList: paraList}) } // MaxRecordsPerRequest 一次 cmd=2 最多获取的记录数 const MaxRecordsPerRequest = 32 // NewGetRecordRequest cmd=2,可一次获取多条记录,设备打成一个包上传 func NewGetRecordRequest(deviceType, sn string, indexes []int) (Request, error) { if len(indexes) == 0 { return Request{}, fmt.Errorf("record index is empty") } if len(indexes) > MaxRecordsPerRequest { return Request{}, fmt.Errorf("too many records: %d (max %d)", len(indexes), MaxRecordsPerRequest) } seen := map[int]bool{} for _, i := range indexes { if i < 1 || i > 255 { return Request{}, fmt.Errorf("invalid record index %d", i) } if seen[i] { return Request{}, fmt.Errorf("duplicate record index %d", i) } seen[i] = true } return newDeviceRequest(CmdGetRecord, deviceType, sn, RequestInfo{Index: indexes}) } // NewQueryParamRequest cmd=3 func NewQueryParamRequest(deviceType, sn string, idList []int) (Request, error) { if len(idList) == 0 { return Request{}, fmt.Errorf("idList is empty") } for _, id := range idList { if _, ok := FunctionCodeNames[id]; !ok { return Request{}, fmt.Errorf("invalid function code %d", id) } } return newDeviceRequest(CmdQueryParam, deviceType, sn, RequestInfo{IdList: idList}) } // NewExitRequest cmd=4,结束设备当前处理 func NewExitRequest() Request { return Request{Cmd: CmdExit, Data: struct{}{}} } /************** 外层解析辅助 **************/ // envelopeHead 只取路由需要的外层字段,dataArea 保留原始字节 type envelopeHead struct { CmdType uint `json:"cmdType"` EquipNo string `json:"equipNo"` DataArea json.RawMessage `json:"dataArea"` } // SplitEnvelope 从完整报文中取出 cmdType、equipNo 和原始 dataArea func SplitEnvelope(raw []byte) (cmdType uint, equipNo string, dataArea json.RawMessage, err error) { var h envelopeHead if err = json.Unmarshal(raw, &h); err != nil { return 0, "", nil, err } return h.CmdType, h.EquipNo, h.DataArea, nil } /************** 378 指令应答 **************/ type Ack struct { Cmd int `json:"cmd"` Result int `json:"result"` // 样本成功为 0 } func ParseAck(dataArea []byte) (Ack, error) { var a Ack err := json.Unmarshal(dataArea, &a) return a, err } func (a Ack) Ok() bool { return a.Result == 0 } /************** 379 业务回复 **************/ type HexItem struct { Code int `json:"code"` HexData string `json:"hexData"` ResultOne string `json:"resultOne"` } type IndexItem struct { Index int `json:"index"` // 数据类型码:4=故障数据,5=波形数据 ResultOne string `json:"resultOne"` } type IdItem struct { Id int `json:"id"` IdData string `json:"idData"` ResultOne string `json:"resultOne"` } type SetResultItem struct { Id int `json:"id"` ResultOne string `json:"resultOne"` } type ReplyInfo struct { Sn string `json:"sn"` HexList []HexItem `json:"hexList,omitempty"` IndexInfo []IndexItem `json:"indexInfo,omitempty"` IdInfo []IdItem `json:"idInfo,omitempty"` SetResultList []SetResultItem `json:"setResultList,omitempty"` } type ReplyGroup struct { Type string `json:"type"` Info []ReplyInfo `json:"info"` } // Reply 379 的 dataArea(FaultFilter 回复的 data 被展开到 dataArea 顶层) type Reply struct { Cmd int `json:"cmd"` Result string `json:"result"` Filename string `json:"filename,omitempty"` List []ReplyGroup `json:"list,omitempty"` } func ParseReply(dataArea []byte) (Reply, error) { var r Reply err := json.Unmarshal(dataArea, &r) return r, err } // Ok 全部子项成功 func (r Reply) Ok() bool { return r.Result == ResultOk } // Busy 设备正在处理其他命令 func (r Reply) Busy() bool { return r.Result == ResultBusy } // IsErrorReply 命令未进入业务处理(busy、格式错误、sn 不存在等),此时没有 list func (r Reply) IsErrorReply() bool { return r.Result != ResultOk && r.Result != ResultError } // FindInfo 按设备类型和 SN 查找回复项;aGate 系统内唯一,不比较 SN func (r Reply) FindInfo(deviceType, sn string) (*ReplyInfo, bool) { for gi := range r.List { g := &r.List[gi] if g.Type != deviceType { continue } for ii := range g.Info { if deviceType == DeviceTypeAGate || g.Info[ii].Sn == sn { return &g.Info[ii], true } } } return nil, false } /************** 607~610 文件上传 **************/ type UploadCall struct { FaultRecordID int64 `json:"faultRecordID"` FileName string `json:"fileName"` Url string `json:"url"` LogTime string `json:"logTime"` } func ParseUploadCall(dataArea []byte) (UploadCall, error) { var u UploadCall err := json.Unmarshal(dataArea, &u) return u, err } // UploadAck 608 上传指令应答 type UploadAck struct { Result int `json:"result"` Id int64 `json:"id"` // 平台召唤记录 ID(即 607.faultRecordID) Filename string `json:"filename"` } // ParseUploadAck 解析 608。 // 旧版本设备把 id 字段名误写为 result(同一对象出现两个 result,第二个为 id),后续版本会修复; // 解析前先把第二个 result 键改名为 id,新旧版本都能按同一结构解析。 func ParseUploadAck(dataArea []byte) (UploadAck, error) { var a UploadAck fixed, err := renameDuplicateKey(dataArea, "result", "id") if err != nil { return a, fmt.Errorf("608 dataArea: %w", err) } err = json.Unmarshal(fixed, &a) return a, err } // renameDuplicateKey 将对象顶层第二次出现的 key 改名为 newKey;没有重复时原样返回 func renameDuplicateKey(obj []byte, key, newKey string) ([]byte, error) { dec := json.NewDecoder(bytes.NewReader(obj)) if t, err := dec.Token(); err != nil || t != json.Delim('{') { return nil, fmt.Errorf("not a JSON object") } quoted := []byte(`"` + key + `"`) seen := false for dec.More() { kt, err := dec.Token() if err != nil { return nil, err } end := int(dec.InputOffset()) if k, _ := kt.(string); k == key { start := end - len(quoted) if seen && start >= 0 && bytes.Equal(obj[start:end], quoted) { out := make([]byte, 0, len(obj)) out = append(out, obj[:start]...) out = append(out, `"`+newKey+`"`...) return append(out, obj[end:]...), nil } seen = true } var skip json.RawMessage if err := dec.Decode(&skip); err != nil { return nil, err } } return obj, nil } // UploadDone 609 type UploadDone struct { Result int `json:"result"` // 样本成功为 0 FileName string `json:"fileName"` } func ParseUploadDone(dataArea []byte) (UploadDone, error) { var u UploadDone err := json.Unmarshal(dataArea, &u) return u, err } func (u UploadDone) Ok() bool { return u.Result == 0 } // UploadFileName 由 379.filename 推得上传文件名 func UploadFileName(batch string) string { return BatchFromUploadFileName(batch) + ".tar.bz2" } // BatchFromUploadFileName 由 609.fileName 去掉扩展名得到批次名 func BatchFromUploadFileName(fileName string) string { return strings.TrimSuffix(fileName, ".tar.bz2") }