diff --git a/apps/modbus/include/modbus_ast.hrl b/apps/modbus/include/modbus_ast.hrl index 8f84e32..d885280 100644 --- a/apps/modbus/include/modbus_ast.hrl +++ b/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 = [] }). \ No newline at end of file diff --git a/apps/modbus/src/modbus_alarm.erl b/apps/modbus/src/modbus_alarm.erl index 7ad8a3d..89c6f7d 100644 --- a/apps/modbus/src/modbus_alarm.erl +++ b/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]) diff --git a/apps/modbus/src/modbus_device.erl b/apps/modbus/src/modbus_device.erl index ea2b01e..12f8541 100644 --- a/apps/modbus/src/modbus_device.erl +++ b/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; -parse_value(<>, <<"int16">>) -> - Val; -parse_value(<>, <<"uint16">>) -> - Val; -parse_value(<>, <<"int32">>) -> - Val; -parse_value(<>, <<"uint32">>) -> - Val; -parse_value(<>, <<"float32">>) -> - Val; -parse_value(<>, <<"float32">>) -> - Val. \ No newline at end of file +parse_value(<>, _, Scale) -> + Val * Scale; +parse_value(<>, <<"int16">>, Scale) -> + Val * Scale; +parse_value(<>, <<"uint16">>, Scale) -> + Val * Scale; +parse_value(<>, <<"int32">>, Scale) -> + Val * Scale; +parse_value(<>, <<"uint32">>, Scale) -> + Val * Scale; +parse_value(<>, <<"float32">>, Scale) -> + Val * Scale; +parse_value(<>, <<"float32">>, Scale) -> + Val * Scale. \ No newline at end of file diff --git a/apps/modbus/src/modbus_parser.erl b/apps/modbus/src/modbus_parser.erl index f32d140..c8b4a78 100644 --- a/apps/modbus/src/modbus_parser.erl +++ b/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 { diff --git a/apps/modbus/src/modbus_processor.erl b/apps/modbus/src/modbus_processor.erl deleted file mode 100644 index 603b612..0000000 --- a/apps/modbus/src/modbus_processor.erl +++ /dev/null @@ -1,113 +0,0 @@ -%%%------------------------------------------------------------------- -%%% @author anlicheng -%%% @copyright (C) 2025, -%%% @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 -%%%=================================================================== diff --git a/apps/modbus/src/modbus_service.erl b/apps/modbus/src/modbus_service.erl index 99214d1..b1964c2 100644 --- a/apps/modbus/src/modbus_service.erl +++ b/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}) -> %% 推送告警信息 diff --git a/modbus.conf b/modbus.conf index 5fd093d..7dbf2c0 100644 --- a/modbus.conf +++ b/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"; - } } \ No newline at end of file