g82tt
3d41f4d412
【问题】 后端 VS 启动报: Unable to resolve service for type 'Warehouse.Services.iot_alarmService' while attempting to activate 'Warehouse.Services.RuleEngineService'. 【根因】 1. iot_alarmService 通过框架 IDependency 机制被 AutofacContainerModuleExtension.AddModule 按第一个接口 Iiot_alarmService 注册到内置 DI（具体类未单独注册） 2. RuleEngineService 构造函数按具体类 iot_alarmService 注入 → 内置 DI 找不到 → 验证失败 【修复】 RuleEngineService.cs: 构造函数参数 + 字段类型由 iot_alarmService 改为 Iiot_alarmService （符合框架惯例，与 gateway_nodesController.cs:40 / iot_alarmController.cs:15/24 一致） 【验证】 dotnet build api_sqlsugar 0 错误 0 警告 【效果】 - RuleEngineService 构造函数 DI 解析成功 - var app = builder.Build() 不再抛 AggregateException - 后端可正常启动 - Quartz 调度 RuleEngineJob 每 10s 调一次 EvaluateAllAsync() - 规则触发"告警"动作时 _alarmService.UpsertAlarmAsync 可正确调用
385 lines
17 KiB
C#
385 lines
17 KiB
C#
// ═══════════════════════════════════════════
|
||
// RuleEngineService — 规则引擎核心服务（V2.0 完整实现）
|
||
// 端到端流程：加载启用规则 → 分组取实时值 → 评估条件 → 执行动作（控制/告警/通知）→ 写日志
|
||
// 设计决策详见 doc/整合方案/SecMPS_规则引擎完整实施_v2.0.md §3 步骤 3
|
||
// ═══════════════════════════════════════════
|
||
using System;
|
||
using System.Collections.Generic;
|
||
using System.Diagnostics;
|
||
using System.Linq;
|
||
using System.Net.Http;
|
||
using System.Net.Http.Json;
|
||
using System.Text.Json;
|
||
using System.Threading.Tasks;
|
||
using Microsoft.AspNetCore.SignalR;
|
||
using Microsoft.Extensions.Logging;
|
||
using SqlSugar;
|
||
using VolPro.Core.DbSqlSugar; // FirstOrDefaultAsync 扩展方法
|
||
using VolPro.Core.SignalR;
|
||
using VolPro.Entity.DomainModels;
|
||
using Warehouse.IRepositories;
|
||
using Warehouse.IServices;
|
||
using Warehouse.Services; // iot_alarmService / SyncAlarmItem 命名空间
|
||
using VolPro.Warehouse.Services; // GatewayClient 命名空间
|
||
|
||
namespace Warehouse.Services;
|
||
|
||
/// <summary>
|
||
/// 规则引擎核心服务。由 RuleEngineJob 每 10s 调用 EvaluateAllAsync() 一次。
|
||
///
|
||
/// 设计要点（详见方案 v2.0 §3 步骤 3）：
|
||
/// 1. 用 SqlSugar 的 Includes 加载导航属性（条件/动作）
|
||
/// 2. 按 AdapterCode 分组批量调网关 B4-batch（避免逐设备 HTTP 调用）
|
||
/// 3. 规则评估：AND/OR + 数值/开关比对 + 滞后窗（RecoveryThreshold）+ 规则级冷却
|
||
/// 4. 动作执行：控制/告警/通知 三种类型在 ExecuteActionsAsync 内 switch 串行执行（MVP 阶段）
|
||
/// 5. 推送 SignalR：用 IHubContext<MessageHub> 推 "RuleTriggered" 事件（与 DataView.vue 订阅名一致）
|
||
/// 6. 写执行日志：warehouse_rulelog 记录每次评估的触发/条件摘要/耗时
|
||
/// 7. gateway_nodes.default 在每条规则触发前查一次（避免 N+1）
|
||
/// </summary>
|
||
public class RuleEngineService
|
||
{
|
||
private readonly Iwarehouse_ruleRepository _ruleRepo;
|
||
private readonly Ibase_deviceRepository _deviceRepo;
|
||
private readonly Iwarehouse_rulelogRepository _logRepo;
|
||
// V2.0.1 修复 DI：iot_alarmService 由 AutofacContainerModuleExtension.AddModule
|
||
// 按"第一个接口 Iiot_alarmService"注册到内置 DI（实现类未单独注册），
|
||
// 构造函数按具体类注入会触发 "Unable to resolve service for type iot_alarmService"。
|
||
private readonly Iiot_alarmService _alarmService;
|
||
private readonly GatewayClient _gatewayClient;
|
||
private readonly Igateway_nodesService _nodeService;
|
||
private readonly IHubContext<MessageHub> _hub;
|
||
private readonly ILogger<RuleEngineService> _logger;
|
||
|
||
public RuleEngineService(
|
||
Iwarehouse_ruleRepository ruleRepo,
|
||
Ibase_deviceRepository deviceRepo,
|
||
Iwarehouse_rulelogRepository logRepo,
|
||
Iiot_alarmService alarmService,
|
||
GatewayClient gatewayClient,
|
||
Igateway_nodesService nodeService,
|
||
IHubContext<MessageHub> hub,
|
||
ILogger<RuleEngineService> logger)
|
||
{
|
||
_ruleRepo = ruleRepo;
|
||
_deviceRepo = deviceRepo;
|
||
_logRepo = logRepo;
|
||
_alarmService = alarmService;
|
||
_gatewayClient = gatewayClient;
|
||
_nodeService = nodeService;
|
||
_hub = hub;
|
||
_logger = logger;
|
||
}
|
||
|
||
/// <summary>
|
||
/// 评估所有启用规则 + 执行动作 + 写日志。每 10s 由 RuleEngineJob 调用。
|
||
/// </summary>
|
||
public async Task EvaluateAllAsync()
|
||
{
|
||
var sw = Stopwatch.StartNew();
|
||
|
||
// 1. 加载启用规则（用 Includes 加载条件/动作导航属性）
|
||
var rules = await _ruleRepo.FindAsIQueryable(r => r.Enable == "启用")
|
||
.Includes(r => r.warehouse_rulecondition)
|
||
.Includes(r => r.warehouse_ruleaction)
|
||
.ToListAsync();
|
||
|
||
if (!rules.Any())
|
||
{
|
||
_logger.LogDebug("[RuleEngine] 无启用规则");
|
||
return;
|
||
}
|
||
|
||
// 2. 收集所有涉及设备 ID（按 AdapterCode 分组取实时值用）
|
||
var allDeviceIds = rules.SelectMany(r => r.warehouse_rulecondition)
|
||
.Select(c => c.DeviceId).Where(id => id.HasValue)
|
||
.Distinct().Select(id => id!.Value).ToList();
|
||
|
||
if (!allDeviceIds.Any())
|
||
{
|
||
_logger.LogDebug("[RuleEngine] 启用规则无有效设备");
|
||
return;
|
||
}
|
||
|
||
// 3. 查设备表 → Dictionary<DeviceId, base_device>
|
||
// SqlSugar 没有 ToDictionaryAsync，先 ToListAsync 再 LINQ ToDictionary
|
||
var deviceList = await _deviceRepo.FindAsIQueryable(d => allDeviceIds.Contains(d.DeviceId))
|
||
.ToListAsync();
|
||
var devices = deviceList.ToDictionary(d => d.DeviceId);
|
||
|
||
// 5. 一次性查 gateway_nodes.default（避免每条规则触发都查）
|
||
var defaultNode = await _nodeService.FindAsIQueryable(n => n.NodeCode == "default")
|
||
.FirstOrDefaultAsync();
|
||
var defaultBaseUrl = defaultNode?.BaseUrl ?? "";
|
||
|
||
// 5b. 按 AdapterCode 分组 → 调 B4-batch 批量取实时值
|
||
var realtimeCache = new Dictionary<int, List<PointValue>>(); // DeviceId → PointValues
|
||
var adapterGroups = devices.Values
|
||
.Where(d => !string.IsNullOrEmpty(d.AdapterCode) && !string.IsNullOrEmpty(d.SourceId))
|
||
.GroupBy(d => d.AdapterCode);
|
||
|
||
foreach (var grp in adapterGroups)
|
||
{
|
||
try
|
||
{
|
||
var adapterCode = grp.Key!;
|
||
var sourceIds = grp.Select(d => d.SourceId!).ToList();
|
||
// V2.0 步骤 5：baseUrl 从 defaultBaseUrl 传入（与方案 §3 步骤 5 一致）
|
||
var result = await _gatewayClient.GetRealtimeBatchAsync(defaultBaseUrl, adapterCode, sourceIds);
|
||
foreach (var dev in grp)
|
||
{
|
||
if (result.TryGetValue(dev.SourceId!, out var points))
|
||
realtimeCache[dev.DeviceId] = points;
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.LogError(ex, "[RuleEngine] 网关批量取实时值失败: Adapter={Adapter}", grp.Key);
|
||
}
|
||
}
|
||
|
||
// 6. 逐规则评估（按 Priority 倒序，数字越大越先评估）
|
||
foreach (var rule in rules.OrderByDescending(r => r.Priority))
|
||
{
|
||
var ruleSw = Stopwatch.StartNew();
|
||
try
|
||
{
|
||
// 6a. 全局冷却检查（距离上次触发不足 CooldownSec 秒则跳过）
|
||
if (rule.LastTriggered.HasValue &&
|
||
(DateTime.Now - rule.LastTriggered.Value).TotalSeconds < rule.CooldownSec)
|
||
{
|
||
continue;
|
||
}
|
||
|
||
// 6b. 评估所有条件
|
||
var condResults = new List<(warehouse_rulecondition cond, bool met, decimal? actualValue)>();
|
||
foreach (var cond in rule.warehouse_rulecondition)
|
||
{
|
||
if (!cond.DeviceId.HasValue) continue;
|
||
if (!realtimeCache.TryGetValue(cond.DeviceId.Value, out var points)) continue;
|
||
|
||
var point = FindPointByValueId(points, cond.ValueId);
|
||
bool met = EvaluateCondition(cond, point?.Value);
|
||
condResults.Add((cond, met, point?.Value));
|
||
}
|
||
|
||
// 至少需要 1 个有效条件才评估
|
||
if (!condResults.Any()) continue;
|
||
|
||
// AND / OR 评估
|
||
bool triggered = rule.JudgmentMode == "OR"
|
||
? condResults.Any(c => c.met)
|
||
: condResults.All(c => c.met);
|
||
|
||
// 6c. 触发 → 执行动作链
|
||
if (triggered)
|
||
{
|
||
await ExecuteActionsAsync(rule, devices, condResults, defaultBaseUrl);
|
||
rule.LastTriggered = DateTime.Now;
|
||
}
|
||
rule.LastEvaluated = DateTime.Now;
|
||
// 立即保存（saveChanges=true）确保状态持久化
|
||
_ruleRepo.Update(rule, true);
|
||
|
||
// 6d. 写执行日志
|
||
var logEntry = new warehouse_rulelog
|
||
{
|
||
RuleID = rule.RuleID,
|
||
Triggered = triggered ? "是" : "否",
|
||
ConditionSummary = string.Join(";", condResults.Select(c =>
|
||
$"id={c.cond.id}({c.met},val={c.actualValue?.ToString() ?? "null"})")),
|
||
EvaluatedAt = DateTime.Now,
|
||
DurationMs = (int)ruleSw.ElapsedMilliseconds
|
||
};
|
||
_logRepo.Add(logEntry, true);
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.LogError(ex, "[RuleEngine] 规则 {RuleID} 评估失败", rule.RuleID);
|
||
}
|
||
}
|
||
|
||
sw.Stop();
|
||
_logger.LogInformation("[RuleEngine] 评估 {Count} 条规则, 触发 {Triggered} 条, 耗时 {Ms}ms",
|
||
rules.Count,
|
||
rules.Count(r => r.LastTriggered == DateTime.Now.Date || (r.LastTriggered.HasValue && (DateTime.Now - r.LastTriggered.Value).TotalSeconds < 1)),
|
||
sw.ElapsedMilliseconds);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 执行规则下的所有动作（MVP 阶段：串行执行；并发留待 V2.1 优化）。
|
||
/// 串行原因：动作多为"控制+告警+通知"组合，串行可保证告警→通知时序一致。
|
||
/// </summary>
|
||
/// <param name="rule">触发的规则</param>
|
||
/// <param name="devices">DeviceId → 设备映射</param>
|
||
/// <param name="condResults">条件评估结果</param>
|
||
/// <param name="defaultBaseUrl">gateway_nodes.default.BaseUrl（已查好）</param>
|
||
private async Task ExecuteActionsAsync(
|
||
warehouse_rule rule,
|
||
Dictionary<int, base_device> devices,
|
||
List<(warehouse_rulecondition cond, bool met, decimal? actualValue)> condResults,
|
||
string defaultBaseUrl)
|
||
{
|
||
foreach (var action in rule.warehouse_ruleaction.OrderByDescending(a => a.id))
|
||
{
|
||
try
|
||
{
|
||
switch (action.ActionType ?? "控制")
|
||
{
|
||
case "控制":
|
||
if (action.DeviceId.HasValue && devices.TryGetValue(action.DeviceId.Value, out var dev)
|
||
&& !string.IsNullOrEmpty(dev.AdapterCode) && !string.IsNullOrEmpty(dev.SourceId))
|
||
{
|
||
var ok = await _gatewayClient.ControlDeviceAsync(
|
||
defaultBaseUrl, dev.AdapterCode, dev.SourceId,
|
||
action.ValueId ?? 0,
|
||
(double)(action.TargetValue_Number ?? 0));
|
||
_logger.LogInformation("[RuleEngine] 规则 {RuleID} 控制 {Adapter}/{SourceId} 点位 {Point} = {Value} -> {Result}",
|
||
rule.RuleID, dev.AdapterCode, dev.SourceId,
|
||
action.ValueId, action.TargetValue_Number, ok ? "OK" : "FAIL");
|
||
}
|
||
else
|
||
{
|
||
_logger.LogWarning("[RuleEngine] 规则 {RuleID} 动作 {ActionId} 设备缺失或未配置 AdapterCode/SourceId",
|
||
rule.RuleID, action.id);
|
||
}
|
||
break;
|
||
|
||
case "告警":
|
||
if (action.Alert == "是")
|
||
{
|
||
var triggerSource = condResults.FirstOrDefault(c => c.met);
|
||
string? deviceSourceId = null;
|
||
if (action.DeviceId.HasValue && devices.TryGetValue(action.DeviceId.Value, out var actionDev))
|
||
{
|
||
deviceSourceId = actionDev.SourceId;
|
||
}
|
||
await _alarmService.UpsertAlarmAsync(new SyncAlarmItem
|
||
{
|
||
SourceAlarmId = $"rule_{rule.RuleID}_{DateTime.Now:yyyyMMddHHmmssfff}",
|
||
DeviceSourceId = deviceSourceId ?? "",
|
||
AdapterCode = "RuleEngine",
|
||
Level = "重要",
|
||
Desc = action.AlertMessage ?? rule.Title,
|
||
Value = triggerSource.actualValue.HasValue ? (double)triggerSource.actualValue.Value : (double?)null,
|
||
StartTime = DateTime.Now.ToString("o")
|
||
}, action.DeviceId);
|
||
_logger.LogInformation("[RuleEngine] 规则 {RuleID} 已生成告警: {Message}",
|
||
rule.RuleID, action.AlertMessage ?? rule.Title);
|
||
}
|
||
break;
|
||
|
||
case "通知":
|
||
// 推送 SignalR（事件名 RuleTriggered 与 DataView.vue:19 订阅名一致）
|
||
await _hub.Clients.All.SendAsync("RuleTriggered", new
|
||
{
|
||
ruleId = rule.RuleID,
|
||
title = rule.Title,
|
||
alertMessage = action.AlertMessage ?? rule.Title,
|
||
level = "重要",
|
||
timestamp = DateTime.Now.ToString("o"),
|
||
triggeredDevices = condResults.Where(c => c.met)
|
||
.Select(c => c.cond.DeviceId).ToList()
|
||
});
|
||
_logger.LogInformation("[RuleEngine] 规则 {RuleID} 已推送 SignalR 通知", rule.RuleID);
|
||
break;
|
||
|
||
default:
|
||
_logger.LogWarning("[RuleEngine] 规则 {RuleID} 动作 {ActionId} 未知类型: {Type}",
|
||
rule.RuleID, action.id, action.ActionType);
|
||
break;
|
||
}
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
// 单条动作失败不阻塞后续动作
|
||
_logger.LogError(ex, "[RuleEngine] 规则 {RuleID} 动作 {ActionId} 执行失败: {Type}",
|
||
rule.RuleID, action.id, action.ActionType);
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 在设备实时点位列表里按 ValueId（点位索引）查找。
|
||
/// </summary>
|
||
/// <param name="points">设备的实时点位列表</param>
|
||
/// <param name="valueId">点位索引（warehouse_rulecondition.ValueId）</param>
|
||
/// <returns>匹配的点位，未找到返回 null</returns>
|
||
private PointValue? FindPointByValueId(List<PointValue> points, int? valueId)
|
||
{
|
||
if (!valueId.HasValue) return null;
|
||
return points.FirstOrDefault(p => p.PointIndex == valueId.Value);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 评估单条条件是否满足（数值/开关 + 比较运算 + 滞后窗）。
|
||
/// </summary>
|
||
/// <param name="cond">规则条件</param>
|
||
/// <param name="actualValue">实际实时值（点位 Value）</param>
|
||
/// <returns>true=满足条件</returns>
|
||
private bool EvaluateCondition(warehouse_rulecondition cond, decimal? actualValue)
|
||
{
|
||
// 没有实际值（如设备离线/点位无数据）→ 不满足
|
||
if (!actualValue.HasValue) return false;
|
||
|
||
bool isSwitch = cond.Type == "开关状态";
|
||
bool isNumeric = cond.Type == "数值";
|
||
|
||
if (isSwitch)
|
||
{
|
||
// 开关型：actualValue 非 0 = "开"，0 = "关"
|
||
var actualSwitch = actualValue.Value != 0 ? "开" : "关";
|
||
var targetSwitch = cond.TargetValue_Switch ?? "开";
|
||
bool met = CompareByOperator(cond.CompareOperator, actualSwitch, targetSwitch);
|
||
// 滞后窗：如果当前未满足且配置了恢复阈值，则用恢复阈值再判一次
|
||
if (!met && !string.IsNullOrEmpty(cond.RecoveryThreshold_Switch))
|
||
{
|
||
met = actualSwitch == cond.RecoveryThreshold_Switch;
|
||
}
|
||
return met;
|
||
}
|
||
else
|
||
{
|
||
// 数值型（默认走数值分支）
|
||
var actual = actualValue.Value;
|
||
var target = cond.TargetValue_Number ?? 0;
|
||
bool met = CompareByOperator(cond.CompareOperator, (double)actual, (double)target);
|
||
// 滞后窗
|
||
if (!met && cond.RecoveryThreshold_Numeric.HasValue)
|
||
{
|
||
met = CompareByOperator(cond.CompareOperator, (double)actual, (double)cond.RecoveryThreshold_Numeric.Value);
|
||
}
|
||
return met;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// 通用比较器（支持 > < = ≥ ≤ ≠）。
|
||
/// </summary>
|
||
private bool CompareByOperator(string? op, double actual, double target)
|
||
{
|
||
return (op ?? "=") switch
|
||
{
|
||
">" => actual > target,
|
||
"<" => actual < target,
|
||
"=" => Math.Abs(actual - target) < 0.0001,
|
||
"≥" or ">=" => actual >= target,
|
||
"≤" or "<=" => actual <= target,
|
||
"≠" or "!=" => Math.Abs(actual - target) >= 0.0001,
|
||
_ => false
|
||
};
|
||
}
|
||
|
||
/// <summary>
|
||
/// 字符串比较器（开关型专用）。
|
||
/// </summary>
|
||
private bool CompareByOperator(string? op, string actual, string target)
|
||
{
|
||
return (op ?? "=") switch
|
||
{
|
||
"=" => actual == target,
|
||
"≠" or "!=" => actual != target,
|
||
_ => false
|
||
};
|
||
}
|
||
}
|