package service import ( "fmt" "strconv" "strings" "sync" "time" "laic-backend/common" "laic-backend/logger" ) // TelemetryPoint 无人机遥测点(对应 device_telemetry 超级表) type TelemetryPoint struct { Ts int64 `json:"ts"` Longitude float64 `json:"longitude"` Latitude float64 `json:"latitude"` Altitude float64 `json:"altitude"` GroundSpeed float64 `json:"groundSpeed"` Roll float64 `json:"roll"` Pitch float64 `json:"pitch"` Yaw float64 `json:"yaw"` BatteryPct int `json:"batteryPct"` BatteryV float64 `json:"batteryV"` Satellites int `json:"satellites"` GpsQuality string `json:"gpsQuality"` LinkQuality int `json:"linkQuality"` FlightMode string `json:"flightMode"` Armed int8 `json:"armed"` } const ( telemetryBatchSize = 500 telemetryFlushInterval = 10 * time.Second ) type telemetryRecord struct { DockID string Point TelemetryPoint } // TelemetryStore 遥测缓冲:累计 500 条或 10 秒后批量写入 TDengine type TelemetryStore struct { mu sync.Mutex buffer []telemetryRecord droneSN sync.Map // dockID -> droneSN(来自 state/drone 上报) } var DefaultTelemetryStore = &TelemetryStore{} // SetDroneSN 记录机巢当前无人机序列号(作为 device_telemetry 的 tag) func (s *TelemetryStore) SetDroneSN(dockID, sn string) { if sn != "" { s.droneSN.Store(dockID, sn) } } // Append 追加一个遥测点,达到阈值立即刷盘 func (s *TelemetryStore) Append(dockID string, p TelemetryPoint) { s.mu.Lock() s.buffer = append(s.buffer, telemetryRecord{DockID: dockID, Point: p}) if len(s.buffer) >= telemetryBatchSize { s.flushLocked() } s.mu.Unlock() } // Start 启动定时刷盘协程 func (s *TelemetryStore) Start() { go func() { ticker := time.NewTicker(telemetryFlushInterval) defer ticker.Stop() for range ticker.C { s.mu.Lock() s.flushLocked() s.mu.Unlock() } }() } func (s *TelemetryStore) flushLocked() { if len(s.buffer) == 0 { return } records := s.buffer s.buffer = nil s.write(records) } func (s *TelemetryStore) write(records []telemetryRecord) { if common.TD == nil { return } groups := make(map[string][]TelemetryPoint, len(records)) for _, r := range records { groups[r.DockID] = append(groups[r.DockID], r.Point) } for dockID, points := range groups { s.insert(dockID, s.droneSNValue(dockID), points) } } func (s *TelemetryStore) droneSNValue(dockID string) string { if v, ok := s.droneSN.Load(dockID); ok { return v.(string) } return "" } // DroneSN 返回机巢当前无人机序列号(来自 state/drone 上报) func (s *TelemetryStore) DroneSN(dockID string) string { return s.droneSNValue(dockID) } func (s *TelemetryStore) insert(dockID, droneSN string, points []TelemetryPoint) { defer func() { if r := recover(); r != nil { logger.ERROR("TDengine 写入 panic(已恢复,避免进程崩溃)", fmt.Errorf("%v", r)) } }() var b strings.Builder b.WriteString("INSERT INTO ") b.WriteString(sanitizeTable(dockID)) b.WriteString(" USING device_telemetry TAGS ('") b.WriteString(escapeTD(dockID)) b.WriteString("', '") b.WriteString(escapeTD(droneSN)) b.WriteString("') VALUES ") for i, p := range points { if i > 0 { b.WriteString(", ") } ts := p.Ts if ts <= 0 { ts = time.Now().UnixMilli() } fmt.Fprintf(&b, "(%d, %s, %s, %s, %s, %s, %s, %s, %d, %s, %d, '%s', %d, '%s', %d)", ts, ftoa(p.Longitude), ftoa(p.Latitude), ftoa(p.Altitude), ftoa(p.GroundSpeed), ftoa(p.Roll), ftoa(p.Pitch), ftoa(p.Yaw), p.BatteryPct, ftoa(p.BatteryV), p.Satellites, escapeTD(p.GpsQuality), p.LinkQuality, escapeTD(p.FlightMode), p.Armed, ) } if _, err := common.TD.Exec(b.String()); err != nil { logger.ERROR("TDengine 批量写入遥测失败", err) } } func ftoa(v float64) string { return strconv.FormatFloat(v, 'f', -1, 64) } func escapeTD(s string) string { return strings.ReplaceAll(s, "'", "\\'") } // sanitizeTable 将 dockID 转为 TDengine 子表名(仅保留字母数字) func sanitizeTable(dockID string) string { var b strings.Builder b.WriteString("d_") for _, r := range dockID { if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') { b.WriteRune(r) } else { b.WriteByte('_') } } return b.String() }