实现 GeneralDeviceService

internal/app/service/device/device_service.go

@@ -6,15 +6,15 @@ import "git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
 type DeviceAction string
 
 var (
-	DeviceActionStart = "start" // 启动
-	DeviceActionStop  = "stop"  // 停止
+	DeviceActionStart DeviceAction = "start" // 启动
+	DeviceActionStop  DeviceAction = "stop"  // 停止
 )
 
 // 指令类型
 type Method string
 
 var (
-	MethodPing = "ping" // ping指令
+	MethodSwitch Method = "switch" // 启停指令
 )
 
 // Service 抽象了一组方法用于控制设备行为

internal/app/service/device/general_device_service.go

@@ -1,20 +1,103 @@
 package device
 
 import (
+	"fmt"
+	"strconv"
+
+	"git.huangwc.com/pig/pig-farm-controller/internal/app/service/device/proto"
+	"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"
+	gproto "google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/anypb"
 )
 
 type GeneralDeviceService struct {
+	deviceRepo repository.DeviceRepository
+	logger     *logs.Logger
+
+	deviceID uint // 区域主控的设备ID
+
+	// regionalController 是执行命令的区域主控, 所有的指令都会发往区域主控
+	regionalController *models.Device
+
 	comm transport.Communicator
 }
 
-func (g *GeneralDeviceService) Switch(device models.Device, action DeviceAction) error {
-	//TODO implement me
-	panic("implement me")
+// NewGeneralDeviceService 创建一个通用设备服务
+func NewGeneralDeviceService(deviceID uint, deviceRepo repository.DeviceRepository, logger *logs.Logger, comm transport.Communicator) *GeneralDeviceService {
+	return &GeneralDeviceService{
+		deviceID:   deviceID,
+		deviceRepo: deviceRepo,
+		logger:     logger,
+		comm:       comm,
+	}
 }
 
-func (g *GeneralDeviceService) Ping() error {
-	//TODO implement me
-	panic("implement me")
+func (g *GeneralDeviceService) Switch(device models.Device, action DeviceAction) error {
+
+	// 校验设备参数及生成指令
+	if *device.ParentID != g.deviceID {
+		return fmt.Errorf("设备 %v(id=%v) 的上级区域主控是(id=%v), 不是当前区域主控(id=%v)下属设备, 无法执行指令", device.Name, device.ID, device.ParentID, g.deviceID)
+	}
+
+	if !device.SelfCheck() {
+		return fmt.Errorf("设备 %v(id=%v) 缺少必要信息, 无法发送指令", device.Name, device.ID)
+	}
+
+	deviceInfo := make(map[string]interface{})
+	if err := device.ParseProperties(&deviceInfo); err != nil {
+		return fmt.Errorf("解析设备 %v(id=%v) 配置失败: %v", device.Name, device.ID, err)
+	}
+
+	busNumber, err := strconv.Atoi(fmt.Sprintf("%v", deviceInfo[models.BusNumber]))
+	if err != nil {
+		return fmt.Errorf("无效的总线号: %v", err)
+	}
+	busAddress, err := strconv.Atoi(fmt.Sprintf("%v", deviceInfo[models.BusAddress]))
+	if err != nil {
+		return fmt.Errorf("无效的总线地址: %v", err)
+	}
+	relayChannel, err := strconv.Atoi(fmt.Sprintf("%v", deviceInfo[models.RelayChannel]))
+	if err != nil {
+		return fmt.Errorf("无效的继电器通道: %v", err)
+	}
+
+	data, err := anypb.New(&proto.Switch{
+		DeviceAction: string(action),
+		BusNumber:    int32(busNumber),
+		BusAddress:   int32(busAddress),
+		RelayChannel: int32(relayChannel),
+	})
+	if err != nil {
+		return fmt.Errorf("创建指令失败: %v", err)
+	}
+
+	instruction := &proto.Instruction{
+		Method: proto.MethodType_SWITCH,
+		Data:   data,
+	}
+
+	// 获取自身LoRa设备ID, 因为可能变更, 所以每次都现获取
+	thisDevice, err := g.deviceRepo.FindByID(g.deviceID)
+	if err != nil {
+		return fmt.Errorf("获取区域主控(id=%v)信息失败: %v", g.deviceID, err)
+	}
+	if !thisDevice.SelfCheck() {
+		return fmt.Errorf("区域主控 %v(id=%v) 缺少必要信息, 无法发送指令", thisDevice.Name, thisDevice.ID)
+	}
+	thisDeviceinfo := make(map[string]interface{})
+	if err := thisDevice.ParseProperties(&thisDeviceinfo); err != nil {
+		return fmt.Errorf("解析区域主控 %v(id=%v) 配置失败: %v", device.Name, device.ID, err)
+	}
+	loraAddress := fmt.Sprintf("%v", thisDeviceinfo[models.LoRaAddress])
+
+	// 生成消息并发送
+	message, err := gproto.Marshal(instruction)
+	if err != nil {
+		return fmt.Errorf("序列化指令失败: %v", err)
+	}
+	return g.comm.Send(loraAddress, message)
+
 }

internal/infra/models/device.go

@@ -39,6 +39,18 @@ const (
 	SubTypeWaterCurtain DeviceSubType = "water_curtain"
 )
 
+// 设备属性名大全
+var (
+
+	// 普通开关式设备
+	BusNumber    = "bus_number"    // 总线号
+	BusAddress   = "bus_address"   // 总线地址
+	RelayChannel = "relay_channel" // 继电器通道号
+
+	// 区域主控
+	LoRaAddress = "lora_address" // 区域主控 LoRa 地址, 如果使用LoRa网关也可能是LoRa网关记录的设备ID
+)
+
 // --- Properties 结构体定义 ---
 
 // LoraProperties 定义了区域主控的特有属性
@@ -95,3 +107,31 @@ func (d *Device) ParseProperties(v interface{}) error {
 	}
 	return json.Unmarshal(d.Properties, v)
 }
+
+// SelfCheck 进行参数自检, 返回检测结果
+// 方法会根据自身类型进行参数检查, 参数不全时返回false
+// TODO 没写单测
+func (d *Device) SelfCheck() bool {
+
+	properties := make(map[string]interface{})
+	if err := d.ParseProperties(&properties); err != nil {
+		return false
+	}
+
+	has := func(key string) bool {
+		_, ok := properties[key]
+		return ok
+	}
+
+	switch d.SubType {
+	case SubTypeFan:
+		if !has(BusNumber) || !has(BusAddress) || !has(RelayChannel) {
+			return false
+		}
+	default:
+		// 不应该有类型未知的设备
+		return false
+	}
+
+	return true
+}

internal/infra/transport/lora/chirp_stack.go

@@ -56,7 +56,7 @@ func NewChirpStackTransport(config ChirpStackConfig, logger *logs.Logger) *Chirp
 	}
 }
 
