python实现onenet平台上传、获取数据

编程入门行业动态更新时间:2024-10-10 05:19:32

python实现onenet平台<a href=https://www.elefans.com/category/jswz/34/1771255.html style= 上传、获取数据"/>

python实现onenet平台上传、获取数据

向onenet平台传数据

代码只实现了简单的上传的功能，没有做其他的功能，比如说数据的产生什么的，所以上传的数据是直接给出的

"""
@author:wgc
@Time:2019/5/10 15:19
@Version: v1.0
"""import urllib.request
import json
import time
from time import sleep
#设备ID
deviceId = "525627026"
APIKey = "Sda0nUGoDiV4TfgZh=S8gZLALZ0="#上传函数
def http_put_data(data):url = "/" + deviceId + '/datapoints'd = time.strftime('%Y-%m-%dT%H:%M:%S')values = {"datastreams": [ {"id": "CO2", "datapoints": [{"value": data}]}, {"id": "PM25", "datapoints": [{"value": data}]},{"id": "PM10", "datapoints": [{"value": data}]},{"id": "VOC", "datapoints": [{"value": data}]} ]}jdata = json.dumps(values).encode("utf-8")request = urllib.request.Request(url, jdata)request.add_header('api-key', APIKey)request.get_method = lambda: 'POST'request = urllib.request.urlopen(request)return request.read()if __name__ == '__main__':R = http_put_data(10)print(R)

效果

这个返回结果也只是直接把平台自定义的返回输出一下，这个结果就表明已经成功了，如果上传出错，他会告诉我们错误的个数及提示错误

平台上的效果

--------------------------2021-4-17更新部分-------------------

onenet平台查看数据

1、登陆onenet平台 --> 控制台
链接：/

2、进入设备管理：鼠标滑到’全部产品服务‘那里 --> 增值服务 --> 设备管理DMP
3、查看数据
1）通过设备概览查看数据量（可以用来检测是否上传成功）

这里只能看到数据数看不到具体的数值，可以通过2）看到具体数据。
2）查看具体数值

如果不断地传输数据就可以看到实时动态，这个我应该是做了，后面整理好可以分享出来。
--------------------------更新部分到这里结束------------------

从平台上获取数据

源码

"""
@author:wgc
@Time:2019/5/9 13:55
@Version: v1.0
"""import requests
import json#设备ID
deviceId = "525627026"
APIKey = "Sda0nUGoDiV4TfgZh=S8gZLALZ0="# 基本设置
url = "/"+deviceId+"/datastreams"
headers = {'api-key': APIKey}# 获得结果并打印
r = requests.get(url, headers=headers)
t: str = r.text#print(t)
params = json.loads(t)
#上面这个语句是将我们获得的内容转成数据字典，这样转是因为我们接收到的内容具有数据字典的形式
#转换成数据字典利于我们后面的操作
#print params['error'][]#print(type(params))
#如果执行上面这条语句我们可以看到返回的结果是dict,也就是我们已经成功转换x = params['data']
#这个语句是从数据转换后的数据字典中获取我们需要的数据,从结果上看params是一个list
#在data前面的都只是一些描述内容，参考教程：('环境参数'+'\t\t\t\t'+'更新时间'+'\t\t\t\t\t'+'数值')
#接下来是获取不同的数据流
for index,values in enumerate(x):#只需要更新时间，id和值，所以这里对获得的数据字典做一下更改#print(values)#这里得到的values也是一个数据字典#因为在onenet那边对这些数据没有给出来，而且也没有意义，所以我们就不在这里显示，因此现将其删除del values['unit']del values['unit_symbol']del values['create_time']#print(values.items())#print(values['update_at'])#这里不知道为什么直接使用values.get('update_at','')和values.get('current_value','')#或者用values['update_at']和values['current_value']报：KeyError错误，而且if里面的那条语句会执行#所以我们通过get方法解决，其中要注意的是，如果没有给定第二个参数，那么默认输出NONEa= str(values.get('update_at',''))b= str(values.get('current_value',''))#因为如果有更新时间就会有相应最新的值，所以这里只用其中一个作为判断条件if (a != ""):if (values['id'] == 'PM25' or values['id'] == 'PM10' ):print(str(values['id']) + '\t\t\t' + a + '\t\t\t' + b)else:print(str(values['id']) + '\t\t\t\t' + a + '\t\t\t'+ b)else:if(values['id'] == 'VOC' or values['id'] == 'VOC'):print(values['id']+ '\t\t\t' +'目前还没有收到任何数据')else:print(values['id'] + '\t\t' + '目前还没有收到任何数据')