pig/pig-farm-controller
1
0
Fork 0
Code Issues 10 Pull Requests Actions Packages Projects Releases Wiki Activity

issue_9 #14

Merged
huang merged 11 commits from issue_9 into main 2025-09-25 00:16:50 +08:00
Conversation 0 Commits 11 Files Changed 18 +284 -145
8 changed files with 284 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 21fb9c7e57 - Show all commits

2
internal/app/api/api.go
View File

@@ -53,6 +53,8 @@ func NewAPI(cfg config.ServerConfig,
userRepo repository.UserRepository, userRepo repository.UserRepository,
deviceRepository repository.DeviceRepository, deviceRepository repository.DeviceRepository,
planRepository repository.PlanRepository, planRepository repository.PlanRepository,
sensorDataRepository repository.SensorDataRepository,
executionLogRepository repository.ExecutionLogRepository,
tokenService token.TokenService, tokenService token.TokenService,
listenHandler transport.ListenHandler, listenHandler transport.ListenHandler,
analysisTaskManager *task.AnalysisPlanTaskManager) *API { analysisTaskManager *task.AnalysisPlanTaskManager) *API {

76
internal/app/service/transport/chirp_stack.go
View File

@@ -4,8 +4,12 @@ import (
"encoding/json" "encoding/json"
"io" "io"
"net/http" "net/http"
"time"
"git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "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"
"gorm.io/datatypes"
) )
// ChirpStackListener 主动发送的请求的event字段, 这个字段代表事件类型 // ChirpStackListener 主动发送的请求的event字段, 这个字段代表事件类型
@@ -23,11 +27,15 @@ const (
// ChirpStackListener 是一个监听器, 用于监听ChirpStack反馈的设备上行事件 // ChirpStackListener 是一个监听器, 用于监听ChirpStack反馈的设备上行事件
type ChirpStackListener struct { type ChirpStackListener struct {
logger *logs.Logger logger *logs.Logger
sensorDataRepo repository.SensorDataRepository
deviceRepo repository.DeviceRepository
} }
func NewChirpStackListener(logger *logs.Logger) *ChirpStackListener { func NewChirpStackListener(logger *logs.Logger, sensorDataRepo repository.SensorDataRepository, deviceRepo repository.DeviceRepository) *ChirpStackListener {
return &ChirpStackListener{ return &ChirpStackListener{
logger: logger, logger: logger,
sensorDataRepo: sensorDataRepo,
deviceRepo: deviceRepo,
} }
} }
@@ -129,13 +137,44 @@ func (c *ChirpStackListener) handler(data []byte, eventType string) {
// handleUpEvent 处理上行数据事件 // handleUpEvent 处理上行数据事件
func (c *ChirpStackListener) handleUpEvent(event *UpEvent) { func (c *ChirpStackListener) handleUpEvent(event *UpEvent) {
c.logger.Infof("处理 'up' 事件: %+v", event) c.logger.Infof("处理 'up' 事件: %+v", event)
// 记录信号强度
// 根据业务逻辑，一个猪场只有一个网关，所以 RxInfo 中通常只有一个元素，或者 gateway_id 都是相同的。
// 因此，我们只取第一个 RxInfo 中的信号数据即可。
if len(event.RxInfo) > 0 {
rx := event.RxInfo[0] // 取第一个接收到的网关信息
// 构建 SignalMetrics 结构体
signalMetrics := models.SignalMetrics{
RSSI: rx.Rssi,
SNR: rx.Snr,
}
c.recordSensorData(event.DeviceInfo.DevEui, event.Time, models.SensorDataTypeSignalMetrics, signalMetrics)
} else {
c.logger.Warnf("处理 'up' 事件时未找到 RxInfo，无法记录信号数据。DevEui: %s", event.DeviceInfo.DevEui)
}
// 在这里添加您的业务逻辑 // 在这里添加您的业务逻辑
} }
// handleStatusEvent 处理设备状态事件 // handleStatusEvent 处理设备状态事件
func (c *ChirpStackListener) handleStatusEvent(event *StatusEvent) { func (c *ChirpStackListener) handleStatusEvent(event *StatusEvent) {
c.logger.Infof("处接收到理 'status' 事件: %+v", event) c.logger.Infof("处接收到理 'status' 事件: %+v", event)
// 在这里添加您的业务逻辑
// 记录信号强度
signalMetrics := models.SignalMetrics{
Margin: event.Margin,
}
c.recordSensorData(event.DeviceInfo.DevEui, event.Time, models.SensorDataTypeSignalMetrics, signalMetrics)
// 记录 电量
batteryLevel := models.BatteryLevel{
BatteryLevel: event.BatteryLevel,
BatteryLevelUnavailable: event.BatteryLevelUnavailable,
ExternalPower: event.ExternalPower,
}
c.recordSensorData(event.DeviceInfo.DevEui, event.Time, models.SensorDataTypeBatteryLevel, batteryLevel)
} }
// handleAckEvent 处理下行确认事件 // handleAckEvent 处理下行确认事件
@@ -163,8 +202,6 @@ func (c *ChirpStackListener) handleLogEvent(event *LogEvent) {
c.logger.Warnf("ChirpStack 日志: [未知级别: %s] %s %s (DevEui: %s)", c.logger.Warnf("ChirpStack 日志: [未知级别: %s] %s %s (DevEui: %s)",
event.Level, event.Code, event.Description, event.DeviceInfo.DevEui) event.Level, event.Code, event.Description, event.DeviceInfo.DevEui)
} }
// 在这里添加您的业务逻辑
} }
// handleJoinEvent 处理入网事件 // handleJoinEvent 处理入网事件
@@ -190,3 +227,34 @@ func (c *ChirpStackListener) handleIntegrationEvent(event *IntegrationEvent) {
c.logger.Infof("接收到 'integration' 事件: %+v", event) c.logger.Infof("接收到 'integration' 事件: %+v", event)
// 在这里添加您的业务逻辑 // 在这里添加您的业务逻辑
} }
// recordSensorData 是一个通用方法，用于将传感器数据存入数据库。
func (c *ChirpStackListener) recordSensorData(devEui string, eventTime time.Time, dataType models.SensorDataType, data interface{}) {
// 1. 查找设备
device, err := c.deviceRepo.FindByDevEui(devEui)
if err != nil {
c.logger.Warnf("记录传感器数据失败：无法通过 DevEui '%s' 找到设备: %v", devEui, err)
return
}
// 2. 序列化数据结构体为 JSON
jsonData, err := json.Marshal(data)
if err != nil {
c.logger.Errorf("记录传感器数据失败：序列化数据为 JSON 时出错: %v", err)
return
}
// 3. 构建 SensorData 模型
sensorData := &models.SensorData{
Time: eventTime,
DeviceID: device.ID,
RegionalControllerID: *device.ParentID,
SensorDataType: dataType, // 设置传感器数据类型
Data: datatypes.JSON(jsonData),
}
// 4. 调用仓库创建记录
if err := c.sensorDataRepo.Create(sensorData); err != nil {
c.logger.Errorf("记录传感器数据失败：存入数据库时出错: %v", err)
}
}

