提供lora公共逻辑

2025-12-01 14:32:50 +08:00
parent 5113e5953a
commit d430307b48
7 changed files with 484 additions and 583 deletions
--- a/internal/infra/transport/lora/lora_mesh_uart_passthrough_transport.go
+++ b/internal/infra/transport/lora/lora_mesh_uart_passthrough_transport.go
@@ -4,25 +4,20 @@ import (
 	"bytes"
 	"context"
 	"encoding/binary"
-	"encoding/json"
 	"fmt"
 	"io"
-	"math"
 	"strconv"
 	"sync"
 	"time"

 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/config"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs"
-	"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
-	"git.huangwc.com/pig/pig-farm-controller/internal/infra/repository"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport"
 	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto"

 	"github.com/google/uuid"
 	"github.com/tarm/serial"
 	gproto "google.golang.org/protobuf/proto"
-	"gorm.io/datatypes"
 )

 // transportState 定义了传输层的内部状态
@@ -43,9 +38,10 @@ type message struct {

 // LoRaMeshUartPassthroughTransport 实现了 transport.Communicator 和 transport.Listener 接口
 type LoRaMeshUartPassthroughTransport struct {
-	ctx    context.Context
-	config config.LoraMeshConfig
-	port   *serial.Port
+	selfCtx context.Context
+	config  config.LoraMeshConfig
+	port    *serial.Port
+	handler transport.UpstreamHandler // 依赖注入的统一业务处理器

 	mu    sync.Mutex // 用于保护对外的公共方法（如Send）的并发调用
 	state transportState
@@ -59,12 +55,6 @@ type LoRaMeshUartPassthroughTransport struct {
 	currentRecvSource   uint16                       // 当前正在接收的源地址
 	reassemblyTimeout   *time.Timer                  // 分片重组的超时定时器
 	reassemblyTimeoutCh chan uint16                  // 当超时触发时，用于传递源地址
-
-	// --- 依赖注入的仓库 ---
-	areaControllerRepo    repository.AreaControllerRepository
-	pendingCollectionRepo repository.PendingCollectionRepository
-	deviceRepo            repository.DeviceRepository
-	sensorDataRepo        repository.SensorDataRepository
 }

 // sendRequest 封装了一次发送请求
@@ -91,10 +81,7 @@ type reassemblyBuffer struct {
 func NewLoRaMeshUartPassthroughTransport(
 	ctx context.Context,
 	config config.LoraMeshConfig,
-	areaControllerRepo repository.AreaControllerRepository,
-	pendingCollectionRepo repository.PendingCollectionRepository,
-	deviceRepo repository.DeviceRepository,
-	sensorDataRepo repository.SensorDataRepository,
+	handler transport.UpstreamHandler,
 ) (*LoRaMeshUartPassthroughTransport, error) {
 	c := &serial.Config{
 		Name:        config.UARTPort,
@@ -108,20 +95,15 @@ func NewLoRaMeshUartPassthroughTransport(
 	}

 	t := &LoRaMeshUartPassthroughTransport{
-		ctx:                 ctx,
+		selfCtx:             logs.AddCompName(ctx, "LoRaMeshUartPassthroughTransport"),
 		config:              config,
 		port:                port,
+		handler:             handler,
 		state:               stateIdle,
 		stopChan:            make(chan struct{}),
 		sendChan:            make(chan *sendRequest),
 		reassemblyBuffers:   make(map[uint16]*reassemblyBuffer),
 		reassemblyTimeoutCh: make(chan uint16, 1),
-
-		// 注入依赖
-		areaControllerRepo:    areaControllerRepo,
-		pendingCollectionRepo: pendingCollectionRepo,
-		deviceRepo:            deviceRepo,
-		sensorDataRepo:        sensorDataRepo,
 	}

 	return t, nil
@@ -129,10 +111,11 @@ func NewLoRaMeshUartPassthroughTransport(

 // Listen 启动后台监听协程（非阻塞）
 func (t *LoRaMeshUartPassthroughTransport) Listen(ctx context.Context) error {
-	loraCtx, logger := logs.Trace(ctx, t.ctx, "Listen")
+	// 注意：这里的 loraCtx 是从 selfCtx 派生的，因为它代表了这个组件自身的生命周期
+	loraCtx, logger := logs.Trace(ctx, t.selfCtx, "Listen")
 	t.wg.Add(1)
 	go t.workerLoop(loraCtx)
-	logger.Info("LoRa传输层工作协程已启动")
+	logger.Info("LoRa Mesh 传输层工作协程已启动")
 	return nil
 }

@@ -167,7 +150,7 @@ func (t *LoRaMeshUartPassthroughTransport) Stop(ctx context.Context) error {

 // workerLoop 是核心的状态机和调度器
 func (t *LoRaMeshUartPassthroughTransport) workerLoop(ctx context.Context) {
-	loraCtx, logger := logs.Trace(ctx, t.ctx, "workerLoop")
+	loraCtx, logger := logs.Trace(ctx, t.selfCtx, "workerLoop")

 	defer t.wg.Done()

@@ -218,7 +201,7 @@ func (t *LoRaMeshUartPassthroughTransport) workerLoop(ctx context.Context) {

 // runIdleState 处理空闲状态下的逻辑，主要是检查并启动发送任务
 func (t *LoRaMeshUartPassthroughTransport) runIdleState(ctx context.Context) {
-	loraCtx := logs.AddFuncName(ctx, t.ctx, "Listen")
+	loraCtx, _ := logs.Trace(ctx, t.selfCtx, "runIdleState")

 	select {
 	case req := <-t.sendChan:
@@ -234,10 +217,10 @@ func (t *LoRaMeshUartPassthroughTransport) runIdleState(ctx context.Context) {

 // runReceivingState 处理接收状态下的逻辑，主要是检查超时
 func (t *LoRaMeshUartPassthroughTransport) runReceivingState(ctx context.Context) {
-	logger := logs.TraceLogger(ctx, t.ctx, "runReceivingState")
+	_, logger := logs.Trace(ctx, t.selfCtx, "runReceivingState")
 	select {
 	case sourceAddr := <-t.reassemblyTimeoutCh:
-		logger.Warnf("接收来自 0x%04X 的消息超时", sourceAddr)
+		logger.Warnw("接收消息超时", "sourceAddr", fmt.Sprintf("0x%04X", sourceAddr))
 		delete(t.reassemblyBuffers, sourceAddr)
 		t.state = stateIdle
 	default:
@@ -247,7 +230,7 @@ func (t *LoRaMeshUartPassthroughTransport) runReceivingState(ctx context.Context

 // executeSend 执行完整的发送流程（分片、构建、写入）
 func (t *LoRaMeshUartPassthroughTransport) executeSend(ctx context.Context, req *sendRequest) (*transport.SendResult, error) {
-	logger := logs.TraceLogger(ctx, t.ctx, "executeSend")
+	_, logger := logs.Trace(ctx, t.selfCtx, "executeSend")
 	chunks := splitPayload(req.payload, t.config.MaxChunkSize)
 	totalChunks := uint8(len(chunks))

@@ -266,7 +249,7 @@ func (t *LoRaMeshUartPassthroughTransport) executeSend(ctx context.Context, req
 		frame.WriteByte(currentChunk)                           // 当前包序号
 		frame.Write(chunk)                                      // 数据块

-		logger.Debugf("构建LoRa数据包: %v", frame.Bytes())
+		logger.Debugw("构建LoRa数据包", "bytes", frame.Bytes())
 		_, err := t.port.Write(frame.Bytes())
 		if err != nil {
 			return nil, fmt.Errorf("写入串口失败: %w", err)
@@ -282,9 +265,9 @@ func (t *LoRaMeshUartPassthroughTransport) executeSend(ctx context.Context, req

 // handleFrame 处理一个从串口解析出的完整物理帧
 func (t *LoRaMeshUartPassthroughTransport) handleFrame(ctx context.Context, frame []byte) {
-	loraCtx, logger := logs.Trace(ctx, t.ctx, "handleFrame")
+	reqCtx, logger := logs.Trace(ctx, t.selfCtx, "handleFrame")
 	if len(frame) < 8 {
-		logger.Warnf("收到了一个无效长度的帧: %d", len(frame))
+		logger.Warnw("收到了一个无效长度的帧", "length", len(frame))
 		return
 	}

@@ -301,7 +284,9 @@ func (t *LoRaMeshUartPassthroughTransport) handleFrame(ctx context.Context, fram
 			DestAddr:   fmt.Sprintf("%04X", destAddr),
 			Payload:    chunkData,
 		}
-		go t.handleUpstreamMessage(loraCtx, msg)
+		// 使用分离的上下文进行异步处理
+		detachedCtx := logs.DetachContext(reqCtx)
+		go t.handleUpstreamMessage(detachedCtx, msg)
 		return
 	}

@@ -326,18 +311,21 @@ func (t *LoRaMeshUartPassthroughTransport) handleFrame(ctx context.Context, fram
 				t.reassemblyTimeoutCh <- sourceAddr
 			})
 		} else {
-			logger.Warnf("在空闲状态下收到了一个来自 0x%04X 的非首包分片，已忽略。", sourceAddr)
+			logger.Warnw("在空闲状态下收到了一个非首包分片，已忽略", "sourceAddr", fmt.Sprintf("0x%04X", sourceAddr))
 		}

 	case stateReceiving:
 		if sourceAddr != t.currentRecvSource {
-			logger.Warnf("正在接收来自 0x%04X 的数据时，收到了另一个源 0x%04X 的分片，已忽略。", t.currentRecvSource, sourceAddr)
+			logger.Warnw("正在接收数据时，收到了另一个源的分片，已忽略",
+				"currentSource", fmt.Sprintf("0x%04X", t.currentRecvSource),
+				"newSource", fmt.Sprintf("0x%04X", sourceAddr),
+			)
 			return
 		}

 		buffer, ok := t.reassemblyBuffers[sourceAddr]
 		if !ok {
-			logger.Errorf("内部错误: 处于接收状态，但没有为 0x%04X 找到缓冲区", sourceAddr)
+			logger.Errorw("内部错误: 处于接收状态，但没有找到缓冲区", "sourceAddr", fmt.Sprintf("0x%04X", sourceAddr))
 			t.state = stateIdle // 重置状态
 			return
 		}
@@ -362,165 +350,43 @@ func (t *LoRaMeshUartPassthroughTransport) handleFrame(ctx context.Context, fram
 				DestAddr:   fmt.Sprintf("%04X", destAddr),
 				Payload:    fullPayload.Bytes(),
 			}
-			go t.handleUpstreamMessage(loraCtx, msg)
+			// 使用分离的上下文进行异步处理
+			detachedCtx := logs.DetachContext(reqCtx)
+			go t.handleUpstreamMessage(detachedCtx, msg)

 			// 清理并返回空闲状态
 			delete(t.reassemblyBuffers, sourceAddr)
 			t.state = stateIdle
 		}
 	default:
-		logger.Errorf("内部错误: 状态机处于未知状态 %d", t.state)
+		logger.Errorw("内部错误: 状态机处于未知状态", "state", t.state)
 	}
 }

 // handleUpstreamMessage 在独立的协程中处理单个上行的、完整的消息。
+// 【已重构】此方法现在只负责解析和委托，不包含任何业务逻辑。
 func (t *LoRaMeshUartPassthroughTransport) handleUpstreamMessage(ctx context.Context, msg *message) {
-	loraCtx, logger := logs.Trace(ctx, t.ctx, "handleUpstreamMessage")
-
-	logger.Infof("开始处理来自 %s 的上行消息", msg.SourceAddr)
+	reqCtx, logger := logs.Trace(ctx, t.selfCtx, "handleUpstreamMessage")
+	logger.Infow("开始适配上行消息并委托", "sourceAddr", msg.SourceAddr)

 	// 1. 解析外层 "信封"
 	var instruction proto.Instruction
 	if err := gproto.Unmarshal(msg.Payload, &instruction); err != nil {
-		logger.Errorf("解析上行 Instruction Protobuf 失败: %v, 源地址: %s, 原始数据: %x", err, msg.SourceAddr, msg.Payload)
+		logger.Errorw("解析上行 Instruction Protobuf 失败",
+			"sourceAddr", msg.SourceAddr,
+			"error", err,
+			"rawData", fmt.Sprintf("%x", msg.Payload),
+		)
 		return
 	}

-	// 2. 使用 type switch 从 oneof payload 中提取 CollectResult
-	var collectResp *proto.CollectResult
-	switch p := instruction.GetPayload().(type) {
-	case *proto.Instruction_CollectResult:
-		collectResp = p.CollectResult
-	default:
-		// 如果上行的数据不是采集结果，记录日志并忽略
-		logger.Infof("收到一个非采集响应的上行指令 (类型: %T)，无需处理。源地址: %s", p, msg.SourceAddr)
-		return
-	}
-
-	if collectResp == nil {
-		logger.Errorf("从 Instruction 中提取的 CollectResult 为 nil。源地址: %s", msg.SourceAddr)
-		return
-	}
-
-	correlationID := collectResp.CorrelationId
-	logger.Infof("成功解析采集响应 (CorrelationID: %s)，包含 %d 个值。", correlationID, len(collectResp.Values))
-
-	// 3. 查找区域主控 (注意：LoRa Mesh 的 SourceAddr 对应于区域主控的 NetworkID)
-	areaController, err := t.areaControllerRepo.FindByNetworkID(loraCtx, msg.SourceAddr)
-	if err != nil {
-		logger.Errorf("处理上行消息失败：无法通过源地址 '%s' 找到区域主控设备: %v", msg.SourceAddr, err)
-		return
-	}
-	if err := areaController.SelfCheck(); err != nil {
-		logger.Errorf("处理上行消息失败：区域主控 %v(ID: %d) 未通过自检: %v", areaController.Name, areaController.ID, err)
-		return
-	}
-
-	// 4. 根据 CorrelationID 查找待处理请求
-	pendingReq, err := t.pendingCollectionRepo.FindByCorrelationID(loraCtx, correlationID)
-	if err != nil {
-		logger.Errorf("处理采集响应失败：无法找到待处理请求 (CorrelationID: %s): %v", correlationID, err)
-		return
-	}
-
-	// 检查状态，防止重复处理
-	if pendingReq.Status != models.PendingStatusPending && pendingReq.Status != models.PendingStatusTimedOut {
-		logger.Warnf("收到一个已处理过的采集响应 (CorrelationID: %s, Status: %s)，将忽略。", correlationID, pendingReq.Status)
-		return
-	}
-
-	// 5. 匹配数据并存入数据库
-	deviceIDs := pendingReq.CommandMetadata
-	values := collectResp.Values
-	if len(deviceIDs) != len(values) {
-		logger.Errorf("数据不匹配：下行指令要求采集 %d 个设备，但上行响应包含 %d 个值 (CorrelationID: %s)", len(deviceIDs), len(values), correlationID)
-		err = t.pendingCollectionRepo.UpdateStatusToFulfilled(loraCtx, correlationID, time.Now())
-		if err != nil {
-			logger.Errorf("处理采集响应失败：无法更新待处理请求 (CorrelationID: %s) 的状态为完成: %v", correlationID, err)
-		}
-		return
-	}
-
-	for i, deviceID := range deviceIDs {
-		rawSensorValue := values[i]
-
-		if math.IsNaN(float64(rawSensorValue)) {
-			logger.Warnf("设备 (ID: %d) 上报了一个无效的 NaN 值，已跳过当前值的记录。", deviceID)
-			continue
-		}
-
-		dev, err := t.deviceRepo.FindByID(loraCtx, deviceID)
-		if err != nil {
-			logger.Errorf("处理采集数据失败：无法找到设备 (ID: %d): %v", deviceID, err)
-			continue
-		}
-		if err := dev.SelfCheck(); err != nil {
-			logger.Warnf("跳过设备 %d，因其未通过自检: %v", dev.ID, err)
-			continue
-		}
-		if err := dev.DeviceTemplate.SelfCheck(); err != nil {
-			logger.Warnf("跳过设备 %d，因其设备模板未通过自检: %v", dev.ID, err)
-			continue
-		}
-
-		var valueDescriptors []*models.ValueDescriptor
-		if err := dev.DeviceTemplate.ParseValues(&valueDescriptors); err != nil {
-			logger.Warnf("跳过设备 %d，因其设备模板的 Values 属性解析失败: %v", dev.ID, err)
-			continue
-		}
-		if len(valueDescriptors) == 0 {
-			logger.Warnf("跳过设备 %d，因其设备模板缺少 ValueDescriptor 定义", dev.ID)
-			continue
-		}
-		valueDescriptor := valueDescriptors[0]
-
-		parsedValue := rawSensorValue*valueDescriptor.Multiplier + valueDescriptor.Offset
-
-		var dataToRecord interface{}
-		switch valueDescriptor.Type {
-		case models.SensorTypeTemperature:
-			dataToRecord = models.TemperatureData{TemperatureCelsius: parsedValue}
-		case models.SensorTypeHumidity:
-			dataToRecord = models.HumidityData{HumidityPercent: parsedValue}
-		case models.SensorTypeWeight:
-			dataToRecord = models.WeightData{WeightKilograms: parsedValue}
-		default:
-			logger.Warnf("未知的传感器类型 '%s'，将使用通用格式记录", valueDescriptor.Type)
-			dataToRecord = map[string]float32{"value": parsedValue}
-		}
-
-		t.recordSensorData(loraCtx, areaController.ID, dev.ID, time.Now(), valueDescriptor.Type, dataToRecord)
-		logger.Infof("成功记录传感器数据: 设备ID=%d, 类型=%s, 原始值=%f, 解析值=%.2f", dev.ID, valueDescriptor.Type, rawSensorValue, parsedValue)
-	}
-
-	// 6. 更新请求状态为“已完成”
-	if err := t.pendingCollectionRepo.UpdateStatusToFulfilled(loraCtx, correlationID, time.Now()); err != nil {
-		logger.Errorf("更新待采集请求状态为 'fulfilled' 失败 (CorrelationID: %s): %v", correlationID, err)
-	} else {
-		logger.Infof("成功完成并关闭采集请求 (CorrelationID: %s)", correlationID)
-	}
-}
-
-// recordSensorData 是一个通用方法，用于将传感器数据存入数据库。
-func (t *LoRaMeshUartPassthroughTransport) recordSensorData(ctx context.Context, areaControllerID uint32, sensorDeviceID uint32, eventTime time.Time, sensorType models.SensorType, data interface{}) {
-	loraCtx, logger := logs.Trace(ctx, t.ctx, "recordSensorData")
-
-	jsonData, err := json.Marshal(data)
-	if err != nil {
-		logger.Errorf("记录传感器数据失败：序列化数据为 JSON 时出错: %v", err)
-		return
-	}
-
-	sensorData := &models.SensorData{
-		Time:             eventTime,
-		DeviceID:         sensorDeviceID,
-		AreaControllerID: areaControllerID,
-		SensorType:       sensorType,
-		Data:             datatypes.JSON(jsonData),
-	}
-
-	if err := t.sensorDataRepo.Create(loraCtx, sensorData); err != nil {
-		logger.Errorf("记录传感器数据失败：存入数据库时出错: %v", err)
+	// 2. 委托给统一处理器
+	// 注意：对于 LoRa Mesh，目前只处理业务指令，没有单独的状态或ACK事件。
+	if err := t.handler.HandleInstruction(reqCtx, msg.SourceAddr, &instruction); err != nil {
+		logger.Errorw("委托上行指令给统一处理器失败",
+			"sourceAddr", msg.SourceAddr,
+			"error", err,
+		)
 	}
 }

--- a/internal/infra/transport/transport.go
+++ b/internal/infra/transport/transport.go
@@ -3,6 +3,8 @@ package transport
 import (
 	"context"
 	"time"
+
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/transport/proto"
 )

 // Communicator 用于其他设备通信
@@ -35,3 +37,17 @@ type Listener interface {
 	// Stop 用于停止监听
 	Stop(ctx context.Context) error
 }
+
+// UpstreamHandler 定义了处理所有来源的上行数据的统一协约。
+// 任何实现了上行消息监听的基础设施（如串口、MQTT客户端），都应该在收到消息后调用此接口的实现者。
+// 这样，基础设施层只负责“接收和解析”，而将“业务处理”的控制权交给了上层。
+type UpstreamHandler interface {
+	// HandleInstruction 处理来自设备的、已解析为Instruction的业务指令。
+	HandleInstruction(ctx context.Context, sourceAddr string, instruction *proto.Instruction) error
+
+	// HandleStatus 处理非业务指令的设备状态更新，例如信号强度、电量等。
+	HandleStatus(ctx context.Context, sourceAddr string, status map[string]interface{}) error
+
+	// HandleAck 处理对下行指令的确认（ACK）事件。
+	HandleAck(ctx context.Context, sourceAddr string, deduplicationID string, acknowledged bool, eventTime time.Time) error
+}