---- 北向消息接口没变,但是南向和设备通讯,不是通用构架,所以自己实现了MQTT订阅,和南向数据发送 -------------------------------------------------------------------------------------------------------- ----------------------------------- 直接操作脚本内部逻辑 ------------------------------------------------- ---- |外部 Broker| --(可选)--> [cloud_mqtt APP][UI WEB] ----> |内部 Broker| ----> [protocalpaser 全脚本] ---- |外部 Broker| <--(可选)-- [cloud_mqtt APP][UI WEB] <---- |内部 Broker| <---- [protocalpaser 全脚本] ----------------------------------- 直接操作设备,但通过脚本解析逻辑 ------------------------------------------------- ---- |外部 Broker| --(可选)--> [cloud_mqtt APP][UI WEB] ----> |内部 Broker| ----> [protocalpaser 全脚本] ---- CAN设备 <---- [485 cache] <---- |内部 Broker| <---- [protocalpaser 全脚本] ---- CAN设备 ----> [485 cache] ----> |内部 Broker| ----> [protocalpaser 全脚本] ---- |外部 Broker| <--(可选)-- [cloud_mqtt APP][UI WEB] <---- |内部 Broker| <---- [protocalpaser 全脚本] require "log" require "sys" require "utils" require "patch" require "pack" siganl = require 'posix.signal' syswait = require 'posix.sys.wait' unistd = require 'posix.unistd' mqtt=require "mosquitto" base64=require("base64") dyutils=require("dyutils") lpack=require("lua_pack") require "internal_api" require"lsqlite3" RACK_PATH = "/app/" PROCJET = "PANA" WHITELIST_FILE = RACK_PATH .. "whitelist.json" -- DATA_TYPE_PROPERTY = 0, -- DATA_TYPE_EVENT, -- DATA_TYPE_SERVICE, LOG_LEVEL = log.LOGLEVEL_TRACE local last_report = nil local gw_sn=nil local nodes_status={} local service_info = {} local message_cache = {} local period_list = {} local gw_port = "TCP50001" local white_dtu_list = {} local last_cmd = {time=0,func=0,addr=0,regaddr=0} ------------------------------------------------------------------ ---------------------------- 业务触发配置表描述 ------------------------- ------------------------------------------------------------------ ----! identifier 服务标识,通过这个标识可以合并不同的采样包 ----! read_write 描述操作的读写 ----! class 采集数据类型:alarm/info,用于标记数据记录到不同的cache ----! regs.reginfo 寄存器名称(或名称的数组),寄存器地址,寄存器格式化符号,数值缩放比例 ----! regs.batch 批量采样个数,批量采样自累加值,格式化输出:字符串/数组 ----! regs.map 数值缩放原始值最小,原始值最大,输出值最小,输出值最大 ----! regs.level 报警等级(或者等级的数组) ------------------------------------------------------------------ global_point_table = { { sample_interval = 30, report_interval = 0, identifier = "SystemInformationRequest", read_write = "R", class = "realdata", regs = { { reginfo = {"Current",0,">s",0.01}, }, { reginfo = {"VoltageOfPack",1,">S",0.01}, }, { reginfo = {"SOC",2,">S"}, }, { reginfo = {"SOH",3,">S"}, }, { reginfo = {"ReMainCapacity",4,">S",0.01}, }, { reginfo = {"DesignCapacity",5,">S",0.01}, }, { reginfo = {"FullCapacity",6,">S",0.01}, }, { reginfo = {"BatteryCycle",7,">S"}, }, { reginfo = {"reserve1",8,">S"}, }, { reginfo = { { "batteryCellCellOV_W","batteryCellCellLV_W", "batteryPackOV_W","batteryPackLV_W", "chargingOC_W","dischargeOC_W", nil,nil, "chargeHighTemp_W","dischargeHighTemp_W", "chargeLowTemp_W","dischargELowTemp_W", "envHighTemp_W","envLowTemp_W", "MOSFETHighTemp_W","SOCLow_W" },9,">S"} }, { reginfo = { { "batteryCellOV_P","batteryCellLV_P", "batteryPackOV_P","batteryPackLV_P", "chargingOC_P","dischargeOC_P", "shortCircuit_P","chargeHighTemp_P", "chargeHighTemp_P","dischargeHighTemp_P", "chargeLowTemp_P","dischargELowTemp_P", "MOSFETHighTemp_P","envHighTemp_P", "envLowTemp_P",nil },0xA,">S"} }, { reginfo = { { nil,nil,nil,nil,nil,nil,nil,nil, "charging","discharging", "chargeMOS","dischargeMOS", "currentLimit", nil,nil, "heaterRunning" },0xB,">S"} }, } }, { sample_interval = 120, report_interval = 0, identifier = "PackCellVol", read_write = "R", regs = { { reginfo = {"Values",15,">S"}, batch = {15,1} } } }, { sample_interval = 120, report_interval = 0, identifier = "PackCellTemp", read_write = "R", regs = { { reginfo = {"Values",31,">s",0.1}, batch = {4,1} }, { reginfo = {"MOS",35,">s",0.1}, }, { reginfo = {"Env",36,">s",0.1}, } } }, { sample_interval = 0, report_interval = 0, identifier = "FetchHistory", read_write = "R", regs = { { reginfo = {"Date",10000,">I"}, }, { reginfo = {"Values",10002,">S"}, batch = {100,1} } } }, { sample_interval = 0, report_interval = 0, identifier = "BMS_Manufacturer", read_write = "R", regs = { { reginfo = {"Model_SN",0x00A0,"A20"}, } } }, { sample_interval = 0, report_interval = 0, identifier = "PACK_SN", read_write = "R", regs = { { reginfo = {"PACK_SN",0x00AA,"A20"}, } } } } local function add_period_service(node_sn) for _,service in ipairs(global_point_table) do if service.read_write == "R" then if not service.sample_interval or service.sample_interval == 0 then break end local period={ identifier = service.identifier, sn = node_sn, intv_sample = service.sample_interval, -- 直接使用上报周期做采样 intv_report = service.report_interval, -- 直接使用上报周期做采样 last_sample = 0, last_report = 0, heartbeat = os.time() } table.insert(period_list,period) end end end local function del_period_service(node_sn) for k,v in ipairs(period_list)do if v and v.sn and v.sn == node_sn then table.remove(period_list,k,1) end end end local function UpdateAddrSN(sn,addr) local update_sn for _,v in pairs(nodes) do if v.sn == sn then v.term_addr = addr for k,vv in ipairs(period_list)do if vv and vv.sn and vv.sn == sn then vv.heartbeat = os.time() break end end for _,vv in pairs(nodes) do if vv.sn ~= sn and vv.term_addr == addr then vv.term_addr = nil end end return end end table.insert(nodes,{sn = sn,term_addr = addr}) add_period_service(sn) end local function FetchAddrBySN(sn) for _,v in pairs(nodes) do if v.sn == sn then return v.term_addr end end end local function FetchSnByAddr(addr) for _,v in pairs(nodes) do if v.term_addr == addr then return v.sn end end end local function DeleteAddrBySN(sn) for i,v in ipairs(nodes) do if v.sn == sn then del_period_service(sn) return table.remove( nodes, i, 1) end end end local function FormatLenght(format) local len if string.upper(format) == "C" then len = 1 elseif string.upper(string.sub(format,2,2)) == "S" then len = 2 elseif string.upper(string.sub(format,2,2)) == "I" then len = 4 elseif string.upper(string.sub(format,1,1)) == "A" then len = tonumber(string.sub(format,2)) else log.warn("port2pp", "format not support",format) return 0 end return len end local function check_symbol(sn) if not sn then return nil end if not nodes_status[sn] then nodes_status[sn] = {} end if not nodes_status[sn].rx_cnt then nodes_status[sn].rx_cnt = 0 end if not nodes_status[sn].tx_cnt then nodes_status[sn].tx_cnt = 0 end if not nodes_status[sn].resp_cnt then nodes_status[sn].resp_cnt = 0 end if not nodes_status[sn].last_rcv then nodes_status[sn].last_rcv = 0 end if not nodes_status[sn].sn then nodes_status[sn].sn = sn end if not nodes_status[sn].loss_ratio then nodes_status[sn].loss_ratio = nil end if not nodes_status[sn].online then nodes_status[sn].online = nil end if not nodes_status[sn].last_online then nodes_status[sn].last_online = nil end return true end function update(sn) if check_symbol(sn) then nodes_status[sn].online = false nodes_status[sn].last_online = nodes_status[sn].online end end function tx_add(sn) if check_symbol(sn) then nodes_status[sn].tx_cnt = nodes_status[sn].tx_cnt + 1 end end function rx_add(sn) if check_symbol(sn) then nodes_status[sn].rx_cnt = nodes_status[sn].rx_cnt + 1 nodes_status[sn].last_rcv = os.time() end end function resp_add(sn) if check_symbol(sn) then nodes_status[sn].resp_cnt = nodes_status[sn].resp_cnt + 1 nodes_status[sn].last_rcv = os.time() end end function deepCopy(object) local lookup_table = {} local function _copy(object) if type(object) ~= "table" then return object elseif lookup_table[object] then return lookup_table[object] end local new_table = {} lookup_table[object] = new_table for key, value in pairs(object) do new_table[_copy(key)] = _copy(value) end return setmetatable(new_table, getmetatable(object)) end return _copy(object) end --! @brief 模块功能：485cache 组包 --! @param addr: 设备地址 --! @param func: 功能码 --! @param regaddr: 寄存器地址 --! @param regsize: 寄存个数 --! @param data: 写操作时：寄存器个数 *2 字节;读操作时：0 --! @return --! @retval raw_data: 返回二进制数据 function ex_pack(addr,func,regaddr,regsize,data) log.info("485cache_api",addr,func,string.format( "0x%04X",regaddr) ,regsize,data and string.toHex(data)) local raw_data=nil if not addr or not func or not regaddr or not regsize then log.warn("ex_pack","修改设备层级") return nil end raw_data = bpack("CC>S>S",addr,func,regaddr,regsize) -- 有数据增加二进制数据 if data and #data > 0 then if func == 0x01 and #data ~= regsize/2 then log.warn("ex_pack",string.format("buff size and regsize are conflict, size:%d,regsize:%d",#data,regsize)) return nil end for i = 1, #data do raw_data = raw_data .. bpack("C",data:byte(i)) end end raw_data = raw_data .. bpack(">S",dyutils.CRC16(raw_data,#raw_data)) last_cmd = {time=os.time(),func=func,addr=addr,regaddr=regaddr} return raw_data end --! @brief 模块功能：485cache 拆包 --! @param raw_data: 输入二进制数据 --! @return --! @retval result: 返回结果 --! @retval level: 设备层级 --! @retval addr: 设备地址 --! @retval func: 功能码 --! @retval regaddr: 寄存器地址 --! @retval regsize: 寄存个数 --! @retval data: 写操作时：寄存器个数 *2 字节;读操作时：0 function ex_unpack(raw_data) --取出入参-- local obj={} local _,addr,func,regaddr,regsize,skip_len if os.difftime(os.time(),last_cmd.time) > 10 then log.warn("ex_unpack","recv timeout") return nil end if last_cmd.func and last_cmd.func == 0x03 then _,addr,func,bytes = bunpack(raw_data,"CCC") regaddr = last_cmd.regaddr regsize = bytes/2 skip_len = 3 elseif last_cmd.func and last_cmd.func == 0x10 and last_cmd.func == func then _,addr,func,regaddr,regsize = bunpack(raw_data,"CC>S>S") skip_len = 6 else log.warn("ex_unpack","modbus last_cmd canot find",last_cmd.func) return end if last_cmd.func ~= func then log.warn("ex_unpack","modbus function code",string.format("send cmd:0x%02X recv cmd:0x%02X",last_cmd.func,func)) return nil end if last_cmd.addr ~= addr then log.warn("ex_unpack","modbus slave addr",string.format("send cmd:0x%02X recv cmd:0x%02X",last_cmd.addr,addr)) return nil end local crc = dyutils.CRC16(raw_data,#raw_data - 2) local _,crc_pack = bunpack(string.sub(raw_data,#raw_data - 1),">S") if (crc ~= crc_pack) then log.warn("ex_unpack",string.format("Data crc check failed cal:0x%04X packet:0x%04X",crc,crc_pack)) return nil end if #raw_data - skip_len <= 2 then -- log.warn("ex_unpack",#raw_data,skip_len) return nil end local data = string.sub(raw_data,skip_len+1,-3) log.info("485cache_resopnse",addr,func,regaddr,regsize,string.toHex(data)) return true,addr,func,regaddr,regsize,data end local function convert2json(regaddr,regsize,raw_data) local service_index -- 遍历寄存器得到服务模型 for i,v in pairs(global_point_table) do for ii,vv in pairs(v.regs) do if vv.reginfo[2] == regaddr then service_index = i end end end if not service_index then log.warn("port2pp", "not found index in point table") return end local obj = {} local levels = {} local identifier local skip_reg for reg = regaddr,regaddr + regsize - 1 do -- 跳转寄存器一旦被赋值,只有到了跳转到指定的寄存器才继续,否则直接退出 if not skip_reg or reg == skip_reg then -- 逐个寄存器查找效率地下 skip_reg = nil local v = global_point_table[service_index] local format = nil local key = nil local scale = nil local level = nil local batch = nil identifier = v.identifier class = v.class -- 找出寄存器名称和长度 local found = false for ii,vv in pairs(v.regs) do if vv.reginfo[2] == reg then format = vv.reginfo[3] key = vv.reginfo[1] scale = vv.reginfo[4] level = vv.level batch = vv.batch found = true break end end if not found then ---- 如果没有找到这个寄存器,说明寄存器数量不对,直接返回空数据 return end if not format or not key then log.warn("port2pp", string.format("register info not found,regaddr:0x%04X",reg or -1)) else -- 计算出字段长度 local len = FormatLenght(format) -- 拆分转为对象 log.info("485cache_raw_data",string.toHex(string.sub(raw_data,1,len)),format,len,string.format("regaddr:0x%04X",reg)) if type(key) == "table" then local _, value= bunpack(raw_data,format) if not value then log.warn("port2pp", "value cannot convert") return end raw_data = string.sub(raw_data,len + 1 ,-1) for pos,sub_key in pairs(key) do if not pos or pos <= 0 then log.warn("port2pp", "bunpack size error",format) return end if sub_key then obj[sub_key] = bit.band(bit.rshift(value,pos - 1),1) end if sub_key and level then levels[sub_key] = level end end elseif type(key) == "string" then if not batch then local _, value= bunpack(raw_data,format) if not value then -- log.warn("port2pp", "value cannot convert in batch process.",string.format("startaddr 0x%04X,currentaddr0x%04X, size %d,setp %d",reg,currreg,size,step)) return end raw_data = string.sub(raw_data,len + 1 ,-1) skip_reg = reg + len/2 if scale then if type(value) ~= "number" then log.warn("port2pp", "value was except as a number type.") return end obj[key] = value * scale else if type(value) == "string" then value = string.match(value,"%w+") end obj[key] = value end else local size = batch[1] or 1 local step = batch[2] or 1 obj[key] = {} for i = 1,size,step do local currreg = reg + (i - 1) -- log.info("port2pp", "=================",string.format("startaddr 0x%04X,currentaddr 0x%04X, size %d,setp %d",reg,currreg,size,step)) local _, value= bunpack(raw_data,format) if not value then log.warn("port2pp", "value cannot convert in batch process.",string.format("startaddr 0x%04X,currentaddr0x%04X, size %d,setp %d",reg,currreg,size,step)) return end raw_data = string.sub(raw_data,len + 1 ,-1) skip_reg = currreg + 1 if scale then table.insert(obj[key] , value * scale) else table.insert(obj[key] , value) end end end end end end end return identifier,obj,class,levels end --! @brief 模块功能：内部服务上报,或者服务请求的应答,用于IOT侧上行数据 --! @param gw_sn: 网关SN --! @param dev_sn: 表示设备SN --! @param port: 设备port --! @param identifier: 服务标识 --! @param param: 数据的具体内容,类型是table --! @param mi: 消息流水号,默认为0 --! @param time: 采集时间,默认是当前时间 function service_response(gw_sn,dev_sn,port,identifier,param,mi,time) local topic = string.format( "/sys/11320465/%s/service",gw_sn) local obj_send={} local json_str obj_send['mi'] = mi or 0 -- obj_send['len'] = #data obj_send['identifier'] = identifier obj_send['sn'] = dev_sn obj_send['time'] = time or os.time() -- obj_send["data_type"] = 2 obj_send["tags"] = param and type(param) == "table" and param or cjson.decode(param) json_str = cjson.encode(obj_send) log.info("service_response",topic,json_str) client:publish(topic, json_str) end -- function nodes_status_report(gw_sn,param) -- local topic = string.format( "/sys/11320465/%s/service",gw_sn) -- local obj_send={} -- local json_str -- obj_send['mi'] = 0 -- obj_send['time'] = os.time() -- obj_send["status"] = param -- json_str = cjson.encode(obj_send) -- log.info("nodes_status",topic,json_str) -- client:publish(topic, json_str) -- end local function port2pp(socket,input) log.info("response:",string.format("socket:%s",socket),string.toHex(input)) local result,addr,func,regaddr,regsize,data = ex_unpack(input) if result then local dev_sn = FetchSnByAddr(socket) if not dev_sn then ---- socket 没绑定设备 设备 log.warn("mqtt.recv",string.format( "not bind sn on socket:%d",obj.socket)) return end if bit.rshift(func,7) ~= 0 then --差错寄存器 log.warn("port2pp", "received a fail response !") return elseif func == 0x10 then --写寄存器响应 log.info("port2pp","write register response.",string.format("regaddr:0x%04X,regsize:%d",regaddr,regsize)) elseif func == 0x03 then --读寄存器响应 log.info("port2pp","read register response.",string.format("regaddr:0x%04X,regsize:%d",regaddr,regsize)) local identifier,obj,class,levels = convert2json(regaddr,regsize,data) if not identifier or not obj then log.warn("port2pp", "object convert failed!") return end -- 设备下没有任何数据chache if not message_cache[dev_sn] then message_cache[dev_sn] = {} local services = {} services = {identifier = identifier,obj = obj,class = class,levels = levels} log.info("port2pp","append device and service to message cache") table.insert( message_cache[dev_sn], services ) else -- 合并相同identifier local found for _,services in ipairs(message_cache[dev_sn]) do if services and services.identifier == identifier then for k,v in pairs(obj) do services.obj[k] = v end log.info("port2pp","update service to message cache",cjson.encode(services.obj),cjson.encode(obj)) found = true end end -- 有cache ,但是identifier不存在应该时新的服务 if not found then local services = {} services = {identifier = identifier,obj = obj,class = class,levels = levels} log.info("port2pp","append service to message cache") table.insert( message_cache[dev_sn], services ) end end if not service_info.mi or service_info.mi == 0 then if period_report_check(dev_sn,identifier) then service_response(gw_sn,dev_sn,gw_port,identifier,obj,0) end else resp_add(dev_sn) service_response(gw_sn,dev_sn,gw_port,identifier,obj,service_info.mi) end rx_add(dev_sn) -- sys.timerStart(service_cache_timeout,300) else log.info("port2pp","function code invalid",func) end end end function read_from_485cache(gw_sn,port,socket,addr,regaddr,regsize) local topic = string.format("GW/218800002001_TCP50001/%d/transparent",socket) -- local payload = ex_pack(addr,0x3,regaddr,0x80-0x40) local payload = ex_pack(addr,0x3,regaddr,regsize) log.info("read:",topic,string.toHex(payload)) client:publish(topic, cjson.encode({socket=socket,payload=base64.encode(payload)})) -- local test_data = string.fromHex("0203143132333435363738393031323334353637383930C019") -- port2pp(test_data) end function write_to_485cache(gw_sn,port,socket,addr,regaddr,regsize,data) local topic = string.format("GW/218800002001_TCP50001/%d/transparent",socket) local payload = ex_pack(addr,0x10,regaddr,regsize,data) log.info("write:",topic,string.toHex(payload)) client:publish(topic, cjson.encode({socket=socket,payload=base64.encode(payload)})) end local function iot2pp(json_str, len) local err_code=0 --取出入参-- log.info("iot2pp","input:"..json_str) local obj=cjson.decode(json_str) local identifier=obj.identifier local server_period=obj.server_period local dev_sn=obj.sn --编码-- if obj.identifier == "null" then log.warn("iot2pp","Need device sn in payload") return 0, "", 0 end -- 不能直接解析,并发往端口,使用脚本的处理的内容 if obj.identifier == "ClusterInfo" then -- replyClusterInfoRequest(gw_sn,dev_sn,gw_port,obj.identifier,param,obj.mi) return else local socket = FetchAddrBySN(dev_sn) tx_add(dev_sn) if not socket then log.warn("iot2pp","Can not found term_addr") return nil end local addr = 1 -- 找服务 for i,v in pairs(global_point_table) do if v.identifier == obj.identifier then if not v.regs then log.warn("iot2pp","regs not found in service") return end local function segment_process(output,rw,identifier,mi) if rw == "W" then if not output.reg_data then log.warn("iot2pp","not find any binary data") return end log.debug("write reg data:",string.toHex(output.reg_data)) write_to_485cache(gw_sn,gw_port,socket,addr,output.regaddr,output.regsize,output.reg_data) elseif rw == "R" then read_from_485cache(gw_sn,gw_port,socket,addr,output.regaddr,output.regsize) end service_info.time = os.time() service_info.identifier = identifier service_info.mi = mi end local last_reg =nil local temp = {} for ii,vv in ipairs(v.regs) do -- 新的地址开头 if not temp.regaddr then temp.regaddr = vv.reginfo[2] temp.regsize = 0 temp.reg_data = '' end -- 判断地址是否连续 if (vv.reginfo[2] - temp.regaddr ~= temp.regsize) then segment_process(temp,v.read_write,obj.identifier,obj.mi) --缓存状态重新赋值 temp.regaddr = vv.reginfo[2] temp.regsize = 0 temp.reg_data = '' end local batch = vv.batch if not vv.batch then -- 写数据才需要从报文中获取字段value if v.read_write == "W" then -- 取出并叠加 local value = obj[vv.reginfo[1]] if not value then log.warn("iot2pp","can not find the key from message,key:",vv.reginfo[1]) return end temp.reg_data = temp.reg_data .. bpack(vv.reginfo[3],value) end local datasize = FormatLenght(vv.reginfo[3])/2 temp.regsize = temp.regsize + datasize else -- 批量下发需要循环 local size = vv.batch[1] or 1 local step = vv.batch[2] or 1 if v.read_write == "W" then local values = obj[vv.reginfo[1]] if not values then log.warn("iot2pp","can not find the key from message,key:",vv.reginfo[1]) return end end for i = 1,size,step do if v.read_write == "W" then -- 取出并叠加 if not values[i] then log.warn("iot2pp","index was invalid from message,key:",vv.reginfo[1],"index:",i) return end temp.reg_data = temp.reg_data .. bpack(vv.reginfo[3],values[i]) end temp.regsize = temp.regsize + 1 end end end -- 根据读写分类调用底层接口 segment_process(temp,v.read_write,obj.identifier,obj.mi) output = {} end end end return -1, nil, 0 end function period_report_check(sn,identifier) for k,v in ipairs(period_list)do if v then if os.difftime(os.time(),v.last_report) >= v.intv_report then v.last_sample = os.time() return true end end end end function white_list(json_str, len) local err_code=0 --取出入参-- local obj=cjson.decode(json_str) --编码-- if obj.nodes then white_dtu_list = {} log.info("nodes config","white dtu list:", "------------------") for i,obj in ipairs(obj.nodes) do table.insert(white_dtu_list,obj.sn) update(obj.sn.."_1") print("dtu sn:", obj.sn) end if #white_dtu_list >= 1 then io.writeFile(WHITELIST_FILE, cjson.encode(white_dtu_list), 'w') end end end function on_external_msg(topic, payload) if string.find(topic,"service/set") then iot2pp(payload,#payload) elseif string.find(topic,"white_list") then ----二进制数据流解析 white_list(payload,#payload) elseif string.find(topic,"transparent") then ----二进制数据流解析 local obj = cjson.decode(payload) if obj and obj.payload and obj.socket then local hex = base64.decode(obj.payload) if not obj.socket or not obj.payload then log.warn("mqtt.recv","frame error") return end if string.find(hex,PROCJET) then ---- 判断心跳 local dtu_sn = string.sub(hex,#PROCJET + 1) local found = false for _,sn_white in ipairs(white_dtu_list) do if sn_white == dtu_sn then found = true break end end if not found then DeleteAddrBySN(dtu_sn.."_1") log.info("dtu",string.format("dtu:%s without whitelist",dtu_sn)) else UpdateAddrSN(dtu_sn.."_1",obj.socket) log.info("dtu",string.format("dtu:%s heartbeat!",dtu_sn)) end else port2pp(obj.socket,hex) end end else log.info("due message:",topic, payload) end end function period_sample_poll() local curr = os.time() for k,v in ipairs(period_list)do if v then if (v.heartbeat == 0 or os.difftime(curr,v.heartbeat) < 120) and os.difftime(curr,v.last_sample) >= v.intv_sample then v.last_sample = curr log.info("lnxall_config.period_sample",v.sn,v.identifier,v.intv_sample) local obj = {} obj.identifier = v.identifier obj.sn = v.sn obj.timestamp = curr obj.mi=0 return obj end end end end local function handler() local pid, status, code = syswait.wait(-1, syswait.WNOHANG) log.error('system exit', pid, status, code) --打印不出来 os.exit(0) end local function load_config() local obj = {} if io.exists(WHITELIST_FILE) then local str = io.readFile(WHITELIST_FILE) if str and #str > 0 then obj = cjson.decode(str or '') if obj and #obj >= 1 then for _,dtu_sn in ipairs(obj) do table.insert(white_dtu_list,dtu_sn) update(dtu_sn.."_1") print("dtu sn:", dtu_sn) end end end end end function main(nodes_cfg,sn) log.info("main_loop", "=============== MOXIGE RUN 1.1 ============") gw_sn = sn if not gw_sn then -- gw_sn = io.getGatewayID() gw_sn = "218800000010" end log.info("main_loop", string.format("Try Run At Shell\nsn='%s'\nnodes='%s'\nmain_loop(nodes,sn)",gw_sn or "NOT SN",nodes_cfg or "")) siganl.signal(siganl.SIGKILL,handler) siganl.signal(siganl.SIGHUP,handler) local subscribe={string.format("/sys/11320465/%s/service/set",gw_sn),string.format("/sys/11320465/%s/white_list",gw_sn),"F/218800002001_TCP50001/+/transparent"} internal_api.mqtt_session("mqtt.lnxall.com",3883,"localuser","dywl@galaxy",subscribe,on_external_msg) -- parser_period_service() load_config() sys.taskInit(function () while true do sys.wait(300) -- sys.waitUntil("JJ_NET_RECV_" .. "DownLinkMsgResp", 2000) --如果等到发送消息的应答,或者超时才继续下一条轮询 local payload = period_sample_poll() if payload then -- sys.publish("JJ_NET_RECV_" .. "DownLinkMsg",payload) local json_str = cjson.encode(payload) iot2pp(json_str, #json_str) end end end) sys.taskInit(function () while true do sys.wait(3000) -- for sn,status in pairs(nodes_status) do -- for k, v in pairs(status) do -- log.debug('Devices.printStatus', 'SN:', sn, k, v) -- end -- end local change = false for sn,status in pairs(nodes_status) do local offtime = 60*3 status.login_time = 0 if offtime == 0 or status.last_rcv and offtime and os.time() < (status.last_rcv + offtime) then status.online = true else status.online = false end -- if status.tx_cnt and status.tx_cnt == 0 then status.loss_ratio = 0 else status.loss_ratio = (status.tx_cnt - status.resp_cnt) * 100 / status.tx_cnt end -- 判断是否离线标记 if status.online ~= status.last_online then status.last_online = status.online change = true end end if not last_report then last_report = 0 end if change or os.difftime(os.time(),last_report) >= 3600 then last_report = os.time() local report = {} report.status={} for sn,status in pairs(nodes_status) do local sta = deepCopy(status) -- 去掉不用的变量 sta.last_online = nil sta.resp_cnt = nil table.insert(report.status,sta) end local str = cjson.encode(report) report = nil if str then log.info("===============", str) local topic = string.format( "/sys/11320465/%s/nodes_status",gw_sn) client:publish(topic, str) end end end end) --启动系统框架 sys.init(0, 0) sys.run() end main(arg[1],arg[2])