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