# -*- encoding: utf-8 -*- try: import network import ujson import time from aliyunIoT import Device from esp32 import NVS import esp import ota import machine module_exist = True except ImportError: module_exist = False on_trigger_succeed = False on_download_succeed = False on_verify_succeed = False on_upgrade_succeed = False on_connect_wifi_succeed = True on_connect_lp_succeed = False device_dyn_resigter_succeed = False # wifi热点信息 ssid = "SSID" pwd = "PASSWORLD" # 定义需要升级的模块和版本号 module_name = 'default' default_ver = '1.0.0' #当前版本号 update_ver = '1.3.0' #升级后版本号 # 三元组信息 productKey = "输入你的产品key" productSecret = "输入你的产品密钥" deviceName = "输入你的设备名称" deviceSecret = "" app_store_file = 'test.zip' #py文件升级必须压缩为zip文件 firmeware_store_file = 'micropython.bin' #固件升级后缀必须为.bin # 定义升级包的下载和安装路径，其中url,hash_type和hash 会通过服务端推送被保存下来 info = { 'url': '', #如果升级micropython固件，填写firmeware_store_file #如果升级python脚本文件，填写app_store_file 'store_path': app_store_file, 'install_path': '/', #安装路径必须为根目录 'length': 0, 'hash_type': '', 'hash': '' } aliyunIoT_Device = {} class Ota(object): def assertTrue(self, x, msg=''): if not msg: msg = "Expected %r to be True" % x assert x, msg def assertFalse(self, x, msg=''): if not msg: msg = "Expected %r to be False" % x assert not x, msg # ota 消息推送的接受函数 def on_trigger(data): global info global on_trigger_succeed print('***** on_trigger *****'); print(f'data:{data}') # 保存服务端推送的ota信息 info['url'] = data['url'] info['length'] = data['length'] info['module_name'] = data['module_name'] info['version'] = data['version'] info['hash'] = data['hash'] info['hash_type'] = data['hash_type'] # 开始ota 包下载 dl_data = {} dl_data['url'] = info['url'] dl_data['store_path'] = info['store_path'] on_trigger_succeed = True print(f'dl_data:{dl_data}') ota.download(dl_data) def on_download(data): global info global on_download_succeed print('***** on_verify *****'); if data >= 0: on_download_succeed = True # 开始ota包校验 param = {} param['length'] = info['length'] param['store_path'] = info['store_path'] param['hash_type'] = info['hash_type'] param['hash'] = info['hash'] ota.verify(param) # ota 升级包校验结果回调函数 def on_verify(data): # global info global on_verify_succeed print('***** on_verify *****'); print(f'verify_data:{data}') if data >= 0: on_verify_succeed = True # 开始ota升级 param = {} param['length'] = info['length'] param['store_path'] = info['store_path'] param['install_path'] = info['install_path'] ota.upgrade(param) # ota 升级包结果回调函数 def on_upgrade(data): global on_upgrade_succeed print('***** on_upgrade *****'); print(data,"-------------------------------------------------------------------------------------------------------------------------------------") if data >= 0: print('***** upgrade succeed *****'); on_upgrade_succeed = True report_info = { "device_handle": aliyunIoT_Device.getDeviceHandle(), "product_key": productKey, "device_name": deviceName, "module_name": module_name, "version": update_ver } ret = ota.report(report_info) time.sleep(3) print('Start reboot...') machine.reset() def test_ota_download_app(self): global aliyunIoT_Device global on_trigger global on_download global on_verify global on_upgrade global on_trigger_succeed global on_download_succeed global on_verify_succeed global on_upgrade_succeed global productKey global deviceName global deviceSecret global default_ver global update_ver global on_connect_lp_succeed func_name = 'test_ota_download_app' if module_exist: if on_connect_lp_succeed: report_info = { "device_handle": aliyunIoT_Device.getDeviceHandle(), "product_key": productKey, "device_name": deviceName, "module_name": module_name, "version": default_ver } ret = ota.report(report_info) time.sleep(1) def test_ota_init(self): global aliyunIoT_Device global on_connect_lp_succeed func_name = 'test_ota_init' if module_exist: if on_connect_lp_succeed: data_handle = {} data_handle['device_handle'] = aliyunIoT_Device.getDeviceHandle() ret = ota.init(data_handle) self.assertTrue('Function %s test pass!' % func_name) else: self.assertFalse('Function %s test failed!' % func_name) else: self.assertFalse('module not exist') #当iot设备连接到物联网平台的时候触发'connect' 事件 def on_connect(data): global on_connect_lp_succeed print('***** connect lp succeed****') on_connect_lp_succeed = True #当连接断开时，触发'disconnect'事件 def on_disconnect(): global on_connect_lp_succeed print('****aliyun is disconnected****') on_connect_lp_succeed = True #当iot云端下发属性设置时，触发'props'事件 def on_props(request): print('clound req data is {}'.format(request)) #当iot云端调用设备service时，触发'service'事件 def on_service(id,request): print('clound req id is {} , req is {}'.format(id,request)) #当设备跟iot平台通信过程中遇到错误时，触发'error'事件 def on_error(err): print('err msg is {} '.format(err)) #动态注册回调函数 def on_dynreg_cb(data): print('aliyun dynreg success') global deviceSecret,device_dyn_resigter_succeed deviceSecret = data print(f'devicesecret:{deviceSecret}') device_dyn_resigter_succeed = True # 连接物联网平台 def dyn_register_device(productKey,productSecret,deviceName): global on_dynreg_cb,aliyunIoT_Device,deviceSecret,device_dyn_resigter_succeed global nvs key = '_devicesecret' deviceSecretdict = bytearray(32) try: nvs.get_blob(key, deviceSecretdict) except OSError as e: print(f"Error: {e}") if(all(x == 0 for x in deviceSecretdict)) : key_info = { 'productKey': productKey , 'productSecret': productSecret , 'deviceName': deviceName } if not device_dyn_resigter_succeed: aliyunIoT_Device.register(key_info,on_dynreg_cb) else: # 动态注册一个设备，获取设备的deviceSecret #下面的if防止多次注册，当前若是注册过一次了，重启设备再次注册就会卡住， deviceSecret = deviceSecretdict.decode('utf-8') print("deviceSecret:",deviceSecret) def connect_network(): wlan = network.WLAN(network.STA_IF) wlan.active(True) time.sleep(2) wlan.connect(ssid, pwd) while not wlan.isconnected(): pass print('network config: ', wlan.ifconfig()) if __name__ == '__main__': nvs = None esp.osdebug(5) # 连接网络 connect_network() nvs = NVS('aos_kv') #初始化物联网平台Device类，获取device实例 aliyunIoT_Device = Device() if deviceName is not None and len(deviceName) > 0 : #动态注册一个设备 dyn_register_device(productKey,productSecret,deviceName) else: print("can not dynamic reg") while deviceSecret is None: time.sleep(0.2) print('dynamic reg success:' + deviceSecret) key_info = { 'region' : 'cn-shanghai' , 'productKey': productKey , 'deviceName': deviceName , 'deviceSecret': deviceSecret , 'keepaliveSec': 60, } #打印设备信息 print(key_info) aliyunIoT_Device.on(aliyunIoT_Device.ON_CONNECT,on_connect) aliyunIoT_Device.on(aliyunIoT_Device.ON_DISCONNECT,on_disconnect) aliyunIoT_Device.on(aliyunIoT_Device.ON_PROPS,on_props) aliyunIoT_Device.on(aliyunIoT_Device.ON_SERVICE,on_service) aliyunIoT_Device.on(aliyunIoT_Device.ON_ERROR,on_error) aliyunIoT_Device.connect(key_info) # time.sleep_ms(6000) ota.on(1, Ota.on_trigger) ota.on(2, Ota.on_download) ota.on(3, Ota.on_verify) ota.on(4, Ota.on_upgrade) while True: if on_connect_lp_succeed == True: #上报版本信息 FwVersion={} FwVersion["FwVersion"]="ed1_haas_v1.0" FwVersionStr=ujson.dumps(FwVersion) data0={ 'params':FwVersionStr } # time.sleep_ms(1000) aliyunIoT_Device.postProps(data0) # time.sleep_ms(3000) print('************** ota test *****************') app_ota = Ota(); app_ota.test_ota_init(); # time.sleep_ms(1000) app_ota.test_ota_download_app(); break else: time.sleep(1) while True: time.sleep_ms(3000) # print('test')