fix

apps/modbus/src/modbus_device.erl

@@ -13,12 +13,13 @@
 -behaviour(gen_server).
 
 %% API
--export([start_link/1]).
+-export([start_link/2]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
 
 -record(state, {
+    parent_pid :: pid(),
     device :: #modbus_device{}
 }).
 
@@ -27,10 +28,10 @@
 %%%===================================================================
 
 %% @doc Spawns the server and registers the local name (unique)
--spec(start_link(Device :: #modbus_device{}) ->
+-spec(start_link(ParentPid :: pid(), Device :: #modbus_device{}) ->
     {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
-start_link(Device = #modbus_device{}) ->
-    gen_server:start_link(?MODULE, [], []).
+start_link(ParentPid, Device = #modbus_device{}) when is_pid(ParentPid) ->
+    gen_server:start_link(?MODULE, [ParentPid, Device], []).
 
 %%%===================================================================
 %%% gen_server callbacks
@@ -41,9 +42,17 @@ start_link(Device = #modbus_device{}) ->
 -spec(init(Args :: term()) ->
     {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term()} | ignore).
-init([Device]) ->
+init([ParentPid, Device = #modbus_device{metrics = Metrics, poll_interval = PollInterval}]) ->
+    %% 处理采集项目
 
-    {ok, #state{device = Device}}.
+    #modbus_metric{
+        name = Name,
+        address = Address,
+        type = Type,
+        unit = Uint
+    } = hd(Metrics),
+
+    {ok, #state{parent_pid = ParentPid, device = Device}}.
 
 %% @private
 %% @doc Handling call messages
@@ -73,6 +82,16 @@ handle_cast(_Request, State = #state{}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
+handle_info({timeout, _, poll_ticker}, State = #state{parent_pid = ParentPid,
+    device = #modbus_device{slave_id = SlaveId, poll_interval = PollInterval, metrics = Metrics}}) ->
+    poll_ticker(PollInterval),
+
+    %% 读取采集项目
+    lists:foreach(fun(#modbus_metric{address = Address}) ->
+        ParentPid ! {read, SlaveId, Address}
+                  end, Metrics),
+
+    {noreply, State};
 handle_info(_Info, State = #state{}) ->
     {noreply, State}.
 
@@ -97,3 +116,6 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
 %%%===================================================================
 %%% Internal functions
 %%%===================================================================
+
+poll_ticker(PollInterval) when is_integer(PollInterval) ->
+    erlang:start_timer(PollInterval, self(), poll_ticker).

apps/modbus/src/modbus_service.erl

@@ -13,7 +13,8 @@
 -behaviour(gen_server).
 
 %% rtu指令
--define(CONNECT, 16#01).
+-define(MODBUS_CONNECT, 16#01).
+-define(MODBUS_READ, 16#02).
 
 %% API
 -export([start_link/1]).
@@ -63,6 +64,8 @@ init([AST = #ast{modbus = Modbus, devices = Devices}]) ->
     % Res = connect(Transport),
     % lager:debug("connect res: ~p", [Res]),
 
+    {ok, DevicePid} = modbus_device:start_link(hd(Devices)),
+    lager:debug("device pid: ~p", [DevicePid]),
 
     %{ok, AccessLogPid} = modbus_logger:start_link(AccessLog),
     %{ok, ErrorLogPid} = modbus_logger:start_link(ErrorLog),
@@ -97,6 +100,12 @@ handle_cast(_Request, State = #state{}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
+%% 发送读取指令
+handle_info({read, SlaveId, Address}, State = #state{port = Port}) ->
+    ReadCmd = <<?MODBUS_READ:8, SlaveId:8, Address:16>>,
+    Port !  {self(), {command, ReadCmd}},
+    ok;
+
 handle_info({Port, {data, Data}}, State = #state{port = Port}) ->
     lager:debug("port data is: ~p", [Data]),
     {noreply, State};
@@ -131,7 +140,7 @@ connect(#modbus{transport = #modbus_transport_rtu{port = Port, baudrate = Baudra
     Port = erlang:open_port({spawn_executable, RealExecCmd}, [binary, {packet, 2}, exit_status]),
 
     Len0 = byte_size(Port),
-    ConnectCmd = <<?CONNECT:8, Len0:8, Port/binary, Baudrate:32, Stopbits:8, Parity:8, Timeout:32>>,
+    ConnectCmd = <<?MODBUS_CONNECT:8, Len0:8, Port/binary, Baudrate:32, Stopbits:8, Parity:8, Timeout:32>>,
     %% 建立连接
     Port ! {self(), {command, ConnectCmd}},
     ok;

apps/modbus/src/modbus_service2.erl

@@ -0,0 +1,134 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 20. 6月 2025 16:38
+%%%-------------------------------------------------------------------
+-module(modbus_service2).
+-author("anlicheng").
+-include("modbus_ast.hrl").
+
+-behaviour(gen_statem).
+
+%% API
+-export([start_link/1]).
+-export([test/0]).
+
+%% gen_statem callbacks
+-export([init/1, format_status/2, state_name/3, handle_event/4, terminate/3, code_change/4, callback_mode/0]).
+
+%% rtu指令
+-define(MODBUS_CONNECT, 16#01).
+-define(MODBUS_READ, 16#02).
+
+%% 当前的状态
+-define(DISCONNECTED, disconnected).
+-define(CONNECTED, connected).
+
+-record(state, {
+    port,
+    access_log_pid :: pid() | undefined,
+    error_log_pid :: pid() | undefined
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+test() ->
+    {ok, Config} = file:read_file("/usr/local/code/cloudkit/modbus/modbus.conf"),
+    %lager:debug("config is: ~ts", [Config]),
+    {ok, AST} = modbus_parser:parse(Config),
+    start_link(AST).
+
+%% @doc Creates a gen_statem process which calls Module:init/1 to
+%% initialize. To ensure a synchronized start-up procedure, this
+%% function does not return until Module:init/1 has returned.
+start_link(AST = #ast{}) ->
+    gen_statem:start_link(?MODULE, [AST], []).
+
+%%%===================================================================
+%%% gen_statem callbacks
+%%%===================================================================
+
+%% @private
+%% @doc Whenever a gen_statem is started using gen_statem:start/[3,4] or
+%% gen_statem:start_link/[3,4], this function is called by the new
+%% process to initialize.
+init([AST = #ast{modbus = Modbus, devices = Devices}]) ->
+    lager:debug("modbus is: ~p", [Modbus]),
+    Device = hd(Devices),
+    lager:debug("devices is: ~p", [Device#modbus_device.metrics]),
+    % Res = connect(Transport),
+    % lager:debug("connect res: ~p", [Res]),
+
+    {ok, DevicePid} = modbus_device:start_link(hd(Devices)),
+    lager:debug("device pid: ~p", [DevicePid]),
+
+    %{ok, AccessLogPid} = modbus_logger:start_link(AccessLog),
+    %{ok, ErrorLogPid} = modbus_logger:start_link(ErrorLog),
+
+    {ok, ?DISCONNECTED, #state{}}.
+
+%% @private
+%% @doc This function is called by a gen_statem when it needs to find out
+%% the callback mode of the callback module.
+callback_mode() ->
+    handle_event_function.
+
+%% @private
+%% @doc If callback_mode is handle_event_function, then whenever a
+%% gen_statem receives an event from call/2, cast/2, or as a normal
+%% process message, this function is called.
+
+handle_event(info, {read, SlaveId, Address}, ?CONNECTED, State = #state{port = Port}) ->
+    ReadCmd = <<?MODBUS_READ:8, SlaveId:8, Address:16>>,
+    Port !  {self(), {command, ReadCmd}},
+    {keep_state, State};
+
+%% 从port读取数据
+handle_event(info, {Port, {data, Data}}, ?CONNECTED, State = #state{port = Port}) ->
+    lager:debug("port data is: ~p", [Data]),
+    {keep_state, State};
+
+handle_event(_EventType, _EventContent, _StateName, State = #state{}) ->
+    NextStateName = the_next_state_name,
+    {next_state, NextStateName, State}.
+
+%% @private
+%% @doc This function is called by a gen_statem 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_statem terminates with
+%% Reason. The return value is ignored.
+terminate(_Reason, _StateName, _State = #state{}) ->
+    ok.
+
+%% @private
+%% @doc Convert process state when code is changed
+code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
+    {ok, StateName, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+connect(#modbus{transport = #modbus_transport_rtu{port = Port, baudrate = Baudrate, stopbits = Stopbits, parity = Parity, timeout = Timeout}}) ->
+    RealExecCmd = "",
+    Port = erlang:open_port({spawn_executable, RealExecCmd}, [binary, {packet, 2}, exit_status]),
+
+    Len0 = byte_size(Port),
+    ConnectCmd = <<?MODBUS_CONNECT:8, Len0:8, Port/binary, Baudrate:32, Stopbits:8, Parity:8, Timeout:32>>,
+    %% 建立连接
+    Port ! {self(), {command, ConnectCmd}},
+    ok;
+
+connect(#modbus{transport = #modbus_transport_tcp{host = Host, port = Port, timeout = Timeout0}}) ->
+    Timeout = case is_integer(Timeout0) andalso Timeout0 > 0 of
+                  true ->
+                      Timeout0;
+                  false ->
+                      2000
+              end,
+    gen_tcp:connect(Host, Port, [binary], Timeout).