cjson=require("cjson") MQTT=require("mqtt_library") base64=require("base64") dyutils=require("dyutils") bit=require("bit") lpack=require("lua_pack") syslog=require("syslog") bcd=require("bcd") sqlite3=require("lsqlite3") bnot = bit.bnot band, bor, bxor = bit.band, bit.bor, bit.bxor lshift, rshift, rol = bit.lshift, bit.rshift, bit.rol --MQTT.Utility.set_debug(true) broker_addr = "localhost" broker_port = 1883 block_size = 192 time_last = 0 fail_cnt = 0 function on_mqtt_msg(topic, payload) -- application specific code --print("receive ", topic, payload); payload_obj = cjson.decode(payload) if (payload_obj['src_identifier'] == '__transparent') then coroutine.yield(topic, payload_obj) end end function mqtt_listen() while(true) do if(time_last ~= 0 and os.time() >= (time_last + 10)) then coroutine.yield(nil, nil) end mqtt_client:handler() end end -- send data to dev_sn function send_data_to_sn(mqtt_client, gw_sn, dev_sn, mi, data, obj_send) topic = obj_send['topic'] obj_send['mi'] = mi obj_send['len'] = #data obj_send['data_b64'] = base64.encode(data) json_str = cjson.encode(obj_send) --print(json_str) mqtt_client:publish(topic, json_str) time_last = os.time() end -- send data to dev_sn function send_upgrade_process(mqtt_client, gw_sn, dev_sn, mi, Upgrade_Status, Upgrade_Progress) topic = string.format("ipc/%s/upgrade/device/%s/data_filtered/event/__Upgrade_Progress", gw_sn, dev_sn) local payload={} payload['identifier'] = '__Upgrade_Progress' payload['sn'] = dev_sn payload['mi'] = mi payload['time'] = os.time() local progress = {} progress['Upgrade_Status'] = Upgrade_Status progress['Upgrade_Progress'] = Upgrade_Progress payload['tags'] = progress json_str = cjson.encode(payload) --print(json_str) mqtt_client:publish(topic, json_str) end function upgrade_node(json_str, gw_sn, firmware, send_out_json) local obj=cjson.decode(json_str) local obj_send = cjson.decode(send_out_json) local ret = 0 mqtt_client = MQTT.client.create(broker_addr, broker_port, on_mqtt_msg) mqtt_client:connect("upgrade_xieneng") topic = string.format("ipc/%s/+/device/%s/data/raw_data", gw_sn, obj['sn']) mqtt_client:subscribe({topic}) mqtt = coroutine.create(mqtt_listen) -- 获取firmware的version信息 firmware_version = string.match(firmware,'[(Se)|(Auto)]MasterV(%d.%d.%d.%d).bin') print(firmware_version) addr = get_target_addr(obj) -- 进入升级模式 while (false) do mi = 1000 data = string.format("0x%04x EnterBoot", addr); --data = '0x1000 ShowVersion' send_data_to_sn(mqtt_client, gw_sn, obj['sn'], mi, data, obj_send) ret, topic, payload_obj = coroutine.resume(mqtt) if (payload_obj and payload_obj.sn == obj['sn'] and payload_obj.mi == mi) then local data = base64.decode(payload_obj['data_b64']) if (data == 'true') then print("enter boot") else print("EnterBoot failed") return -1 end send_upgrade_process(mqtt_client, gw_sn, obj['sn'], mi, "start", 0); break end end -- get version while (true) do mi = 1000 data = string.format("0x%04x ShowVersion", addr); --data = '0x1000 ShowVersion' send_data_to_sn(mqtt_client, gw_sn, obj['sn'], mi, data, obj_send) ret, topic, payload_obj = coroutine.resume(mqtt) if (payload_obj and payload_obj.sn == obj['sn'] and payload_obj.mi == mi) then local data = base64.decode(payload_obj['data_b64']) pv = string.match(data, 'pv%s=%s(%d+)') mv = string.match(data, 'mv%s=%s(%d+)') sv = string.match(data, 'sv%s=%s(%d+)') cv = string.match(data, 'cv%s=%s(%d+)') cur_version = string.format('%d.%d.%d.%d', pv, mv, sv, cv) if (cur_version == firmware_version) then -- 版本号与当前版本一致,不进行升级 print("same version") return 0 end send_upgrade_process(mqtt_client, gw_sn, obj['sn'], mi, "ready", 0); break end end file = io.open(firmware, "r") local bin_buf = file:read("*a") -- send data file:seek("set") local offset = 0 local last_progress = 0 while (true) do if (offset >= #bin_buf) then break end block_data = string.sub(bin_buf, offset+1, offset+block_size) if not block_data then print("read data failed") break end --print(dyutils.bin_str_to_hex(block_data)) --print(#block_data) block_pkt = bpack("CS",block_pkt,crc) --print(dyutils.bin_str_to_hex(bin_str)) while (true) do mi = mi + 1 send_data_to_sn(mqtt_client, gw_sn, obj['sn'], mi, bin_str, obj_send) ret, topic, payload_obj = coroutine.resume(mqtt) if (payload_obj and payload_obj.sn == obj['sn'] and payload_obj.mi == mi) then local data = base64.decode(payload_obj['data_b64']) --print(dyutils.bin_str_to_hex(data)) offset = offset + #block_data _,_,_,_,_,_,result = bunpack(data, "CSCCCI"); --print("result:", result) if (result == offset) then fail_cnt = 0 progress = math.modf((offset * 100)/#bin_buf) if (progress ~= last_progress) then send_upgrade_process(mqtt_client, gw_sn, obj['sn'], mi, "upgrading", progress); last_progress = progress end break else fail_cnt = fail_cnt + 1 if (fail_cnt > 10) then send_upgrade_process(mqtt_client, gw_sn, obj['sn'], mi, "failed", 0) return -1 end end else fail_cnt = fail_cnt + 1 if (fail_cnt > 10) then send_upgrade_process(mqtt_client, gw_sn, obj['sn'], mi, "failed", 0) return -1 end end end end --发送校验 -- local file_check = dyutils.CalculateXorCrc(bin_buf, #bin_buf) --print(file_check) local check_data = bpack("CA>S",check_data,crc) --print(dyutils.bin_str_to_hex(bin_str)) while (true) do mi = mi + 1 send_data_to_sn(mqtt_client, gw_sn, obj['sn'], mi, bin_str, obj_send) ret, topic, payload_obj = coroutine.resume(mqtt) if (payload_obj and payload_obj.sn == obj['sn'] and payload_obj.mi == mi) then local data = base64.decode(payload_obj['data_b64']) --print(dyutils.bin_str_to_hex(data)) offset = offset + #block_data _,_,_,_,_,_,result = bunpack(data, "CSCCCI"); print("result:", result) ret = result break end end if (ret == 0) then if (obj['Reset'] == 1) then data = string.format("0x%04x reset", addr); send_data_to_sn(mqtt_client, gw_sn, obj['sn'], mi, data, obj_send) end end mi = mi + 1 if (ret == 0) then send_upgrade_process(mqtt_client, gw_sn, obj['sn'], mi, "success", 100) else send_upgrade_process(mqtt_client, gw_sn, obj['sn'], mi, "failed", 0) end return ret end function get_target_addr(obj) -- 先关心单播方式升级 addr = 0 addr = bit.bor(addr, bit.lshift(obj['L3_DevId'], 12)) -- (14 .. 12) addr = bit.bor(addr, bit.lshift(obj['L2_DevId'], 7)) -- (11 .. 7) addr = bit.bor(addr, obj['L1_DevId']) -- (6 .. 0) return addr end --upgrade_node('{"identifier":"__Upgrade","Action":"start","L1_DevId":0,"sn":"22014C00023401","L3_DevId":1,"L2_DevId":0,"mi":34004323,"URL":"http://10.3.1.229:8001/SeMasterV2.5.7.8.bin","MD5":"bbaa3ea3bea9730dd4d33a6c0603cc92","timestamp":1584091544}', '22014C000234', '/tmp/SeMasterV2.5.7.8.bin')