python实现阿里云物联网平台历史数据查询

互联网金融投资需谨慎，查看平台资质和历史表现 #生活技巧# #理财投资建议# #互联网金融知识#

python实现阿里云物联网平台历史数据查询

物联网平台爱好者阅读

阿里云官方文档提供物联网平台访问历史数据的API，但是操作起来不方便，最多显示100行数据，利用一个下午的时间，自己写了一份代码，控制台数据查询数据时间段。权当让大家偷懒的工具，多余的不说，直接上代码。代码由两部分组成，时间戳日期互换模块和主函数代码:

时间戳日期互换模块 代码片.

import time ''' 时间戳转换成日期格式 1）将时间时间戳转换成时间元组 2）通过strftime(）将时间元组转换成格式化时间 ''' def timestamp_to_format(timestamp=None, format='%Y-%m-%d %H:%M:%S'): # try: if timestamp: time_tuple = time.localtime(timestamp) #print('time_tuple:', time_tuple) # print('type(time_tuple):',type(time_tuple)) res = time.strftime(format, time_tuple) else: res = time.strftime(format) return res ''' 将格式化时间转换为时间戳 1）将日期转换成时间元组 2）转换成时间戳 ''' def timeformat_to_timestamp(timeformat=None,format = '%Y-%m-%d %H:%M:%S'): # try: if timeformat: time_tuple = time.strptime(timeformat,format) res = time.mktime(time_tuple) #转成时间戳 else: res = time.time() #获取当前时间戳 return int(res) #print("timeformat_to_timestamp('2018-04-12 09:09:45')的时间戳：",timeformat_to_timestamp('2018-04-12 09:09:45')) if __name__ == '__main__': ac = timestamp_to_format(1615811334) # print(ac,'\t',at) print(ac) format = input("请输入时间：") at = timeformat_to_timestamp(format) print(at) # print('时间戳1234567转换成日期格式为：',timestamp_to_format(1234567) #以下为主函数部分 from aliyunsdkcore.client import AcsClient from aliyunsdkcore.acs_exception.exceptions import ClientException from aliyunsdkcore.acs_exception.exceptions import ServerException from aliyunsdkiot.request.v20180120.QueryDevicePropertiesDataRequest import QueryDevicePropertiesDataRequest import json import time_transform import time """ { "NextValid":true, "NextTime":1616414594866, "RequestId":"4F83BDAF-A66C-4B53-828B-BC721C4B2209", "PropertyDataInfos": {"PropertyDataInfo": [ {"Identifier":"Distance", "List": {"PropertyInfo": [ {"Value":"65.56","Time":1616083267025}, {"Value":"65.10","Time":1616083335520}, {"Value":"65.10","Time":1616083404606}, {"Value":"65.45","Time":1616083473543}, {"Value":"67.92","Time":1616086865662}, {"Value":"69.39","Time":1616086934633}, {"Value":"70.33","Time":1616087003654}, {"Value":"69.55","Time":1616087072515}, {"Value":"70.13","Time":1616090465884}, ] } } ] }, "Code":"","Success":true } """ def check(StartTime, EedTime): #阿里云控制台可以查到AccessKey ID和AccessKey Secret client = AcsClient('AccessKey ID', 'AccessKey Secret', 'cn-shanghai') request = QueryDevicePropertiesDataRequest() request.set_accept_format("json") request.set_DeviceName("YAN76-5") """ print("起始时间应早于终止时间,时间格式:xxxx-xx-xx xx:xx:xx") format = input("请输入起始时间：") ad = time_transform.timeformat_to_timestamp(format)*1000 print(ad) request.set_StartTime(ad) # request.set_EndTime(1618230534000) # print("起始时间应早于终止时间") format = input("请输入终止时间：") ap = time_transform.timeformat_to_timestamp(format)*1000 print(ap) request.set_EndTime(ap) """ request.set_StartTime(StartTime) request.set_EndTime(EedTime) request.set_Asc("1") request.set_PageSize("10") request.set_Identifiers(["Distance"]) # request.set_Identifiers(["Distance","Distance1"]) request.set_ProductKey("a1gMuOnqrr7") # request.set_IotId("b01a617629f73a72d8867f9232633ae0") response = client.do_action_with_exception(request) # python2: print(response) # print(response[0], response[1], response[2], response[3], # response[4], response[5], response[6], response[7]) # print('\n') byte = response str = byte.decode() # print(type(str)) # print(len(str)) # print(str) # print(type(str)) # print(str[10], str[11], str[12], str[13], str[14], str[15]) # print('\n') json_str = json.loads(str) # print(json_str) # print(type(json_str)) # print(json_str['PropertyDataInfos']) # print('\n') # print(type(json_str),print(json_str['NextValid'])) if json_str['NextValid'] == True: # json_str['NextTime'] = time_transform.timestamp_to_format((json_str['NextTime'])/1000) # print(time_transform.timestamp_to_format( # (json_str['NextTime'])/1000)) # print(json_str['NextTime']) a = json_str['PropertyDataInfos'] # print('\n') # print(a['PropertyDataInfo'], type(a['PropertyDataInfo'])) # print('\n') b = a['PropertyDataInfo'] # print(b[0], type(b[0])) # print('\n') c = b[0] # print(c['List'], type(c['List'])) d = c['List'] # print('\n') # print(d['PropertyInfo'], type(d['PropertyInfo'])) e = d['PropertyInfo'] # f=e[0] # g=time_transform.timestamp_to_format((f['Time'])/1000) # print(g) i = 0 while i < 10: # print(i, '\t', e[i], time_transform.timestamp_to_format( print(e[i], time_transform.timestamp_to_format((e[i]['Time'])/1000)) i += 1 # print('\n') return json_str['NextTime'] if json_str['NextValid'] == False: print("查询结束") return None if __name__ == '__main__': print("起始时间应早于终止时间,时间格式:xxxx-xx-xx xx:xx:xx") format = input("请输入起始时间：") ad = time_transform.timeformat_to_timestamp(format)*1000 #ad = 1617638400000 print(ad) # request.set_StartTime(ad) # request.set_EndTime(1618230534000) # print("起始时间应早于终止时间") format = input("请输入终止时间：") ap = time_transform.timeformat_to_timestamp(format)*1000 #ap = 1618070400000 print(ap) # request.set_EndTime(ap) ac = ap-ad print(ac) while ac > 600000: an = check(ad, ap) # time.sleep(0.5) # print(an) ad = an if ad == None: print("执行完毕") break input("等待指令") # return ad

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197

网址：python实现阿里云物联网平台历史数据查询 https://www.yuejiaxmz.com/news/view/378691

相关内容

基于阿里云物联网平台
阿里云物联网平台学习（二）之场景联动
ESP32C3对接阿里云生活物联网平台
【阿里云生活物联网架构师专题 ⑦】阿里云物联网平台的网关
（二）stm32单片机连接阿里云生活物联网平台/物联网平台（附代码）
阿里云IoT发布生活物联网平台2.0，智能硬件开发效率提高5倍
【阿里云生活物联网架构师专题 ①】esp32 sdk 直连接入阿里云物联网平台，实现天猫精灵语音控制；
阿里云物联网生活平台简介
WIFI设备接入阿里云物联网平台
【阿里云生活物联网架构师专题 ⑧】基于在 SpringBoot Java私有云上接收阿里云物联网设备的各种状态数据，实现m2m流转；

随便看看