798 lines
28 KiB
Go
798 lines
28 KiB
Go
package mqtt
|
|
|
|
import (
|
|
"crypto/tls"
|
|
"encoding/binary"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"hr_receiver/config"
|
|
"hr_receiver/models"
|
|
whgwv3pb "hr_receiver/proto/v3"
|
|
"log"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
mqtt "github.com/eclipse/paho.mqtt.golang"
|
|
"github.com/gorilla/websocket"
|
|
"google.golang.org/protobuf/proto"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
)
|
|
|
|
const (
|
|
wearableSampleEncodingCompactV1 = 1
|
|
wearableUlHasHr = 1 << 0
|
|
wearableUlHasStep = 1 << 1
|
|
wearableRecordHasTimingError = 1 << 2
|
|
wearableRecordReservedMask = 1 << 3
|
|
wearableUlBatteryShift = 4
|
|
wearableHrConfMask = 0x03
|
|
wearableHrActive = 1 << 2
|
|
wearableHrOnSkin = 1 << 3
|
|
wearableHrReservedMask = 0xF0
|
|
)
|
|
|
|
type DebugV3Status struct {
|
|
Active bool `json:"active"`
|
|
ClientConnected bool `json:"clientConnected"`
|
|
PersistToDatabase bool `json:"persistToDatabase"`
|
|
Region string `json:"region"`
|
|
SubscriberCount int `json:"subscriberCount"`
|
|
}
|
|
|
|
type DebugV3GatewayStatusRecord struct {
|
|
Identifier string `json:"identifier"`
|
|
Topic string `json:"topic"`
|
|
ProtocolVersion int `json:"protocolVersion"`
|
|
RegionID uint32 `json:"regionId"`
|
|
GatewayMAC string `json:"gatewayMac"`
|
|
GatewayIPv4Addr string `json:"gatewayIpv4Addr"`
|
|
GatewayActiveUplink int32 `json:"gatewayActiveUplink"`
|
|
GatewayCellularIMEI string `json:"gatewayCellularImei"`
|
|
GatewayCellularRSSI int32 `json:"gatewayCellularRssi"`
|
|
GatewayCellularBER int32 `json:"gatewayCellularBer"`
|
|
BootCount uint32 `json:"bootCount"`
|
|
UptimeMs uint32 `json:"uptimeMs"`
|
|
DurationMsSinceLastPacket uint32 `json:"durationMsSinceLastPacket"`
|
|
RxCount uint32 `json:"rxCount"`
|
|
BatteryVoltageMV uint32 `json:"batteryVoltageMv"`
|
|
BatterySOCPercentage uint32 `json:"batterySocPercentage"`
|
|
ChargingRatePercentage int32 `json:"chargingRatePercentage"`
|
|
ReceivedAt int64 `json:"receivedAt"`
|
|
}
|
|
|
|
type DebugV3WearableSampleRecord struct {
|
|
Identifier string `json:"identifier"`
|
|
Topic string `json:"topic"`
|
|
ProtocolVersion int `json:"protocolVersion"`
|
|
RegionID uint32 `json:"regionId"`
|
|
GatewayMAC string `json:"gatewayMac"`
|
|
GatewayIPv4Addr string `json:"gatewayIpv4Addr"`
|
|
GatewayActiveUplink int32 `json:"gatewayActiveUplink"`
|
|
GatewayCellularIMEI string `json:"gatewayCellularImei"`
|
|
GatewayCellularRSSI int32 `json:"gatewayCellularRssi"`
|
|
GatewayCellularBER int32 `json:"gatewayCellularBer"`
|
|
ReportSeq uint32 `json:"reportSeq"`
|
|
FrameID uint32 `json:"frameId"`
|
|
FrameUnixTimeUs uint64 `json:"frameUnixTimeUs"`
|
|
RadioSubDevID uint32 `json:"radioSubDevId"`
|
|
RfFrequencyHz uint32 `json:"rfFrequencyHz"`
|
|
SampleEncoding uint32 `json:"sampleEncoding"`
|
|
SampleCount uint32 `json:"sampleCount"`
|
|
SampleIndex int `json:"sampleIndex"`
|
|
NodeID uint32 `json:"nodeId"`
|
|
BeltAddr string `json:"beltAddr"`
|
|
Flags uint32 `json:"flags"`
|
|
Battery uint32 `json:"battery"`
|
|
HasTimingError bool `json:"hasTimingError"`
|
|
TimingErrorUs int32 `json:"timingErrorUs"`
|
|
HasHeartRate bool `json:"hasHeartRate"`
|
|
HeartRate uint32 `json:"heartRate"`
|
|
HrConfidence int `json:"hrConfidence"`
|
|
IsActive bool `json:"isActive"`
|
|
IsOnSkin bool `json:"isOnSkin"`
|
|
HasStepCount bool `json:"hasStepCount"`
|
|
StepCount uint32 `json:"stepCount"`
|
|
RssiSyncX2Neg uint32 `json:"rssiSyncX2Neg"`
|
|
SignalRSSINeg float64 `json:"signalRssiNeg"`
|
|
ReceivedAt int64 `json:"receivedAt"`
|
|
}
|
|
|
|
type DebugV3RadioFrameRecord struct {
|
|
Identifier string `json:"identifier"`
|
|
Topic string `json:"topic"`
|
|
ProtocolVersion int `json:"protocolVersion"`
|
|
Kind string `json:"kind"`
|
|
GatewayMAC string `json:"gatewayMac"`
|
|
HubBusID uint32 `json:"hubBusId"`
|
|
HubSubDevID uint32 `json:"hubSubDevId"`
|
|
PacketStatusKind string `json:"packetStatusKind"`
|
|
SignalRSSINeg float64 `json:"signalRssiNeg"`
|
|
SNR float64 `json:"snr"`
|
|
RawSignalRSSIX2Neg uint32 `json:"rawSignalRssiX2Neg"`
|
|
RawSnrPktX4 int32 `json:"rawSnrPktX4"`
|
|
RawFskRssiSyncX2Neg uint32 `json:"rawFskRssiSyncX2Neg"`
|
|
RawFskRssiAvgX2Neg uint32 `json:"rawFskRssiAvgX2Neg"`
|
|
HubRadioMode string `json:"hubRadioMode"`
|
|
HubRadioBW int32 `json:"hubRadioBw"`
|
|
HubRadioSF uint32 `json:"hubRadioSf"`
|
|
HubRadioFrequencyMHz float64 `json:"hubRadioFrequencyMHz"`
|
|
HubGfskBitrateBps uint32 `json:"hubGfskBitrateBps"`
|
|
HubGfskDeviationHz uint32 `json:"hubGfskDeviationHz"`
|
|
HubGfskRxBandwidthHz uint32 `json:"hubGfskRxBandwidthHz"`
|
|
HubGfskPayloadLength uint32 `json:"hubGfskPayloadLength"`
|
|
DataHex string `json:"dataHex"`
|
|
DataLength int `json:"dataLength"`
|
|
GatewayUptimeMs uint32 `json:"gatewayUptimeMs"`
|
|
RxSeq uint32 `json:"rxSeq"`
|
|
RxDoneHubUptimeUs uint64 `json:"rxDoneHubUptimeUs"`
|
|
IrqToForwardUs uint32 `json:"irqToForwardUs"`
|
|
ReceivedAt int64 `json:"receivedAt"`
|
|
}
|
|
|
|
type DebugV3Event struct {
|
|
CardKey string `json:"cardKey"`
|
|
Kind string `json:"kind"`
|
|
RegionID uint32 `json:"regionId"`
|
|
ReceivedAt int64 `json:"receivedAt"`
|
|
Topic string `json:"topic"`
|
|
GatewayStatus *DebugV3GatewayStatusRecord `json:"gatewayStatus,omitempty"`
|
|
WearableHealthSample *DebugV3WearableSampleRecord `json:"wearableHealthSample,omitempty"`
|
|
RadioNodeGeneralRsp *DebugV3RadioFrameRecord `json:"radioNodeGeneralRsp,omitempty"`
|
|
RadioUnrecognizedFrame *DebugV3RadioFrameRecord `json:"radioUnrecognizedFrame,omitempty"`
|
|
}
|
|
|
|
type packetStatusSnapshotV3 struct {
|
|
kind string
|
|
signalRSSINeg float64
|
|
snr float64
|
|
rawSignalRSSIX2Neg uint32
|
|
rawSnrPktX4 int32
|
|
rawFskRssiSyncX2Neg uint32
|
|
rawFskRssiAvgX2Neg uint32
|
|
}
|
|
|
|
type radioParametersSnapshotV3 struct {
|
|
mode string
|
|
loraBw int32
|
|
loraSf uint32
|
|
loraFrequencyMHz float64
|
|
gfskBitrateBps uint32
|
|
gfskFrequencyMHz float64
|
|
gfskDeviationHz uint32
|
|
gfskRxBandwidthHz uint32
|
|
gfskPayloadLength uint32
|
|
}
|
|
|
|
type DebugV3Service struct {
|
|
cfg config.MQTTConfig
|
|
client mqtt.Client
|
|
db *gorm.DB
|
|
mu sync.RWMutex
|
|
persistToDatabase bool
|
|
subscribers map[*websocket.Conn]struct{}
|
|
active bool
|
|
}
|
|
|
|
var globalDebugV3Service *DebugV3Service
|
|
|
|
func InitDebugV3Service(db *gorm.DB, cfg config.MQTTConfig) {
|
|
globalDebugV3Service = &DebugV3Service{
|
|
cfg: cfg,
|
|
db: db,
|
|
subscribers: make(map[*websocket.Conn]struct{}),
|
|
}
|
|
}
|
|
|
|
func GetDebugV3Service() *DebugV3Service {
|
|
return globalDebugV3Service
|
|
}
|
|
|
|
func (s *DebugV3Service) Status() DebugV3Status {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return DebugV3Status{
|
|
Active: s.active,
|
|
ClientConnected: s.client != nil && s.client.IsConnected(),
|
|
PersistToDatabase: s.persistToDatabase,
|
|
Region: s.cfg.Region,
|
|
SubscriberCount: len(s.subscribers),
|
|
}
|
|
}
|
|
|
|
func (s *DebugV3Service) Start(persistToDatabase bool) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
|
|
if s.active && s.client != nil && s.client.IsConnected() {
|
|
s.persistToDatabase = persistToDatabase
|
|
return nil
|
|
}
|
|
if err := validateConfig(s.cfg); err != nil {
|
|
return err
|
|
}
|
|
|
|
client, err := s.connectLocked(persistToDatabase)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.client = client
|
|
s.persistToDatabase = persistToDatabase
|
|
s.active = true
|
|
return nil
|
|
}
|
|
|
|
func (s *DebugV3Service) Stop() {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.client != nil && s.client.IsConnected() {
|
|
s.client.Disconnect(250)
|
|
}
|
|
s.client = nil
|
|
s.active = false
|
|
s.persistToDatabase = false
|
|
}
|
|
|
|
func (s *DebugV3Service) AddSubscriber(conn *websocket.Conn) {
|
|
s.mu.Lock()
|
|
s.subscribers[conn] = struct{}{}
|
|
s.mu.Unlock()
|
|
}
|
|
|
|
func (s *DebugV3Service) RemoveSubscriber(conn *websocket.Conn) {
|
|
s.mu.Lock()
|
|
delete(s.subscribers, conn)
|
|
s.mu.Unlock()
|
|
_ = conn.Close()
|
|
}
|
|
|
|
func (s *DebugV3Service) connectLocked(persistToDatabase bool) (mqtt.Client, error) {
|
|
opts := mqtt.NewClientOptions()
|
|
scheme := "tcp"
|
|
if s.cfg.UseTLS {
|
|
scheme = "ssl"
|
|
opts.SetTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12})
|
|
}
|
|
broker := fmt.Sprintf("%s://%s:%d", scheme, s.cfg.Host, s.cfg.Port)
|
|
opts.AddBroker(broker)
|
|
opts.SetClientID(fmt.Sprintf("%s-debug-v3-%d", s.cfg.ClientIDPrefix, time.Now().UnixNano()))
|
|
opts.SetUsername(s.cfg.Username)
|
|
opts.SetPassword(s.cfg.Password)
|
|
opts.SetKeepAlive(60 * time.Second)
|
|
opts.SetAutoReconnect(false)
|
|
opts.SetConnectRetry(false)
|
|
opts.SetDefaultPublishHandler(s.handleMessage)
|
|
opts.SetOnConnectHandler(func(client mqtt.Client) {
|
|
if err := s.subscribe(client); err != nil {
|
|
log.Printf("mqtt v3 debug subscribe failed: %v", err)
|
|
return
|
|
}
|
|
log.Printf("mqtt v3 debug connected to %s persist=%v", broker, persistToDatabase)
|
|
})
|
|
opts.SetConnectionLostHandler(func(client mqtt.Client, err error) {
|
|
log.Printf("mqtt v3 debug connection lost: %v", err)
|
|
s.mu.Lock()
|
|
if s.client == client {
|
|
s.client = nil
|
|
s.active = false
|
|
}
|
|
s.mu.Unlock()
|
|
})
|
|
|
|
client := mqtt.NewClient(opts)
|
|
token := client.Connect()
|
|
if !token.WaitTimeout(15 * time.Second) {
|
|
return nil, fmt.Errorf("mqtt v3 debug connect timeout")
|
|
}
|
|
if err := token.Error(); err != nil {
|
|
return nil, err
|
|
}
|
|
return client, nil
|
|
}
|
|
|
|
func (s *DebugV3Service) subscribe(client mqtt.Client) error {
|
|
topics := []string{
|
|
fmt.Sprintf("/whgw/v3/region/%s/gateway/+/telemetry/wearable-health-telemetry", s.cfg.Region),
|
|
"/whgw/v3/gateway/+/telemetry/status",
|
|
"/whgw/v3/gateway/+/telemetry/radio-node-general-rsp",
|
|
"/whgw/v3/gateway/+/telemetry/radio-unrecognized-frame",
|
|
}
|
|
for _, topic := range topics {
|
|
token := client.Subscribe(topic, byte(s.cfg.QoS), s.handleMessage)
|
|
if !token.WaitTimeout(10 * time.Second) {
|
|
return fmt.Errorf("mqtt v3 debug subscribe timeout for topic %s", topic)
|
|
}
|
|
if err := token.Error(); err != nil {
|
|
return fmt.Errorf("mqtt v3 debug subscribe topic %s: %w", topic, err)
|
|
}
|
|
log.Printf("mqtt v3 debug subscribed: %s", topic)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *DebugV3Service) handleMessage(_ mqtt.Client, msg mqtt.Message) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Printf("mqtt v3 debug handle panic topic=%s err=%v", msg.Topic(), r)
|
|
}
|
|
}()
|
|
if len(msg.Payload()) == 0 {
|
|
return
|
|
}
|
|
|
|
now := time.Now().UnixMilli()
|
|
if strings.Contains(msg.Topic(), "/telemetry/wearable-health-telemetry") {
|
|
s.handleWearableHealthTelemetry(msg.Topic(), msg.Payload(), now)
|
|
return
|
|
}
|
|
|
|
var packet whgwv3pb.GatewayTelemetryMsg
|
|
if err := proto.Unmarshal(msg.Payload(), &packet); err != nil {
|
|
log.Printf("mqtt v3 debug payload parse failed topic=%s err=%v", msg.Topic(), err)
|
|
return
|
|
}
|
|
|
|
switch payload := packet.Choice.(type) {
|
|
case *whgwv3pb.GatewayTelemetryMsg_NtfGatewayStatus:
|
|
record := buildGatewayStatusRecordV3(payload.NtfGatewayStatus, msg.Topic(), now)
|
|
s.maybePersistGatewayStatus(record)
|
|
s.broadcast(DebugV3Event{
|
|
CardKey: fmt.Sprintf("%d-%s", record.RegionID, record.GatewayMAC),
|
|
GatewayStatus: &record,
|
|
Kind: "gateway_status",
|
|
ReceivedAt: now,
|
|
RegionID: record.RegionID,
|
|
Topic: msg.Topic(),
|
|
})
|
|
case *whgwv3pb.GatewayTelemetryMsg_NtfRadioNodeGeneralRsp:
|
|
record := buildRadioFrameRecordV3("radio_node_general_rsp", payload.NtfRadioNodeGeneralRsp.GetFrame(), msg.Topic(), now)
|
|
s.broadcast(DebugV3Event{
|
|
CardKey: record.Identifier,
|
|
Kind: "radio_node_general_rsp",
|
|
ReceivedAt: now,
|
|
Topic: msg.Topic(),
|
|
RadioNodeGeneralRsp: &record,
|
|
})
|
|
case *whgwv3pb.GatewayTelemetryMsg_NtfRadioUnrecognizedFrame:
|
|
record := buildRadioFrameRecordV3("radio_unrecognized_frame", payload.NtfRadioUnrecognizedFrame, msg.Topic(), now)
|
|
s.broadcast(DebugV3Event{
|
|
CardKey: record.Identifier,
|
|
Kind: "radio_unrecognized_frame",
|
|
ReceivedAt: now,
|
|
Topic: msg.Topic(),
|
|
RadioUnrecognizedFrame: &record,
|
|
})
|
|
default:
|
|
log.Printf("mqtt v3 debug payload ignored topic=%s", msg.Topic())
|
|
}
|
|
}
|
|
|
|
func (s *DebugV3Service) handleWearableHealthTelemetry(topic string, payload []byte, now int64) {
|
|
var batch whgwv3pb.WearableHealthTelemetryBatch
|
|
if err := proto.Unmarshal(payload, &batch); err != nil {
|
|
log.Printf("mqtt v3 wearable batch parse failed topic=%s err=%v", topic, err)
|
|
return
|
|
}
|
|
|
|
samples, err := decodeWearableHealthSamples(&batch, topic, now)
|
|
if err != nil {
|
|
log.Printf("mqtt v3 wearable batch decode failed topic=%s err=%v", topic, err)
|
|
return
|
|
}
|
|
|
|
for _, sample := range samples {
|
|
s.maybePersistWearableSample(sample)
|
|
event := DebugV3Event{
|
|
CardKey: fmt.Sprintf("%d-%d", sample.RegionID, sample.NodeID),
|
|
Kind: "wearable_health_sample",
|
|
ReceivedAt: now,
|
|
RegionID: sample.RegionID,
|
|
Topic: topic,
|
|
WearableHealthSample: &sample,
|
|
}
|
|
s.broadcast(event)
|
|
}
|
|
}
|
|
|
|
func (s *DebugV3Service) maybePersistWearableSample(sample DebugV3WearableSampleRecord) {
|
|
if sample.HasHeartRate {
|
|
record := models.MqttHeartRateRecord{
|
|
Identifier: sample.Identifier + ":hr",
|
|
Topic: sample.Topic,
|
|
RegionID: sample.RegionID,
|
|
GatewayMAC: sample.GatewayMAC,
|
|
GatewaySchemaVersion: 0,
|
|
GatewayActiveUplink: sample.GatewayActiveUplink,
|
|
GatewayCellularIMEI: sample.GatewayCellularIMEI,
|
|
GatewayCellularRSSI: sample.GatewayCellularRSSI,
|
|
GatewayCellularBER: sample.GatewayCellularBER,
|
|
BandID: sample.NodeID,
|
|
BeltAddr: sample.BeltAddr,
|
|
PacketNum: sample.FrameID,
|
|
HeartRate: int(sample.HeartRate),
|
|
HrConfidence: sample.HrConfidence,
|
|
IsActive: sample.IsActive,
|
|
IsOnSkin: sample.IsOnSkin,
|
|
Battery: sample.Battery,
|
|
PacketStatusSource: "wearable_batch_v1",
|
|
SignalRSSINeg: sample.SignalRSSINeg,
|
|
SNR: 0,
|
|
RawSignalRSSIX2Neg: sample.RssiSyncX2Neg,
|
|
RawSnrPktX4: 0,
|
|
HubBusID: 0,
|
|
HubSubDevID: sample.RadioSubDevID,
|
|
HubRadioBW: 0,
|
|
HubRadioSF: 0,
|
|
HubRadioFrequencyMHz: float64(sample.RfFrequencyHz) / 1_000_000,
|
|
ReceivedAt: sample.ReceivedAt,
|
|
}
|
|
s.maybePersist(&record)
|
|
}
|
|
if sample.HasStepCount {
|
|
record := models.MqttStepCountRecord{
|
|
Identifier: sample.Identifier + ":step",
|
|
Topic: sample.Topic,
|
|
RegionID: sample.RegionID,
|
|
GatewayMAC: sample.GatewayMAC,
|
|
GatewaySchemaVersion: 0,
|
|
GatewayActiveUplink: sample.GatewayActiveUplink,
|
|
GatewayCellularIMEI: sample.GatewayCellularIMEI,
|
|
GatewayCellularRSSI: sample.GatewayCellularRSSI,
|
|
GatewayCellularBER: sample.GatewayCellularBER,
|
|
BandID: sample.NodeID,
|
|
BeltAddr: sample.BeltAddr,
|
|
PacketNum: sample.FrameID,
|
|
StepCount: sample.StepCount,
|
|
PacketStatusSource: "wearable_batch_v1",
|
|
SignalRSSINeg: sample.SignalRSSINeg,
|
|
SNR: 0,
|
|
RawSignalRSSIX2Neg: sample.RssiSyncX2Neg,
|
|
RawSnrPktX4: 0,
|
|
HubBusID: 0,
|
|
HubSubDevID: sample.RadioSubDevID,
|
|
HubRadioBW: 0,
|
|
HubRadioSF: 0,
|
|
HubRadioFrequencyMHz: float64(sample.RfFrequencyHz) / 1_000_000,
|
|
ReceivedAt: sample.ReceivedAt,
|
|
}
|
|
s.maybePersist(&record)
|
|
}
|
|
}
|
|
|
|
func (s *DebugV3Service) maybePersistGatewayStatus(record DebugV3GatewayStatusRecord) {
|
|
modelRecord := models.MqttGatewayStatusRecord{
|
|
Identifier: record.Identifier,
|
|
Topic: record.Topic,
|
|
RegionID: record.RegionID,
|
|
GatewayMAC: record.GatewayMAC,
|
|
GatewaySchemaVersion: 0,
|
|
GatewayActiveUplink: record.GatewayActiveUplink,
|
|
GatewayCellularIMEI: record.GatewayCellularIMEI,
|
|
GatewayCellularRSSI: record.GatewayCellularRSSI,
|
|
GatewayCellularBER: record.GatewayCellularBER,
|
|
BootCount: record.BootCount,
|
|
UptimeMs: record.UptimeMs,
|
|
DurationMsSinceLastPacket: record.DurationMsSinceLastPacket,
|
|
RxCount: record.RxCount,
|
|
BatteryVoltageMV: record.BatteryVoltageMV,
|
|
BatterySOCPercentage: record.BatterySOCPercentage,
|
|
ChargingRatePercentage: record.ChargingRatePercentage,
|
|
ReceivedAt: record.ReceivedAt,
|
|
}
|
|
s.maybePersist(&modelRecord)
|
|
}
|
|
|
|
func (s *DebugV3Service) maybePersist(record interface{}) {
|
|
s.mu.RLock()
|
|
enabled := s.persistToDatabase
|
|
s.mu.RUnlock()
|
|
if !enabled {
|
|
return
|
|
}
|
|
tx := s.db.Clauses(clause.OnConflict{DoNothing: true}).Create(record)
|
|
if tx.Error != nil {
|
|
log.Printf("mqtt v3 debug persist failed type=%T err=%v", record, tx.Error)
|
|
}
|
|
}
|
|
|
|
func (s *DebugV3Service) broadcast(event DebugV3Event) {
|
|
payload, err := json.Marshal(event)
|
|
if err != nil {
|
|
log.Printf("mqtt v3 debug marshal failed err=%v", err)
|
|
return
|
|
}
|
|
|
|
s.mu.RLock()
|
|
conns := make([]*websocket.Conn, 0, len(s.subscribers))
|
|
for conn := range s.subscribers {
|
|
conns = append(conns, conn)
|
|
}
|
|
s.mu.RUnlock()
|
|
|
|
for _, conn := range conns {
|
|
if err := conn.WriteMessage(websocket.TextMessage, payload); err != nil {
|
|
log.Printf("mqtt v3 debug websocket send failed err=%v", err)
|
|
s.RemoveSubscriber(conn)
|
|
}
|
|
}
|
|
}
|
|
|
|
func decodeWearableHealthSamples(batch *whgwv3pb.WearableHealthTelemetryBatch, topic string, now int64) ([]DebugV3WearableSampleRecord, error) {
|
|
if batch.GetSampleEncoding() != wearableSampleEncodingCompactV1 {
|
|
return nil, fmt.Errorf("unsupported sample_encoding=%d", batch.GetSampleEncoding())
|
|
}
|
|
|
|
gatewayInfo := batch.GetGatewayInfo()
|
|
regionID := gatewayInfo.GetRegionId()
|
|
if regionID == 0 {
|
|
regionID = parseRegionFromTopic(topic)
|
|
}
|
|
gatewayMAC := formatMAC(gatewayInfo.GetGatewayMac())
|
|
raw := batch.GetSamples()
|
|
records := make([]DebugV3WearableSampleRecord, 0, batch.GetSampleCount())
|
|
for offset, sampleIndex := 0, 0; offset < len(raw); sampleIndex++ {
|
|
record, consumed, err := parseWearableHealthSample(raw[offset:], gatewayInfo, batch, topic, now, regionID, gatewayMAC, sampleIndex)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("sample_index=%d offset=%d: %w", sampleIndex, offset, err)
|
|
}
|
|
offset += consumed
|
|
records = append(records, record)
|
|
}
|
|
return records, nil
|
|
}
|
|
|
|
func parseWearableHealthSample(raw []byte, gatewayInfo *whgwv3pb.GatewayInfo, batch *whgwv3pb.WearableHealthTelemetryBatch, topic string, now int64, regionID uint32, gatewayMAC string, sampleIndex int) (DebugV3WearableSampleRecord, int, error) {
|
|
if len(raw) < 3 {
|
|
return DebugV3WearableSampleRecord{}, 0, fmt.Errorf("record too short")
|
|
}
|
|
|
|
nodeID := raw[0]
|
|
flags := raw[1]
|
|
rssiSyncX2Neg := raw[2]
|
|
if (flags & wearableRecordReservedMask) != 0 {
|
|
return DebugV3WearableSampleRecord{}, 0, fmt.Errorf("reserved flags set: 0x%02x", flags)
|
|
}
|
|
|
|
offset := 3
|
|
timingErrorUs := int16(0)
|
|
hasTimingError := (flags & wearableRecordHasTimingError) != 0
|
|
if hasTimingError {
|
|
if len(raw) < offset+2 {
|
|
return DebugV3WearableSampleRecord{}, 0, fmt.Errorf("timing error field truncated")
|
|
}
|
|
timingErrorUs = int16(binary.LittleEndian.Uint16(raw[offset : offset+2]))
|
|
offset += 2
|
|
}
|
|
|
|
hasHr := (flags & wearableUlHasHr) != 0
|
|
hasStep := (flags & wearableUlHasStep) != 0
|
|
hrStatus := byte(0)
|
|
hrValue := byte(0)
|
|
if hasHr {
|
|
if len(raw) < offset+2 {
|
|
return DebugV3WearableSampleRecord{}, 0, fmt.Errorf("hr block truncated")
|
|
}
|
|
hrStatus = raw[offset]
|
|
if (hrStatus & wearableHrReservedMask) != 0 {
|
|
return DebugV3WearableSampleRecord{}, 0, fmt.Errorf("reserved hr status bits set: 0x%02x", hrStatus)
|
|
}
|
|
hrValue = raw[offset+1]
|
|
offset += 2
|
|
}
|
|
|
|
stepValue := uint16(0)
|
|
if hasStep {
|
|
if len(raw) < offset+2 {
|
|
return DebugV3WearableSampleRecord{}, 0, fmt.Errorf("step block truncated")
|
|
}
|
|
stepValue = binary.LittleEndian.Uint16(raw[offset : offset+2])
|
|
offset += 2
|
|
}
|
|
|
|
batteryQ4 := (flags >> wearableUlBatteryShift) & 0x0F
|
|
batteryPercent := uint32((uint16(batteryQ4)*100 + 7) / 15)
|
|
nodeID32 := uint32(nodeID)
|
|
|
|
record := DebugV3WearableSampleRecord{
|
|
Identifier: fmt.Sprintf("v3:wearable:%d:%s:%d:%d:%d", regionID, gatewayMAC, batch.GetFrameId(), nodeID32, sampleIndex),
|
|
Topic: topic,
|
|
ProtocolVersion: 3,
|
|
RegionID: regionID,
|
|
GatewayMAC: gatewayMAC,
|
|
GatewayIPv4Addr: formatIPv4(gatewayInfo.GetIpv4Addr()),
|
|
GatewayActiveUplink: int32(gatewayInfo.GetActiveUplink()),
|
|
GatewayCellularIMEI: gatewayInfo.GetCellularModem().GetImei(),
|
|
GatewayCellularRSSI: gatewayInfo.GetCellularModem().GetCsqRssi(),
|
|
GatewayCellularBER: gatewayInfo.GetCellularModem().GetCsqBer(),
|
|
ReportSeq: batch.GetReportSeq(),
|
|
FrameID: batch.GetFrameId(),
|
|
FrameUnixTimeUs: batch.GetFrameUnixTimeUs(),
|
|
RadioSubDevID: batch.GetRadioSubDevId(),
|
|
RfFrequencyHz: batch.GetRfFrequencyHz(),
|
|
SampleEncoding: batch.GetSampleEncoding(),
|
|
SampleCount: batch.GetSampleCount(),
|
|
SampleIndex: sampleIndex,
|
|
NodeID: nodeID32,
|
|
BeltAddr: fmt.Sprintf("%d-%d", regionID, nodeID32),
|
|
Flags: uint32(flags),
|
|
Battery: batteryPercent,
|
|
HasTimingError: hasTimingError,
|
|
TimingErrorUs: int32(timingErrorUs),
|
|
HasHeartRate: hasHr,
|
|
HeartRate: uint32(hrValue),
|
|
HrConfidence: int(hrStatus & wearableHrConfMask),
|
|
IsActive: (hrStatus & wearableHrActive) != 0,
|
|
IsOnSkin: (hrStatus & wearableHrOnSkin) != 0,
|
|
HasStepCount: hasStep,
|
|
StepCount: uint32(stepValue),
|
|
RssiSyncX2Neg: uint32(rssiSyncX2Neg),
|
|
SignalRSSINeg: wearableRssiToDBm(rssiSyncX2Neg),
|
|
ReceivedAt: now,
|
|
}
|
|
return record, offset, nil
|
|
}
|
|
|
|
func wearableRssiToDBm(value uint8) float64 {
|
|
if value == 0 {
|
|
return 0
|
|
}
|
|
return -float64(value) / 2
|
|
}
|
|
|
|
func buildGatewayStatusRecordV3(status *whgwv3pb.GatewayStatus, topic string, now int64) DebugV3GatewayStatusRecord {
|
|
gatewayInfo := status.GetInfo()
|
|
regionID := gatewayInfo.GetRegionId()
|
|
if regionID == 0 {
|
|
regionID = parseRegionFromTopic(topic)
|
|
}
|
|
gatewayMAC := formatMAC(gatewayInfo.GetGatewayMac())
|
|
|
|
return DebugV3GatewayStatusRecord{
|
|
Identifier: fmt.Sprintf("v3:gateway:%d:%s:%d:%d:%d", regionID, gatewayMAC, status.GetStat().GetBootCount(), status.GetStat().GetUptimeMs(), status.GetStat().GetRxCount()),
|
|
Topic: topic,
|
|
ProtocolVersion: 3,
|
|
RegionID: regionID,
|
|
GatewayMAC: gatewayMAC,
|
|
GatewayIPv4Addr: formatIPv4(gatewayInfo.GetIpv4Addr()),
|
|
GatewayActiveUplink: int32(gatewayInfo.GetActiveUplink()),
|
|
GatewayCellularIMEI: gatewayInfo.GetCellularModem().GetImei(),
|
|
GatewayCellularRSSI: gatewayInfo.GetCellularModem().GetCsqRssi(),
|
|
GatewayCellularBER: gatewayInfo.GetCellularModem().GetCsqBer(),
|
|
BootCount: status.GetStat().GetBootCount(),
|
|
UptimeMs: status.GetStat().GetUptimeMs(),
|
|
DurationMsSinceLastPacket: status.GetStat().GetDurationMsSinceLastPacket(),
|
|
RxCount: status.GetStat().GetRxCount(),
|
|
BatteryVoltageMV: status.GetStat().GetBatteryInfo().GetVoltageMv(),
|
|
BatterySOCPercentage: status.GetStat().GetBatteryInfo().GetSocPercentage(),
|
|
ChargingRatePercentage: status.GetStat().GetBatteryInfo().GetChargingRatePercentage(),
|
|
ReceivedAt: now,
|
|
}
|
|
}
|
|
|
|
func buildRadioFrameRecordV3(kind string, frame *whgwv3pb.RadioRxFrame, topic string, now int64) DebugV3RadioFrameRecord {
|
|
radioData := frame.GetRadioData()
|
|
hubInfo := radioData.GetHubInfo()
|
|
status := parsePacketStatusV3(radioData.GetPacketStatus())
|
|
radio := parseRadioParametersV3(hubInfo.GetRadioParameters())
|
|
gatewayMAC := parseGatewayMACFromTopic(topic)
|
|
|
|
return DebugV3RadioFrameRecord{
|
|
Identifier: fmt.Sprintf("v3:%s:%s:%d:%d", kind, gatewayMAC, hubInfo.GetSubDevId(), frame.GetRxSeq()),
|
|
Topic: topic,
|
|
ProtocolVersion: 3,
|
|
Kind: kind,
|
|
GatewayMAC: gatewayMAC,
|
|
HubBusID: hubInfo.GetBusId(),
|
|
HubSubDevID: hubInfo.GetSubDevId(),
|
|
PacketStatusKind: status.kind,
|
|
SignalRSSINeg: status.signalRSSINeg,
|
|
SNR: status.snr,
|
|
RawSignalRSSIX2Neg: status.rawSignalRSSIX2Neg,
|
|
RawSnrPktX4: status.rawSnrPktX4,
|
|
RawFskRssiSyncX2Neg: status.rawFskRssiSyncX2Neg,
|
|
RawFskRssiAvgX2Neg: status.rawFskRssiAvgX2Neg,
|
|
HubRadioMode: radio.mode,
|
|
HubRadioBW: radio.loraBw,
|
|
HubRadioSF: radio.loraSf,
|
|
HubRadioFrequencyMHz: chooseFrequencyMHz(radio),
|
|
HubGfskBitrateBps: radio.gfskBitrateBps,
|
|
HubGfskDeviationHz: radio.gfskDeviationHz,
|
|
HubGfskRxBandwidthHz: radio.gfskRxBandwidthHz,
|
|
HubGfskPayloadLength: radio.gfskPayloadLength,
|
|
DataHex: strings.ToUpper(hex.EncodeToString(radioData.GetData())),
|
|
DataLength: len(radioData.GetData()),
|
|
GatewayUptimeMs: frame.GetGatewayUptimeMs(),
|
|
RxSeq: frame.GetRxSeq(),
|
|
RxDoneHubUptimeUs: radioData.GetRxTiming().GetRxDoneHubUptimeUs(),
|
|
IrqToForwardUs: radioData.GetRxTiming().GetIrqToForwardUs(),
|
|
ReceivedAt: now,
|
|
}
|
|
}
|
|
|
|
func parsePacketStatusV3(status *whgwv3pb.PacketStatus) packetStatusSnapshotV3 {
|
|
if status == nil {
|
|
return packetStatusSnapshotV3{}
|
|
}
|
|
if lora := status.GetLora(); lora != nil {
|
|
return packetStatusSnapshotV3{
|
|
kind: "lora",
|
|
signalRSSINeg: -float64(lora.GetSignalRssiX2Neg()) / 2,
|
|
snr: float64(lora.GetSnrPktX4()) / 4,
|
|
rawSignalRSSIX2Neg: lora.GetSignalRssiX2Neg(),
|
|
rawSnrPktX4: lora.GetSnrPktX4(),
|
|
}
|
|
}
|
|
if fsk := status.GetFsk(); fsk != nil {
|
|
return packetStatusSnapshotV3{
|
|
kind: "fsk",
|
|
signalRSSINeg: -float64(fsk.GetRssiSyncX2Neg()) / 2,
|
|
rawFskRssiSyncX2Neg: fsk.GetRssiSyncX2Neg(),
|
|
rawFskRssiAvgX2Neg: fsk.GetRssiAvgX2Neg(),
|
|
}
|
|
}
|
|
return packetStatusSnapshotV3{}
|
|
}
|
|
|
|
func parseRadioParametersV3(params *whgwv3pb.RadioParameters) radioParametersSnapshotV3 {
|
|
if params == nil {
|
|
return radioParametersSnapshotV3{}
|
|
}
|
|
if lora := params.GetLora(); lora != nil {
|
|
return radioParametersSnapshotV3{
|
|
mode: "lora",
|
|
loraBw: int32(lora.GetBw()),
|
|
loraSf: lora.GetSf(),
|
|
loraFrequencyMHz: float64(lora.GetFrequencyMhz()),
|
|
}
|
|
}
|
|
if gfsk := params.GetGfsk(); gfsk != nil {
|
|
return radioParametersSnapshotV3{
|
|
mode: "gfsk",
|
|
gfskBitrateBps: gfsk.GetBitrateBps(),
|
|
gfskFrequencyMHz: float64(gfsk.GetFrequencyMhz()),
|
|
gfskDeviationHz: gfsk.GetFrequencyDeviationHz(),
|
|
gfskRxBandwidthHz: gfsk.GetRxBandwidthHz(),
|
|
gfskPayloadLength: gfsk.GetPayloadLength(),
|
|
}
|
|
}
|
|
return radioParametersSnapshotV3{}
|
|
}
|
|
|
|
func chooseFrequencyMHz(radio radioParametersSnapshotV3) float64 {
|
|
if radio.mode == "gfsk" {
|
|
return radio.gfskFrequencyMHz
|
|
}
|
|
return radio.loraFrequencyMHz
|
|
}
|
|
|
|
func parseGatewayMACFromTopic(topic string) string {
|
|
parts := strings.Split(topic, "/")
|
|
for i := 0; i < len(parts)-1; i++ {
|
|
if parts[i] == "gateway" {
|
|
return formatTopicGatewayMAC(parts[i+1])
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
func formatTopicGatewayMAC(value string) string {
|
|
trimmed := strings.TrimSpace(value)
|
|
if len(trimmed) != 12 {
|
|
return strings.ToUpper(trimmed)
|
|
}
|
|
chunks := make([]string, 0, 6)
|
|
for i := 0; i < len(trimmed); i += 2 {
|
|
chunks = append(chunks, strings.ToUpper(trimmed[i:i+2]))
|
|
}
|
|
return strings.Join(chunks, ":")
|
|
}
|
|
|
|
func formatIPv4(value uint32) string {
|
|
if value == 0 {
|
|
return ""
|
|
}
|
|
return fmt.Sprintf("%d.%d.%d.%d", byte(value>>24), byte(value>>16), byte(value>>8), byte(value))
|
|
}
|