规则引擎

规则引擎的计算规则是：首先会通过 ruleSets 参数，设定一组规则，字典格式，key 表示规则的键值，为 NULL 时，此条规则为默认规则。ruleSetColumn 设置输入数据中命中规则的字段名，当字段值等于 ruleSet 中的 key 值时表示命中此规则，然后根据 policy 的设置输出引擎的计算结果，此时如果制定了 callback 则同时会触发回调处理逻辑。ruleSetColumn 若没指定或者没有命中规则，那么使用默认规则处理。

采样电力设备的电流、电压、和温度，设置规则集进行告警监控。

采集的数据存放到流数据表中，规则引擎通过订阅流数据表来获取实时数据，并进行规则检查，将检查结果输出到另外一个表中。

实现步骤如下：

(1) 定义流表存放采集的数据

share streamTable(1000:0,`time`pointId`voltage`current`temprature,[TIMESTAMP,STRING,DOUBLE,DOUBLE,DOUBLE]) as inputTable

(2) 定义规则引擎的输出流表以及规则引擎回调函数处理后的流表

// 创建流表 保存规则引擎的输出数据
share streamTable(1000:0,`time`pointId`voltage`current`temprature`inputTime`rule,[TIMESTAMP,STRING,DOUBLE,DOUBLE,DOUBLE,NANOTIMESTAMP,BOOL[]]) as outputTable
// 创建流表 保存规则引擎回调函数处理后数据
share streamTable(1000:0,`time`pointId`voltage`current`temprature`inputTime`outputTime`comment,[TIMESTAMP,STRING,DOUBLE,DOUBLE,DOUBLE,NANOTIMESTAMP,NANOTIMESTAMP,STRING]) as resultTable

(3) 定义规则集

