fix

apps/modbus/include/modbus_ast.hrl

@@ -71,7 +71,8 @@
     name,
     address,
     type,
-    unit
+    unit,
+    scale = 1
 }).
 
 -record(modbus_control, {
@@ -114,6 +115,5 @@
 -record(ast, {
     modbus,
     devices = [],
-    processors = [],
     alarms = []
 }).

apps/modbus/src/modbus_alarm.erl

@@ -82,7 +82,7 @@ handle_call(_Request, _From, State = #state{}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_cast({push_var, Key, Val}, State = #state{alarm = #modbus_alarm{condition = Condition, hold_time = HoldTime}, history = History}) ->
+handle_cast({push_var, Key, Val}, State = #state{parent_pid = ParentPid, alarm = #modbus_alarm{name = AlarmName, condition = Condition, hold_time = HoldTime}, history = History}) ->
     case binary:match(Condition, Key) of
         nomatch ->
             {noreply, State};
@@ -95,6 +95,8 @@ handle_cast({push_var, Key, Val}, State = #state{alarm = #modbus_alarm{condition
             NHistory = [{Timestamp, IsSatisfied}|History],
             case hold(NHistory, HoldTime) of
                 true ->
+                    %% 告警
+                    ParentPid ! {device_alarm, AlarmName, HoldTime, Val},
                     lager:warning("[push_var] hold will trigger");
                 false ->
                     lager:debug("[push_var] IsSatisfied: ~p, hold will not trigger", [IsSatisfied])

apps/modbus/src/modbus_device.erl

@@ -110,13 +110,15 @@ handle_info({request_reply, Ref, Val}, State = #state{parent_pid = ParentPid, de
         error ->
             {noreply, State};
         {MetricName, NInflight} ->
-            #modbus_metric{type = Type} = maps:get(MetricName, MetricsMap),
+            #modbus_metric{name = MetricName, unit = Uint, type = Type, scale = Scale} = maps:get(MetricName, MetricsMap),
             lager:debug("[modbus_device] metric_name: ~p, get value: ~p", [MetricName, Val]),
-            %% todo 还需要解决数据的上传问题
+
-            RealVal = parse_value(Val, Type),
+            RealVal = parse_value(Val, Type, Scale),
             lager:debug("[modbus_device] real val is: ~p", [RealVal]),
 
-            %% 解决值的广播问题
+            %% 还需要解决数据的上传问题
+            ParentPid ! {device_report, DeviceName, MetricName, RealVal, Uint},
+            %% 解决值的广播问题, 给报警处理
             Key = <<"$", DeviceName/binary, ".", MetricName/binary>>,
             ParentPid ! {broadcast, Key, Val},
 
@@ -179,17 +181,17 @@ get_register_count(<<"float64">>) ->
 get_register_count(_) ->
     1.
 
-parse_value(<<Val:8>>, _) ->
+parse_value(<<Val:8>>, _, Scale) ->
-    Val;
+    Val * Scale;
-parse_value(<<Val:16>>, <<"int16">>) ->
+parse_value(<<Val:16>>, <<"int16">>, Scale) ->
-    Val;
+    Val * Scale;
-parse_value(<<Val:16>>, <<"uint16">>) ->
+parse_value(<<Val:16>>, <<"uint16">>, Scale) ->
-    Val;
+    Val * Scale;
-parse_value(<<Val:32>>, <<"int32">>) ->
+parse_value(<<Val:32>>, <<"int32">>, Scale) ->
-    Val;
+    Val * Scale;
-parse_value(<<Val:32/unsigned-integer>>, <<"uint32">>) ->
+parse_value(<<Val:32/unsigned-integer>>, <<"uint32">>, Scale) ->
-    Val;
+    Val * Scale;
-parse_value(<<Val:32/big-float>>, <<"float32">>) ->
+parse_value(<<Val:32/big-float>>, <<"float32">>, Scale) ->
-    Val;
+    Val * Scale;
-parse_value(<<Val:64/big-float>>, <<"float32">>) ->
+parse_value(<<Val:64/big-float>>, <<"float32">>, Scale) ->
-    Val.
+    Val * Scale.

apps/modbus/src/modbus_parser.erl

@@ -47,14 +47,6 @@ parse(Input) when is_binary(Input) ->
                     _ -> false
                 end
                                    end, Trees),
-            Processors = lists:filter(fun(E) ->
-                case E of
-                    #modbus_processor{} ->
-                        true;
-                    _ ->
-                        false
-                end
-                                      end, Trees),
             Alarms = lists:filter(fun(E) ->
                 case E of
                     #modbus_alarm{} ->
@@ -67,7 +59,7 @@ parse(Input) when is_binary(Input) ->
             case length(Modbus) == 1 of
                 true ->
                     NDevices = lists:map(fun(Device) -> merge_io(Device, Templates) end, Devices),
-                    AST = #ast{modbus = hd(Modbus), devices = NDevices, processors = Processors, alarms = Alarms },
+                    AST = #ast{modbus = hd(Modbus), devices = NDevices, alarms = Alarms },
                     {ok, AST};
                 false ->
                     {error, modbus_block_error}
@@ -193,13 +185,6 @@ parse_ast0(#block{ident = <<"device", Name0/binary>>, props = Props}) ->
         metrics = maps:get(<<"metrics">>, MapProps, undefined),
         controls = maps:get(<<"controls">>, MapProps, undefined)
     };
-parse_ast0(#block{ident = <<"processor", Name0/binary>>, props = Props}) ->
-    MapProps = parse_ast1(Props),
-    #modbus_processor {
-        name = string:trim(Name0),
-        input = map_of_binary(<<"input">>, MapProps, <<"">>),
-        transform = maps:get(<<"transform">>, MapProps, undefined)
-    };
 parse_ast0(#block{ident = <<"alarm", Name0/binary>>, props = Props}) ->
     MapProps = parse_ast1(Props, []),
     #modbus_alarm {

apps/modbus/src/modbus_processor.erl

@@ -1,113 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author anlicheng
-%%% @copyright (C) 2025, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 24. 6月 2025 10:49
-%%%-------------------------------------------------------------------
--module(modbus_processor).
--author("anlicheng").
--include("modbus_ast.hrl").
-
--behaviour(gen_server).
-
-%% API
--export([start_link/2]).
-
-%% gen_server callbacks
--export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
-
--define(SERVER, ?MODULE).
-
--record(state, {
-    parent_pid :: pid(),
-    processor :: #modbus_processor{}
-}).
-
-%%%===================================================================
-%%% API
-%%%===================================================================
-
-%% @doc Spawns the server and registers the local name (unique)
--spec(start_link(ParentPid :: pid(), Processor :: #modbus_processor{}) ->
-    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
-start_link(ParentPid, Processor = #modbus_processor{}) when is_pid(ParentPid) ->
-    gen_server:start_link(?MODULE, [ParentPid, Processor], []).
-
-%%%===================================================================
-%%% gen_server callbacks
-%%%===================================================================
-
-%% @private
-%% @doc Initializes the server
--spec(init(Args :: term()) ->
-    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term()} | ignore).
-init([ParentPid, Processor = #modbus_processor{input = Input}]) ->
-    %% 处理input
-    lager:debug("processor is: ~p", [Processor]),
-
-    case binary:split(Input, <<".">>) of
-        [<<$$, DeviceName/binary>>, Metric] ->
-            lager:debug("device_name is: ~p, metric is: ~p", [DeviceName, Metric]);
-        _ ->
-            lager:debug("invalid input: ~p", [Input])
-    end,
-
-
-
-    {ok, #state{parent_pid = ParentPid, processor = Processor}}.
-
-%% @private
-%% @doc Handling call messages
--spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
-    State :: #state{}) ->
-    {reply, Reply :: term(), NewState :: #state{}} |
-    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
-    {noreply, NewState :: #state{}} |
-    {noreply, NewState :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
-    {stop, Reason :: term(), NewState :: #state{}}).
-handle_call(_Request, _From, State = #state{}) ->
-    {reply, ok, State}.
-
-%% @private
-%% @doc Handling cast messages
--spec(handle_cast(Request :: term(), State :: #state{}) ->
-    {noreply, NewState :: #state{}} |
-    {noreply, NewState :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term(), NewState :: #state{}}).
-handle_cast(_Request, State = #state{}) ->
-    {noreply, State}.
-
-%% @private
-%% @doc Handling all non call/cast messages
--spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
-    {noreply, NewState :: #state{}} |
-    {noreply, NewState :: #state{}, timeout() | hibernate} |
-    {stop, Reason :: term(), NewState :: #state{}}).
-handle_info(_Info, State = #state{}) ->
-    {noreply, State}.
-
-%% @private
-%% @doc This function is called by a gen_server when it is about to
-%% terminate. It should be the opposite of Module:init/1 and do any
-%% necessary cleaning up. When it returns, the gen_server terminates
-%% with Reason. The return value is ignored.
--spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
-    State :: #state{}) -> term()).
-terminate(_Reason, _State = #state{}) ->
-    ok.
-
-%% @private
-%% @doc Convert process state when code is changed
--spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
-    Extra :: term()) ->
-    {ok, NewState :: #state{}} | {error, Reason :: term()}).
-code_change(_OldVsn, State = #state{}, _Extra) ->
-    {ok, State}.
-
-%%%===================================================================
-%%% Internal functions
-%%%===================================================================

apps/modbus/src/modbus_service.erl

@@ -45,8 +45,6 @@
 
     %% #{slaveId => DevicePid}
     devices_map = #{},
-    %% 所有的处理过程
-    processor_pids = [],
     %% alarms
     alarm_pids = [],
 
@@ -174,6 +172,11 @@ handle_event(info, {timeout, _, delay_locking}, ?CONNECTED, State = #state{mode
     lager:debug("[modbus_service] delay unlock trigger next"),
     {keep_state, State#state{mode = Mode#rtu_mode{is_busy = false}}, [{next_event, info, read_next}]};
 
+%% todo 数据上报
+handle_event(info, {device_report, DeviceName, Name, Value, Uint}, ?CONNECTED, State) ->
+    lager:debug("[modbus_service] get report_data device_name: ~p, name: ~p, value: ~p, uint: ~p", [DeviceName, Name, Value, Uint]),
+    {keep_state, State};
+
 %% 广播消息
 handle_event(info, {broadcast, Key, Val}, ?CONNECTED, State = #state{alarm_pids = AlarmPids}) ->
     %% 推送告警信息

modbus.conf

@@ -160,50 +160,10 @@ device xyz {
     }
 }
 
-processor temperature_processor {
-    # 输入源
-    input $boiler_controller.temperature;
-
-    # 数据转换
-    transform {
-        # 线性转换: raw * 0.1 + 2.5
-        linear 0.1 2.5;
-
-        # 滤波
-        moving_average 5;
-
-        # 范围检查
-        validate {
-            min -10.0;
-            max 150.0;
-            action log;  # or 'discard', 'replace_with: value'
-        }
-    }
-
-    # 输出目标
-    #output {
-    #    mqtt "sensors/boiler/temp";
-    #    influxdb "plant_metrics" measurement="temperature";
-    #}
-}
-
 alarm high_temperature {
     # 触发条件
     condition $boiler_controller.temperature > 90.0;
 
     # 持续判定
     hold_time 30s;
-
-    # 动作
-    actions {
-        log "CRITICAL: Boiler temperature too high!";
-        mqtt "alarms/boiler";
-        email "ops@example.com";
-        exec "/usr/local/bin/shutdown_boiler.sh";
-    }
-
-    # 恢复动作
-    recovery_actions {
-        log "Boiler temperature back to normal";
-    }
 }