add docker env

.args.yaml

@@ -0,0 +1,26 @@
+metrics:
+  -
+    key: serial
+    type: string
+    label: "空调设备ID"
+    config:
+      default: ""
+      maxlen: 20
+  -
+    key: device_uuid
+    type: string
+    label: "设备UUID"
+    config:
+      default: ""
+      maxlen: 50
+
+boot: "./boot.sh"
+# stop can be used for execute stop action
+stop: "./stop.sh"
+
+# coll / collect for collection, has data upload
+# stream has transform
+# up / upload for data upload
+service_type: "coll"
+
+run_as_root: true

Dockerfile

@@ -0,0 +1,3 @@
+FROM erlang:25.3
+
+CMD /data/aircon/bin/aircon foreground

apps/aircon/src/aircon.app.src

@@ -1,5 +1,5 @@
 {application, aircon,
- [{description, "An OTP application"},
+ [{description, "aircon gateway application"},
   {vsn, "0.1.0"},
   {registered, []},
   {mod, {aircon_app, []}},

apps/aircon/src/aircon_app.erl

@@ -11,7 +11,6 @@
 
 start(_StartType, _StartArgs) ->
     set_logger_level(),
-    logger:warning("call me here start logger"),
     aircon_sup:start_link().
 
 stop(_State) ->
@@ -22,6 +21,8 @@ stop(_State) ->
 set_logger_level() ->
     logger:set_application_level(kernel, notice),
     logger:set_application_level(stdlib, notice),
-    logger:set_application_level(emqtt, debug),
+    logger:set_application_level(emqtt, notice),
-    logger:set_application_level(aircon, debug),
+
+    {ok, Level} = application:get_env(aircon, logger_level),
+    logger:set_application_level(aircon, Level),
     ok.

apps/aircon/src/aircon_args.erl

@@ -69,7 +69,7 @@ init([]) ->
     {ok, Metrics} = efka_client:request_metric(),
     try convert_metric(Metrics) of
         {ok, MetricMap} ->
-            logger:debug("[aircon_args] init load metric_map: ~p", [MetricMap]),
+            logger:info("[aircon_args] init load metric_map: ~p", [MetricMap]),
             {ok, Param} = efka_client:request_param(),
 
             {ok, #state{metrics = MetricMap, param = Param}}
@@ -111,7 +111,7 @@ handle_cast({push_param, Param}, State = #state{}) ->
 handle_cast({push_metric, Metrics}, State = #state{}) ->
     try convert_metric(Metrics) of
         {ok, MetricMap} ->
-            logger:debug("[aircon_args] push metric_map: ~p", [MetricMap]),
+            logger:info("[aircon_args] push metric_map: ~p", [MetricMap]),
             {noreply, State#state{metrics = MetricMap}}
     catch _:_ ->
         {noreply, State}

apps/aircon/src/aircon_device.erl

@@ -96,24 +96,27 @@ handle_call(_Request, _From, State = #state{}) ->
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_cast({metric_data, Message}, State = #state{device_uuid = DeviceUUID, data_counter = DataCounter, status = Status}) ->
     case catch jiffy:decode(Message, [return_maps]) of
-        #{<<"properties">> := Props0} ->
+        Props0 when is_list(Props0) ->
-            Props = lists:map(fun(Fields) -> transform(Fields#{<<"device_uuid">> => DeviceUUID}) end, Props0),
+            case is_valid_props(Props0) of
-            Info = iolist_to_binary(jiffy:encode(Props, [force_utf8])),
+                true ->
-            case catch efka_client:send_metric_data(Props, #{}) of
+                    Props = lists:map(fun(Fields) -> Fields#{<<"device_uuid">> => DeviceUUID} end, Props0),
-                {ok, _} ->
+                    Info = iolist_to_binary(jiffy:encode(Props, [force_utf8])),
-                    aircon_logger:write([<<"OK">>, Info]);
+                    case catch efka_client:send_metric_data(Props, #{}) of
-                _ ->
+                        {ok, _} ->
-                    aircon_logger:write([<<"ERROR">>, Info])
+                            aircon_logger:write([<<"OK">>, Info]);
-            end,
+                        _ ->
+                            aircon_logger:write([<<"ERROR">>, Info])
+                    end,
 
-            %% 如果设备当前是离线状态，则需要发送上线消息
+                    %% 如果设备当前是离线状态，则需要发送上线消息
-            Status == 0 andalso efka_client:device_online(DeviceUUID),
+                    Status == 0 andalso efka_client:device_online(DeviceUUID),
 
-            {noreply, State#state{data_counter = DataCounter + 1, status = 1}};
+                    {noreply, State#state{data_counter = DataCounter + 1, status = 1}};
-        M when is_map(M) ->
+                false ->
-            logger:notice("[power_device] invalid map: ~p", [M]);
+                    logger:notice("[aircon_device] invalid map: ~p", [Props0])
+            end;
         Error ->
-            logger:notice("[power_device] jiffy decode error: ~p", [Error]),
+            logger:notice("[aircon_device] jiffy decode error: ~p", [Error]),
             {noreply, State}
     end.
 
@@ -155,23 +158,10 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
-%% 网关上来的数据写反了： value是类型，unit才是数值
+-spec is_valid_props(Props :: list()) -> boolean().
-%% [{"value":"int","unit":44081,"type":"AI","timestamp":1730799797,"name":"使用次数","label":"次","key":"use_times","device_uuid":"30409239002349977617259608405456"}]
+is_valid_props(Props) when is_list(Props) ->
--spec transform(Prop :: map()) -> map().
+    lists:all(fun is_valid_props0/1, Props).
-transform(Prop = #{<<"key">> := <<"total_power">>, <<"unit">> := Unit}) ->
+is_valid_props0(#{<<"key">> := Key, <<"value">> := _, <<"unit">> := Unit, <<"type">> := Type, <<"name">> := Name, <<"label">> := Label}) ->
-    Prop#{<<"name">> => <<"总能耗"/utf8>>, <<"value">> => Unit, <<"unit">> => 16#02};
+    is_binary(Key) andalso is_integer(Unit) andalso is_binary(Name) andalso is_binary(Label) andalso is_binary(Type);
-transform(Prop = #{<<"key">> := <<"total_runtime">>, <<"unit">> := Unit}) ->
+is_valid_props0(_) ->
-    Prop#{<<"name">> => <<"总运行时间"/utf8>>, <<"value">> => Unit, <<"unit">> => ?UNIT0};
+    false.
-transform(Prop = #{<<"key">> := <<"use_times">>, <<"unit">> := Unit}) ->
-    Prop#{<<"name">> => <<"使用次数"/utf8>>, <<"label">> => <<"次"/utf8>>, <<"value">> => Unit, <<"unit">> => ?UNIT0};
-transform(Prop = #{<<"key">> := <<"light_switch">>, <<"unit">> := Unit}) ->
-    Prop#{<<"name">> => <<"开关"/utf8>>, <<"value">> => Unit, <<"unit">> => ?UNIT0};
-transform(Prop = #{<<"key">> := <<"light_brightness">>, <<"unit">> := Unit}) when is_integer(Unit) ->
-    Label = iolist_to_binary([<<"亮度设置为:"/utf8>>, integer_to_binary(Unit)]),
-    Prop#{<<"name">> => <<"亮度"/utf8>>, <<"label">> => Label, <<"value">> => Unit, <<"unit">> => ?UNIT0};
-transform(Prop = #{<<"key">> := <<"light_change_time">>, <<"unit">> := Unit}) ->
-    Prop#{<<"name">> => <<"变亮变暗时间"/utf8>>, <<"value">> => Unit, <<"unit">> => ?UNIT0};
-transform(Prop = #{<<"key">> := <<"light_status">>, <<"unit">> := Unit}) ->
-    Prop#{<<"name">> => <<"是否损坏"/utf8>>, <<"value">> => Unit, <<"unit">> => ?UNIT0};
-transform(Prop = #{<<"unit">> := Unit}) ->
-    Prop#{<<"value">> => Unit, <<"unit">> => ?UNIT0}.

apps/aircon/src/aircon_mqtt_publisher.erl

@@ -57,18 +57,17 @@ start_link() ->
 init([]) ->
     %% 建立到emqx服务器的连接
     Opts = emqx_opts(<<"aircon-data-publisher">>),
-    logger:debug("[mqtt_publisher] opts is: ~p", [Opts]),
+    logger:info("[mqtt_publisher] start with opts is: ~p", [Opts]),
     case emqtt:start_link(Opts) of
         {ok, ConnPid} ->
-            logger:debug("[mqtt_publisher] start conntecting, pid: ~p", [ConnPid]),
             {ok, _} = emqtt:connect(ConnPid),
-            logger:debug("[mqtt_publisher] connect success"),
+            logger:info("[mqtt_publisher] connect success, pid: ~p", [ConnPid]),
             {ok, #state{conn_pid = ConnPid}};
         ignore ->
-            logger:debug("[mqtt_publisher] connect emqx get ignore"),
+            logger:warning("[mqtt_publisher] connect emqx get ignore"),
             {stop, ignore};
         {error, Reason} ->
-            logger:debug("[mqtt_publisher] connect emqx get error: ~p", [Reason]),
+            logger:warning("[mqtt_publisher] connect emqx get error: ~p", [Reason]),
             {stop, Reason}
     end.
 

apps/aircon/src/aircon_mqtt_subscriber.erl

@@ -51,22 +51,21 @@ start_link() ->
 init([]) ->
     %% 建立到emqx服务器的连接
     Opts = emqx_opts(<<"aircon-data-subscriber">>),
-    logger:debug("[mqtt_subscriber] opts is: ~p", [Opts]),
+    logger:info("[mqtt_subscriber] start with opts is: ~p", [Opts]),
     case emqtt:start_link(Opts) of
         {ok, ConnPid} ->
-            logger:debug("[mqtt_subscriber] start conntecting, pid: ~p", [ConnPid]),
             {ok, _} = emqtt:connect(ConnPid),
-            logger:debug("[mqtt_subscriber] connect success"),
+            logger:info("[mqtt_subscriber] connect success, pid: ~p", [ConnPid]),
             %% 监听和设备的全部事件
             SubscribeResult = emqtt:subscribe(ConnPid, ?Topics),
-            logger:debug("[mqtt_subscriber] subscribe topics: ~p, result is: ~p", [?Topics, SubscribeResult]),
+            logger:info("[mqtt_subscriber] subscribe topics: ~p, result is: ~p", [?Topics, SubscribeResult]),
 
             {ok, #state{conn_pid = ConnPid}};
         ignore ->
-            logger:debug("[mqtt_subscriber] connect emqx get ignore"),
+            logger:warning("[mqtt_subscriber] connect emqx get ignore"),
             {stop, ignore};
         {error, Reason} ->
-            logger:debug("[mqtt_subscriber] connect emqx get error: ~p", [Reason]),
+            logger:warning("[mqtt_subscriber] connect emqx get error: ~p", [Reason]),
             {stop, Reason}
     end.
 
@@ -99,7 +98,7 @@ handle_cast(_Request, State = #state{}) ->
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_info({disconnect, ReasonCode, Properties}, State = #state{}) ->
-    logger:debug("[mqtt_subscriber] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]),
+    logger:warning("[mqtt_subscriber] Recv a DISONNECT packet - ReasonCode: ~p, Properties: ~p", [ReasonCode, Properties]),
     {stop, disconnected, State};
 %% 必须要做到消息的快速分发，数据的json反序列需要在host进程进行
 handle_info({publish, #{packet_id := _PacketId, payload := Payload, qos := Qos, topic := Topic}}, State = #state{conn_pid = _ConnPid}) ->
@@ -187,12 +186,12 @@ emqx_opts(ClientSuffix) when is_binary(ClientSuffix) ->
 dispatch(DeviceMac, Message) when is_binary(DeviceMac), is_binary(Message) ->
     case aircon_args:get_device_uuid(DeviceMac) of
         error ->
-            logger:notice("[mqtt_subscriber] device_mac: ~p, device_uuid not found", [DeviceMac]);
+            logger:warning("[mqtt_subscriber] device_mac: ~p, device_uuid not found", [DeviceMac]);
         {ok, DeviceUUID} ->
             case aircon_device_sup:ensure_device_started(DeviceUUID) of
                 {ok, DevicePid} ->
                     aircon_device:metric_data(DevicePid, Message);
                 {error, Reason} ->
-                    logger:notice("[mqtt_subscriber] start device get error: ~p", [Reason])
+                    logger:warning("[mqtt_subscriber] start device get error: ~p", [Reason])
             end
     end.

apps/aircon/src/efka_client.erl

@@ -6,6 +6,7 @@
 %%% Created : 28. 8月 2023 15:39
 %%%-------------------------------------------------------------------
 -module(efka_client).
+-feature(maybe_expr, enable).
 -author("aresei").
 
 -behaviour(gen_server).
@@ -144,12 +145,15 @@ start_link(RegisterName, Host, Port) when is_binary(RegisterName), is_list(Host)
     {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term()} | ignore).
 init([RegisterName, Host, Port]) ->
-    {ok, Socket} = gen_tcp:connect(Host, Port, [binary, {packet, 4}, {active, true}]),
+    maybe
-    ok = gen_tcp:controlling_process(Socket, self()),
+        logger:info("[efka_client] connect ~p:~p, register name: ~p", [Host, Port, RegisterName]),
-    case do_register(RegisterName, Socket) of
+        {ok, Socket} ?= gen_tcp:connect(Host, Port, [binary, {packet, 4}, {active, true}]),
-        ok ->
+        ok = gen_tcp:controlling_process(Socket, self()),
-            {ok, #state{packet_id = 1, host = Host, port = Port, socket = Socket}};
+        ok ?= do_register(RegisterName, Socket),
-        {error, Reason} ->
+        {ok, #state{packet_id = 1, host = Host, port = Port, socket = Socket}}
+    else
+        Reason ->
+            logger:notice("[efka_client] init get error: ~p", [Reason]),
             {stop, Reason}
     end.
 
@@ -173,7 +177,7 @@ do_register(RegisterName, Socket) ->
             end
     after
         ?EFKA_REQUEST_TIMEOUT ->
-            {error, timeout}
+            {error, register_timeout}
     end.
 
 %% @private

boot.sh

@@ -0,0 +1,3 @@
+#! /bin/bash
+
+docker run -e TZ=Asia/Shanghai --hostname=aircon --net=host --restart=always -v /data/docker/aircon/:/data/aircon/ aircon:latest

build.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+rebar3 compile && rebar3 release && rebar3 tar
+
+tar -czvf aircon_gateway.tgz boot.sh stop.sh Dockerfile .args.yaml

config/sys-dev.config

@@ -3,6 +3,9 @@
     %% 离线判断时间间隔，单位：(秒)
     {heartbeat_ticker, 120},
 
+    %% 设置当前系统的日志级别
+    {logger_level, debug},
+
     {emqx_server, [
       {host, "118.178.229.213"},
       {port, 1883},

config/sys-prod.config

@@ -3,6 +3,9 @@
     %% 离线判断时间间隔，单位：(秒)
     {heartbeat_ticker, 120},
 
+    %% 设置当前系统的日志级别
+    {logger_level, debug},
+
     {emqx_server, [
       {host, "172.30.37.212"},
       {port, 1883},
@@ -14,8 +17,35 @@
     ]},
 
     {efka_server, [
-      {host, "39.98.184.67"},
+      {host, "localhost"},
       {port, 3361}
     ]}
+  ]},
+
+  {kernel, [
+    %% 设置 Logger 的 primary log level
+    {logger_level, debug},
+    {logger, [
+      {handler, default, logger_std_h,
+        #{
+          level => debug,
+          formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
+        }
+      },
+
+      {handler, disk, logger_disk_log_h,
+        #{
+          level => debug,
+          config => #{
+          file => "log/debug.log",
+          max_no_files => 10,
+          max_no_bytes => 524288000
+        },
+          formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
+        }
+      }
+
+    ]}
   ]}
+
 ].

stop.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+docker ps -a  | grep "aircon:latest" | awk -F " " '{print $1}' | xargs docker stop | xargs docker rm