Merge branch 'master' of github.com:dgiot/dgiot

.gitignore

@@ -96,6 +96,8 @@ apps/dgiot_viewx/
 apps/dgiot_sec/
 apps/dgiot_demo/
 apps/dgiot_jingmao/
+apps/dgiot_atom/
+apps/dgiot_12184/
 node_modules/
 package-lock.json
 erlang_ls.config

apps/dgiot_api/src/handler/dgiot_system_handler.erl

@@ -182,7 +182,7 @@ do_request(post_plugin_app, #{<<"Action">> := Action, <<"App">> := App}, _Contex
             {200, #{<<"error">> => <<"license error">>}}
     end;
 
-%% iot_hub 概要: 升级插件dgiot_system_handler
+%% iot_hub 概要: 升级插件
 %% OperationId:post_station_data
 %% 请求:POST /iotapi/post_station_data
 do_request(post_upgrade_plugin, #{<<"file">> := #{<<"filename">> := Filename, <<"fullpath">> := Fullpath}}, _Context, _Req) ->

apps/dgiot_atomgit/README.md

@@ -3,7 +3,6 @@
 
 ![开源大赛.jpg](https://dgiot-1253666439.cos.ap-shanghai-fsi.myqcloud.com/kaiyuandasai.png)
 
-
 <a href="https://www.bilibili.com/video/BV13B421z7NP/?spm_id_from=333.999.0.0"><img src="https://dgiot-1253666439.cos.ap-shanghai-fsi.myqcloud.com/dgiot_doc/bilibidasai.jpg" /></a>
 
 ### DGIOT物联网开源平台，承办了工信部开放原子的开源大赛，
@@ -63,3 +62,4 @@
 3：可信接入与网络安全算法
 
 4：设备精细化能耗分析与优化
+

apps/dgiot_atomgit/include/dgiot_atomgit.hrl

@@ -1,25 +1,127 @@
-%%--------------------------------------------------------------------
+-define(ATOMGIT, <<"ATOMGIT">>).
-%% Copyright (c) 2020 DGIOT Technologies Co., Ltd. All Rights Reserved.
-%%
-%% Licensed under the Apache License, Version 2.0 (the "License");
-%% you may not use this file except in compliance with the License.
-%% You may obtain a copy of the License at
-%%
-%%     http://www.apache.org/licenses/LICENSE-2.0
-%%
-%% Unless required by applicable law or agreed to in writing, software
-%% distributed under the License is distributed on an "AS IS" BASIS,
-%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-%% See the License for the specific language governing permissions and
-%% limitations under the License.
-%%--------------------------------------------------------------------
-
-
--define(DEMO, <<"DEMO">>).
-
 
 -record(state, {
     id,
-    env = #{},
+    devaddr = <<>>,
-    search = <<"IP">>
+    heartcount = 0,
+    regtype = <<>>,
+    head = "xxxxxx0eee",
+    len = 0,
+    app = <<>>,
+    product = <<>>,
+    deviceId = <<>>,
+    scale = 10,
+    temperature = 0,
+    env = <<>>,
+    dtutype = <<>>,
+    hb = 60,
+    maxaddr = 0
 }).
+
+%%设置超时时间及重发次数
+-define(SetTimeOutReSendTimes, 1000).
+%%提取现场机时间
+-define(GetSceneDeviceTime, 1011).
+%%上传现场机时间
+-define(UploadSceneDeviceTime, 1011).
+%%设置现场机时间
+-define(SetSceneDeviceTime, 1012).
+%%现场机时间校准请求
+-define(SceneDeviceTimeCalibration, 1013).
+%%提取实时数据间隔
+-define(GetRtdDataInterval, 1061).
+%%上传实时数据间隔
+-define(UploadRtdDataInterval, 1061).
+%%设置实时数据间隔
+-define(SetRtdDataInterval, 1062).
+%%提取分钟数据间隔
+-define(GetMinuteDataInterval, 1063).
+%%上传分钟数据间隔
+-define(UploadMinuteDataInterval, 1063).
+%%设置分钟数据间隔
+-define(SetMinuteDataInterval, 1064).
+%%用于设置现场机的密码
+-define(SetSceneDevicePassword, 1072).
+%%取污染物实时数据
+-define(GetRtdData, 2011).
+%%上传污染物实时数据
+-define(UploadRtdData, 2011).
+%%停止察看污染物实时数据
+-define(StopRtdData, 2012).
+%%取设备运行状态数据
+-define(GetDeviceRunState, 2021).
+%%上传设备运行状态数据
+-define(UploadDeviceRunState, 2021).
+%%停止察看设备运行状态
+-define(StopDeviceRunState, 2022).
+%%取污染物日历史数据
+-define(GetDayData, 2031).
+%%上传污染物日历史数据
+-define(UploadDayData, 2031).
+%%取设备运行时间日历史数据
+-define(GetDeviceRunTimeDayData, 2041).
+%%上传设备运行时间日历史数据
+-define(UploadDeviceRunTimeDayData, 2041).
+%%取污染物分钟数据
+-define(GetMinuteData, 2051).
+%%上传污染物分钟数据
+-define(UploadMinuteData, 2051).
+%%取污染物小时数据
+-define(GetHourData, 2061).
+%%上传污染物小时数据
+-define(UploadHourData, 2061).
+%%上传数采仪开机时间
+-define(UploadComputerPowerOnTime, 2081).
+%%零点校准量程校准
+-define(RangeCalibration, 3011).
+%%即时采样
+-define(TakeSampleImmediately, 3012).
+%%启动清洗/反吹
+-define(StartClear, 3013).
+%%比对采样
+-define(CompareSample, 3014).
+%%超标留样
+-define(LeaveSuperstandardSample, 3015).
+%%上传超标留样信息
+-define(UploadSuperstandardSample, 3015).
+%%设置采样时间周期
+-define(SetSampleTimeInterval, 3016).
+%%提取采样时间周期
+-define(GetSampleTimeInterval, 3017).
+%%上传采样时间周期
+-define(UploadSampleTimeInterval, 3017).
+%%提取出样时间
+-define(GetSampleTime, 3018).
+%%上传出样时间
+-define(UploadSampleTime, 3018).
+%%提取设备唯一标识
+-define(GetSceneDeviceUUID, 3019).
+%%上传设备唯一标识
+-define(UploadSceneDeviceUUID, 3019).
+%%提取现场机信息
+-define(GetSceneDeviceInfo, 3020).
+%%上传现场机信息
+-define(UploadSceneDeviceInfo, 3020).
+%%设置现场机参数
+-define(SetSceneDeviceParam, 3021).
+%%取污染物周期数据
+-define(GetCycleData, 8051).
+%%上传污染物周期数据
+-define(UploadCycleData, 8051).
+%%请求应答
+-define(RequestResponse, 9011).
+%%执行结果
+-define(ExecuteResponse, 9012).
+%%通知应答
+-define(NoticeResponse, 9013).
+%%数据应答
+-define(DataResponse, 9014).
+%%心跳
+-define(HeartBeat, 9021).
+
+%%commandType
+-define(Request, 1).  %%请求命令
+-define(Upload, 2).  %% 上传命令
+-define(Notice, 3).  %% 通知命令
+-define(Other, 4). %% 其他
+-define(None, 5).  %%不支持

apps/dgiot_atomgit/priv/swagger/swagger_atomgit.json

@@ -1,10 +1,10 @@
 {
   "definitions": {},
   "paths": {
-    "/demotest": {
+    "/atomgitapi": {
       "post": {
-        "summary": "demo新增设备",
+        "summary": "atomgit接口",
-        "description": "demo测试接口",
+        "description": "atomgit接口",
         "parameters": [
           {
             "in": "body",

apps/dgiot_atomgit/src/dgiot_atomgit.app.src

@@ -1,5 +1,5 @@
 {application, dgiot_atomgit,
-    [{description, "DGIOT DEMO"},
+    [{description, "DGIOT ATOMGIT"},
         {vsn, "4.3.0"},
         {registered, []},
         {mod, {dgiot_atomgit_app, []}},

apps/dgiot_atomgit/src/dgiot_atomgit.erl

@@ -20,22 +20,89 @@
 -include_lib("dgiot/include/logger.hrl").
 
 -export([
-    demo_test/1
+    atomgitapi/1
+    , create_product/0
+    , create_device/3
+    , save_td/0
+    , query/0
 ]).
 
 -define(APP, ?MODULE).
 
 
-demo_test(#{<<"name">> := _Name, <<"devaddr">> := _Devaddr} = Args) ->
+atomgitapi(#{<<"name">> := _Name, <<"devaddr">> := _Devaddr} = Args) ->
     io:format("~s ~p Args = ~p.~n", [?FILE, ?LINE, Args]),
     {ok, Args};
 
-demo_test(Args) ->
+atomgitapi(Args) ->
     io:format("~s ~p Args = ~p.~n", [?FILE, ?LINE, Args]),
-    {error}.
+    error.
-
+
-
+
-
+%% 创建产品
+create_product() ->
+    dgiot_product:create_product(#{
+        <<"name">> => <<"添加产品例子"/utf8>>,
+        <<"devType">> => <<"dgiot">>,
+        <<"category">> => #{<<"objectId">> => <<"5ca6049839">>, <<"__type">> => <<"Pointer">>, <<"className">> => <<"Category">>},
+        <<"desc">> => <<"atom test">>,
+        <<"config">> => #{<<"interval">> => -1},
+        <<"channel">> => #{},
+        <<"thing">> => #{},
+        <<"ACL">> => #{<<"role:开发者"/utf8>> => #{<<"read">> => true, <<"write">> => true}},
+        <<"nodeType">> => 0,
+        <<"productSecret">> => dgiot_utils:random()
+    }).
+
+%% 创建设备
+create_device(ProductId, DTUMAC, DTUIP) ->
+    case dgiot_product:lookup_prod(ProductId) of
+        {ok, #{<<"ACL">> := Acl, <<"devType">> := DevType}} ->
+            dgiot_device:create_device(#{
+                <<"devaddr">> => DTUMAC,
+                <<"name">> => <<DevType/binary, "_", DTUMAC/binary>>,
+                <<"ip">> => DTUIP,
+                <<"isEnable">> => true,
+                <<"product">> => ProductId,
+                <<"ACL">> => Acl,
+                <<"status">> => <<"ONLINE">>,
+                <<"brand">> => DevType,
+                <<"devModel">> => DevType
+            });
+        _ ->
+            pass
+    end.
+
+%% 存td库
+save_td() ->
+%%     通过任务通道存td
+    dgiot_task:save_td(<<"产品objectId">>, <<"设备地址">>, #{<<"物模型标识符"/utf8>> => <<"value">>}, #{}),
+%%    直接存td
+    Sql = dgiot_tdengine:format_sql(<<"产品objectId">>, <<"设备地址">>, [#{<<"物模型标识符"/utf8>> => <<"value">>}]),
+    dgiot_tdengine_adapter:save_sql(<<"产品objectId">>, Sql).
+
+
+%% 查询
+query() ->
+%%    按产品Id查询产品详情
+    dgiot_parse:get_object(<<"Product">>, <<"产品id">>),
+%%    查询产品列表
+    dgiot_parse:query_object(<<"Product">>, #{<<"count">> => <<"objectId">>, <<"limit">> => 100, <<"where">> => #{}}),
+%%    按设备Id查询设备详情
+    dgiot_parse:get_object(<<"Device">>, <<"设备id">>),
+%%    查询设备列表
+    case dgiot_parse:query_object(<<"Device">>, #{<<"count">> => <<"objectId">>, <<"limit">> => 100, <<"where">> => #{}}) of
+        {ok, #{<<"count">> := Count, <<"results">> := Results}} ->
+            io:format("~s ~p Count = ~p.~n", [?FILE, ?LINE, Count]),
+%%            count 总数, results 设备列表
+%%            遍历设备列表
+            lists:foldl(fun(#{<<"objectId">> := DeviceId, <<"name">> := Name}, _) ->
+%%              打印设备objectId, 设备名称
+                io:format("~s ~p DeviceId ~p => Name = ~p.~n", [?FILE, ?LINE, DeviceId, Name])
+                        end, #{}, Results);
+        _ ->
+            pass
+    end.
 
 
 

apps/dgiot_atomgit/src/dgiot_atomgit_channel.erl

@@ -21,7 +21,7 @@
 -include_lib("dgiot/include/dgiot_socket.hrl").
 -include_lib("dgiot/include/logger.hrl").
 -include("dgiot_atomgit.hrl").
--define(TYPE, <<"CUSTOMER">>).
+-define(TYPE, <<"ATOMGIT">>).
 %% API
 -export([start/2]).
 
@@ -33,19 +33,31 @@
     cType => ?TYPE,
     type => ?PROTOCOL_CHL,
     title => #{
-        zh => <<"demo测试通道"/utf8>>
+        zh => <<"ATOMGIT测试通道"/utf8>>
     },
     description => #{
-        zh => <<"demo测试通道"/utf8>>
+        zh => <<"ATOMGIT测试通道"/utf8>>
     }
 }).
 %% 注册通道参数
 -params(#{
+    <<"port">> => #{
+        order => 1,
+        type => integer,
+        required => true,
+        default => 20660,
+        title => #{
+            zh => <<"端口"/utf8>>
+        },
+        description => #{
+            zh => <<"侦听端口"/utf8>>
+        }
+    },
     <<"ico">> => #{
         order => 102,
         type => string,
         required => false,
-        default => <<"/dgiot_file/shuwa_tech/zh/product/dgiot/channel/demo.jpg">>,
+        default => <<"/dgiot_file/shuwa_tech/zh/product/dgiot/channel/atomgit.jpg">>,
         title => #{
             en => <<"channel ICO">>,
             zh => <<"通道ICO"/utf8>>
@@ -62,12 +74,26 @@ start(ChannelId, ChannelArgs) ->
     dgiot_channelx:add(?TYPE, ChannelId, ?MODULE, ChannelArgs).
 
 %% 通道初始化
-init(?TYPE, ChannelId, Args) ->
+init(?TYPE, ChannelId, #{
+    <<"port">> := Port,
+    <<"product">> := Products
+} = Args) ->
+    {ProdcutId, App} =
+        case get_app(Products) of
+            [{ProdcutId1, App1} | _] ->
+                {ProdcutId1, App1};
+            [] ->
+                {<<>>, <<>>};
+            _ ->
+                {<<>>, <<>>}
+        end,
     State = #state{
         id = ChannelId,
-        env = Args
+        env = Args,
+        app = App,
+        product = ProdcutId
     },
-    {ok, State, []}.
+    {ok, State, dgiot_atomgit_tcp:start(Port, State)}.
 
 handle_init(State) ->
     {ok, State}.
@@ -77,9 +103,39 @@ handle_event(EventId, Event, State) ->
     ?LOG(error, "EventId ~p Event ~p", [EventId, Event]),
     {ok, State}.
 
+handle_message({dlink_properties_report, ProductId, DevAddr, Buff}, #state{id = ChannelId} = State) ->
+    dgiot_bridge:send_log(ChannelId, ProductId, "~s ~p ~p mqtt revice from ~p => ProductId ~p ", [?FILE, ?LINE, dgiot_datetime:format("YYYY-MM-DD HH:NN:SS"), Buff, ProductId]),
+    case dgiot_atomgit_decoder:parse_frame(Buff, []) of
+        {ok, [#{<<"devaddr">> := DevAddr} = Data | _]} ->
+            NewData = dgiot_dlink_proctol:parse_payload(ProductId, Data),
+            dgiot_bridge:send_log(ChannelId, ProductId, DevAddr, "~s ~p ~p revice from ~p~n save td => ProductId ~p DevAddr ~p ~ts ",
+                [?FILE, ?LINE, dgiot_datetime:format("YYYY-MM-DD HH:NN:SS"), Buff, ProductId, DevAddr, unicode:characters_to_list(dgiot_json:encode(NewData))]),
+            dgiot_task:save_td(ProductId, DevAddr, NewData, #{});
+        _ ->
+            pass
+    end,
+    {ok, State};
+
 handle_message(_Message, State) ->
     {ok, State}.
 
 stop(_ChannelType, _ChannelId, _State) ->
     ok.
 
+get_app(Products) ->
+    lists:map(fun({ProdcutId, #{<<"ACL">> := Acl}}) ->
+        Predicate = fun(E) ->
+            case E of
+                <<"role:", _/binary>> -> true;
+                _ -> false
+            end
+                    end,
+        App =
+            case lists:filter(Predicate, maps:keys(Acl)) of
+                [<<"role:", Name/binary>> | _] ->
+                    Name;
+                _ ->
+                    <<"dgiot">>
+            end,
+        {ProdcutId, App}
+              end, Products).

apps/dgiot_atomgit/src/dgiot_atomgit_tcp.erl

@@ -0,0 +1,74 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 DGIOT Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+-module(dgiot_atomgit_tcp).
+-author("stoneliu").
+-include("dgiot_atomgit.hrl").
+-include_lib("dgiot/include/dgiot_socket.hrl").
+-include_lib("dgiot/include/logger.hrl").
+
+-define(MAX_BUFF_SIZE, 1024).
+
+-export([
+    start/2
+]).
+
+%% TCP callback
+-export([init/1, handle_info/2, handle_cast/2, handle_call/3, terminate/2]).
+
+start(Port, State) ->
+    dgiot_tcp_server:child_spec(?MODULE, dgiot_utils:to_int(Port), State).
+
+init(#tcp{state = #state{id = ChannelId}} = TCPState) ->
+    case dgiot_bridge:get_products(ChannelId) of
+        {ok, _TYPE, _ProductIds} ->
+            {ok, TCPState};
+        {error, not_find} ->
+            {stop, not_find_channel}
+    end.
+
+%% 设备连接后，第一条报文做设备地址
+handle_info({tcp, Buff}, #tcp{socket = Socket, state = #state{id = ChannelId, devaddr = <<>>, product = ProductId} = State} = TCPState) ->
+    DTUIP = dgiot_utils:get_ip(Socket),
+    DtuAddr = dgiot_utils:binary_to_hex(Buff),
+    DeviceId = dgiot_parse_id:get_deviceid(ProductId, DtuAddr),
+    dgiot_atomgit:create_device(ProductId, DtuAddr, DTUIP),
+    dgiot_bridge:send_log(ChannelId, ProductId, DtuAddr, "~s ~p DTU login DtuAddr:~p", [?FILE, ?LINE, DtuAddr]),
+    {noreply, TCPState#tcp{buff = <<>>, register = true, clientid = DeviceId, state = State#state{devaddr = DtuAddr, deviceId = DeviceId}}};
+
+handle_info({tcp, Buff}, #tcp{state = #state{id = ChannelId, devaddr = Devaddr, product = ProductId} = _State} = TCPState) ->
+    dgiot_bridge:send_log(ChannelId, ProductId, "~s ~p ~p tcp revice from ~p => ProductId ~p ", [?FILE, ?LINE, dgiot_datetime:format("YYYY-MM-DD HH:NN:SS"), Buff, ProductId]),
+    case dgiot_atomgit_decoder:parse_frame(Buff, []) of
+        {ok, [#{<<"devaddr">> := Devaddr} = Data | _]} ->
+            NewData = dgiot_dlink_proctol:parse_payload(ProductId, Data),
+            dgiot_bridge:send_log(ChannelId, ProductId, Devaddr, "~s ~p ~p revice from ~p~n save td => ProductId ~p DevAddr ~p ~ts ", [?FILE, ?LINE, dgiot_datetime:format("YYYY-MM-DD HH:NN:SS"), Buff, ProductId, Devaddr, unicode:characters_to_list(dgiot_json:encode(NewData))]),
+            dgiot_task:save_td(ProductId, Devaddr, NewData, #{});
+        _O ->
+            pass
+    end,
+    {noreply, TCPState};
+
+handle_info(_Info, TCPState) ->
+    {noreply, TCPState}.
+
+handle_call(_Msg, _From, TCPState) ->
+    {reply, ok, TCPState}.
+
+handle_cast(_Msg, TCPState) ->
+    {noreply, TCPState}.
+
+terminate(_Reason, _TCPState) ->
+    ok.
+

apps/dgiot_atomgit/src/handler/dgiot_atomgit_handler.erl

@@ -21,19 +21,19 @@
 
 
 %% API
--export([swagger_dahua/0]).
+-export([swagger_atomgit/0]).
 -export([handle/4]).
 
 %% API描述
 %% 支持二种方式导入
 %% 示例:
 %% 1. Metadata为map表示的JSON,
-%%    dgiot_http_server:bind(<<"/dahua>>, ?MODULE, [], Metadata)
+%%    dgiot_http_server:bind(<<"/atomgit>>, ?MODULE, [], Metadata)
 %% 2. 从模块的priv/swagger/下导入
-%%    dgiot_http_server:bind(<<"/swagger_dahua.json">>, ?MODULE, [], priv)
+%%    dgiot_http_server:bind(<<"/swagger_atomgit.json">>, ?MODULE, [], priv)
-swagger_dahua() ->
+swagger_atomgit() ->
     [
-        dgiot_http_server:bind(<<"/swagger_demo.json">>, ?MODULE, [], priv)
+        dgiot_http_server:bind(<<"/swagger_atomgit.json">>, ?MODULE, [], priv)
     ].
 
 
@@ -79,8 +79,8 @@ handle(OperationID, Args, Context, Req) ->
 %% System 概要: demo测试接口 描述:demo测试接口
 %% OperationId:post_demotest
 %% 请求:POST /iotapi/post_demotest
-do_request(post_demotest, Args, _Context, _Req) ->
+do_request(post_atomgitapi, Args, _Context, _Req) ->
-    case dgiot_atomgit:demo_test(Args) of
+    case dgiot_atomgit:atomgitapi(Args) of
         {ok, Data} ->
             {ok, #{<<"data">> => Data, <<"status">> => 0, <<"msg">> => <<"success">>}};
         _ ->

apps/dgiot_atomgit/src/protocol/dgiot_atomgit_decoder.erl

@@ -0,0 +1,153 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 DGIOT Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(dgiot_atomgit_decoder).
+-include_lib("dgiot_atomgit.hrl").
+-include_lib("dgiot/include/logger.hrl").
+-protocol([?ATOMGIT]).
+-define(CRLF, "\r\n").
+%% API
+-export([parse_frame/2, to_frame/1]).
+
+parse_frame(Buff, Opts) ->
+    parse_frame(Buff, [], Opts).
+
+parse_frame(<<>>, Acc, _Opts) ->
+    {ok, Acc};
+
+%% HJ 212-2017
+%% 6.3 通讯协议数据结构
+%%所有的通讯包都是由 ASCII 码（汉字除外，采用 UTF-8 码，8 位，1 字节）字符组成。通讯协议数
+%%据结构如图 4 所示。
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+%% 定 义         |    类型        |  长度         |          描 述                                                       |
+%------------------------------------------------------------------------------------------------------------------------
+%%      包头     |    字符        |    2          |     固定为##                                                        |
+%-------------------------------------------------------------------------------------------------------------------------
+%%    数据段长度 |    十进制整型   |    2          |     数据段的 ASCII 字符数，例如：长 255，则写为“0255”                 |
+%-------------------------------------------------------------------------------------------------------------------------
+%%    数据段     |    字符        |  0 ≤ n ≤ 1024 |  变长的数据，详见 6.3.2 章节的表 3《数据段结构组成表》                 |
+%-------------------------------------------------------------------------------------------------------------------------
+%%    CRC 校验   |   十六进制整数  | 4             |  数据段的校验结果，CRC 校验算法见附录 A。接收到一条命令，             |
+%%               |                |               |    如果 CRC 错误，执行结束                                          |
+%-------------------------------------------------------------------------------------------------------------------------
+%%    包尾     |    字符          |  2             |  固定为<CR><LF>（回车、换行）                                      |
+%-------------------------------------------------------------------------------------------------------------------------
+parse_frame(<<"##", Length:4/binary, Tail/binary>>, Acc, State) ->
+    Len = binary_to_integer(Length, 10),
+    {Rest1, Acc1} =
+        case Len > -1 andalso Len < 1025 of
+            true ->
+                case Tail of
+                    <<UserZone:Len/binary, Crc:4/binary, ?CRLF, Rest/binary>> ->
+                        CheckCrc = dgiot_atomgit_utils:crc16(UserZone),
+                        case Crc of
+                            CheckCrc ->
+                                {Rest, Acc ++ [parse_userzone(UserZone, State)]};
+                            _ ->
+                                {<<>>, Acc}
+                        end;
+                    _ ->
+                        {<<>>, Acc}
+                end;
+
+            _ ->
+                {<<>>, Acc}
+        end,
+    parse_frame(Rest1, Acc1, State);
+
+parse_frame(<<_:8, Data/binary>> = _Rest, Acc, Opts) ->
+    parse_frame(Data, Acc, Opts).
+
+
+%%6.3.2 数据段结构组成
+%% 数据段结构组成见表 3，表 3 中“长度”包含字段名称、‘=’、字段内容三部分内容。
+%%  表 3 数据段结构组成表
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+%|   名称             | 类型     |    长度 |    描述                                                                     |
+%|----------------------------------------------------------------------------------------------------------------------
+%| 请求编码 QN        | 字符     | 20      | 精确到毫秒的时间戳:QN=YYYYMMDDhhmmsszzz，用来唯一标识一次命令交互             |
+%------------------------------------------------------------------------------------------------------------------------
+%| 系统编码 ST        | 字符     | 5       | ST=系统编码, 系统编码取值详见 6.6.1 章节的表 5《系统编码表》                  |
+%|----------------------------------------------------------------------------------------------------------------------
+%| 命令编码 CN        | 字符     |  7      | CN=命令编码, 命令编码取值详见 6.6.5 章节的表 9《命令编码表》                  |
+%|----------------------------------------------------------------------------------------------------------------------
+%|  访问密码          |   字符   |   9     | PW=访问密码                                                                 |
+%|----------------------------------------------------------------------------------------------------------------------|
+%|                    |          |       | MN=设备唯一标识，这个标识固化在设备中，用于唯一标识一个设备。MN 由 EPC-96        |
+%|                    |          |       |  编码转化的字符串组成，即 MN 由 24 个 0~9，A~F 的字符组成                       |
+%|                    |          |       |     _______________________________________________________________          |
+%| 设备唯一标识 MN     |   字符   | 27    |     |              EPC-96 编码结构                                 |         |
+%|                    |          |       |     |------------------------------------------------------------- |        |
+%|                    |          |       |     |   名称   |  标头  | 厂商识别代码 | 对象分类代码 |  序列号      |         |
+%|                    |          |       |      ---------------------------------------------------------------         |
+%|                    |          |       |     | 长度(比特 | 8     |   28         |   24        |  36         |         |
+%|----------------------------------------------------------------------------------------------------------------------|
+%|                    |          |       |   Flag=标志位，这个标志位包含标准版本号、是否拆分包、数据是否应答。             |
+%|                    |          |       |   ——————————————————————————————————————————                                 |
+%|                    |          |       |   | V5 | V4 | V3 | V2 | V1 | V0 | D | A    |                                 |
+%|                    |          |       |   -------------------------------------------                                |
+%|  拆分包及应答标志   | 整数     | 8     |   V5~V0：标准版本号；Bit：000000 表示标准 HJ/T 212-2005，000001                  |
+%|      Flag          | （0-255）|       |   表示本次标准修订版本号。                                                      |
+%|                    |          |       |   A：命令是否应答；Bit：1-应答，0-不应答。                                      |
+%|                    |          |       |   D：是否有数据包序号；Bit：1-数据包中包含包号和总包数两部分,                    |
+%|                    |          |       |     0-数据包中不包含包号和总包数两部分。                                       |
+%|                    |          |       |  示例：Flag=7 表示标准版本为本次修订版本号，数据段需要拆分并且命令需要应答       |
+%|----------------------------------------------------------------------------------------------------------------------|
+%| 总包数PNUM         |  字符     |  9     |  PNUM 指示本次通讯中总共包含的包数,注：不分包时可以没有本字段，与标志位有关      |
+%|----------------------------------------------------------------------------------------------------------------------|
+%|   包号 PNO        |    字符    |   8    | PNO 指示当前数据包的包号,注：不分包时可以没有本字段，与标志位有关
+%|----------------------------------------------------------------------------------------------------------------------|
+%|  指令参数 CP      |    字符     | 0≤n≤950| CP=&&数据区&&，数据区定义见 6.3.3 章节                                      |
+%|----------------------------------------------------------------------------------------------------------------------|
+%%parse_userzone(<<"QN=",QN:17/binary,";ST=", ST:2/binary, ";CN=", CN:4/binary, ";PW=", PWD:6/binary, ";MN=", MN:24/binary, ";Flag=", Flag:2/binary, PNUM:9/binary, PNO:8/binary, ";CP=", CP/binary>>, _State) ->
+%%  ##0331QN=20240204193300000;ST=31;CN=2011;PW=123456;MN=60436377;Flag=4;CP=&&DataTime=20240204193300;a34004-Rtd=10,a34004-Flag=N;a34002-Rtd=12,a34002-Flag=N;a34001-Rtd=26,a34001-Flag=N;a01001-Rtd=1.8,a01001-Flag=N;a01002-Rtd=91.0,a01002-Flag=N;a01007-Rtd=3.9,a01007-Flag=N;a01008-Rtd=314.00,a01008-Flag=N;a01006-Rtd=102.36,a01006-Flag=N;&&7F80\r\n
+parse_userzone(UserZone, _State) ->
+    lists:foldl(fun(X, Acc) ->
+        case X of
+            <<"MN=", Devaddr/binary>> ->
+                Acc#{<<"devaddr">> => Devaddr};
+            <<"CP=&&", CP/binary>> ->
+                Acc#{<<"CP">> => dgiot_atomgit_utils:get_cps(CP)};
+            _ ->
+                NewX =
+                    case re:split(X, <<",">>) of
+                        [First, _] ->
+                            First;
+                        _ ->
+                            X
+                    end,
+                case re:split(NewX, <<"=">>) of
+                    [K, V] ->
+                        Acc#{K => V};
+                    _ -> Acc
+                end
+        end
+                end, #{}, re:split(UserZone, <<";">>)).
+
+to_frame(#{<<"QN">> := QN, <<"ST">> := ST, <<"CN">> := CN, <<"PW">> := PW, <<"MN">> := MN, <<"Flag">> := Flag, <<"CP">> := CP, <<"PNUM">> := PNUM, <<"PNO">> := PNO}) ->
+    Rdata = <<"QN=", QN/binary, ";ST=", ST/binary, ";CN=", CN/binary, ";PW=", PW/binary, ";MN=", MN/binary,
+        ";Flag=", Flag/binary, ";PNUM=", PNUM/binary, ";PNO=", PNO/binary, ";CP=&&", CP/binary, "&&">>,
+    Len = dgiot_atomgit_utils:get_len(Rdata),
+    Crc = dgiot_atomgit_utils:crc16(Rdata),
+    <<"##", Len/binary, Rdata/binary, Crc/binary, "\r\n">>;
+
+to_frame(#{<<"QN">> := QN, <<"ST">> := ST, <<"CN">> := CN, <<"PW">> := PW, <<"MN">> := MN, <<"Flag">> := Flag, <<"CP">> := CP}) ->
+    Rdata = <<"QN=", QN/binary, ";ST=", ST/binary, ";CN=", CN/binary, ";PW=", PW/binary, ";MN=", MN/binary, ";Flag=", Flag/binary, ";CP=&&", CP/binary, "&&">>,
+    Len = dgiot_atomgit_utils:get_len(Rdata),
+    Crc = dgiot_atomgit_utils:crc16(Rdata),
+    <<"##", Len/binary, Rdata/binary, Crc/binary, "\r\n">>.
+

apps/dgiot_atomgit/src/protocol/dgiot_atomgit_utils.erl

@@ -0,0 +1,361 @@
+%%--------------------------------------------------------------------
+%% Copyright (c) 2020-2021 DGIOT Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http) ->%%www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+%%--------------------------------------------------------------------
+
+-module(dgiot_atomgit_utils).
+-include_lib("dgiot_atomgit.hrl").
+-include_lib("dgiot/include/logger.hrl").
+%% API
+-export([
+    get_sys_type/1,
+    get_status_type/1,
+    get_result_type/1,
+    get_return_type/1,
+    get_len/1,
+    crc16/1,
+    get_cps/1,
+    getCommandType/1
+]).
+
+
+%%6.6.1.2 系统编码方法
+%%系统编码（见表 5）由两位取值 0~9、A~Z 的字符表示。
+%% 6．6．1．2 污染物在线监控 系统编码表
+%%---------------------------------------------------------------------------------------------------------------------------------------
+%%地表水质量监测 21
+%%空气质量监测 22
+%%声环境质量监测 23
+%%地下水质量监测 24
+%%土壤质量监测 25
+%%海水质量监测 26
+%%挥发性有机物监测 27
+%%大气环境污染源 31
+%%地表水体环境污染源 32
+%%地下水体环境污染源 33
+%%海洋环境污染源 34
+%%土壤环境污染源 35
+%%声环境污染源 36
+%%振动环境污染源 37
+%%放射性环境污染源 38
+%%工地扬尘污染源 39
+%%电磁环境污染源 41
+%%烟气排放过程监控 51
+%%污水排放过程监控 52
+%%系统交互 91 用于现场机和上位机的交互
+%%---------------------------------------------------------------------------------------------------------------------------------------
+
+get_sys_type(21) ->
+    <<"地表水质量监测"/utf8>>;
+get_sys_type(22) ->
+    <<"空气质量监测"/utf8>>;
+get_sys_type(23) ->
+    <<"声环境质量监测"/utf8>>;
+get_sys_type(24) ->
+    <<"地下水质量监测"/utf8>>;
+get_sys_type(25) ->
+    <<"土壤质量监测"/utf8>>;
+get_sys_type(26) ->
+    <<"海水质量监测"/utf8>>;
+get_sys_type(27) ->
+    <<"挥发性有机物监测"/utf8>>;
+get_sys_type(31) ->
+    <<"大气环境污染源"/utf8>>;
+get_sys_type(32) ->
+    <<"地表水体环境污染源"/utf8>>;
+get_sys_type(33) ->
+    <<"地下水体环境污染源"/utf8>>;
+get_sys_type(34) ->
+    <<"海洋环境污染源"/utf8>>;
+get_sys_type(35) ->
+    <<"土壤环境污染源"/utf8>>;
+get_sys_type(36) ->
+    <<"声环境污染源"/utf8>>;
+get_sys_type(37) ->
+    <<"振动环境污染源"/utf8>>;
+get_sys_type(38) ->
+    <<"放射性环境污染源"/utf8>>;
+get_sys_type(39) ->
+    <<"工地扬尘污染源"/utf8>>;
+get_sys_type(41) ->
+    <<"电磁环境污染源"/utf8>>;
+get_sys_type(51) ->
+    <<"烟气排放过程监控"/utf8>>;
+get_sys_type(52) ->
+    <<"污水排放过程监控"/utf8>>;
+get_sys_type(91) ->
+    <<"系统交互"/utf8>>.
+
+
+%%-------------------------------------------------------------------------------------
+%%6.6.2 执行结果定义（可扩充）
+%%执行结果定义如表 6 所示。
+%%表 6 执行结果定义表
+%%编号 描述 备注
+%% 1 执行成功
+%% 2 执行失败，但不知道原因
+%% 3 命令请求条件错误
+%% 4 通讯超时
+%% 5 系统繁忙不能执行
+%% 6 系统故障
+%% 100 没有数据
+%%------------------------------------------------------------------------------------
+
+get_result_type(1) ->
+    <<"执行成功">>;
+get_result_type(2) ->
+    <<"执行失败，但不知道原因">>;
+get_result_type(3) ->
+    <<"命令请求条件错误">>;
+get_result_type(4) ->
+    <<"通讯超时">>;
+get_result_type(5) ->
+    <<"系统繁忙不能执行">>;
+get_result_type(6) ->
+    <<"系统故障">>;
+get_result_type(100) ->
+    <<"没有数据">>;
+get_result_type(_) ->
+    <<"可扩充">>.
+
+%%6.6.3 请求命令返回（可扩充）
+%%请求命令返回如表 7 所示。
+%%表 7 请求命令返回表
+%%1 准备执行请求
+%%2 请求被拒绝
+%%3 PW 错误
+%%4 MN 错误
+%%5 ST 错误
+%%6 Flag 错误
+%%7 QN 错误
+%%8 CN 错误
+%%9 CRC 校验错误
+%%100 未知错误
+get_return_type(1) ->
+    <<"准备执行请求"/utf8>>;
+get_return_type(2) ->
+    <<"请求被拒绝"/utf8>>;
+get_return_type(3) ->
+    <<"PW 错误"/utf8>>;
+get_return_type(4) ->
+    <<"MN 错误"/utf8>>;
+get_return_type(5) ->
+    <<"ST 错误"/utf8>>;
+get_return_type(6) ->
+    <<"Flag 错误"/utf8>>;
+get_return_type(7) ->
+    <<"QN 错误"/utf8>>;
+get_return_type(8) ->
+    <<"CN 错误"/utf8>>;
+get_return_type(9) ->
+    <<"CRC 校验错误"/utf8>>;
+get_return_type(100) ->
+    <<"未知错误"/utf8>>;
+get_return_type(Type) ->
+    BinType = dgiot_utils:to_binary(Type),
+    <<"可扩充"/utf8, "_", BinType/binary>>.
+
+%%6.6.4 数据标记（可扩充）
+%%数据标记如表 8 所示。
+%% 表 8 数据标记表
+%%N 在线监控（监测）仪器仪表工作正常
+%%F 在线监控（监测）仪器仪表停运
+%%M 在线监控（监测）仪器仪表处于维护期间产生的数据
+%%S 手工输入的设定值
+%%D 在线监控（监测）仪器仪表故障
+%%C 在线监控（监测）仪器仪表处于校准状态
+%%T 在线监控（监测）仪器仪表采样数值超过测量上限
+%%B 在线监控（监测）仪器仪表与数采仪通讯异常
+
+get_status_type("N") ->
+    <<"在线监控（监测）仪器仪表工作正常"/utf8>>;
+get_status_type("F") ->
+    <<"在线监控（监测）仪器仪表停运"/utf8>>;
+get_status_type("M") ->
+    <<"在线监控（监测）仪器仪表处于维护期间产生的数据"/utf8>>;
+get_status_type("S") ->
+    <<"手工输入的设定值"/utf8>>;
+get_status_type("D") ->
+    <<"在线监控（监测）仪器仪表故障"/utf8>>;
+get_status_type("C") ->
+    <<"在线监控（监测）仪器仪表处于校准状态"/utf8>>;
+get_status_type("T") ->
+    <<"在线监控（监测）仪器仪表采样数值超过测量上限"/utf8>>;
+get_status_type("B") ->
+    <<"在线监控（监测）仪器仪表与数采仪通讯异常"/utf8>>;
+get_status_type(Type) ->
+    BinType = dgiot_utils:to_binary(Type),
+    <<"可扩充"/utf8, "_", BinType/binary>>.
+
+
+%%6.6.5.2 命令编码方法
+%%命令编码用 4 位阿拉伯数字表示，如表 9 所示。
+%%表 9 命令编码表
+
+%%设置超时时间及重发次数
+getCommandType(?SetTimeOutReSendTimes) ->
+    ?Request;
+%%提取现场机时间
+getCommandType(?GetSceneDeviceTime) ->
+    ?Request;
+%%设置现场机时间
+getCommandType(?SetSceneDeviceTime) ->
+    ?Request;
+%%现场机时间校准请求
+getCommandType(?SceneDeviceTimeCalibration) ->
+    ?Notice;
+%%提取实时数据间隔
+getCommandType(?GetRtdDataInterval) ->
+    ?Request;
+%%设置实时数据间隔
+getCommandType(?SetRtdDataInterval) ->
+    ?Request;
+%%提取分钟数据间隔
+getCommandType(?GetMinuteDataInterval) ->
+    ?Request;
+%%设置分钟数据间隔
+getCommandType(?SetMinuteDataInterval) ->
+    ?Request;
+%%用于设置现场机的密码
+getCommandType(?SetSceneDevicePassword) ->
+    ?Request;
+%%取污染物实时数据
+getCommandType(?GetRtdData) ->
+    ?Request;
+%%停止察看污染物实时数据
+getCommandType(?StopRtdData) ->
+    ?Notice;
+%%取设备运行状态数据
+getCommandType(?GetDeviceRunState) ->
+    ?Request;
+%%停止察看设备运行状态
+getCommandType(?StopDeviceRunState) ->
+    ?Request;
+%%取污染物日历史数据
+getCommandType(?GetDayData) ->
+    ?Request;
+%%取设备运行时间日历史数据
+getCommandType(?GetDeviceRunTimeDayData) ->
+    ?Request;
+%%取污染物分钟数据
+getCommandType(?GetMinuteData) ->
+    ?Request;
+%%取污染物小时数据
+getCommandType(?GetHourData) ->
+    ?Request;
+%%上传数采仪开机时间
+getCommandType(?UploadComputerPowerOnTime) ->
+    ?Upload;
+%%零点校准量程校准
+getCommandType(?RangeCalibration) ->
+    ?Request;
+%%即时采样
+getCommandType(?TakeSampleImmediately) ->
+    ?Request;
+%%启动清洗/反吹
+getCommandType(?StartClear) ->
+    ?Request;
+%%比对采样
+getCommandType(?CompareSample) ->
+    ?Request;
+%%超标留样
+getCommandType(?LeaveSuperstandardSample) ->
+    ?Request;
+%%设置采样时间周期
+getCommandType(?SetSampleTimeInterval) ->
+    ?Request;
+%%提取采样时间周期
+getCommandType(?GetSampleTimeInterval) ->
+    ?Request;
+%%提取出样时间
+getCommandType(?GetSampleTime) ->
+    ?Request;
+%%提取设备唯一标识
+getCommandType(?GetSceneDeviceUUID) ->
+    ?Request;
+%%提取现场机信息
+getCommandType(?GetSceneDeviceInfo) ->
+    ?Request;
+%%设置现场机参数
+getCommandType(?SetSceneDeviceParam) ->
+    ?Request;
+%%取污染物周期数据
+getCommandType(?GetCycleData) ->
+    ?Request;
+%%请求应答
+getCommandType(?RequestResponse) ->
+    ?Other;
+%%执行结果
+getCommandType(?ExecuteResponse) ->
+    ?Other;
+%%通知应答
+getCommandType(?NoticeResponse) ->
+    ?Other;
+%%数据应答
+getCommandType(?DataResponse) ->
+    ?Other;
+%%心跳
+getCommandType(?HeartBeat) ->
+    ?Other;
+getCommandType(_) ->
+    ?None.
+
+get_len(Rdata) ->
+    Len = byte_size(Rdata),
+    list_to_binary(lists:flatten(io_lib:format("~4.10.0b", [Len]))).
+
+
+%%CRC 校验（Cyclic Redundancy Check）是一种数据传输错误检查方法。本标准采用 ANSI CRC16，
+%%简称 CRC16。
+%%CRC16 码由传输设备计算后加入到数据包中。接收设备重新计算接收数据包的 CRC16 码，并与接
+%%收到的 CRC16 码比较，如果两值不同，则有误。
+%%CRC16 校验字节的生成步骤如下：
+%%1) CRC16 校验寄存器赋值为 0xFFFF；
+%%2) 取被校验串的第一个字节赋值给临时寄存器；
+%%3) 临时寄存器与 CRC16 校验寄存器的高位字节进行“异或”运算，赋值给 CRC16 校验寄存器；
+%%4) 取 CRC16 校验寄存器最后一位赋值给检测寄存器；
+%%5) 把 CRC16 校验寄存器右移一位；
+%%6) 若检测寄存器值为 1，CRC16 校验寄存器与多项式 0xA001 进行“异或”运算，赋值给 CRC16
+%%校验寄存器；
+%%7) 重复步骤 4~6，直至移出 8 位；
+%%8) 取被校验串的下一个字节赋值给临时寄存器；
+%%9) 重复步骤 3~8，直至被校验串的所有字节均被校验；
+%%10) 返回 CRC16 校验寄存器的值。
+%%校验码按照先高字节后低字节的顺序存放。
+crc16(Buff) -> crc16(Buff, 16#FFFF).
+crc16(<<>>, Crc) ->
+    list_to_binary(string:to_upper(lists:flatten(io_lib:format("~4.16.0b", [Crc]))));
+crc16(<<B:8, Rest/binary>>, Crc) ->
+    NewCrc =
+        lists:foldl(fun(_, CrcRegister) ->
+            Check = CrcRegister band 16#0001,
+            New = CrcRegister bsr 1,
+            case Check of
+                1 ->
+                    New bxor 16#A001;
+                0 ->
+                    New
+            end
+                    end, (Crc bsr 8) bxor B, lists:seq(1, 8)),
+    crc16(Rest, NewCrc).
+
+get_cps(ParamCodes) ->
+    CP1 = re:replace(ParamCodes, <<"&&">>, <<"">>),
+    lists:foldl(fun(X, Acc) ->
+        case re:split(X, <<"=">>) of
+            [K, V] ->
+                Acc#{K => V};
+            _ -> Acc
+        end
+                end, #{}, re:split(CP1, ",")).

apps/dgiot_task/src/dgiot_task.erl

@@ -20,8 +20,85 @@
 -include_lib("dgiot_bridge/include/dgiot_bridge.hrl").
 
 -export([start/1, start/2, send/3, get_pnque_len/1, save_pnque/4, get_pnque/1, del_pnque/1, save_td/4, merge_cache_data/3, save_cache_data/2]).
--export([get_props/1, get_control/3, get_collection/4, get_calculated/2, get_instruct/2, get_storage/2, string2value/2, string2value/3]).
+-export([get_props/1, get_control/3, get_collection/4, get_calculated/4, get_instruct/2, get_storage/2, string2value/2, string2value/3]).
 -export([save_td_no_match/4]).
+
+%% 注册协议类型
+-protocol_type(#{
+    cType => <<"TASKSTATISTICS">>,
+    type => <<"TASKSTATISTICS">>,
+    colum => 10,
+    title => #{
+        zh => <<"任务统计"/utf8>>
+    },
+    description => #{
+        zh => <<"任务统计"/utf8>>
+    }
+}).
+
+-params(#{
+    <<"type">> => #{
+        order => 1,
+        type => string,
+        required => true,
+        default => #{<<"value">> => <<"duration">>, <<"label">> => <<"时长累加"/utf8>>},
+        enum => [
+            #{<<"value">> => <<"duration">>, <<"label">> => <<"时长累加"/utf8>>},
+            #{<<"value">> => <<"frequency">>, <<"label">> => <<"次数累加"/utf8>>}
+        ],
+        title => #{
+            zh => <<"条件"/utf8>>
+        },
+        description => #{
+            zh => <<"条件"/utf8>>
+        }
+    },
+    <<"key">> => #{
+        order => 2,
+        type => string,
+        required => true,
+        default => <<"key"/utf8>>,
+        title => #{
+            zh => <<"物模型标识符"/utf8>>
+        },
+        description => #{
+            zh => <<"统计的物模型标识符"/utf8>>
+        }
+    },
+    <<"comparetype">> => #{
+        order => 3,
+        type => string,
+        required => true,
+        default => #{<<"value">> => <<"EQ">>, <<"label">> => <<"等于"/utf8>>},
+        enum => [
+            #{<<"value">> => <<"LT">>, <<"label">> => <<"小于"/utf8>>},
+            #{<<"value">> => <<"LE">>, <<"label">> => <<"小于等于"/utf8>>},
+            #{<<"value">> => <<"GT">>, <<"label">> => <<"大于"/utf8>>},
+            #{<<"value">> => <<"GE">>, <<"label">> => <<"大于等于"/utf8>>},
+            #{<<"value">> => <<"EQ">>, <<"label">> => <<"等于"/utf8>>},
+            #{<<"value">> => <<"NE">>, <<"label">> => <<"不等于"/utf8>>}
+        ],
+        title => #{
+            zh => <<"条件"/utf8>>
+        },
+        description => #{
+            zh => <<"条件"/utf8>>
+        }
+    },
+    <<"value">> => #{
+        order => 4,
+        type => string,
+        required => true,
+        default => <<"1">>,
+        title => #{
+            zh => <<"值"/utf8>>
+        },
+        description => #{
+            zh => <<"物模型比较值"/utf8>>
+        }
+    }
+}).
+
 start(ChannelId) ->
     lists:map(fun(Y) ->
         case Y of
@@ -60,18 +137,108 @@ send(ProductId, DevAddr, Payload) ->
             dgiot_client:send(ChannelId, DevAddr, Topic, Payload)
     end.
 
+%% 比较统计值
+compare(KeyValue, <<"LT">>, Value) ->
+    KeyValue < Value;
+compare(KeyValue, <<"LE">>, Value) ->
+    KeyValue =< Value;
+compare(KeyValue, <<"GT">>, Value) ->
+    KeyValue > Value;
+compare(KeyValue, <<"GE">>, Value) ->
+    KeyValue >= Value;
+compare(KeyValue, <<"EQ">>, Value) ->
+    KeyValue == Value;
+compare(KeyValue, <<"NE">>, Value) ->
+    KeyValue =/= Value;
+compare(_, _, _) ->
+    false.
+
+%% 查询上次值
+%% select last(devaddr) as devaddr FROM  _24b9b4bc50._1c9966755d;
+get_last_value(ProductId, DevAddr, Identifier) ->
+    case dgiot_data:get({last_value, ProductId, DevAddr, Identifier}) of
+        not_find ->
+            case dgiot_tdengine:get_channel(ProductId) of
+                {ok, Channel} ->
+                    dgiot_tdengine:transaction(Channel,
+                        fun(Context) ->
+                            DB = dgiot_tdengine:get_database(Channel, ProductId),
+                            DeviceId = dgiot_parse_id:get_deviceid(ProductId, DevAddr),
+                            Sql = <<"select last(", Identifier/binary, ") as ", Identifier/binary, " FROM ", DB/binary, "_", DeviceId/binary, ";">>,
+                            case dgiot_tdengine_pool:run_sql(Context#{<<"channel">> => Channel}, execute_query, Sql) of
+                                {ok, #{<<"results">> := [#{Identifier := Value} | _]}} when Value =/= null ->
+                                    dgiot_utils:to_int(Value);
+                                _ ->
+                                    0
+                            end
+                        end);
+                _ ->
+                    0
+            end;
+        Value ->
+            dgiot_utils:to_int(Value)
+    end.
+
+%% 统计时长
+get_statistic(ProductId, DevAddr, Identifier, KeyValue, #{<<"type">> := <<"duration">>, <<"comparetype">> := Comparetype, <<"value">> := Value}, Acc) ->
+    Last_Value = get_last_value(ProductId, DevAddr, Identifier),
+    case compare(KeyValue, Comparetype, Value) of
+        true ->
+            Last_Value = get_last_value(ProductId, DevAddr, Identifier),
+            Time =
+                case dgiot_data:get({last_time, ProductId, DevAddr, Identifier}) of
+                    {true, OldTime} ->
+                        dgiot_datetime:now_secs() - OldTime;
+                    _ ->
+                        0
+                end,
+            dgiot_data:insert({last_time, ProductId, DevAddr, Identifier}, {true, dgiot_datetime:now_secs()}),
+            dgiot_data:insert({last_value, ProductId, DevAddr, Identifier}, Last_Value + Time),
+            Acc#{Identifier => Last_Value + Time};
+        _ ->
+            dgiot_data:insert({last_time, ProductId, DevAddr, Identifier}, {false, dgiot_datetime:now_secs()}),
+            Acc#{Identifier => Last_Value}
+    end;
+
+%% 次数累加
+get_statistic(ProductId, DevAddr, Identifier, KeyValue, #{<<"type">> := <<"frequency">>, <<"comparetype">> := Comparetype, <<"value">> := Value}, Acc) ->
+    Num = get_last_value(ProductId, DevAddr, Identifier),
+    case compare(KeyValue, Comparetype, Value) of
+        true ->
+            dgiot_data:insert({last_value, ProductId, DevAddr, Identifier}, Num + 1),
+            Acc#{Identifier => Num + 1};
+        _ ->
+            Acc#{Identifier => Num}
+    end;
+
+get_statistic(_, _, _, _, _, Acc) ->
+    Acc.
+
 %%获取计算值，必须返回物模型里面的数据表示，不能用寄存器地址
-get_calculated(Calculated, Props) ->
+get_calculated(ProductId, DevAddr, Calculated, Props) ->
     lists:foldl(fun(X, Acc) ->
         case Acc of
             error ->
                 Acc;
             _ ->
                 case X of
+                    #{<<"isaccumulate">> := true,
+                        <<"isstorage">> := true,
+                        <<"identifier">> := Identifier,
+                        <<"dataForm">> := #{<<"strategy">> := <<"计算值"/utf8>>},
+                        <<"dataSource">> := #{<<"key">> := Key} = DataSource
+                    } ->
+                        case maps:get(Key, Calculated, not_find) of
+                            not_find ->
+                                Acc;
+                            KeyValue ->
+                                get_statistic(ProductId, DevAddr, Identifier, dgiot_utils:to_int(KeyValue), DataSource, Acc)
+                        end;
                     #{<<"isstorage">> := true,
-                        <<"identifier">> := Identifier, <<"dataForm">> := #{
+                        <<"identifier">> := Identifier,
-                        <<"strategy">> := <<"计算值"/utf8>>, <<"collection">> := Collection},
+                        <<"dataForm">> := #{<<"strategy">> := <<"计算值"/utf8>>, <<"collection">> := Collection},
-                        <<"dataType">> := #{<<"type">> := Type, <<"specs">> := Specs}} ->
+                        <<"dataType">> := #{<<"type">> := Type, <<"specs">> := Specs}
+                    } ->
                         Str1 = maps:fold(fun(K, V, Acc2) ->
                             Str = re:replace(Acc2, dgiot_utils:to_list(<<"%%{", K/binary, "}">>), dgiot_utils:to_list(V), [global, {return, list}]),
                             re:replace(Str, "%{s}", dgiot_utils:to_list(V), [global, {return, list}])
@@ -317,7 +484,7 @@ save_td(ProductId, DevAddr, Ack, _AppData) ->
             %%            计算上报值
             Collection = dgiot_task:get_collection(ProductId, [], CacheData, Props),
             %%            计算计算值
-            AllData = dgiot_task:get_calculated(Collection, Props),
+            AllData = dgiot_task:get_calculated(ProductId, DevAddr, Collection, Props),
             %%            过滤存储值
             Storage = dgiot_task:get_storage(AllData, Props),
             case Interval > 0 of
@@ -402,7 +569,7 @@ save_td_no_match(ProductId, DevAddr, Ack, AppData) ->
 %%            计算上报值
             Collection = dgiot_task:get_collection(ProductId, [], Ack, Props),
 %%            计算计算值
-            Calculated = dgiot_task:get_calculated(Collection, Props),
+            Calculated = dgiot_task:get_calculated(ProductId, DevAddr, Collection, Props),
 %%            过滤存储值
             Storage = dgiot_task:get_storage(Calculated, Props),
             DeviceId = dgiot_parse_id:get_deviceid(ProductId, DevAddr),