fix alarm

apps/modbus/src/modbus_alarm.erl

@@ -10,45 +10,71 @@
 -include("modbus_ast.hrl").
 -export([new/1, maybe_alarm/3]).
 
+%% 最大的个数
+-define(MAX_HISTORY_NUM, 2000).
+
+-export([test/0]).
+
 -record(alarm_mod, {
     alarm :: #modbus_alarm{},
+    parsed_condition :: binary(),
     %% 存储历史值，用来做持续性判断, {timestamp, bool}
-    history = []
+    history = [],
+    history_num = 0
 }).
 
+test() ->
+    {ok, Key0, X} = new(#modbus_alarm{name = <<"high_temperature">>, condition = <<"$temperature > 90.0 && $temperature =< 2000">>, hold_time = 5000}),
+    lists:foldl(fun(Id, Acc) ->
+        timer:sleep(1000),
+        case maybe_alarm(Key0, 100.0 + Id, Acc) of
+            {alarm, {AlarmName, HoldTime}, NAcc} ->
+                lager:debug("[modbus_alarm] triggered alarm_name: ~p, hold_time: ~p", [AlarmName, HoldTime]),
+                NAcc;
+            {ignore, NAcc} ->
+                NAcc
+        end
+                end, X, lists:seq(1, 11)).
+
 -spec new(Alarm :: #modbus_alarm{}) -> error | {ok, Key :: binary(), AlarmMod :: #alarm_mod{}}.
 new(Alarm = #modbus_alarm{condition = Condition0}) ->
-    Condition = binary_to_list(Condition0),
+    Condition = condition_operators(Condition0),
-    case re:run(Condition, "\\$(\\w+)", [global, {capture, all}, list]) of
+    case catch re:run(binary_to_list(Condition), "\\$(\\w+)", [global, {capture, all, list}]) of
         {match, Captures} ->
-            Vars = [Var || [_|Var] <- Captures],
+            Vars = [Var || [_, Var] <- Captures],
             NVars = sets:to_list(sets:from_list(Vars)),
             case length(NVars) =:= 1 of
                 true ->
-                    {ok, hd(NVars), #alarm_mod{alarm = Alarm, history = []}};
+                    {ok, list_to_binary(hd(NVars)), #alarm_mod{alarm = Alarm, parsed_condition = Condition, history = []}};
                 false ->
                     error
-            end
+            end;
+        nomatch ->
+            error
     end.
 
 -spec maybe_alarm(Key0 :: binary(), Val :: any(), AlarmMod :: #alarm_mod{}) ->
     {alarm, {AlarmName :: binary(), HoldTime :: integer()}, AlarmMod :: #alarm_mod{}} | {ignore, AlarmMod :: #alarm_mod{}}.
-maybe_alarm(Key0, Val, AlarmMod = #alarm_mod{alarm = #modbus_alarm{name = AlarmName, condition = Condition, hold_time = HoldTime}, history = History}) when is_binary(Key0) ->
+maybe_alarm(Key0, Val0, AlarmMod = #alarm_mod{parsed_condition = Condition, alarm = #modbus_alarm{name = AlarmName, hold_time = HoldTime}, history = History, history_num = HistoryNum}) when is_binary(Key0) ->
     Key = <<$$, Key0/binary>>,
+    case as_binary(Val0) of
+        {ok, Val} ->
+            Expr0 = binary:replace(Condition, Key, Val, [global]),
+            Expr = iolist_to_binary(<<Expr0/binary, ".">>),
+            IsSatisfied = eval_condition(binary_to_list(Expr)),
 
-    Expr0 = binary:replace(Condition, Key, Val, [global]),
+            Timestamp = timestamp_ms(),
-    Expr = iolist_to_binary(<<Expr0/binary, ".">>),
+            NHistory = [{Timestamp, IsSatisfied}|History],
-    IsSatisfied = eval_condition(binary_to_list(Expr)),
+            case hold(NHistory, HoldTime) of
-
+                true ->
-    Timestamp = modbus_util:current_seconds(),
+                    %% 告警
-    NHistory = [{Timestamp, IsSatisfied}|History],
+                    {alarm, {AlarmName, HoldTime}, AlarmMod#alarm_mod{history = [{Timestamp, IsSatisfied}], history_num = 1}};
-    case hold(NHistory, HoldTime) of
+                false ->
-        true ->
+                    {ignore, history_limit(AlarmMod#alarm_mod{history = NHistory, history_num = HistoryNum + 1})}
-            %% 告警
+            end;
-            {alarm, {AlarmName, HoldTime}, AlarmMod#alarm_mod{history = NHistory}};
+        error ->
-        false ->
+            lager:warning("[modbus_alarm] alarm_name: ~p, invalid val: ~p", [AlarmName, Val0]),
-            lager:debug("[push_var] IsSatisfied: ~p, hold will not trigger", [IsSatisfied]),
+            {ignore, AlarmMod}
-            {ignore, AlarmMod#alarm_mod{history = NHistory}}
     end.
 
 %%%===================================================================
@@ -71,8 +97,49 @@ eval_condition(Condition) when is_list(Condition) ->
 
 %% 用来做持续性判断
 -spec hold(History :: list(), HoldTime :: integer()) -> boolean().
-hold(History, HoldTime) when is_list(History), is_integer(HoldTime) ->
+hold(History, 0) when is_list(History) ->
-    PrefixItems = lists:takewhile(fun({_, Bool}) -> Bool end, History),
+    {_, Bool} = hd(History),
+    Bool;
+hold(History, HoldTime) when is_list(History), is_integer(HoldTime), HoldTime > 0 ->
+    Timestamps = lists:map(fun({T0, _}) -> T0 end, lists:takewhile(fun({_, Bool}) -> Bool end, History)),
+    MinTimestamp = lists:last(Timestamps),
     %% 判断是否存在一个元素的时间
-    Current = modbus_util:current_seconds(),
+    Current = timestamp_ms(),
-    lists:any(fun({T0, _}) -> T0 =< Current - HoldTime end, PrefixItems).
+    MinTimestamp =< Current - HoldTime.
+
+-spec as_binary(Val :: any()) -> error | {ok, Bin :: binary()}.
+as_binary(Val) when is_binary(Val) ->
+    {ok, Val};
+as_binary(Val) when is_list(Val) ->
+    {ok, list_to_binary(Val)};
+as_binary(Val) when is_integer(Val) ->
+    {ok, integer_to_binary(Val)};
+as_binary(Val) when is_float(Val) ->
+    Bin = list_to_binary(float_to_list(Val, [{decimals, 4}, compact])),
+    {ok, Bin};
+as_binary(Val) when is_boolean(Val) ->
+    {ok, atom_to_binary(Val)};
+as_binary(_) ->
+    error.
+
+%% 时间，精确到毫秒
+timestamp_ms() ->
+    {Mega, Seconds, Micro} = os:timestamp(),
+    (Mega * 1000000 + Seconds) * 1000 + Micro div 1000.
+
+%% 历史值限制
+-spec history_limit(Mod :: #alarm_mod{}) -> NMod :: #alarm_mod{}.
+history_limit(Mod = #alarm_mod{history = History, history_num = HistoryNum}) when HistoryNum > ?MAX_HISTORY_NUM ->
+    Mod#alarm_mod{history = lists:sublist(History, 1, ?MAX_HISTORY_NUM), history_num = ?MAX_HISTORY_NUM};
+history_limit(Mod) ->
+    Mod.
+
+%% condition支持的运算符号
+condition_operators(Condition) when is_binary(Condition) ->
+    Operators = [
+        {<<" and ">>, <<" andalso ">>},
+        {<<" or ">>, <<" orelse ">>},
+        {<<" && ">>, <<" andalso ">>},
+        {<<" || ">>, <<" orelse ">>}
+    ],
+    lists:foldl(fun({Search, Replace}, Acc0) -> binary:replace(Acc0, Search, Replace, [global]) end, Condition, Operators).

apps/modbus/src/modbus_parser.erl

@@ -51,6 +51,7 @@ parse(Input) when is_binary(Input) ->
             case length(Modbus) == 1 of
                 true ->
                     NDevices = lists:map(fun(Device) -> merge_io(Device, Templates) end, Devices),
+                    lager:debug("[devices] ~p", [NDevices]),
                     AST = #ast{modbus = hd(Modbus), devices = NDevices},
                     {ok, AST};
                 false ->
@@ -160,8 +161,9 @@ parse_ast0(#block{ident = <<"device_io", Name0/binary>>, props = Props}) ->
         poll_interval = map_of_time(<<"poll_interval">>, MapProps, 0),
         retries = map_of_integer(<<"retries">>, MapProps, 0),
         retry_timeout = maps:get(<<"retry_timeout">>, MapProps, undefined),
-        metrics = maps:get(<<"metrics">>, MapProps, undefined),
+        metrics = maps:get(<<"metrics">>, MapProps, []),
-        controls = maps:get(<<"controls">>, MapProps, undefined)
+        controls = maps:get(<<"controls">>, MapProps, []),
+        alarms = maps:get(<<"alarms">>, MapProps, [])
     };
 parse_ast0(#block{ident = <<"device", Name0/binary>>, props = Props}) ->
     MapProps = parse_ast1(Props),
@@ -174,8 +176,9 @@ parse_ast0(#block{ident = <<"device", Name0/binary>>, props = Props}) ->
         poll_interval = map_of_time(<<"poll_interval">>, MapProps, 0),
         retries = map_of_integer(<<"retries">>, MapProps, 0),
         retry_timeout = maps:get(<<"retry_timeout">>, MapProps, undefined),
-        metrics = maps:get(<<"metrics">>, MapProps, undefined),
+        metrics = maps:get(<<"metrics">>, MapProps, []),
-        controls = maps:get(<<"controls">>, MapProps, undefined)
+        controls = maps:get(<<"controls">>, MapProps, []),
+        alarms = maps:get(<<"alarms">>, MapProps, [])
     }.
 
 parse_ast1(Props) ->
@@ -232,8 +235,8 @@ parse_ast1([#block{ident = <<"alarms">>, props = Controls0}|T], Acc) ->
         PropsMap = maps:from_list(Props),
         #modbus_alarm{
             name = AlarmName,
-            condition = map_of_binary(<<"condition">>, PropsMap, undefined),
+            condition = map_of_binary(<<"condition">>, PropsMap, <<"">>),
-            hold_time = map_of_time(<<"hold_time">>, PropsMap, undefined)
+            hold_time = map_of_time(<<"hold_time">>, PropsMap, 0)
         }
                          end, Controls0),
     parse_ast1(T, [{<<"alarms">>, Alarms}|Acc]);
@@ -328,7 +331,9 @@ map_of_time(Key, M, Def) ->
                         <<"ms">> ->
                             Digit;
                         <<"s">> ->
-                            Digit * 1000
+                            Digit * 1000;
+                        <<"min">> ->
+                            Digit * 60 * 1000
                     end;
                 _ ->
                     Def
@@ -361,4 +366,5 @@ merge_io(Device = #modbus_device{device_io = IOName}, Templates) ->
             };
         false ->
             Device
-    end.
+    end.
+

modbus.conf

@@ -75,6 +75,14 @@ device_io t1 {
            # 持续判定
            hold_time 30s;
        }
+
+        high_temperature {
+          # 触发条件
+          condition $pressure > 10.0;
+
+          # 持续判定
+          hold_time 50s;
+      }
    }
 
 }