214
internal/app/service/transport/chirp_stack_types.go
View File

@@ -10,189 +10,189 @@ import (
// DeviceInfo 包含了所有事件中通用的设备信息。 // DeviceInfo 包含了所有事件中通用的设备信息。
// 基于 aiserver.proto v4 (integration) // 基于 aiserver.proto v4 (integration)
type DeviceInfo struct { type DeviceInfo struct {
TenantID string `json:"tenantId"` TenantID string `json:"tenant_id"` // 租户ID
TenantName string `json:"tenantName"` TenantName string `json:"tenant_name"` // 租户名称
ApplicationID string `json:"applicationId"` ApplicationID string `json:"application_id"` // 应用ID
ApplicationName string `json:"applicationName"` ApplicationName string `json:"application_name"` // 应用名称
DeviceProfileID string `json:"deviceProfileId"` DeviceProfileID string `json:"device_profile_id"` // 设备配置文件ID
DeviceProfileName string `json:"deviceProfileName"` DeviceProfileName string `json:"device_profile_name"` // 设备配置文件名称
DeviceName string `json:"deviceName"` DeviceName string `json:"device_name"` // 设备名称
DevEui string `json:"devEui"` DevEui string `json:"dev_eui"` // 设备EUI (十六进制编码)
DeviceClassEnabled string `json:"deviceClassEnabled,omitempty"` // Class A, B, or C DeviceClassEnabled string `json:"device_class_enabled,omitempty"` // 设备启用的LoRaWAN类别 (A, B, 或 C)
Tags map[string]string `json:"tags"` Tags map[string]string `json:"tags"` // 用户定义的标签
} }
// Location 包含了地理位置信息。 // Location 包含了地理位置信息。
type Location struct { type Location struct {
Latitude float64 `json:"latitude"` Latitude float64 `json:"latitude"` // 纬度
Longitude float64 `json:"longitude"` Longitude float64 `json:"longitude"` // 经度
Altitude float64 `json:"altitude"` Altitude float64 `json:"altitude"` // 海拔
} }
// --- 可复用的子结构体 --- // --- 可复用的子结构体 ---
// UplinkRelayRxInfo 包含了上行中继接收信息。 // UplinkRelayRxInfo 包含了上行中继接收信息。
type UplinkRelayRxInfo struct { type UplinkRelayRxInfo struct {
DevEui string `json:"devEui"` DevEui string `json:"dev_eui"` // 中继设备的DevEUI
Frequency uint32 `json:"frequency"` Frequency uint32 `json:"frequency"` // 接收频率
Dr uint32 `json:"dr"` Dr uint32 `json:"dr"` // 数据速率
Snr int32 `json:"snr"` Snr int32 `json:"snr"` // 信噪比
Rssi int32 `json:"rssi"` Rssi int32 `json:"rssi"` // 接收信号强度指示
WorChannel uint32 `json:"worChannel"` WorChannel uint32 `json:"wor_channel"` // Work-on-Relay 通道
} }
// KeyEnvelope 包装了一个加密的密钥。 // KeyEnvelope 包装了一个加密的密钥。
// 基于 common.proto // 基于 common.proto
type KeyEnvelope struct { type KeyEnvelope struct {
KEKLabel string `json:"kekLabel,omitempty"` KEKLabel string `json:"kek_label,omitempty"` // 密钥加密密钥 (KEK) 标签
AESKey string `json:"aesKey,omitempty"` // Base64 编码的加密密钥 AESKey string `json:"aes_key,omitempty"` // Base64 编码的加密密钥
} }
// JoinServerContext 包含了 Join-Server 上下文。 // JoinServerContext 包含了 Join-Server 上下文。
// 基于 common.proto // 基于 common.proto
type JoinServerContext struct { type JoinServerContext struct {
SessionKeyID string `json:"sessionKeyId"` SessionKeyID string `json:"session_key_id"` // 会话密钥ID
AppSKey *KeyEnvelope `json:"appSKey,omitempty"` AppSKey *KeyEnvelope `json:"app_s_key,omitempty"` // 应用会话密钥
} }
// UplinkRxInfo 包含了上行接收信息。 // UplinkRxInfo 包含了上行接收信息。
type UplinkRxInfo struct { type UplinkRxInfo struct {
GatewayID string `json:"gatewayId"` GatewayID string `json:"gateway_id"` // 接收到上行数据的网关ID
UplinkID uint32 `json:"uplinkId"` UplinkID uint32 `json:"uplink_id"` // 上行ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 接收时间
Rssi int `json:"rssi"` Rssi int `json:"rssi"` // 接收信号强度指示
Snr float64 `json:"snr"` Snr float64 `json:"snr"` // 信噪比
Channel int `json:"channel"` Channel int `json:"channel"` // 接收通道
Location *Location `json:"location"` Location *Location `json:"location"` // 网关位置
Context string `json:"context"` Context string `json:"context"` // 上下文信息
Metadata map[string]string `json:"metadata"` Metadata map[string]string `json:"metadata"` // 元数据
} }
// LoraModulationInfo 包含了 LoRa 调制的具体参数。 // LoraModulationInfo 包含了 LoRa 调制的具体参数。
type LoraModulationInfo struct { type LoraModulationInfo struct {
Bandwidth int `json:"bandwidth"` Bandwidth int `json:"bandwidth"` // 带宽
SpreadingFactor int `json:"spreadingFactor"` SpreadingFactor int `json:"spreading_factor"` // 扩频因子
CodeRate string `json:"codeRate"` CodeRate string `json:"code_rate"` // 编码率
Polarization bool `json:"polarizationInvert,omitempty"` Polarization bool `json:"polarization_invert,omitempty"` // 极化反转
} }
// Modulation 包含了具体的调制信息。 // Modulation 包含了具体的调制信息。
type Modulation struct { type Modulation struct {
Lora LoraModulationInfo `json:"lora"` Lora LoraModulationInfo `json:"lora"` // LoRa 调制信息
} }
// UplinkTxInfo 包含了上行发送信息。 // UplinkTxInfo 包含了上行发送信息。
type UplinkTxInfo struct { type UplinkTxInfo struct {
Frequency int `json:"frequency"` Frequency int `json:"frequency"` // 发送频率
Modulation Modulation `json:"modulation"` Modulation Modulation `json:"modulation"` // 调制信息
} }
// DownlinkTxInfo 包含了下行发送信息。 // DownlinkTxInfo 包含了下行发送信息。
type DownlinkTxInfo struct { type DownlinkTxInfo struct {
Frequency int `json:"frequency"` Frequency int `json:"frequency"` // 发送频率
Power int `json:"power"` Power int `json:"power"` // 发送功率
Modulation Modulation `json:"modulation"` Modulation Modulation `json:"modulation"` // 调制信息
} }
// ResolvedLocation 包含了地理位置解析结果。 // ResolvedLocation 包含了地理位置解析结果。
type ResolvedLocation struct { type ResolvedLocation struct {
Latitude float64 `json:"latitude"` Latitude float64 `json:"latitude"` // 纬度
Longitude float64 `json:"longitude"` Longitude float64 `json:"longitude"` // 经度
Altitude float64 `json:"altitude"` Altitude float64 `json:"altitude"` // 海拔
Source string `json:"source"` Source string `json:"source"` // 位置来源
Accuracy int `json:"accuracy"` Accuracy int `json:"accuracy"` // 精度
} }
// --- 事件专属结构体 --- // --- 事件专属结构体 ---
// UpEvent 对应 ChirpStack 的 "up" 事件。 // UpEvent 对应 ChirpStack 的 "up" 事件。
type UpEvent struct { type UpEvent struct {
DeduplicationID string `json:"deduplicationId"` DeduplicationID string `json:"deduplication_id"` // 去重ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
DevAddr string `json:"devAddr"` DevAddr string `json:"dev_addr"` // 设备地址
ADR bool `json:"adr"` ADR bool `json:"adr"` // 自适应数据速率 (ADR) 是否启用
DR int `json:"dr"` DR int `json:"dr"` // 数据速率
FCnt uint32 `json:"fCnt"` FCnt uint32 `json:"f_cnt"` // 帧计数器
FPort uint8 `json:"fPort"` FPort uint8 `json:"f_port"` // 端口
Confirmed bool `json:"confirmed"` Confirmed bool `json:"confirmed"` // 是否是确认帧
Data string `json:"data"` Data string `json:"data"` // Base64 编码的原始负载数据
Object json.RawMessage `json:"object"` Object json.RawMessage `json:"object"` // 解码后的JSON对象负载
RxInfo []UplinkRxInfo `json:"rxInfo"` RxInfo []UplinkRxInfo `json:"rx_info"` // 接收信息列表
TxInfo UplinkTxInfo `json:"txInfo"` TxInfo UplinkTxInfo `json:"tx_info"` // 发送信息
RelayRxInfo *UplinkRelayRxInfo `json:"relayRxInfo,omitempty"` RelayRxInfo *UplinkRelayRxInfo `json:"relay_rx_info,omitempty"` // 中继接收信息
JoinServerContext *JoinServerContext `json:"joinServerContext,omitempty"` JoinServerContext *JoinServerContext `json:"join_server_context,omitempty"` // Join-Server 上下文
RegionConfigID string `json:"regionConfigId,omitempty"` RegionConfigID string `json:"region_config_id,omitempty"` // 区域配置ID
} }
// JoinEvent 对应 ChirpStack 的 "join" 事件。 // JoinEvent 对应 ChirpStack 的 "join" 事件。
type JoinEvent struct { type JoinEvent struct {
DeduplicationID string `json:"deduplicationId"` DeduplicationID string `json:"deduplication_id"` // 去重ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
DevAddr string `json:"devAddr"` DevAddr string `json:"dev_addr"` // 设备地址
RelayRxInfo *UplinkRelayRxInfo `json:"relayRxInfo,omitempty"` RelayRxInfo *UplinkRelayRxInfo `json:"relay_rx_info,omitempty"` // 中继接收信息
JoinServerContext *JoinServerContext `json:"joinServerContext,omitempty"` JoinServerContext *JoinServerContext `json:"join_server_context,omitempty"` // Join-Server 上下文
RegionConfigID string `json:"regionConfigId,omitempty"` RegionConfigID string `json:"region_config_id,omitempty"` // 区域配置ID
} }
// AckEvent 对应 ChirpStack 的 "ack" 事件。 // AckEvent 对应 ChirpStack 的 "ack" 事件。
type AckEvent struct { type AckEvent struct {
DeduplicationID string `json:"deduplicationId"` DeduplicationID string `json:"deduplication_id"` // 去重ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
Acknowledged bool `json:"acknowledged"` Acknowledged bool `json:"acknowledged"` // 是否已确认
FCntDown uint32 `json:"fCntDown"` FCntDown uint32 `json:"f_cnt_down"` // 下行帧计数器
QueueItemID string `json:"queueItemId"` QueueItemID string `json:"queue_item_id"` // 队列项ID
} }
// TxAckEvent 对应 ChirpStack 的 "txack" 事件。 // TxAckEvent 对应 ChirpStack 的 "txack" 事件。
type TxAckEvent struct { type TxAckEvent struct {
DownlinkID uint32 `json:"downlinkId"` // 修改: 替换 DeduplicationID DownlinkID uint32 `json:"downlink_id"` // 下行ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
FCntDown uint32 `json:"fCntDown"` FCntDown uint32 `json:"f_cnt_down"` // 下行帧计数器
GatewayID string `json:"gatewayId"` GatewayID string `json:"gateway_id"` // 网关ID
QueueItemID string `json:"queueItemId"` QueueItemID string `json:"queue_item_id"` // 队列项ID
TxInfo DownlinkTxInfo `json:"txInfo"` TxInfo DownlinkTxInfo `json:"tx_info"` // 下行发送信息
} }
// StatusEvent 对应 ChirpStack 的 "status" 事件。 // StatusEvent 对应 ChirpStack 的 "status" 事件。
type StatusEvent struct { type StatusEvent struct {
DeduplicationID string `json:"deduplicationId"` DeduplicationID string `json:"deduplication_id"` // 去重ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
Margin int `json:"margin"` Margin int `json:"margin"` // 链路预算余量 (dB)
ExternalPower bool `json:"externalPowerSource"` ExternalPower bool `json:"external_power_source"` // 设备是否连接外部电源
BatteryLevel float32 `json:"batteryLevel"` BatteryLevel float32 `json:"battery_level"` // 电池剩余电量
BatteryLevelUnavailable bool `json:"batteryLevelUnavailable"` BatteryLevelUnavailable bool `json:"battery_level_unavailable"` // 电池电量是否不可用
} }
// LogEvent 对应 ChirpStack 的 "log" 事件。 // LogEvent 对应 ChirpStack 的 "log" 事件。
type LogEvent struct { type LogEvent struct {
DeduplicationID string `json:"deduplicationId"` DeduplicationID string `json:"deduplication_id"` // 去重ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
Level string `json:"level"` Level string `json:"level"` // 日志级别 (e.g., INFO, WARNING, ERROR)
Code string `json:"code"` Code string `json:"code"` // 日志代码
Description string `json:"description"` Description string `json:"description"` // 日志描述
Context map[string]string `json:"context"` Context map[string]string `json:"context"` // 上下文信息
} }
// LocationEvent 对应 ChirpStack 的 "location" 事件。 // LocationEvent 对应 ChirpStack 的 "location" 事件。
type LocationEvent struct { type LocationEvent struct {
DeduplicationID string `json:"deduplicationId"` DeduplicationID string `json:"deduplication_id"` // 去重ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
Location ResolvedLocation `json:"location"` Location ResolvedLocation `json:"location"` // 解析后的位置信息
} }
// IntegrationEvent 对应 ChirpStack 的 "integration" 事件。 // IntegrationEvent 对应 ChirpStack 的 "integration" 事件。
type IntegrationEvent struct { type IntegrationEvent struct {
DeduplicationID string `json:"deduplicationId"` DeduplicationID string `json:"deduplication_id"` // 去重ID
Time time.Time `json:"time"` Time time.Time `json:"time"` // 事件时间
DeviceInfo DeviceInfo `json:"deviceInfo"` DeviceInfo DeviceInfo `json:"device_info"` // 设备信息
IntegrationName string `json:"integrationName"` IntegrationName string `json:"integration_name"` // 集成名称
EventType string `json:"eventType,omitempty"` EventType string `json:"event_type,omitempty"` // 事件类型
Object json.RawMessage `json:"object"` Object json.RawMessage `json:"object"` // 集成事件的原始JSON负载
} }

7
internal/core/application.go
View File

@@ -69,8 +69,11 @@ func NewApplication(configPath string) (*Application, error) {
// 初始化执行日志仓库 // 初始化执行日志仓库
executionLogRepo := repository.NewGormExecutionLogRepository(storage.GetDB()) executionLogRepo := repository.NewGormExecutionLogRepository(storage.GetDB())
// 初始化传感器数据仓库
sensorDataRepo := repository.NewGormSensorDataRepository(storage.GetDB())
// 初始化设备上行监听器 // 初始化设备上行监听器
listenHandler := transport.NewChirpStackListener(logger) listenHandler := transport.NewChirpStackListener(logger, sensorDataRepo, deviceRepo)
// 初始化计划触发器管理器 // 初始化计划触发器管理器
analysisPlanTaskManager := task.NewAnalysisPlanTaskManager(planRepo, pendingTaskRepo, executionLogRepo, logger) analysisPlanTaskManager := task.NewAnalysisPlanTaskManager(planRepo, pendingTaskRepo, executionLogRepo, logger)
@@ -79,7 +82,7 @@ func NewApplication(configPath string) (*Application, error) {
executor := task.NewScheduler(pendingTaskRepo, executionLogRepo, planRepo, analysisPlanTaskManager, task.TaskFactory, logger, time.Duration(cfg.Task.Interval)*time.Second, cfg.Task.NumWorkers) executor := task.NewScheduler(pendingTaskRepo, executionLogRepo, planRepo, analysisPlanTaskManager, task.TaskFactory, logger, time.Duration(cfg.Task.Interval)*time.Second, cfg.Task.NumWorkers)
// 初始化 API 服务器 // 初始化 API 服务器
apiServer := api.NewAPI(cfg.Server, logger, userRepo, deviceRepo, planRepo, tokenService, listenHandler, analysisPlanTaskManager) apiServer := api.NewAPI(cfg.Server, logger, userRepo, deviceRepo, planRepo, sensorDataRepo, executionLogRepo, tokenService, listenHandler, analysisPlanTaskManager)
// 组装 Application 对象 // 组装 Application 对象
app := &Application{ app := &Application{

30
internal/infra/models/SensorData.go
View File

@@ -1,30 +0,0 @@
package models
import (
"time"
"gorm.io/datatypes"
)
// SensorData 存储所有类型的传感器数据，对应数据库中的 'sensor_data' 表。
type SensorData struct {
// Time 是数据记录的时间戳，作为复合主键的一部分。
// GORM 会将其映射到 'time' TIMESTAMPTZ 列。
Time time.Time `gorm:"primaryKey" json:"time"`
// DeviceID 是传感器的唯一标识符，作为复合主键的另一部分。
// GORM 会将其映射到 'device_id' VARCHAR(50) 列。
DeviceID uint `gorm:"primaryKey" json:"device_id"`
// RegionalControllerID 是上报此数据的区域主控的ID。
// 我们为其添加了数据库索引以优化按区域查询的性能。
RegionalControllerID uint `json:"regional_controller_id"`
// Data 存储一个或多个传感器读数，格式为 JSON。
// GORM 会使用 'jsonb' 类型来创建此列。
Data datatypes.JSON `gorm:"type:jsonb" json:"data"`
}
func (SensorData) TableName() string {
return "sensor_data"
}

56
internal/infra/models/sensor_data.go Normal file
View File

@@ -0,0 +1,56 @@
package models
import (
"time"
"gorm.io/datatypes"
)
// SensorDataType 定义了 SensorData 记录中 Data 字段的整体类型
type SensorDataType string
const (
SensorDataTypeSignalMetrics SensorDataType = "signal_metrics" // 信号强度
SensorDataTypeBatteryLevel SensorDataType = "battery_level" // 电池电量
)
// SignalMetrics 存储信号强度数据
type SignalMetrics struct {
RSSI int `json:"rssi"` // 绝对信号强度（dBm），受距离、障碍物影响
SNR float64 `json:"snr"` // 信号与噪声的相对比率（dB），由 RSSI 减去噪声地板（Noise Floor）
Sensitivity int `json:"sensitivity"` // 网关的最低检测阈值(dBm)
Margin int `json:"margin"` // SNR 相对于接收器灵敏度的余量, Margin = SNR - Sensitivity
}
// BatteryLevel 存储电池电量数据
type BatteryLevel struct {
BatteryLevel float32 `json:"battery_level"` // 电量剩余百分比
BatteryLevelUnavailable bool `json:"battery_level_unavailable"` // 电量数据不可用
ExternalPower bool `json:"external_power"` // 是否使用外部电源
}
// SensorData 存储所有类型的传感器数据，对应数据库中的 'sensor_data' 表。
type SensorData struct {
// Time 是数据记录的时间戳，作为复合主键的一部分。
// GORM 会将其映射到 'time' TIMESTAMPTZ 列。
Time time.Time `gorm:"primaryKey" json:"time"`
// DeviceID 是传感器的唯一标识符，作为复合主键的另一部分。
// GORM 会将其映射到 'device_id' VARCHAR(50) 列。
DeviceID uint `gorm:"primaryKey" json:"device_id"`
// RegionalControllerID 是上报此数据的区域主控的ID。
// 我们为其添加了数据库索引以优化按区域查询的性能。
RegionalControllerID uint `json:"regional_controller_id"`
// SensorDataType 是传感数据的类型
SensorDataType SensorDataType `gorm:"not null;index" json:"sensor_data_type"`
// Data 存储一个或多个传感器读数，格式为 JSON。
// GORM 会使用 'jsonb' 类型来创建此列。
Data datatypes.JSON `gorm:"type:jsonb" json:"data"`
}
func (SensorData) TableName() string {
return "sensor_data"
}

13
internal/infra/repository/device_repository.go
View File

@@ -32,6 +32,9 @@ type DeviceRepository interface {
// Delete 根据主键 ID 删除一个设备 // Delete 根据主键 ID 删除一个设备
Delete(id uint) error Delete(id uint) error
// FindByDevEui 根据 DevEui (存储在 properties JSONB 中的 lora_address) 查找设备 (新增)
FindByDevEui(devEui string) (*models.Device, error)
} }
// gormDeviceRepository 是 DeviceRepository 的 GORM 实现 // gormDeviceRepository 是 DeviceRepository 的 GORM 实现
@@ -108,3 +111,13 @@ func (r *gormDeviceRepository) Update(device *models.Device) error {
func (r *gormDeviceRepository) Delete(id uint) error { func (r *gormDeviceRepository) Delete(id uint) error {
return r.db.Delete(&models.Device{}, id).Error return r.db.Delete(&models.Device{}, id).Error
} }
// FindByDevEui 根据 DevEui (存储在 properties JSONB 中的 lora_address) 查找设备
func (r *gormDeviceRepository) FindByDevEui(devEui string) (*models.Device, error) {
var device models.Device
// 使用 GORM 的 JSONB 查询语法: properties->>'lora_address'
if err := r.db.Where("properties->>'lora_address' = ?", devEui).First(&device).Error; err != nil {
return nil, err // 如果找不到或发生其他错误，GORM 会返回错误
}
return &device, nil
}

27
internal/infra/repository/sensor_data_repository.go Normal file
View File

@@ -0,0 +1,27 @@
package repository
import (
"git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
"gorm.io/gorm"
)
// SensorDataRepository 定义了与传感器数据相关的数据库操作接口。
type SensorDataRepository interface {
Create(sensorData *models.SensorData) error
}
// gormSensorDataRepository 是 SensorDataRepository 的 GORM 实现。
type gormSensorDataRepository struct {
db *gorm.DB
}
// NewGormSensorDataRepository 创建一个新的 SensorDataRepository GORM 实现实例。
// 它直接接收一个 *gorm.DB 实例作为依赖，完全遵循项目中的既定模式。
func NewGormSensorDataRepository(db *gorm.DB) SensorDataRepository {
return &gormSensorDataRepository{db: db}
}
// Create 将一条新的传感器数据记录插入数据库。
func (r *gormSensorDataRepository) Create(sensorData *models.SensorData) error {
return r.db.Create(sensorData).Error
}