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

issue_50 #57

Merged
huang merged 20 commits from issue_50 into main 2025-11-03 17:27:25 +08:00
Conversation 0 Commits 20 Files Changed 483 +149 -149
3 changed files with 149 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 4b0be88fca - Show all commits

26
internal/core/application.go
View File

@@ -44,16 +44,16 @@ func NewApplication(configPath string) (*Application, error) {
apiServer := api.NewAPI(
cfg.Server,
logger,
infra.Repos.UserRepo,
appServices.PigFarmService,
appServices.PigBatchService,
appServices.MonitorService,
appServices.DeviceService,
appServices.PlanService,
appServices.UserService,
infra.TokenService,
appServices.AuditService,
infra.Lora.ListenHandler,
infra.repos.userRepo,
appServices.pigFarmService,
appServices.pigBatchService,
appServices.monitorService,
appServices.deviceService,
appServices.planService,
appServices.userService,
infra.tokenService,
appServices.auditService,
infra.lora.listenHandler,
)
// 4. 组装 Application 对象
@@ -74,7 +74,7 @@ func (app *Application) Start() error {
app.Logger.Info("应用启动中...")
// 1. 启动底层监听器
if err := app.Infra.Lora.LoraListener.Listen(); err != nil {
if err := app.Infra.lora.loraListener.Listen(); err != nil {
return fmt.Errorf("启动 LoRa Mesh 监听器失败: %w", err)
}
@@ -109,12 +109,12 @@ func (app *Application) Stop() error {
app.Domain.planService.Stop()
// 断开数据库连接
if err := app.Infra.Storage.Disconnect(); err != nil {
if err := app.Infra.storage.Disconnect(); err != nil {
app.Logger.Errorw("数据库连接断开失败", "error", err)
}
// 关闭 LoRa Mesh 监听器
if err := app.Infra.Lora.LoraListener.Stop(); err != nil {
if err := app.Infra.lora.loraListener.Stop(); err != nil {
app.Logger.Errorw("LoRa Mesh 监听器关闭失败", "error", err)
}

258
internal/core/component_initializers.go
View File

@@ -26,11 +26,11 @@ import (
// Infrastructure 聚合了所有基础设施层的组件。
type Infrastructure struct {
Storage database.Storage
Repos *Repositories
Lora *LoraComponents
NotifyService domain_notify.Service
TokenService token.Service
storage database.Storage
repos *Repositories
lora *LoraComponents
notifyService domain_notify.Service
tokenService token.Service
}
// initInfrastructure 初始化所有基础设施层组件。
@@ -47,7 +47,7 @@ func initInfrastructure(cfg *config.Config, logger *logs.Logger) (*Infrastructur
return nil, err
}
notifyService, err := initNotifyService(cfg.Notify, logger, repos.UserRepo, repos.NotificationRepo)
notifyService, err := initNotifyService(cfg.Notify, logger, repos.userRepo, repos.notificationRepo)
if err != nil {
return nil, fmt.Errorf("初始化通知服务失败: %w", err)
}
@@ -55,112 +55,112 @@ func initInfrastructure(cfg *config.Config, logger *logs.Logger) (*Infrastructur
tokenService := token.NewTokenService([]byte(cfg.App.JWTSecret))
return &Infrastructure{
Storage: storage,
Repos: repos,
Lora: lora,
NotifyService: notifyService,
TokenService: tokenService,
storage: storage,
repos: repos,
lora: lora,
notifyService: notifyService,
tokenService: tokenService,
}, nil
}
// Repositories 聚合了所有的仓库实例。
type Repositories struct {
UserRepo repository.UserRepository
DeviceRepo repository.DeviceRepository
AreaControllerRepo repository.AreaControllerRepository
DeviceTemplateRepo repository.DeviceTemplateRepository
PlanRepo repository.PlanRepository
PendingTaskRepo repository.PendingTaskRepository
ExecutionLogRepo repository.ExecutionLogRepository
SensorDataRepo repository.SensorDataRepository
DeviceCommandLogRepo repository.DeviceCommandLogRepository
PendingCollectionRepo repository.PendingCollectionRepository
UserActionLogRepo repository.UserActionLogRepository
PigBatchRepo repository.PigBatchRepository
PigBatchLogRepo repository.PigBatchLogRepository
PigFarmRepo repository.PigFarmRepository
PigPenRepo repository.PigPenRepository
PigTransferLogRepo repository.PigTransferLogRepository
PigTradeRepo repository.PigTradeRepository
PigSickPigLogRepo repository.PigSickLogRepository
MedicationLogRepo repository.MedicationLogRepository
RawMaterialRepo repository.RawMaterialRepository
NotificationRepo repository.NotificationRepository
UnitOfWork repository.UnitOfWork
userRepo repository.UserRepository
deviceRepo repository.DeviceRepository
areaControllerRepo repository.AreaControllerRepository
deviceTemplateRepo repository.DeviceTemplateRepository
planRepo repository.PlanRepository
pendingTaskRepo repository.PendingTaskRepository
executionLogRepo repository.ExecutionLogRepository
sensorDataRepo repository.SensorDataRepository
deviceCommandLogRepo repository.DeviceCommandLogRepository
pendingCollectionRepo repository.PendingCollectionRepository
userActionLogRepo repository.UserActionLogRepository
pigBatchRepo repository.PigBatchRepository
pigBatchLogRepo repository.PigBatchLogRepository
pigFarmRepo repository.PigFarmRepository
pigPenRepo repository.PigPenRepository
pigTransferLogRepo repository.PigTransferLogRepository
pigTradeRepo repository.PigTradeRepository
pigSickPigLogRepo repository.PigSickLogRepository
medicationLogRepo repository.MedicationLogRepository
rawMaterialRepo repository.RawMaterialRepository
notificationRepo repository.NotificationRepository
unitOfWork repository.UnitOfWork
}
// initRepositories 初始化所有的仓库。
func initRepositories(db *gorm.DB, logger *logs.Logger) *Repositories {
return &Repositories{
UserRepo: repository.NewGormUserRepository(db),
DeviceRepo: repository.NewGormDeviceRepository(db),
AreaControllerRepo: repository.NewGormAreaControllerRepository(db),
DeviceTemplateRepo: repository.NewGormDeviceTemplateRepository(db),
PlanRepo: repository.NewGormPlanRepository(db),
PendingTaskRepo: repository.NewGormPendingTaskRepository(db),
ExecutionLogRepo: repository.NewGormExecutionLogRepository(db),
SensorDataRepo: repository.NewGormSensorDataRepository(db),
DeviceCommandLogRepo: repository.NewGormDeviceCommandLogRepository(db),
PendingCollectionRepo: repository.NewGormPendingCollectionRepository(db),
UserActionLogRepo: repository.NewGormUserActionLogRepository(db),
PigBatchRepo: repository.NewGormPigBatchRepository(db),
PigBatchLogRepo: repository.NewGormPigBatchLogRepository(db),
PigFarmRepo: repository.NewGormPigFarmRepository(db),
PigPenRepo: repository.NewGormPigPenRepository(db),
PigTransferLogRepo: repository.NewGormPigTransferLogRepository(db),
PigTradeRepo: repository.NewGormPigTradeRepository(db),
PigSickPigLogRepo: repository.NewGormPigSickLogRepository(db),
MedicationLogRepo: repository.NewGormMedicationLogRepository(db),
RawMaterialRepo: repository.NewGormRawMaterialRepository(db),
NotificationRepo: repository.NewGormNotificationRepository(db),
UnitOfWork: repository.NewGormUnitOfWork(db, logger),
userRepo: repository.NewGormUserRepository(db),
deviceRepo: repository.NewGormDeviceRepository(db),
areaControllerRepo: repository.NewGormAreaControllerRepository(db),
deviceTemplateRepo: repository.NewGormDeviceTemplateRepository(db),
planRepo: repository.NewGormPlanRepository(db),
pendingTaskRepo: repository.NewGormPendingTaskRepository(db),
executionLogRepo: repository.NewGormExecutionLogRepository(db),
sensorDataRepo: repository.NewGormSensorDataRepository(db),
deviceCommandLogRepo: repository.NewGormDeviceCommandLogRepository(db),
pendingCollectionRepo: repository.NewGormPendingCollectionRepository(db),
userActionLogRepo: repository.NewGormUserActionLogRepository(db),
pigBatchRepo: repository.NewGormPigBatchRepository(db),
pigBatchLogRepo: repository.NewGormPigBatchLogRepository(db),
pigFarmRepo: repository.NewGormPigFarmRepository(db),
pigPenRepo: repository.NewGormPigPenRepository(db),
pigTransferLogRepo: repository.NewGormPigTransferLogRepository(db),
pigTradeRepo: repository.NewGormPigTradeRepository(db),
pigSickPigLogRepo: repository.NewGormPigSickLogRepository(db),
medicationLogRepo: repository.NewGormMedicationLogRepository(db),
rawMaterialRepo: repository.NewGormRawMaterialRepository(db),
notificationRepo: repository.NewGormNotificationRepository(db),
unitOfWork: repository.NewGormUnitOfWork(db, logger),
}
}
// DomainServices 聚合了所有的领域服务实例。
type DomainServices struct {
PigPenTransferManager pig.PigPenTransferManager
PigTradeManager pig.PigTradeManager
PigSickManager pig.SickPigManager
PigBatchDomain pig.PigBatchService
GeneralDeviceService device.Service
pigPenTransferManager pig.PigPenTransferManager
pigTradeManager pig.PigTradeManager
pigSickManager pig.SickPigManager
pigBatchDomain pig.PigBatchService
generalDeviceService device.Service
taskFactory plan.TaskFactory
PlanExecutionManager plan.ExecutionManager
AnalysisPlanTaskManager plan.AnalysisPlanTaskManager
planExecutionManager plan.ExecutionManager
analysisPlanTaskManager plan.AnalysisPlanTaskManager
planService plan.Service
}
// initDomainServices 初始化所有的领域服务。
func initDomainServices(cfg *config.Config, infra *Infrastructure, logger *logs.Logger) *DomainServices {
// 猪群管理相关
pigPenTransferManager := pig.NewPigPenTransferManager(infra.Repos.PigPenRepo, infra.Repos.PigTransferLogRepo, infra.Repos.PigBatchRepo)
pigTradeManager := pig.NewPigTradeManager(infra.Repos.PigTradeRepo)
pigSickManager := pig.NewSickPigManager(infra.Repos.PigSickPigLogRepo, infra.Repos.MedicationLogRepo)
pigBatchDomain := pig.NewPigBatchService(infra.Repos.PigBatchRepo, infra.Repos.PigBatchLogRepo, infra.Repos.UnitOfWork,
pigPenTransferManager := pig.NewPigPenTransferManager(infra.repos.pigPenRepo, infra.repos.pigTransferLogRepo, infra.repos.pigBatchRepo)
pigTradeManager := pig.NewPigTradeManager(infra.repos.pigTradeRepo)
pigSickManager := pig.NewSickPigManager(infra.repos.pigSickPigLogRepo, infra.repos.medicationLogRepo)
pigBatchDomain := pig.NewPigBatchService(infra.repos.pigBatchRepo, infra.repos.pigBatchLogRepo, infra.repos.unitOfWork,
pigPenTransferManager, pigTradeManager, pigSickManager)
// 通用设备服务
generalDeviceService := device.NewGeneralDeviceService(
infra.Repos.DeviceRepo,
infra.Repos.DeviceCommandLogRepo,
infra.Repos.PendingCollectionRepo,
infra.repos.deviceRepo,
infra.repos.deviceCommandLogRepo,
infra.repos.pendingCollectionRepo,
logger,
infra.Lora.Comm,
infra.lora.comm,
)
// 任务工厂
taskFactory := task.NewTaskFactory(logger, infra.Repos.SensorDataRepo, infra.Repos.DeviceRepo, generalDeviceService)
taskFactory := task.NewTaskFactory(logger, infra.repos.sensorDataRepo, infra.repos.deviceRepo, generalDeviceService)
// 计划任务管理器
analysisPlanTaskManager := plan.NewAnalysisPlanTaskManager(infra.Repos.PlanRepo, infra.Repos.PendingTaskRepo, infra.Repos.ExecutionLogRepo, logger)
analysisPlanTaskManager := plan.NewAnalysisPlanTaskManager(infra.repos.planRepo, infra.repos.pendingTaskRepo, infra.repos.executionLogRepo, logger)
// 任务执行器
planExecutionManager := plan.NewPlanExecutionManager(
infra.Repos.PendingTaskRepo,
infra.Repos.ExecutionLogRepo,
infra.Repos.DeviceRepo,
infra.Repos.SensorDataRepo,
infra.Repos.PlanRepo,
infra.repos.pendingTaskRepo,
infra.repos.executionLogRepo,
infra.repos.deviceRepo,
infra.repos.sensorDataRepo,
infra.repos.planRepo,
analysisPlanTaskManager,
taskFactory,
logger,
@@ -173,75 +173,75 @@ func initDomainServices(cfg *config.Config, infra *Infrastructure, logger *logs.
planService := plan.NewPlanService(planExecutionManager, analysisPlanTaskManager, logger)
return &DomainServices{
PigPenTransferManager: pigPenTransferManager,
PigTradeManager: pigTradeManager,
PigSickManager: pigSickManager,
PigBatchDomain: pigBatchDomain,
GeneralDeviceService: generalDeviceService,
AnalysisPlanTaskManager: analysisPlanTaskManager,
pigPenTransferManager: pigPenTransferManager,
pigTradeManager: pigTradeManager,
pigSickManager: pigSickManager,
pigBatchDomain: pigBatchDomain,
generalDeviceService: generalDeviceService,
analysisPlanTaskManager: analysisPlanTaskManager,
taskFactory: taskFactory,
PlanExecutionManager: planExecutionManager,
planExecutionManager: planExecutionManager,
planService: planService,
}
}
// AppServices 聚合了所有的应用服务实例。
type AppServices struct {
PigFarmService service.PigFarmService
PigBatchService service.PigBatchService
MonitorService service.MonitorService
DeviceService service.DeviceService
PlanService service.PlanService
UserService service.UserService
AuditService audit.Service
pigFarmService service.PigFarmService
pigBatchService service.PigBatchService
monitorService service.MonitorService
deviceService service.DeviceService
planService service.PlanService
userService service.UserService
auditService audit.Service
}
// initAppServices 初始化所有的应用服务。
func initAppServices(infra *Infrastructure, domainServices *DomainServices, logger *logs.Logger) *AppServices {
pigFarmService := service.NewPigFarmService(infra.Repos.PigFarmRepo, infra.Repos.PigPenRepo, infra.Repos.PigBatchRepo, domainServices.PigBatchDomain, infra.Repos.UnitOfWork, logger)
pigBatchService := service.NewPigBatchService(domainServices.PigBatchDomain, logger)
pigFarmService := service.NewPigFarmService(infra.repos.pigFarmRepo, infra.repos.pigPenRepo, infra.repos.pigBatchRepo, domainServices.pigBatchDomain, infra.repos.unitOfWork, logger)
pigBatchService := service.NewPigBatchService(domainServices.pigBatchDomain, logger)
monitorService := service.NewMonitorService(
infra.Repos.SensorDataRepo,
infra.Repos.DeviceCommandLogRepo,
infra.Repos.ExecutionLogRepo,
infra.Repos.PlanRepo,
infra.Repos.PendingCollectionRepo,
infra.Repos.UserActionLogRepo,
infra.Repos.RawMaterialRepo,
infra.Repos.MedicationLogRepo,
infra.Repos.PigBatchRepo,
infra.Repos.PigBatchLogRepo,
infra.Repos.PigTransferLogRepo,
infra.Repos.PigSickPigLogRepo,
infra.Repos.PigTradeRepo,
infra.Repos.NotificationRepo,
infra.repos.sensorDataRepo,
infra.repos.deviceCommandLogRepo,
infra.repos.executionLogRepo,
infra.repos.planRepo,
infra.repos.pendingCollectionRepo,
infra.repos.userActionLogRepo,
infra.repos.rawMaterialRepo,
infra.repos.medicationLogRepo,
infra.repos.pigBatchRepo,
infra.repos.pigBatchLogRepo,
infra.repos.pigTransferLogRepo,
infra.repos.pigSickPigLogRepo,
infra.repos.pigTradeRepo,
infra.repos.notificationRepo,
)
deviceService := service.NewDeviceService(
infra.Repos.DeviceRepo,
infra.Repos.AreaControllerRepo,
infra.Repos.DeviceTemplateRepo,
domainServices.GeneralDeviceService,
infra.repos.deviceRepo,
infra.repos.areaControllerRepo,
infra.repos.deviceTemplateRepo,
domainServices.generalDeviceService,
)
auditService := audit.NewService(infra.Repos.UserActionLogRepo, logger)
planService := service.NewPlanService(logger, infra.Repos.PlanRepo, domainServices.AnalysisPlanTaskManager)
userService := service.NewUserService(infra.Repos.UserRepo, infra.TokenService, infra.NotifyService, logger)
auditService := audit.NewService(infra.repos.userActionLogRepo, logger)
planService := service.NewPlanService(logger, infra.repos.planRepo, domainServices.analysisPlanTaskManager)
userService := service.NewUserService(infra.repos.userRepo, infra.tokenService, infra.notifyService, logger)
return &AppServices{
PigFarmService: pigFarmService,
PigBatchService: pigBatchService,
MonitorService: monitorService,
DeviceService: deviceService,
AuditService: auditService,
PlanService: planService,
UserService: userService,
pigFarmService: pigFarmService,
pigBatchService: pigBatchService,
monitorService: monitorService,
deviceService: deviceService,
auditService: auditService,
planService: planService,
userService: userService,
}
}
// LoraComponents 聚合了所有 LoRa 相关组件。
type LoraComponents struct {
ListenHandler webhook.ListenHandler
Comm transport.Communicator
LoraListener transport.Listener
listenHandler webhook.ListenHandler
comm transport.Communicator
loraListener transport.Listener
}
// initLora 根据配置初始化 LoRa 相关组件。
@@ -256,13 +256,13 @@ func initLora(
if cfg.Lora.Mode == config.LoraMode_LoRaWAN {
logger.Info("当前运行模式: lora_wan。初始化 ChirpStack 监听器和传输层。")
listenHandler = webhook.NewChirpStackListener(logger, repos.SensorDataRepo, repos.DeviceRepo, repos.AreaControllerRepo, repos.DeviceCommandLogRepo, repos.PendingCollectionRepo)
listenHandler = webhook.NewChirpStackListener(logger, repos.sensorDataRepo, repos.deviceRepo, repos.areaControllerRepo, repos.deviceCommandLogRepo, repos.pendingCollectionRepo)
comm = lora.NewChirpStackTransport(cfg.ChirpStack, logger)
loraListener = lora.NewPlaceholderTransport(logger)
} else {
logger.Info("当前运行模式: lora_mesh。初始化 LoRa Mesh 传输层和占位符监听器。")
listenHandler = webhook.NewPlaceholderListener(logger)
tp, err := lora.NewLoRaMeshUartPassthroughTransport(cfg.LoraMesh, logger, repos.AreaControllerRepo, repos.PendingCollectionRepo, repos.DeviceRepo, repos.SensorDataRepo)
tp, err := lora.NewLoRaMeshUartPassthroughTransport(cfg.LoraMesh, logger, repos.areaControllerRepo, repos.pendingCollectionRepo, repos.deviceRepo, repos.sensorDataRepo)
if err != nil {
return nil, fmt.Errorf("无法初始化 LoRa Mesh 模块: %w", err)
}
@@ -271,9 +271,9 @@ func initLora(
}
return &LoraComponents{
ListenHandler: listenHandler,
Comm: comm,
LoraListener: loraListener,
listenHandler: listenHandler,
comm: comm,
loraListener: loraListener,
}, nil
}

14
internal/core/data_initializer.go
View File

@@ -41,7 +41,7 @@ func (app *Application) initializeSystemPlans() error {
predefinedSystemPlans := app.getPredefinedSystemPlans()
// 1. 获取所有已存在的系统计划
existingPlans, _, err := app.Infra.Repos.PlanRepo.ListPlans(repository.ListPlansOptions{
existingPlans, _, err := app.Infra.repos.planRepo.ListPlans(repository.ListPlansOptions{
PlanType: repository.PlanTypeFilterSystem,
}, 1, 99999) // 使用一个较大的 pageSize 来获取所有系统计划
if err != nil {
@@ -66,7 +66,7 @@ func (app *Application) initializeSystemPlans() error {
predefinedPlan.ID = foundExistingPlan.ID
predefinedPlan.ExecuteCount = foundExistingPlan.ExecuteCount
if err := app.Infra.Repos.PlanRepo.UpdatePlan(predefinedPlan); err != nil {
if err := app.Infra.repos.planRepo.UpdatePlan(predefinedPlan); err != nil {
return fmt.Errorf("更新预定义计划 '%s' 失败: %w", predefinedPlan.Name, err)
} else {
app.Logger.Infof("成功更新预定义计划 '%s'。", predefinedPlan.Name)
@@ -74,7 +74,7 @@ func (app *Application) initializeSystemPlans() error {
} else {
// 如果计划不存在, 则创建
app.Logger.Infof("预定义计划 '%s' 不存在，正在创建...", predefinedPlan.Name)
if err := app.Infra.Repos.PlanRepo.CreatePlan(predefinedPlan); err != nil {
if err := app.Infra.repos.planRepo.CreatePlan(predefinedPlan); err != nil {
return fmt.Errorf("创建预定义计划 '%s' 失败: %w", predefinedPlan.Name, err)
} else {
app.Logger.Infof("成功创建预定义计划 '%s'。", predefinedPlan.Name)
@@ -123,7 +123,7 @@ func (app *Application) initializePendingCollections() error {
app.Logger.Info("开始清理所有未完成的采集请求...")
// 直接将所有 'pending' 状态的请求更新为 'timed_out'。
count, err := app.Infra.Repos.PendingCollectionRepo.MarkAllPendingAsTimedOut()
count, err := app.Infra.repos.pendingCollectionRepo.MarkAllPendingAsTimedOut()
if err != nil {
return fmt.Errorf("清理未完成的采集请求失败: %v", err)
} else if count > 0 {
@@ -138,9 +138,9 @@ func (app *Application) initializePendingCollections() error {
// initializePendingTasks 在应用启动时清理并刷新待执行任务列表。
func (app *Application) initializePendingTasks() error {
logger := app.Logger
planRepo := app.Infra.Repos.PlanRepo
pendingTaskRepo := app.Infra.Repos.PendingTaskRepo
executionLogRepo := app.Infra.Repos.ExecutionLogRepo
planRepo := app.Infra.repos.planRepo
pendingTaskRepo := app.Infra.repos.pendingTaskRepo
executionLogRepo := app.Infra.repos.executionLogRepo
planService := app.Domain.planService
logger.Info("开始初始化待执行任务列表...")