-func (c *ChirpStackTransport) Send(deviceID string, payload []byte) error {
+func (c *ChirpStackTransport) Send(address string, payload []byte) error {
 	// 1. 构建 API 请求体。
 	//    - Confirmed: true 表示确认消息, 设为false将不保证消息送达(但可以节约下行容量)。
 	//    - Data: 经过 Base64 编码的数据。
@@ -75,17 +75,17 @@ func (c *ChirpStackTransport) Send(deviceID string, payload []byte) error {
 	//    - WithBody 设置请求体。
 	params := device_service.NewDeviceServiceEnqueueParams().
 		WithTimeout(10 * time.Second).
-		WithQueueItemDevEui(deviceID).
+		WithQueueItemDevEui(address).
 		WithBody(body)
 
 	// 3. 调用生成的客户端方法来发送请求。
 	//    c.authInfo 是您在 NewChirpStackTransport 中创建的认证信息。
 	_, err := c.client.DeviceService.DeviceServiceEnqueue(params, c.authInfo)
 	if err != nil {
-		c.logger.Errorf("设备 %s 调用ChirpStack Enqueue失败: %v", deviceID, err)
+		c.logger.Errorf("设备 %s 调用ChirpStack Enqueue失败: %v", address, err)
 		return err
 	}
 
-	c.logger.Infof("设备 %s 调用ChirpStack Enqueue成功", deviceID)
+	c.logger.Infof("设备 %s 调用ChirpStack Enqueue成功", address)
 	return nil
 }

internal/infra/transport/transport.go

@@ -3,5 +3,5 @@ package transport
 // Communicator 用于其他设备通信
 type Communicator interface {
 	// Send 用于发送一条单向数据(不等待回信)
-	Send(deviceID string, payload []byte) error
+	Send(address string, payload []byte) error
 }