test/dgiot
1
0
Fork 0
mirror of https://gitee.com/dgiiot/dgiot.git synced 2024-11-29 18:57:41 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

release: add dlt376 proctol

Browse Source
This commit is contained in:
lsxredrain 2021-10-08 09:21:59 +08:00
parent f3b173c3f1
commit 1d4e560343
9 changed files with 1086 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CHANGELOG.md
View File

@ -1,8 +1,9 @@
# [](https://github.com/dgiot/dgiot/compare/v4.3.4...v) (2021-10-07)
# [](https://github.com/dgiot/dgiot/compare/v4.3.4...v) (2021-10-08)
### Bug Fixes
* add device log ([f3b173c](https://github.com/dgiot/dgiot/commit/f3b173c3f146e1ddf21038b5026dfd08ae93c95f))
* channel_log ([1e37058](https://github.com/dgiot/dgiot/commit/1e3705876eb905cab215085e7341e050ce822170))
* task_worker bug ([0396163](https://github.com/dgiot/dgiot/commit/039616312126d24e36b03cf339e0e6ffb6f3aa13))

23
apps/dgiot_meter/include/dgiot_meter.hrl
View File

@ -14,8 +14,20 @@
%% limitations under the License.
%%--------------------------------------------------------------------
% 使用说明:
% --------------------------------------------------------------------
% 在 dgiot_task文件 save_pnque(DtuProductId, DtuAddr, ProductId, DevAddr) 函数,用于注册电表的远程控制的topic
% timer:sleep(500),
% TopicCtrl = <<"thingctrl/", ProductId/binary, "/", DevAddr/binary>>,
% dgiot_mqtt:subscribe(TopicCtrl),
% timer:sleep(500),
% TopicStatus = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
% dgiot_mqtt:subscribe(TopicStatus),
% --------------------------------------------------------------------
-define(METER, <<"METER">>).
-define(DLT645, <<"DLT645">>).
-define(DLT376, <<"DLT376">>).
-record(state, {
id,
@ -23,11 +35,20 @@
dtuaddr = <<>>,
step = login,
ref = undefined,
search = <<"quick">>
search = <<"quick">>,
protocol = dlt645
}).
%% Internal Header File
%% @doc dlt376 COMMAND.
-define(DLT376_MS_READ_DATA, 16#5B).
-define(DLT376_MS_READ_DATA_AFN, 16#0C).
-define(DLT376_MS_CONVERT_SEND_AFN, 16#10). %透明转发
-define(DLT376_MS_CTRL_DEV, 16#5A).
-define(DLT376_MS_CTRL_DEV_AFN, 16#05).
%% @doc dlt645 COMMAND.
-define(DLT645_MS_BROADCAST_DATA, 16#08).
-define(DLT645_MS_READ_DATA, 16#11).

127
apps/dgiot_meter/priv/swagger/swagger_meter.json
View File

@ -1,11 +1,126 @@
{
"definitions": {},
"paths": {
},
"tags": [
{
"description": "电力抄表",
"name": "METER"
"name": "METER",
"description": "电力抄表"
}
]
],
"definitions": {
},
"paths": {
"/meter_ctrl": {
"get": {
"security": [],
"summary": "对远程电表进行关闸/合闸(重要：接口调用后间隔15秒以上，避免对同一终端设备频繁操作)",
"description": "对远程电表进行关闸/合闸",
"parameters": [
{
"in": "query",
"name": "pid",
"required": false,
"description": "产品ID",
"type": "string",
"default": "3ea14b556a"
},
{
"in": "query",
"name": "devaddr",
"description": "设备地址",
"required": true,
"type": "string",
"default": "000003146332"
},
{
"in": "query",
"name": "ctrlflag",
"description": "=ture合闸, =false 开闸",
"required": true,
"type": "boolean",
"default": true
},
{
"in": "query",
"name": "devpass",
"description": "终端设备密码",
"required": true,
"type": "string",
"default": "000000"
}
],
"responses": {
"200": {
"description": "Returns operation status"
},
"400": {
"description": "Bad Request"
},
"401": {
"description": "Unauthorized"
},
"403": {
"description": "Forbidden"
},
"500": {
"description": "Server Internal error"
}
},
"tags": [
"METER"
]
}
},
"/meter_ctrl_status": {
"get": {
"security": [],
"summary": "查询拉闸/合闸的状态(重要：接口调用后间隔15秒以上，避免对同一终端设备频繁操作)",
"description": "返回上次拉闸/合闸的时间，连接mqtt服务（连接信息：ws://121.40.78.136:8083/mqtt）<br>订阅topic：thing/productid/devid/status(如 thing/3ea14b556a/000003146332/status) 获取结果<br> 订阅mqtt主题之后，收到格式 ：<br>{\"1D000101\":1632549207}（上次电表跳闸时间）<br> {\"1E000101\":1632549176}(上次电表合闸时间)<br> {\"FEFEFEFE\":0} 合闸、跳闸成功",
"parameters": [
{
"in": "query",
"name": "pid",
"required": false,
"description": "产品ID",
"type": "string",
"default": "3ea14b556a"
},
{
"in": "query",
"name": "ctrlflag",
"description": "=ture合闸, =false 开闸",
"required": true,
"type": "boolean",
"default": true
},
{
"in": "query",
"name": "devaddr",
"description": "设备地址",
"required": true,
"type": "string",
"default": "000003146332"
}
],
"responses": {
"200": {
"description": "Returns operation status"
},
"400": {
"description": "Bad Request"
},
"401": {
"description": "Unauthorized"
},
"403": {
"description": "Forbidden"
},
"500": {
"description": "Server Internal error"
}
},
"tags": [
"METER"
]
}
}
}
}

165
apps/dgiot_meter/src/dgiot_meter.erl
View File

@ -24,6 +24,7 @@
create_dtu/3,
create_dtu/4,
create_meter/4,
create_meter4G/3,
get_sub_device/1
]).
@ -83,6 +84,22 @@ create_meter(MeterAddr, ChannelId, DTUIP, DtuAddr) ->
{DtuProductId, _, _} = dgiot_data:get({dtu, ChannelId}),
dgiot_task:save_pnque(DtuProductId, DtuAddr, ProductId, MeterAddr).
create_meter4G(MeterAddr, ChannelId, DTUIP) ->
{ProductId, ACL, _Properties} = dgiot_data:get({meter, ChannelId}),
Requests = #{
<<"devaddr">> => MeterAddr,
<<"name">> => <<"Meter_", MeterAddr/binary>>,
<<"ip">> => DTUIP,
<<"isEnable">> => true,
<<"product">> => ProductId,
<<"ACL">> => ACL,
<<"status">> => <<"ONLINE">>,
<<"brand">> => <<"Meter", MeterAddr/binary>>,
<<"devModel">> => <<"Meter">>
},
dgiot_device:create_device(Requests),
dgiot_task:save_pnque(ProductId, MeterAddr, ProductId, MeterAddr).
get_sub_device(DtuAddr) ->
Query = #{<<"keys">> => [<<"devaddr">>, <<"product">>],
<<"where">> => #{<<"route.", DtuAddr/binary>> => #{<<"$regex">> => <<".+">>}},
@ -93,12 +110,37 @@ get_sub_device(DtuAddr) ->
_ -> []
end.
parse_frame(dlt645, Buff, Opts) ->
parse_frame(?DLT645, Buff, Opts) ->
{Rest, Frames} = dlt645_decoder:parse_frame(Buff, Opts),
{Rest, lists:foldl(fun(X, Acc) ->
Acc ++ [maps:without([<<"diff">>, <<"send_di">>], X)]
end, [], Frames)};
parse_frame(?DLT376, Buff, Opts) ->
{Rest, Frames} = dlt376_decoder:parse_frame(Buff, Opts),
% ?LOG(warning,"GGM 170 dgiot_meter parse_frame:~p", [Frames]),
{Rest, lists:foldl(fun(X, Acc) ->
Acc ++ [maps:without([<<"diff">>, <<"send_di">>], X)]
end, [], Frames)}.
% DLT376发送抄数指令
to_frame(#{
<<"devaddr">> := Addr,
<<"di">> := Di,
<<"command">> := <<"r">>,
<<"protocol">> := ?DLT376,
<<"data">> := <<"null">>
} = Frame) ->
dlt376_decoder:to_frame(Frame#{
<<"msgtype">> => ?DLT376,
<<"addr">> => dlt376_proctol:decode_of_addr(dgiot_utils:hex_to_binary(Addr)),
<<"data">> => <<>>,
<<"di">> => Di,
<<"command">> => ?DLT376_MS_READ_DATA,
<<"afn">> => ?DLT376_MS_READ_DATA_AFN
});
% DLT645 组装电表抄表指令
to_frame(#{
<<"devaddr">> := Addr,
<<"di">> := Di,
@ -114,6 +156,127 @@ to_frame(#{
<<"command">> => ?DLT645_MS_READ_DATA
});
% DLT645 组装电表控制指令
to_frame(#{
<<"devaddr">> := Addr,
<<"ctrlflag">> := CtrlFlag,
<<"protocol">> := ?DLT645,
<<"devpass">> := DevPass,
<<"apiname">> := get_meter_ctrl
} = Frame) ->
case CtrlFlag of
true ->
PassGrade = <<"02">>,
Di = <<(dgiot_utils:hex_to_binary(DevPass))/binary,(dgiot_utils:hex_to_binary(PassGrade))/binary>>,
Data= <<"111111111C00010101010133">>,
dlt645_decoder:to_frame(Frame#{
<<"msgtype">> => ?DLT645,
<<"addr">> => dlt645_proctol:reverse(dgiot_utils:hex_to_binary(Addr)),
<<"di">> => dlt645_proctol:reverse(Di),
<<"data">> => dgiot_utils:hex_to_binary(Data),
<<"command">> => ?DLT645_MS_FORCE_EVENT
});
false ->
PassGrade = <<"02">>,
Di = <<(dgiot_utils:hex_to_binary(DevPass))/binary,(dgiot_utils:hex_to_binary(PassGrade))/binary>>,
Data = <<"111111111A00010101010133">>,
dlt645_decoder:to_frame(Frame#{
<<"msgtype">> => ?DLT645,
<<"addr">> => dlt645_proctol:reverse(dgiot_utils:hex_to_binary(Addr)),
<<"data">> => dgiot_utils:hex_to_binary(Data),
<<"di">> => dlt645_proctol:reverse(Di),
<<"command">> => ?DLT645_MS_FORCE_EVENT
})
end;
% DLT376 远程电表控制（透明转发）
to_frame(#{
<<"devaddr">> := DevAddr,
<<"ctrlflag">> := CtrlFlag,
<<"devpass">> := DevPass,
<<"protocol">> := ?DLT376,
<<"apiname">> := get_meter_ctrl
} = Frame) ->
Di = <<"00000100">>,
Data= <<16#02,16#6B,16#64,16#64,16#1C,16#00>>,
Data2 = <<16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00>>,
CtrlPayload = to_frame(Frame#{
<<"protocol">> => ?DLT645,
<<"devaddr">> => DevAddr,
<<"ctrlflag">> => CtrlFlag,
<<"devpass">> => DevPass,
<<"apiname">> => get_meter_ctrl
}),
DataNew = <<Data/binary,CtrlPayload/binary,Data2/binary>>,
% ?LOG(info, "GGM 230 to_frame, DataNew ~p~n~n~n",[dgiot_utils:binary_to_hex(DataNew)]),
RetPlayload = dlt376_decoder:to_frame(Frame#{
<<"msgtype">> => ?DLT376,
<<"addr">> => dlt376_proctol:decode_of_addr(dgiot_utils:hex_to_binary(DevAddr)),
<<"data">> => dgiot_utils:binary_to_hex(DataNew),
<<"di">> => Di,
<<"command">> => ?DLT376_MS_READ_DATA,
<<"afn">> => ?DLT376_MS_CONVERT_SEND_AFN
}),
% ?LOG(info, "GGM 231 to_frame, Payload1 ~p~n~n~n",[dgiot_utils:binary_to_hex(RetPlayload)]),
RetPlayload;
% DLT645 组装电表获取上次拉闸合闸的时间
to_frame(#{
<<"devaddr">> := DevAddr,
<<"ctrlflag">> := CtrlFlag,
<<"protocol">> := ?DLT645,
<<"apiname">> := get_meter_ctrl_status
} = Frame) ->
case CtrlFlag of
true ->
Di = <<"1E000101">>,
dlt645_decoder:to_frame(Frame#{
<<"msgtype">> => ?DLT645,
<<"addr">> => dlt645_proctol:reverse(dgiot_utils:hex_to_binary(DevAddr)),
<<"di">> => dlt645_proctol:reverse(dgiot_utils:hex_to_binary(Di)),
<<"data">> => <<>>,
<<"command">> => ?DLT645_MS_READ_DATA
});
false ->
Di = <<"1D000101">>,
dlt645_decoder:to_frame(Frame#{
<<"msgtype">> => ?DLT645,
<<"addr">> => dlt645_proctol:reverse(dgiot_utils:hex_to_binary(DevAddr)),
<<"di">> => dlt645_proctol:reverse(dgiot_utils:hex_to_binary(Di)),
<<"data">> => <<>>,
<<"command">> => ?DLT645_MS_READ_DATA
})
end;
% DLT376 组装电表获取上次拉闸合闸的时间（透明转发）
to_frame(#{
<<"devaddr">> := DevAddr,
<<"ctrlflag">> := CtrlFlag,
<<"protocol">> := ?DLT376,
<<"apiname">> := get_meter_ctrl_status
} = Frame) ->
Di = <<"00000100">>,
Data= <<16#02,16#6B,16#64,16#64,16#10,16#00>>,
Data2 = <<16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00,16#00>>,
CtrlPayload = to_frame(Frame#{
<<"devaddr">> := DevAddr,
<<"ctrlflag">> := CtrlFlag,
<<"protocol">> := ?DLT645,
<<"apiname">> := get_meter_ctrl_status
}),
DataNew = <<Data/binary,CtrlPayload/binary,Data2/binary>>,
% ?LOG(info, "GGM 260 to_frame, DataNew ~p,~n~n~n",[dgiot_utils:binary_to_hex(CtrlPayload)]),
RetPlayload = dlt376_decoder:to_frame(Frame#{
<<"msgtype">> => ?DLT376,
<<"addr">> => dlt376_proctol:decode_of_addr(dgiot_utils:hex_to_binary(DevAddr)),
<<"data">> => dgiot_utils:binary_to_hex(DataNew),
<<"di">> => Di,
<<"command">> => ?DLT376_MS_READ_DATA,
<<"afn">> => ?DLT376_MS_CONVERT_SEND_AFN
}),
RetPlayload;
to_frame(#{
<<"devaddr">> := Addr,
<<"di">> := Di,

63
apps/dgiot_meter/src/dgiot_meter_handler.erl
View File

@ -17,7 +17,8 @@
-author("johnliu").
-behavior(dgiot_rest).
-include_lib("dgiot/include/logger.hrl").
-include("dgiot_meter.hrl").
-dgiot_rest(all).
%% API
-export([swagger_meter/0]).
-export([handle/4, check_auth/3]).
@ -57,16 +58,16 @@ handle(OperationID, Args, Context, Req) ->
end,
{500, Headers, #{<<"error">> => Err}};
ok ->
?LOG(debug,"do request: ~p, ~p ->ok ~n", [OperationID, Args]),
?LOG(debug, "do request: ~p, ~p ->ok ~n", [OperationID, Args]),
{200, Headers, #{}, Req};
{ok, Res} ->
?LOG(debug,"do request: ~p, ~p ->~p~n", [OperationID, Args, Res]),
?LOG(debug, "do request: ~p, ~p ->~p~n", [OperationID, Args, Res]),
{200, Headers, Res, Req};
{Status, Res} ->
?LOG(debug,"do request: ~p, ~p ->~p~n", [OperationID, Args, Res]),
?LOG(debug, "do request: ~p, ~p ->~p~n", [OperationID, Args, Res]),
{Status, Headers, Res, Req};
{Status, NewHeaders, Res} ->
?LOG(debug,"do request: ~p, ~p ->~p~n", [OperationID, Args, Res]),
?LOG(debug, "do request: ~p, ~p ->~p~n", [OperationID, Args, Res]),
{Status, maps:merge(Headers, NewHeaders), Res, Req}
end.
@ -75,14 +76,58 @@ handle(OperationID, Args, Context, Req) ->
%%% 内部函数 Version:API版本
%%%===================================================================
%% TDengine 概要: 获取当前产品下的所有设备数据 描述:获取当前产品下的所有设备数据
%% OperationId:get_td_cid_pid
%% 请求:GET /iotapi/td/prodcut/:productId
do_request(get_meter_ctrl, #{
<<"pid">> := ProductId,
<<"devaddr">> := DevAddr,
<<"ctrlflag">> := CtrlFlag,
<<"devpass">> := DevPass
}, _Context, Req0) ->
get_meter_ctrl(Req0, ProductId, DevAddr, CtrlFlag, DevPass);
do_request(get_meter_ctrl_status, #{
<<"pid">> := ProductId,
<<"ctrlflag">> := CtrlFlag,
<<"devaddr">> := DevAddr
}, _Context, _Req) ->
TopicCtrl = <<"thingctrl/", ProductId/binary, "/", DevAddr/binary>>,
ThingData = #{<<"devaddr">> => DevAddr, <<"ctrlflag">> => CtrlFlag, <<"apiname">> => get_meter_ctrl_status},
Payload = [#{<<"appdata">> => #{}, <<"thingdata">> => ThingData}],
dgiot_mqtt:publish(DevAddr, TopicCtrl, jsx:encode(Payload));
%% 服务器不支持的API接口
do_request(_OperationId, _Args, _Context, _Req) ->
?LOG(info,"_OperationId:~p~n", [_OperationId]),
?LOG(info, "_OperationId:~p~n", [_OperationId]),
{error, <<"Not Allowed.">>}.
get_meter_ctrl(Req0, ProductId, DevAddr, CtrlFlag, DevPass) ->
Sendtopic = <<"thingctrl/", ProductId/binary, "/", DevAddr/binary>>,
ThingData = #{<<"devaddr">> => DevAddr, <<"ctrlflag">> => CtrlFlag, <<"devpass">> => DevPass, <<"apiname">> => get_meter_ctrl},
Payload = [#{<<"pid">> => self(),<<"appdata">> => #{}, <<"thingdata">> => ThingData}],
case dgiot_mqtt:has_routes(Sendtopic) of
true ->
dgiot_mqtt:publish(DevAddr, Sendtopic, jsx:encode(Payload)),
receive
{ctrl_meter, Msg} ->
Resp = cowboy_req:reply(200, #{<<"content-type">> => <<"application/json">>}, jsx:encode(Msg), Req0),
{ok, Resp};
{error} ->
Resp = cowboy_req:reply(200, #{<<"content-type">> => <<"text/plain">>}, <<"CTRL_METER_FAILED">>, Req0),
{ok, Resp}
after 10000 ->
Resp = cowboy_req:reply(200, #{
<<"content-type">> => <<"text/plain">>
}, <<"CTRL_METER_TIMEOUT">>, Req0),
{ok, Resp}
end;
false ->
Resp = cowboy_req:reply(200, #{<<"content-type">> => <<"text/plain">>}, <<"METER_OFFLINE">>, Req0),
{ok, Resp}
end.

421
apps/dgiot_meter/src/dgiot_meter_tcp.erl
View File

@ -31,43 +31,65 @@ start(Port, State) ->
%% tcp server start
%% {ok, State} | {stop, Reason}
init(TCPState) ->
dgiot_metrics:inc(dgiot_meter, <<"dtu_login">>, 1),
{ok, TCPState}.
%%设备登录报文，登陆成功后，开始搜表
%
handle_info({tcp, DtuAddr}, #tcp{socket = Socket, state = #state{id = ChannelId, dtuaddr = <<>>, search = Search} = State} = TCPState) ->
DTUIP = dgiot_utils:get_ip(Socket),
HexDtuAddr = dgiot_utils:binary_to_hex(DtuAddr),
dgiot_meter:create_dtu(HexDtuAddr, ChannelId, DTUIP),
{DtuProductId, _, _} = dgiot_data:get({dtu, ChannelId}),
Topic = <<"profile/", DtuProductId/binary, "/", DtuAddr/binary>>,
dgiot_mqtt:subscribe(Topic),
{NewRef, NewStep} =
case Search of
<<"nosearch">> ->
[dgiot_task:save_pnque(DtuProductId, DtuAddr, Meterproductid, Meteraddr) || #{<<"product">> := Meterproductid, <<"devaddr">> := Meteraddr}
<- dgiot_meter:get_sub_device(DtuAddr)],
{undefined, read_meter};
<<"quick">> ->
dgiot_meter:search_meter(tcp, undefined, TCPState, 0),
{undefined, search_meter};
_ ->
{Ref, Step, _Payload} = dgiot_meter:search_meter(tcp, undefined, TCPState, 1),
{Ref, Step}
end,
{noreply, TCPState#tcp{buff = <<>>, state = State#state{dtuaddr = HexDtuAddr, ref = NewRef, step = NewStep}}};
?LOG(info, "GGM 334 dgiot_meter_tcp, handle_info (登录) ~p,,,~p",[DtuAddr,TCPState]),
{_,[Acc|_]} = dlt376_decoder:parse_frame(DtuAddr,[]),
#{<<"msgtype">> := Protocol} = Acc,
case Protocol of
?DLT376 ->
#{<<"addr">> := MeterAddr} = Acc,
dgiot_meter:create_meter4G(MeterAddr, ChannelId, DTUIP),
{ProductId, _,_} = dgiot_data:get({meter, ChannelId}),
Topic = <<"thing/", ProductId/binary, "/", MeterAddr/binary>>,
dgiot_mqtt:subscribe(Topic), %为这个设备订阅一个mqtt
% timer:sleep(500),
% TopicCtrl = <<"thingctrl/", ProductId/binary, "/", MeterAddr/binary>>,
% dgiot_mqtt:subscribe(TopicCtrl),
{NewRef, NewStep} = {undefined, read_meter},
{noreply, TCPState#tcp{buff = <<>>, state = State#state{dtuaddr = MeterAddr, protocol = ?DLT376,ref = NewRef, step = NewStep}}};
?DLT645 ->
% HexDtuAddr = dgiot_utils:binary_to_hex(DtuAddr),
?LOG(info, "GGM 335 dgiot_meter_tcp, handle_info (DLT645创建) ~p,",[DtuAddr]),
dgiot_meter:create_dtu(DtuAddr, ChannelId, DTUIP),
{DtuProductId, _, _} = dgiot_data:get({dtu, ChannelId}),
Topic = <<"profile/", DtuProductId/binary, "/", DtuAddr/binary>>,
dgiot_mqtt:subscribe(Topic),
{NewRef, NewStep} =
case Search of
<<"nosearch">> ->
[dgiot_task:save_pnque(DtuProductId, DtuAddr, Meterproductid, Meteraddr) || #{<<"product">> := #{<<"objectId">>:=Meterproductid}, <<"devaddr">> := Meteraddr}
<- dgiot_meter:get_sub_device(DtuAddr)],
{undefined, read_meter};
<<"quick">> ->
dgiot_meter:search_meter(tcp, undefined, TCPState, 0),
{undefined, search_meter};
_ ->
{Ref, Step, _Payload} = dgiot_meter:search_meter(tcp, undefined, TCPState, 1),
{Ref, Step}
end,
{noreply, TCPState#tcp{buff = <<>>, state = State#state{dtuaddr = DtuAddr, protocol = ?DLT645,ref = NewRef, step = NewStep}}};
_ ->
?LOG(info, "GGM 334 dgiot_meter_tcp, handle_info (登录s失败) ~p",[dgiot_utils:binary_to_hex(DtuAddr)]),
dgiot_utils:binary_to_hex(DtuAddr)
end;
%%定时器触发搜表
handle_info(search_meter, #tcp{state = #state{ref = Ref} = State} = TCPState) ->
%%定时器触发搜表
handle_info(search_meter, #tcp{state = #state{ref = Ref,protocol = ?DLT645} = State} = TCPState) ->
{NewRef, Step, _Payload} = dgiot_meter:search_meter(tcp, Ref, TCPState, 1),
{noreply, TCPState#tcp{buff = <<>>, state = State#state{ref = NewRef, step = Step}}};
%%ACK报文触发搜表
handle_info({tcp, Buff}, #tcp{socket = Socket, state = #state{id = ChannelId, dtuaddr = DtuAddr, ref = Ref, step = search_meter, search = Search} = State} = TCPState) ->
handle_info({tcp, Buff}, #tcp{socket = Socket, state = #state{id = ChannelId, dtuaddr = DtuAddr, protocol = ?DLT645,ref = Ref, step = search_meter, search = Search} = State} = TCPState) ->
?LOG(info, "from_dev: search_meter Buff ~p", [dgiot_utils:binary_to_hex(Buff)]),
?LOG(info, "from_dev: parse_frame Buff ~p", [dgiot_meter:parse_frame(dlt645, Buff, [])]),
{Rest, Frames} = dgiot_meter:parse_frame(dlt645, Buff, []),
?LOG(info, "from_dev: parse_frame Buff ~p", [dgiot_meter:parse_frame(?DLT645, Buff, [])]),
{Rest, Frames} = dgiot_meter:parse_frame(?DLT645, Buff, []),
lists:map(fun(X) ->
case X of
#{<<"addr">> := Addr} ->
@ -92,19 +114,29 @@ handle_info({tcp, Buff}, #tcp{socket = Socket, state = #state{id = ChannelId, dt
end
end;
%%接受抄表任务命令抄表
handle_info({deliver, _Topic, Msg}, #tcp{state = #state{id = ChannelId, step = read_meter}} = TCPState) ->
%%接受抄表任务命令抄表(下发指令)
handle_info({deliver, _Topic, Msg}, #tcp{state = #state{id = ChannelId}} = TCPState) ->
Payload = dgiot_mqtt:get_payload(Msg),
dgiot_bridge:send_log(ChannelId, "Topic ~p Msg ~p", [dgiot_mqtt:get_topic(Msg), Payload]),
case jsx:is_json(Payload) of
true ->
case binary:split(dgiot_mqtt:get_topic(Msg), <<$/>>, [global, trim]) of
[<<"thing">>, _ProductId, _DevAddr] ->
[#{<<"thingdata">> := ThingData} | _] = jsx:decode(dgiot_mqtt:get_payload(Msg), [{labels, binary}, return_maps]),
Payload = dgiot_meter:to_frame(ThingData),
dgiot_bridge:send_log(ChannelId, "from_task: ~ts: ~ts ", [_Topic, unicode:characters_to_list(dgiot_mqtt:get_payload(Msg))]),
?LOG(info, "task->dev: Payload ~p", [dgiot_utils:binary_to_hex(Payload)]),
dgiot_tcp_server:send(TCPState, Payload);
#tcp{state = #state{protocol = Protocol}} = TCPState,
case Protocol of
?DLT376 ->
[#{<<"thingdata">> := ThingData} | _] = jsx:decode(dgiot_mqtt:get_payload(Msg), [{labels, binary}, return_maps]),
Payload1 = dgiot_meter:to_frame(ThingData),
dgiot_bridge:send_log(ChannelId, "from_task: ~ts: ~ts ", [_Topic, unicode:characters_to_list(dgiot_mqtt:get_payload(Msg))]),
% ?LOG(info, "GGM 214 task->dev: Payload ~p", [dgiot_utils:binary_to_hex(Payload1)]),
dgiot_tcp_server:send(TCPState, Payload1);
?DLT645 ->
[#{<<"thingdata">> := ThingData} | _] = jsx:decode(dgiot_mqtt:get_payload(Msg), [{labels, binary}, return_maps]),
Payload1 = dgiot_meter:to_frame(ThingData),
% ?LOG(info, "GGM 216 dgiot_meter_tcp, Payload1 ~p,~p",[ChannelId,Payload1]),
dgiot_bridge:send_log(ChannelId, "from_task: ~ts: ~ts ", [_Topic, unicode:characters_to_list(dgiot_mqtt:get_payload(Msg))]),
dgiot_tcp_server:send(TCPState, Payload1)
end;
[<<"profile">>, _ProductId, _DtuAddr] ->
case Payload of
#{<<"_dgiotprotocol">> := <<"hex">>} ->
@ -115,6 +147,76 @@ handle_info({deliver, _Topic, Msg}, #tcp{state = #state{id = ChannelId, step = r
_ ->
pass
end;
[<<"thingctrl">>, _, _DtuAddr] ->
#tcp{state = #state{protocol = Protocol}} = TCPState,
?LOG(info, "GGM 250 dgiot_meter_tcp, thingctrl ~p",[TCPState]),
case Protocol of
?DLT376 ->
[#{<<"thingdata">> := ThingData} | _] = jsx:decode(dgiot_mqtt:get_payload(Msg), [{labels, binary}, return_maps]),
% ?LOG(info, "GGM 205 dgiot_meter_tcp, handle_info10 ~p",[ThingData]),
case ThingData of
% 远程控制电表拉闸、合闸
#{<<"devaddr">> := DevAddr,<<"ctrlflag">> := CtrlFlag,<<"devpass">> := DevPass,<<"apiname">> := <<"get_meter_ctrl">>} ->
% ?LOG(info, "GGM 206 dgiot_meter_tcp, handle_info9 ~p,~p,~p,~p",[DevAddr,CtrlFlag,DevPass,Protocol]),
ThingData1 = #{
<<"devaddr">> => DevAddr,
<<"ctrlflag">> => CtrlFlag,
<<"devpass">> => DevPass,
<<"protocol">> => Protocol ,
<<"apiname">> => get_meter_ctrl
},
Payload1 = dgiot_meter:to_frame(ThingData1),
% diot_bridge:send_log(ChannelId, "from_task: ~ts: ~ts ", [_Topic, unicode:characters_to_list(dgiot_mqtt:get_payload(Msg))]),
dgiot_tcp_server:send(TCPState, Payload1);
% 获取上次拉闸、合闸的时间（一次发送两条查询指令）
#{<<"devaddr">> := DevAddr,<<"ctrlflag">> := CtrlFlag,<<"apiname">> := <<"get_meter_ctrl_status">>} ->
?LOG(info, "GGM 235 dgiot_meter_tcp, handle_info9 ~p,~p",[DevAddr,Protocol]),
% 上次合闸时间
ThingData1 = #{
<<"devaddr">> => DevAddr,
<<"ctrlflag">> => CtrlFlag,
<<"protocol">> => Protocol,
<<"apiname">> => get_meter_ctrl_status
},
Payload1 = dgiot_meter:to_frame(ThingData1),
dgiot_tcp_server:send(TCPState, Payload1);
_->
?LOG(info, "GGM 236 dgiot_meter_tcp, handle_info9 ~p",[ThingData]),
pass
end;
?DLT645 ->
[#{<<"thingdata">> := ThingData} | _] = jsx:decode(dgiot_mqtt:get_payload(Msg), [{labels, binary}, return_maps]),
?LOG(info, "GGM 211 dgiot_meter_tcp, handle_info10 ~p",[ThingData]),
case ThingData of
% 远程控制电表拉闸、合闸
#{<<"devaddr">> := DevAddr,<<"ctrlflag">> := CtrlFlag,<<"devpass">> := DevPass,<<"apiname">> := <<"get_meter_ctrl">>} ->
% ?LOG(info, "GGM 212 dgiot_meter_tcp, handle_info9 ~p,~p,~p,~p",[DevAddr,CtrlFlag,DevPass,Protocol]),
ThingData1 = #{
<<"devaddr">> => DevAddr,
<<"ctrlflag">> => CtrlFlag,
<<"devpass">> => DevPass,
<<"protocol">> => Protocol,
<<"apiname">> => get_meter_ctrl
},
Payload1 = dgiot_meter:to_frame(ThingData1),
?LOG(info, "GGM 213 dgiot_meter_tcp, Payload1 ~p,~p",[ChannelId,dgiot_utils:to_hex(Payload1)]),
dgiot_tcp_server:send(TCPState, Payload1);
% 获取上次拉闸、合闸的时间（一次发送两条查询指令）
#{<<"devaddr">> := DevAddr,<<"ctrlflag">> := CtrlFlag,<<"apiname">> := <<"get_meter_ctrl_status">>} ->
?LOG(info, "GGM 236 dgiot_meter_tcp, handle_info9 ~p,~p",[DevAddr,Protocol]),
ThingData1 = #{
<<"devaddr">> => DevAddr,
<<"ctrlflag">> => CtrlFlag,
<<"protocol">> => Protocol,
<<"apiname">> => get_meter_ctrl_status
},
Payload1 = dgiot_meter:to_frame(ThingData1),
?LOG(info, "GGM 237 dgiot_meter_tcp, Payload1 ~p,~p",[ChannelId,dgiot_utils:to_hex(Payload1)]),
dgiot_tcp_server:send(TCPState, Payload1);
_->
pass
end
end;
_ ->
pass
end;
@ -122,27 +224,250 @@ handle_info({deliver, _Topic, Msg}, #tcp{state = #state{id = ChannelId, step = r
end,
{noreply, TCPState};
%% 接收抄表任务的ACK报文
handle_info({tcp, Buff}, #tcp{state = #state{id = ChannelId, step = read_meter}} = TCPState) ->
dgiot_bridge:send_log(ChannelId, "from_dev: ~p ", [dgiot_utils:binary_to_hex(Buff)]),
?LOG(info, "Buff ~p", [dgiot_utils:binary_to_hex(Buff)]),
{Rest, Frames} = dgiot_meter:parse_frame(dlt645, Buff, []),
case Frames of
[#{<<"addr">> := Addr, <<"value">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/post">>,
DeviceId = dgiot_parse:get_deviceid(ProductId, DevAddr),
dgiot_mqtt:publish(DeviceId, Topic, jsx:encode(Value));
handle_info({tcp, Buff}, #tcp{state = #state{id = ChannelId,protocol = Protocol,step = Step}} = TCPState) ->
?LOG(info, "GGM 333 Buff========ALL REST========= ~p,~p,~p~n~n~n", [ChannelId,Step,dgiot_utils:binary_to_hex(Buff)]),
case Protocol of
?DLT376 ->
dgiot_bridge:send_log(ChannelId, "from_dev: ~p ", [dgiot_utils:binary_to_hex(Buff)]),
?LOG(info, "GGM 140 Buff========DLT376========= ~p", [dgiot_utils:binary_to_hex(Buff)]),
{Rest, Frames} = dgiot_meter:parse_frame(?DLT376, Buff, []),
?LOG(info, "GGM 246 Frames ~p", [Frames]),
case Frames of
% 返回抄表数据
[#{<<"di">> :=<<16#01,16#01,16#01,16#10>>,<<"addr">> := Addr, <<"value">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/post">>, % 发送给mqtt进行数据存储
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
_ -> pass
end;
% 返回读取上次合闸时间
[#{<<"di">> :=<<16#1E,16#00,16#01,16#01>>,<<"addr">> := Addr, <<"value">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
?LOG(info, "GGM 246 jsx:encode(Value) ~p", [jsx:encode(Value)]),
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
_ -> pass
end;
% 返回读取上次拉闸时间
[#{<<"di">> :=<<16#1D,16#00,16#01,16#01>>, <<"addr">> := Addr, <<"value">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
?LOG(info, "GGM 245 jsx:encode(Value) ~p", [jsx:encode(Value)]),
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
_ -> pass
end;
% 拉闸，合闸成功
[#{<<"di">> :=<<16#FE,16#FE,16#FE,16#FE>>, <<"addr">> := Addr, <<"value">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
_ -> pass
end;
% 拉闸，合闸失败
[#{<<"di">> :=<<16#FE,16#FE,16#FE,16#FD>>, <<"addr">> := Addr, <<"value">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
_ -> pass
end;
_ -> pass
end;
_ -> pass
end,
{noreply, TCPState#tcp{buff = Rest}};
end,
{noreply, TCPState#tcp{buff = Rest}};
?DLT645 ->
dgiot_bridge:send_log(ChannelId, "from_dev: ~p ", [dgiot_utils:binary_to_hex(Buff)]),
?LOG(info, "GGM 222 Meter Rev Buff========DLT645========= ~p~n~n~n", [dgiot_utils:binary_to_hex(Buff)]),
{Rest, Frames} = dgiot_meter:parse_frame(?DLT645, Buff, []),
% ?LOG(info, "GGM 245 Frames ~p", [Frames]),
case Frames of
% 抄表数据返回
[#{<<"command">> := 16#91,<<"di">>:=<<16#00,16#00,16#00,16#00>>,<<"addr">> := Addr, <<"value">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/post">>,
% ?LOG(info, "GGM 247 jsx:encode(Value) ~p", [jsx:encode(Value)]),
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
_ -> pass
end;
% 查询上一次合闸时间返回
[#{<<"command">>:=16#91,<<"di">> := <<16#1E,16#00,16#01,16#01>>,<<"addr">> := Addr,<<"data">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
Di = <<16#1E,16#00,16#01,16#01>>,
DValue = #{dgiot_utils:to_hex(Di)=>dlt645_decoder:binary_to_dtime_dlt645_bcd(Value)},
?LOG(info, "GGM 248 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
_ -> pass
end;
% 查询上一次拉闸时间返回
[#{<<"command">>:=16#91,<<"di">> := <<16#1D,16#00,16#01,16#01>>,<<"addr">> := Addr,<<"data">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
Di = <<16#1D,16#00,16#01,16#01>>,
DValue = #{dgiot_utils:to_hex(Di)=>dlt645_decoder:binary_to_dtime_dlt645_bcd(Value)},
?LOG(info, "GGM 249 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
_ -> pass
end;
% 拉闸，合闸成功
[#{<<"command">>:=16#9C, <<"addr">> := Addr} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
Di = <<16#FE,16#FE,16#FE,16#FE>>,
DValue = #{dgiot_utils:to_hex(Di)=>0 },
?LOG(info, "GGM 250 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
_ -> pass
end;
% 拉闸，合闸失败
[#{<<"command">>:=16#DC, <<"addr">> := Addr, <<"data">> := Value} | _] ->
case dgiot_data:get({meter, ChannelId}) of
{ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
Di = <<16#FE,16#FE,16#FE,16#FD>>,
DValue = #{dgiot_utils:to_hex(Di)=>dgiot_utils:to_hex(Value)},
?LOG(info, "GGM 251 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
_ -> pass
end;
_ -> pass
end,
{noreply, TCPState#tcp{buff = Rest}};
_ ->
{noreply, TCPState#tcp{buff = <<0,0,0,0,0,0,0,0>>}}
end;
% %% 接收抄表任务的ACK报文DLT376
% handle_info({tcp, Buff}, #tcp{state = #state{id = ChannelId, protocol = ?DLT376,step = read_meter}} = TCPState) ->
% dgiot_bridge:send_log(ChannelId, "from_dev: ~p ", [dgiot_utils:binary_to_hex(Buff)]),
% ?LOG(info, "GGM 140 Buff========DLT376========= ~p", [dgiot_utils:binary_to_hex(Buff)]),
% {Rest, Frames} = dgiot_meter:parse_frame(?DLT376, Buff, []),
% ?LOG(info, "GGM 246 Frames ~p", [Frames]),
% case Frames of
% % 返回抄表数据
% [#{<<"di">> :=<<16#01,16#01,16#01,16#10>>,<<"addr">> := Addr, <<"value">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/post">>, % 发送给mqtt进行数据存储
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
% _ -> pass
% end;
% % 返回读取上次合闸时间
% [#{<<"di">> :=<<16#1E,16#00,16#01,16#01>>,<<"addr">> := Addr, <<"value">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
% ?LOG(info, "GGM 246 jsx:encode(Value) ~p", [jsx:encode(Value)]),
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
% _ -> pass
% end;
% % 返回读取上次拉闸时间
% [#{<<"di">> :=<<16#1D,16#00,16#01,16#01>>, <<"addr">> := Addr, <<"value">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
% ?LOG(info, "GGM 245 jsx:encode(Value) ~p", [jsx:encode(Value)]),
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
% _ -> pass
% end;
% % 拉闸，合闸成功
% [#{<<"di">> :=<<16#FE,16#FE,16#FE,16#FE>>, <<"addr">> := Addr, <<"value">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
% _ -> pass
% end;
% % 拉闸，合闸失败
% [#{<<"di">> :=<<16#FE,16#FE,16#FE,16#FD>>, <<"addr">> := Addr, <<"value">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", Addr/binary, "/status">>,
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
% _ -> pass
% end;
% _ -> pass
% end,
% {noreply, TCPState#tcp{buff = Rest}};
% %% 接收抄表任务的ACK报文DLT645
% handle_info({tcp, Buff}, #tcp{state = #state{id = ChannelId, protocol = ?DLT645,step = read_meter}} = TCPState) ->
% dgiot_bridge:send_log(ChannelId, "from_dev: ~p ", [dgiot_utils:binary_to_hex(Buff)]),
% ?LOG(info, "GGM 222 Meter Rev Buff========DLT645========= ~p~n~n~n", [dgiot_utils:binary_to_hex(Buff)]),
% {Rest, Frames} = dgiot_meter:parse_frame(?DLT645, Buff, []),
% % ?LOG(info, "GGM 245 Frames ~p", [Frames]),
% case Frames of
% % 抄表数据返回
% [#{<<"command">> := 16#91,<<"di">>:=<<16#00,16#00,16#00,16#00>>,<<"addr">> := Addr, <<"value">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/post">>,
% % ?LOG(info, "GGM 247 jsx:encode(Value) ~p", [jsx:encode(Value)]),
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(Value));
% _ -> pass
% end;
% % 查询上一次合闸时间返回
% [#{<<"command">>:=16#91,<<"di">> := <<16#1E,16#00,16#01,16#01>>,<<"addr">> := Addr,<<"data">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
% Di = <<16#1E,16#00,16#01,16#01>>,
% DValue = #{dgiot_utils:to_hex(Di)=>dlt645_decoder:binary_to_dtime_dlt645_bcd(Value)},
% ?LOG(info, "GGM 248 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
% _ -> pass
% end;
% % 查询上一次拉闸时间返回
% [#{<<"command">>:=16#91,<<"di">> := <<16#1D,16#00,16#01,16#01>>,<<"addr">> := Addr,<<"data">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
% Di = <<16#1D,16#00,16#01,16#01>>,
% DValue = #{dgiot_utils:to_hex(Di)=>dlt645_decoder:binary_to_dtime_dlt645_bcd(Value)},
% ?LOG(info, "GGM 249 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
% _ -> pass
% end;
% % 拉闸，合闸成功
% [#{<<"command">>:=16#9C, <<"addr">> := Addr} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
% Di = <<16#FE,16#FE,16#FE,16#FE>>,
% DValue = #{dgiot_utils:to_hex(Di)=>0 },
% ?LOG(info, "GGM 249 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
% _ -> pass
% end;
% % 拉闸，合闸失败
% [#{<<"command">>:=16#DC, <<"addr">> := Addr, <<"data">> := Value} | _] ->
% case dgiot_data:get({meter, ChannelId}) of
% {ProductId, _ACL, _Properties} -> DevAddr = dgiot_utils:binary_to_hex(Addr),
% Topic = <<"thing/", ProductId/binary, "/", DevAddr/binary, "/status">>,
% Di = <<16#FE,16#FE,16#FE,16#FD>>,
% DValue = #{dgiot_utils:to_hex(Di)=>dgiot_utils:to_hex(Value)},
% ?LOG(info, "GGM 250 jsx:encode(Value) ~p", [jsx:encode(DValue)]),
% dgiot_mqtt:publish(DevAddr, Topic, jsx:encode(DValue));
% _ -> pass
% end;
% _ -> pass
% end,
% {noreply, TCPState#tcp{buff = Rest}};
%% 异常报文丢弃
%% {stop, TCPState} | {stop, Reason} | {ok, TCPState} | ok | stop
handle_info(_Info, TCPState) ->
?LOG(info, "GGM 999 Meter Rev Buff========ERROR========= dgiot_meter_tcp, handle_info9, ~p,~p~n~n~n ",[_Info,TCPState]),
{noreply, TCPState}.
handle_call(_Msg, _From, TCPState) ->

262
apps/dgiot_meter/src/proctol/dlt376_decoder.erl Normal file
View File

@ -0,0 +1,262 @@
%%--------------------------------------------------------------------
%% 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.
%%--------------------------------------------------------------------
-module(dlt376_decoder).
-author("gugm").
-include("dgiot_meter.hrl").
-include_lib("dgiot/include/logger.hrl").
-protocol([?DLT376]).
%% API
-export([parse_frame/2, to_frame/1, parse_value/2]).
parse_frame(Buff, Opts) ->
parse_frame(Buff, [], Opts).
parse_frame(<<>>, Acc, _Opts) ->
{<<>>, Acc};
% 对于小于9的消息，独立decode
parse_frame(<<Rest/binary>> = Bin, Acc, _Opts) when byte_size(Rest) == 15 ->
NewFrame = #{
<<"msgtype">> => ?DLT645
},
Acc1 = Acc ++ [NewFrame],
{Bin, Acc1};
%% DLT376协议
%% 68 32 00 32 00 68 C9 14 03 32 63 00 02 73 00 00 01 00 EB 16
parse_frame(<<16#68, _:16, L2_low:6, _:2,L2_high:8, 16#68, C:8, A1:2/bytes,A2:2/bytes,A3:1/bytes, AFN:8, SEQ:8,Rest/binary>> = Bin, Acc, Opts) ->
Len = L2_high * 255 + L2_low,
DLen = Len -8,
case byte_size(Rest) -2 >= DLen of
true ->
case Rest of
<<UserZone:DLen/bytes, Crc:8, 16#16, Rest1/binary>> ->
CheckBuf = <<C:8, A1:2/bytes,A2:2/bytes,A3:1/bytes, AFN:8, SEQ:8, UserZone/binary>>,
CheckCrc = dgiot_utils:get_parity(CheckBuf),
% BinA = dgiot_utils:to_binary(A),
Acc1 =
case CheckCrc =:= Crc of
true ->
Frame = #{
% <<"addr">> => <<"16#00,16#00",dlt645_proctol:reverse(A1)/binary,dlt645_proctol:reverse(A2)/binary>>,
<<"addr">> => dgiot_utils:binary_to_hex(dlt376_proctol:encode_of_addr(A1,A2)), %dlt376_proctol:concrat_binary(dlt645_proctol:reverse(A1),dlt645_proctol:reverse(A2)),
<<"command">> => C,
<<"afn">> => AFN,
<<"datalen">> => DLen,
<<"msgtype">> => ?DLT376
},
case catch (parse_userzone(UserZone, Frame, Opts)) of
{'EXIT', Reason} ->
?LOG(warning,"UserZone error,UserZone:~p, Reason:~p~n", [dgiot_utils:binary_to_hex(UserZone), Reason]),
Acc;
NewFrame ->
Acc ++ [NewFrame]
end;
false ->
Acc
end,
parse_frame(Rest1, Acc1, Opts);
_ ->
parse_frame(Rest, Acc, Opts)
end;
false ->
{Bin, Acc}
end;
parse_frame(<<_:8, Rest/binary>>, Acc, Opts) when byte_size(Rest) > 50 ->
parse_frame(Rest, Acc, Opts);
parse_frame(<<Rest/binary>>, Acc, _Opts) ->
{Rest, Acc}.
parse_userzone(UserZone, #{<<"msgtype">> := ?DLT376} = Frame, _Opts) ->
check_Command(Frame#{<<"data">> => UserZone}).
%% 组装成封包
to_frame(#{
% <<"msgtype">> := ?DLT376,
<<"command">> := C,
<<"addr">> := Addr,
<<"afn">> := AFN
} = Msg) ->
{ok, UserZone} = get_userzone(Msg),
Len = (byte_size(UserZone) + 8) * 4 + 2,
Crc = dgiot_utils:get_parity(<<C:8, Addr:5/bytes, AFN:8, 16#71, UserZone/binary>>),
<<
16#68,
Len:8,
16#00,
Len:8,
16#00,
16#68,
C:8,
Addr:5/bytes,
AFN:8,
16#71,
UserZone/binary,
Crc:8,
16#16
>>.
% DLT376 链路检测，心跳数据
check_Command(State = #{<<"command">> := 16#C9, <<"afn">> := 16#02}) ->
State;
% DLT376 抄表返回的数据
check_Command(State = #{<<"command">> := 16#88, <<"afn">> := 16#0C}) ->
Data = maps:get(<<"data">>, State, <<>>),
case Data of
<<Di:4/bytes,DTime:5/bytes,DNum:1/bytes,DValue:5/bytes,_/bytes>> ->
% {Value, Diff, TopicDI} = parse_value(dlt645_proctol:reverse(Di), Bin),
State1 = #{
<<"di">> => Di,
<<"time">> =>dgiot_utils:to_hex(DTime),
<<"valuenum">> => DNum,
<<"value">> => #{dgiot_utils:to_hex(Di)=>binary_to_value_dlt376_bcd(DValue) },
<<"addr">> => maps:get(<<"addr">>, State, <<>>)
},
State1;
_ ->
State
end;
% DLT376 穿透转发返回
check_Command(State = #{<<"command">> := 16#88, <<"afn">> := 16#10}) ->
Data = maps:get(<<"data">>, State, <<>>),
case Data of
<<_:4/bytes,_:1/bytes,DLen2:8,DLen1:8,Rest/bytes>> ->
DLen = DLen1 * 255 + DLen2,
case Rest of
<<DValue:DLen/bytes,_/bytes>> ->
{_, Frames} = dlt645_decoder:parse_frame(DValue, []),
?LOG(warning,"GGM 160 check_Command:~p", [Frames]),
case Frames of
% 拉闸、合闸返回成功
[#{<<"command">>:=16#9C} | _] ->
Di = <<16#FE,16#FE,16#FE,16#FE>>,
State1 = #{
<<"di">> => Di,%不做处理
<<"value">> => #{dgiot_utils:to_hex(Di)=>0 },
<<"addr">> => maps:get(<<"addr">>, State, <<>>)
},
State1;
% 拉闸、合闸返回失败
[#{<<"command">>:=16#DC,<<"data">> := VData} | _] ->
Di = <<16#FE,16#FE,16#FE,16#FD>>,
State1 = #{
<<"di">> => Di,%不做处理
<<"value">> => #{dgiot_utils:to_hex(Di)=>dgiot_utils:to_hex(VData) },
<<"addr">> => maps:get(<<"addr">>, State, <<>>)
},
State1;
% 查询上一次合闸时间返回
[#{<<"command">>:=16#91,<<"di">> := <<16#1E,16#00,16#01,16#01>>,<<"data">> := VData} | _] ->
Di = <<16#1E,16#00,16#01,16#01>>,
State1 = #{
<<"di">> => Di,
<<"value">> => #{dgiot_utils:to_hex(Di)=>dlt645_decoder:binary_to_dtime_dlt645_bcd(VData) },
<<"addr">> => maps:get(<<"addr">>, State, <<>>)
},
State1;
% 查询上一次拉闸时间返回
[#{<<"command">>:=16#91,<<"di">> := <<16#1D,16#00,16#01,16#01>>,<<"data">> := VData} | _] ->
Di = <<16#1D,16#00,16#01,16#01>>,
State1 = #{
<<"di">> => Di,
<<"value">> => #{dgiot_utils:to_hex(Di)=>dlt645_decoder:binary_to_dtime_dlt645_bcd(VData) },
<<"addr">> => maps:get(<<"addr">>, State, <<>>)
},
State1;
_ ->
pass
end;
_->
pass
end;
_ ->
State
end;
check_Command(State) ->
State.
% DLT376协议中把二进制转化成float
binary_to_value_dlt376_bcd(BinValue) ->
RValue =
case BinValue of
<<Vf3:4,Vf4:4,Vf1:4,Vf2:4,V2:4,V1:4,V4:4,V3:4,V6:4,V5:4,_/binary>> ->
Value = V6 * 100000 + V5 * 10000 + V4 * 1000 + V3 * 100 + V2 * 10 + V1 + Vf1 * 0.1 + Vf2 * 0.01 + Vf3 * 0.001 + Vf4 * 0.0001,
Value;
_ ->
0.0
end,
RValue.
get_userzone(Msg) ->
Di = maps:get(<<"di">>, Msg, <<>>),
Data = maps:get(<<"data">>, Msg, <<>>),
Di2 = dgiot_utils:hex_to_binary(Di),
Data2 = dgiot_utils:hex_to_binary(Data),
UserZone = <<Di2/binary,Data2/binary>>,
{ok, UserZone}.
parse_value(Di, Data) ->
{DI, Diff, SendDi} =
case Di of
<<16#05, 16#06, Di3:8, Di4:8>> when (Di3 >= 1 andalso Di3 =< 8) andalso (Di4 >= 1 andalso Di4 =< 63) ->
{<<16#05, 16#06, Di3:8, 1>>, Di4 - 1, dgiot_utils:binary_to_hex(<<1, Di3:8, 6, 5>>)};
<<16#00, D2:8, 16#FF, Di4:8>> when D2 >= 1 andalso D2 =< 10 andalso (Di4 >= 1 andalso Di4 =< 12) ->
{<<16#00, D2:8, 16#FF, 1>>, Di4 - 1, dgiot_utils:binary_to_hex(<<1, 16#FF, D2:8, 0>>)};
_ ->
{Di, 0, dgiot_utils:binary_to_hex(dlt645_proctol:reverse(Di))}
end,
case dlt645_proctol:parse_data_to_json(DI, Data) of
{Key, Value} ->
ValueMap =
case jsx:is_json(Value) of
false ->
#{Key => Value};
true ->
[{K1, _V1} | _] = Value0 = jsx:decode(Value),
case size(K1) == 8 of
true ->
maps:from_list(Value0);
false ->
maps:from_list(lists:map(fun({K2, V2}) ->
<<Di1:6/binary, _:2/binary, Di2/binary>> = K2,
{<<Di1:6/binary, Di2/binary>>, V2}
end, Value0))
end
end,
{ValueMap, Diff, SendDi};
_ -> {#{}, Diff, SendDi}
end.
% test() ->
% B1 = <<12, 16#68, 16#01, 16#00, 16#00, 16#00, 16#00, 16#00, 16#68, 16#91, 16#08, 16#33, 16#33, 16#3D, 16#33, 16#33, 16#33, 16#33, 16#33, 16#0C, 16#16,
% 1, 3, 36,
% 16#68, 16#18, 16#00, 16#18, 16#00, 16#68, 16#88, 16#00, 16#31, 16#07, 16#02, 16#00, 16#00, 16#01, 16#0c, 16#64, 16#00, 16#00, 16#00, 16#00, 16#01, 16#01, 16#58, 16#23, 16#10, 16#03, 16#16, 16#93, 16#99, 16#02, 16#07, 16#16,
% 16#68, 16#90, 16#F0, 16#55, 16#00, 16#87>>,
% {Rest, Frames} = parse_frame(B1, [], #{<<"vcaddr">> => <<"003107020000">>}),
% io:format("Rest:~p~n", [Frames]),
% B2 = <<16#00, 16#68, 16#12, 16#09, 16#00, 16#40, 16#01, 16#02, 16#00, 16#07, 16#00, 16#18, 16#11, 16#18, 16#D2, 16#16>>,
% {Rest2, Frames2} = parse_frame(<<Rest/binary, B2/binary>>, [], #{vcaddr => <<"00310702">>}),
% io:format("Rest:~p, Frames:~p~n", [Rest2, Frames2]).

61
apps/dgiot_meter/src/proctol/dlt376_proctol.erl Normal file
View File

@ -0,0 +1,61 @@
%%--------------------------------------------------------------------
%% 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.
%%--------------------------------------------------------------------
%% @doc dlt645 Protocol Processor.
-module(dlt376_proctol).
-author("gugm").
-include_lib("dgiot_meter.hrl").
-include_lib("dgiot/include/logger.hrl").
-export([
binary_to_hex/1,
concrat_binary/2,
encode_of_addr/2,
decode_of_addr/1]).
% -record(di_data_A6, {di1, di2, di3, di4,di5}).
binary_to_hex(Id) ->
<< <<Y>> || <<X:4>> <= Id, Y <- integer_to_list(X,16)>>.
% binary 相连接
concrat_binary( Acc,<<>>) -> Acc;
concrat_binary(Acc,<<H:1/binary,Rest/binary >> ) ->
concrat_binary(<<Acc/binary , H/binary>>,Rest).
split_head_bytes(<<Head:2/binary,Rest/binary>>) ->
{Head,Rest}.
% 把地址转化成binary
encode_of_addr(A1,A2) ->
AA = concrat_binary(dlt645_proctol:reverse(A1),dlt645_proctol:reverse(A2)),
AA1 = concrat_binary(<<16#00,16#00>>,AA),
AA1.
%把binary转化成地址
decode_of_addr(A) ->
{_,Rest} = split_head_bytes(A),
{A1,A2} = split_head_bytes(Rest),
AA = concrat_binary(dlt645_proctol:reverse(A1),dlt645_proctol:reverse(A2)),
AA1 = concrat_binary(AA,<<16#14>>),
AA1.

29
apps/dgiot_meter/src/proctol/dlt645_decoder.erl
View File

@ -20,7 +20,7 @@
-protocol([?DLT645]).
%% API
-export([parse_frame/2, to_frame/1, test/0, parse_value/2]).
-export([parse_frame/2, to_frame/1, test/0, parse_value/2,binary_to_dtime_dlt645_bcd/1]).
parse_frame(Buff, Opts) ->
parse_frame(Buff, [], Opts).
@ -81,7 +81,7 @@ parse_userzone(UserZone, #{<<"msgtype">> := ?DLT645} = Frame, _Opts) ->
%% 组装成封包
to_frame(#{
<<"msgtype">> := ?DLT645,
% <<"msgtype">> := ?DLT645,
<<"command">> := C,
<<"addr">> := Addr
} = Msg) ->
@ -152,11 +152,13 @@ check_Command(State = #{<<"command">> := 16#C1, <<"data">> := <<Data/binary>>})
% -define(DLT645_MS_FORCE_EVENT_NAME, <<"1C">>).
% -define(DLT645_SM_FORCE_EVENT_NORM_NAME, <<"9C">>).
% -define(DLT645_SM_FORCE_EVENT_ERRO_NAME, <<"DC">>).
% 远程开闸、拉闸 返回正常
check_Command(State = #{<<"command">> := 16#9C}) ->
State#{
};
% 远程开闸、拉闸 返回正常
check_Command(State = #{<<"command">> := 16#DC}) ->
State#{
@ -207,6 +209,29 @@ parse_value(Di, Data) ->
_ -> {#{}, Diff, SendDi}
end.
%返回数据转化成时间（时间戳（单位：秒））如1554282344
binary_to_dtime_dlt645_bcd(BinValue) ->
RValue =
case BinValue of
<<S1:4,S2:4,MT1:4,MT2:4,H1:4,H2:4,D1:4,D2:4,M1:4,M2:4,Y1:4,Y2:4,_/binary>> ->
Year = 2 * 1000 + Y1 * 10 + Y2 ,
Month = M1 * 10 + M2,
Day = D1 * 10 + D2,
Hour = H1 * 10 + H2,
Minite = MT1 * 10 + MT2,
Second = S1 * 10 + S2,
case Year of
2000 ->
0;
_ ->
Value = calendar:datetime_to_gregorian_seconds({{Year,Month,Day},{Hour,Minite,Second}})-719528*24*3600-8*3600,
Value
end;
_ ->
0
end,
RValue.
test() ->
B1 = <<12, 16#68, 16#01, 16#00, 16#00, 16#00, 16#00, 16#00, 16#68, 16#91, 16#08, 16#33, 16#33, 16#3D, 16#33, 16#33, 16#33, 16#33, 16#33, 16#0C, 16#16,
1, 3, 36,