// 定义规则集
n = 10
pointId=`ID + string(1..n)
voltageHigh=round(double(rand(220..230,n))-rand(0.6,n),2)
voltageLow=round(double(rand(45..50,n)) +rand(0.5,n),2)
currentHigh=round(double(rand(40..45,n))-rand(0.6,n),2)    
currentLow=round(double(rand(7..9,n))+rand(0.5,n),2)    
tempratureHigh=round(double(rand(40..45,n))-rand(0.6,n),2)
tempratureLow=round(double(rand(7..9,n))+rand(0.5,n),2)
pt = table(pointId,voltageHigh,voltageLow,currentHigh,currentLow,tempratureHigh,tempratureLow)

ids = exec pointId from pt
ruleSet = dict(STRING,ANY)
for(i in 0:pt.size()){
    tmp = pt[i]
    keys = tmp.keys()[1:]
    value = tmp.values()[1:]
    a = array(STRING)
    for(j in 0:keys.size()){
        if(keys[j] like "%High"){
            s=strpos(keys[j],"High");
            a.append!(keys[j][0:s]+ ">" + value[j] )
        }else{
            s=strpos(keys[j],"Low");
            a.append!(keys[j][0:s] + "<" + value[j])
        }
    }
    ruleSet[ids[i]] = parseExpr(a) 
}
ruleSet[string(NULL)] = [ <voltage > 100000>]

(4) 定义回调函数

// 定义规则引擎的回调函数
def writeBack(result){
    outputTime = now(true)
    if(result.rule[0]==true){
        s="电压过高"
        insert into resultTable values (result.time,result.pointId,result.voltage,result.current,result.temprature,result.inputTime,outputTime,s)
    }
    if(result.rule[1]==true){
        s="电压过低"
        insert into resultTable values (result.time,result.pointId,result.voltage,result.current,result.temprature,result.inputTime,outputTime,s)
    }
    if(result.rule[2]==true){
        s="电流过高"
        insert into resultTable values (result.time,result.pointId,result.voltage,result.current,result.temprature,result.inputTime,outputTime,s)
    }
    if(result.rule[3]==true){
        s="电流过低"
        insert into resultTable values (result.time,result.pointId,result.voltage,result.current,result.temprature,result.inputTime,outputTime,s)
    }
    if(result.rule[4]==true){
        s="温度过高"
        insert into resultTable values (result.time,result.pointId,result.voltage,result.current,result.temprature,result.inputTime,outputTime,s)
    }
    if(result.rule[5]==true){
        s="温度过低"
        insert into resultTable values (result.time,result.pointId,result.voltage,result.current,result.temprature,result.inputTime,outputTime,s)
    }
}

(5) 创建规则引擎

// 创建规则引擎
colNames = inputTable.schema().colDefs.name join `inputTime
colTypes = inputTable.schema().colDefs.typeString join `NANOTIMESTAMP
schemaTable = table(1:0,colNames,colTypes)
ruleEngine=createRuleEngine(
    name="ruleEngine",
    ruleSets=ruleSet,
    dummyTable=schemaTable,
    outputColumns=["time","pointId","voltage","current","temprature","inputTime"],
    outputTable=outputTable, 
    policy="all", 
    ruleSetColumn="pointId",
    callback=writeBack
)

(6) 订阅流表写入 10 条模拟数据

def handle(msg){
    tmp = select *,now(true) as inputTime from msg  
    getStreamEngine("ruleEngine").append!(tmp)
}
// 订阅数据数据流表
subscribeTable(
    tableName="inputTable", 
    actionName="RuleEngine", 
    handler=handle, 
    msgAsTable=true,
    offset=0
)
// 生成10条模拟数据
n =10
t=table(n:n,`time`pointId`voltage`current`temprature,[TIMESTAMP,STRING,DOUBLE,DOUBLE,DOUBLE])
t["time"] = now() +( 1..10) *1000
t["pointId"]=`ID + string(1..n)
t["voltage"]=round(double(rand(45..225,n))+rand(0.5,n) ,2)
t["current"]=round(double(rand(7..43,n))+rand(0.5,n),2)  
t["temprature"]=round(double(rand(7..43,n))+rand(0.5,n),2) 
// 向流表写入模拟数据
inputTable.append!(t)

查看输入数据表

查看规则引擎的输出数据表

查看规则引擎回调函数处理后的输出表

结果解释：

设备 'ID5' 的规则集为：

(< voltage > 221.86 >,< voltage < 45.16 >,< current > 39.96 >,< current < 7.47 >,< temprature > 44.43 >,< temprature < 8.03 >)

输入数据命中规则集中的 current > 39.96，经回调函数处理，输出“电流过高”；

设备 'ID6' 的规则集为：

(< voltage > 221.89 >,< voltage < 49.05 >,< current > 43.42 >,< current < 7.06 >,< temprature > 41.98 >,< temprature < 7.418>)

输入数据命中规则集中的 temprature > 41.98 ，经回调函数处理，输出“温度过高”；

设备 'ID7' 的规则集为：

(< voltage > 226.61 >,< voltage < 48.40 >,< current > 40.77 >,< current < 9.19 >,< temprature > 41.95 >,< temprature < 8.14 >)

输入数据命中规则集中的 < temprature > 41.95，经回调函数处理，输出“温度过高”；

设备 'ID8' 的规则集为：

(< voltage > 228.97 >,< voltage < 46.137 >,< current > 42.607 >,< current < 7.29 >,< temprature > 44.40 >,< temprature < 9.27 >)

输入数据命中规则集中的 temprature < 9.27，经回调函数处理，输出“温度过高”；

(7) 动态调整规则示例

//修改规则
newRules = [ < voltage > 223.41 >,< voltage < 49.25 >,< current > 36.84 >,< current < 8.4 >,< temprature > 44.60 >,< temprature < 7.43 >]
updateRule("ruleEngine","ID1",newRules)
//插入一条模拟数据
insert into inputTable values (now(), 'ID1',100.15,37.19,22.33)

current= 37.19，之前的规则未触发报警，现在命中更新后的规则 current > 36.84，输出表新增记录：