如何在Go语言中通过Chirpstack API获取设备上行数据？

问题背景 #

需要使用Go语言通过本地Chirpstack的gRPC API获取LoRaWAN设备的上行数据，但官方仅提供了下行队列操作的示例代码，查阅API文档后仍无法实现需求。

实现方案 #

Chirpstack提供两种获取上行数据的方式：查询历史上行消息和实时监听上行事件，以下是具体实现代码：

1. 查询历史上行消息 #

使用DeviceService的ListUplinkMessages方法，可获取指定设备的历史上行数据，支持分页、时间范围过滤等参数。

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
	"github.com/brocaar/chirpstack-api/go/v3/as/external/api"
	"github.com/brocaar/lorawan"
)

// 配置项
var (
	server   = "localhost:8080" // Chirpstack API地址
	devEUI   = lorawan.EUI64{0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01} // 目标设备DevEUI
	apiToken = "YOUR_API_TOKEN" // 从Chirpstack web界面获取的API Token
)

// APIToken 实现gRPC认证接口
type APIToken string

func (a APIToken) GetRequestMetadata(ctx context.Context, url ...string) (map[string]string, error) {
	return map[string]string{
		"authorization": fmt.Sprintf("Bearer %s", a),
	}, nil
}

func (a APIToken) RequireTransportSecurity() bool {
	return false // 使用TLS时改为true
}

func main() {
	// 构建gRPC拨号选项
	dialOpts := []grpc.DialOption{
		grpc.WithBlock(),
		grpc.WithPerRPCCredentials(APIToken(apiToken)),
		grpc.WithInsecure(), // 启用TLS时移除该选项
	}

	// 连接Chirpstack gRPC服务器
	conn, err := grpc.Dial(server, dialOpts...)
	if err != nil {
		panic(fmt.Sprintf("连接失败: %v", err))
	}
	defer conn.Close()

	// 创建DeviceService客户端
	deviceClient := api.NewDeviceServiceClient(conn)

	// 构造查询请求
	req := &api.ListDeviceUplinkMessagesRequest{
		DevEui: devEUI.String(),
		Limit:  10, // 单次查询返回的最大条目数
		// 可选：设置时间范围过滤，比如只查最近24小时的数据
		// StartTime: &timestamppb.Timestamp{Seconds: time.Now().Unix() - 86400},
		// EndTime:   &timestamppb.Timestamp{Seconds: time.Now().Unix()},
	}

	// 调用API获取上行消息
	resp, err := deviceClient.ListUplinkMessages(context.Background(), req)
	if err != nil {
		panic(fmt.Sprintf("查询上行消息失败: %v", err))
	}

	// 遍历输出上行数据
	fmt.Printf("共查询到%d条上行消息:\n", len(resp.Result))
	for idx, msg := range resp.Result {
		fmt.Printf("第%d条:\n", idx+1)
		fmt.Printf("  接收时间: %s\n", msg.RxInfo.Time.AsTime().Format("2006-01-02 15:04:05"))
		fmt.Printf("  FPort: %d\n", msg.FPort)
		fmt.Printf("  上行数据(十六进制): %x\n", msg.Data)
		fmt.Printf("  RSSI: %d dBm\n", msg.RxInfo.Rssi)
		fmt.Printf("  SNR: %.2f dB\n", msg.RxInfo.Snr)
		fmt.Println("------------------------")
	}
}

2. 实时监听上行事件 #

使用EventService的Stream方法，可实时接收设备的上行消息、激活事件等通知，适合需要实时处理数据的场景。

package main

import (
	"context"
	"fmt"
	"google.golang.org/grpc"
	"github.com/brocaar/chirpstack-api/go/v3/as/external/api"
)

// 配置项同历史查询示例
var (
	server   = "localhost:8080"
	apiToken = "YOUR_API_TOKEN"
)

type APIToken string

func (a APIToken) GetRequestMetadata(ctx context.Context, url ...string) (map[string]string, error) {
	return map[string]string{
		"authorization": fmt.Sprintf("Bearer %s", a),
	}, nil
}

func (a APIToken) RequireTransportSecurity() bool {
	return false
}

func main() {
	dialOpts := []grpc.DialOption{
		grpc.WithBlock(),
		grpc.WithPerRPCCredentials(APIToken(apiToken)),
		grpc.WithInsecure(),
	}

	conn, err := grpc.Dial(server, dialOpts...)
	if err != nil {
		panic(fmt.Sprintf("连接失败: %v", err))
	}
	defer conn.Close()

	// 创建EventService客户端
	eventClient := api.NewEventServiceClient(conn)

	// 构造流请求，可通过Topic过滤事件类型（比如只监听上行消息）
	req := &api.StreamEventsRequest{
		Topics: []string{"up"}, // "up"表示上行消息，其他可选值如"join"（激活事件）等
	}

	// 建立事件流
	stream, err := eventClient.Stream(context.Background(), req)
	if err != nil {
		panic(fmt.Sprintf("建立事件流失败: %v", err))
	}

	fmt.Println("开始监听上行事件...")
	for {
		event, err := stream.Recv()
		if err != nil {
			panic(fmt.Sprintf("接收事件失败: %v", err))
		}

		// 处理上行消息事件
		if event.GetUp() != nil {
			upEvent := event.GetUp()
			fmt.Printf("收到上行消息:\n")
			fmt.Printf("  DevEUI: %s\n", upEvent.DevEui)
			fmt.Printf("  接收时间: %s\n", upEvent.RxInfo.Time.AsTime().Format("2006-01-02 15:04:05"))
			fmt.Printf("  FPort: %d\n", upEvent.FPort)
			fmt.Printf("  上行数据(十六进制): %x\n", upEvent.Data)
			fmt.Printf("  RSSI: %d dBm\n", upEvent.RxInfo.Rssi)
			fmt.Printf("  SNR: %.2f dB\n", upEvent.RxInfo.Snr)
			fmt.Println("------------------------")
		}
	}
}

关键说明 #

• API Token获取：登录Chirpstack web界面，进入Applications -> 目标应用 -> API keys页面生成并复制Token。
• 参数调整：历史查询时可通过Limit、StartTime、EndTime等参数控制返回结果；实时监听时可修改Topics字段过滤需要的事件类型。
• TLS配置：如果Chirpstack启用了TLS，需移除grpc.WithInsecure()选项，并添加TLS相关拨号配置。

内容的提问来源于stack exchange，提问作者shaoC