关键词:
本文章为原创,转载请注明出处!
账号:iotos_test 密码:iotos123
代码地址:IOTOSDK-Python: IOTOS Python版本SDK,自带原生接口和采集引擎 (gitee.com)
目录
4.3、根据中台获取的数据及开发文档进行获取数据再将其上传的操作
1、驱动目的
与指定平台上的NB设备对接
2、开发文档要求
3、代码示例
#coding=utf-8
import sys
sys.path.append("..")
from driver import *
import requests
import time
import datetime
from urllib import urlencode
import urllib2
import base64
import json
import urllib
import hmac
from hashlib import sha1
#签名算法
def signature(key, application, timestamp, param, body):
code = "application:" + application + "\\n" + "timestamp:" + timestamp + "\\n"
for v in param:
code += str(v[0]) + ":" + str(v[1]) + "\\n"
if (body is not None) and (body.strip()) :
code += body + '\\n'
print("param=" + str(param))
print("body=" + str(body))
print("code=" + str(code))
#print(binascii.b2a_hex(code))
return base64.b64encode(hash_hmac(key, code, sha1))
def hash_hmac(key, code, sha1):
hmac_code = hmac.new(key.encode(), code.encode(), sha1)
print("hmac_code=" + str(hmac_code.hexdigest()))
return hmac_code.digest()
def getTimeOffset(url):
request = urllib2.Request(url)
start = int(time.time() * 1000)
response = urllib2.urlopen(request)
end = int(time.time() * 1000)
if response is not None:
return int(int(response.headers['x-ag-timestamp']) - (end + start) / 2);
else:
return 0
baseUrl = '###########' # 地址
timeUrl = '##############' # 地址
offset = getTimeOffset(timeUrl)
#发送http请求函数
def sendSDKRequest(path, head, param, body, version, application, MasterKey, key, method=None, isNeedSort=True,isNeedGetTimeOffset=False):
paramList = []
for key_value in param:
paramList.append([key_value, param[key_value]])
print("paramList=" + str(paramList))
if (MasterKey is not None) and (MasterKey.strip()):
paramList.append(['MasterKey', MasterKey])
if isNeedSort:
paramList = sorted(paramList)
headers =
if (MasterKey is not None) and (MasterKey.strip()):
headers['MasterKey'] = MasterKey
headers['application'] = application
headers['Date'] = str(datetime.datetime.now())
headers['version'] = version
# headers['Content-Type'] = Content_Type
# headers['sdk'] = sdk
# headers['Accept'] = Accept
# headers['User-Agent'] = User_Agent
# headers['Accept-Encoding'] = 'gzip,deflate'
temp = dict(param.items())
if (MasterKey is not None) and (MasterKey.strip()):
temp['MasterKey'] = MasterKey
url_params = urlencode(temp)
url = baseUrl + path
if (url_params is not None) and (url_params.strip()):
url = url + '?' + url_params
print("url=" + str(url))
global offset
if isNeedGetTimeOffset:
offset = getTimeOffset(timeUrl)
timestamp = str(int(time.time() * 1000) + offset)
headers['timestamp'] = timestamp
sign = signature(key, application, timestamp, paramList, body)
headers['signature'] = sign
headers.update(head)
print("headers : %s" % (str(headers)))
if (body is not None) and (body.strip()):
request = urllib2.Request(url=url, headers=headers, data=body.encode('utf-8'))
else:
request = urllib2.Request(url=url, headers=headers)
if (method is not None):
request.get_method = lambda: method
response = urllib2.urlopen(request)
if ('response' in vars()):
print("response.code: %d" % (response.code))
return response
else:
return None
def QueryDeviceStatus(appKey, appSecret, body):
path = '/aep_device_status/deviceStatus'
head =
param =
version = '20181031202028'
application = appKey
key = appSecret
response = sendSDKRequest(path, head, param, body, version, application, None, key, 'POST')
if response is not None:
return response.read()
return None
def getDeviceStatusHisInPage(appKey, appSecret, body):
path = '/aep_device_status/getDeviceStatusHisInPage'
head =
param =
version = '20190928013337'
application = appKey
key = appSecret
response = sendSDKRequest(path, head, param, body, version, application, None, key, 'POST')
if response is not None:
return response.read()
return None
class Project(IOTOSDriverI):
def InitComm(self,attrs):
self.setPauseCollect(False)
self.setCollectingOneCircle=True
self.online(True)
try:
# 获取中台配置的参数(必不可少)
self.Nbapplication=self.sysAttrs['config']['param']['application'] # APPKEY
self.Nbkey=self.sysAttrs['config']['param']['key'] # APPScret
self.NbproductId=self.sysAttrs['config']['param']['productId']
self.NbdeviceId=self.sysAttrs['config']['param']['deviceId']
except Exception,e:
self.debug(u'获取参数失败!'+e.message)
def Collecting(self,dataId):
# application = "yp4RWxBkpU" # APPKEY
# key = "hPpzvjeGzd" # APPScret
# productId="15031938"
# deviceId="4b3869025fea421bb5a186b90bce90da"
timearry = (datetime.datetime.now() + datetime.timedelta(days=-29)).timetuple() # 当前时间减去29天后转化为timetuple的格式用于转换成timestamp格式
begin_timestamp = str(int(time.mktime(timearry) * 1000) + offset) # "1624550400000"
end_timestamp = str(int(time.time() * 1000) + offset) # "1624982399000"
page_size = "1"
page_timestamp = ""
body_HisInPage ='"productId":"' + self.NbproductId + '","deviceId":"' + self.NbdeviceId + '","begin_timestamp":"' + begin_timestamp + '","end_timestamp":"' + end_timestamp + '","page_size":' + page_size + ',"page_timestamp":"' + page_timestamp + '"'
#发送请求,处理数据
res = getDeviceStatusHisInPage(self.Nbapplication, self.Nbkey, body_HisInPage)
r = eval(res)
if bin(int(r["deviceStatusList"][0]["start_stop"]))[-3:-2]=='0':
door_value=False
else:
door_value=True
if bin(int(r["deviceStatusList"][0]["start_stop"]))[-4:-3]=='0':
sos_value=False
else:
sos_value=True
try:
self.setValue(u'door', door_value)
self.setValue(u'sos', sos_value)
self.setValue(u'hardware_ver', r["deviceStatusList"][0]["hardware_ver"])
self.setValue(u'voltameter', r["deviceStatusList"][0]["voltameter"])
self.setValue(u'software_ver', r["deviceStatusList"][0]["software_ver"])
self.setValue(u'serial_num', r["deviceStatusList"][0]["serial_num"])
self.setValue(u'mes_len', r["deviceStatusList"][0]["mes_len"])
self.setValue(u'frequency', r["deviceStatusList"][0]["frequency"])
self.setValue(u'IMEI', r["deviceStatusList"][0]["IMEI"])
self.setValue(u'ICCID', r["deviceStatusList"][0]["ICCID"])
self.setValue(u'device_type', r["deviceStatusList"][0]["device_type"])
self.setValue(u'Data', time.strftime("%Y-%m-%d %H:%M:%S",time.localtime(r["deviceStatusList"][0]["Data"])))
self.debug(u'数据上传成功')
except Exception,e:
self.debug(u'数据上传失败'+e.message)
time.sleep(7)
return ()
4、驱动解析
4.1、导入指定依赖包
import sys
sys.path.append("..")
from driver import *
import requests
import time
import datetime
from urllib import urlencode
import urllib2
import base64
import json
import urllib
import hmac
from hashlib import sha1
4.2、从中台中获取数据
4.3、根据中台获取的数据及开发文档进行获取数据再将其上传的操作
4.4、执行效果如下
iotos驱动详解-获取中台中设备驱动的配置信息(代码片段)
本文章为原创,转载请注明出处!登录平台:IOTOS®爱投斯物联中台账号:iotos_test 密码:iotos123代码地址:IOTOSDK-Python:IOTOSPython版本SDK,自带原生接口和采集引擎(gitee.com)1、代码示例#!coding:utf8importjsoni... 查看详情
iotos驱动详解-参数的上传(代码片段)
...com)目录1.上传代码如下1.1、创建通信网关1.2、创建模板驱动1 查看详情
驱动开发实战之tcpclient(代码片段)
...,你需要根据设备方提供的协议,为IoTGateway开发驱动,进行数据交互。文章比较长也可以到官网会有更好的体验,地址:http://iotgateway.net/docs/iotgateway/driver/tcpclient请先浏览上一篇驱动简介:http://iotgateway.net 查看详情
电信天翼物联网平台对接应用服务ctwing(代码片段)
1.创建账号天翼平台2.创建产品 3.添加设备4.在应用管理中新增应用5.应用服务对接SDK使用说明 中国电信物联网开放平台NB网关2.0API参考@Data@ApiModel(value="设备数据推送请求")publicclassDevicePushRequest@ApiModelProperty(na... 查看详情
linux-设备驱动概述(代码片段)
文章目录Linux设备驱动概述1.设备驱动的作用2.无操作系统的设备驱动3.有操作系统时的设备驱动4.Linux设备驱动4.1设备的分类及特点4.2Linux设备驱动与整个软硬件系统的关系4.3Linux设备驱动的重难点5.源代码阅读6.设备驱动:LED... 查看详情
linux-设备驱动概述(代码片段)
文章目录Linux设备驱动概述1.设备驱动的作用2.无操作系统的设备驱动3.有操作系统时的设备驱动4.Linux设备驱动4.1设备的分类及特点4.2Linux设备驱动与整个软硬件系统的关系4.3Linux设备驱动的重难点5.源代码阅读6.设备驱动:LED... 查看详情
虚拟字符设备驱动开发(代码片段)
目录字符设备驱动简介内核驱动操作函数集合字符设备驱动开发步骤驱动模块的加载和卸载字符设备注册与注销实现设备的具体操作函数添加LICENSE和作者信息Linux设备号设备号的组成设备号的分配chrdevbase字符设备驱动开发实验... 查看详情
电信平台对接ctwing具体实现流程(代码片段)
简介:要实现电信平台对接,需要设备、电信平台、系统进行对接,主要介绍后台获取http请求和指令下发,具体流程如下图:一、在电信平台上创建设备、产品。具体参考:官方文档||TCP产品创建创建产品... 查看详情
bsp开发学习2平台设备驱动(代码片段)
文章目录Linux设备驱动模型Linux中的设备驱动模型组成为什么要使用设备驱动模型设备驱动模型的底层架构sysfs目录kobjectkobj_typekset设备驱动模型三大组件总线设备驱动平台设备驱动概述平台设备驱动工作原理核心变量与函数工作... 查看详情
基于openharmonyl2设备,如何用iotdevicesdktiny对接华为云(代码片段)
摘要:本文主要讲解如何基于L2设备对接华为云IoTDA,以DAYU200开发板,采用IoTDeviceSDKTiny对接华为云IoTDA,当然这里也可以采用其他OpenHarmony的富设备。本文分享自华为云社区《基于OpenHarmonyL2设备采用IoTDeviceSDKTiny对... 查看详情
是否有api可供云对接?(代码片段)
...开发者拉取:应用维度、产品维度。应用维度:设备绑定的用户属于开发者在涂鸦云应用中的用户,开发者间接拥有权限操作属于自己应用用户的设备;产品维度:设备所属产品属于开发者在涂鸦云产品中的... 查看详情
linux驱动开发:字符设备驱动开发(代码片段)
文章目录Linux驱动开发:字符设备驱动开发一、字符设备简介二、驱动调用原理三、驱动加载与卸载四、字符设备注册与注销五、设备驱动编写流程六、添加驱动信息Linux驱动开发:字符设备驱动开发一、字符设备简介字... 查看详情
linux驱动开发:字符设备驱动开发(代码片段)
文章目录Linux驱动开发:字符设备驱动开发一、字符设备简介二、驱动调用原理三、驱动加载与卸载四、字符设备注册与注销五、设备驱动编写流程六、添加驱动信息Linux驱动开发:字符设备驱动开发一、字符设备简介字... 查看详情
设备树下的platform驱动编写(代码片段)
目录设备树下的platform驱动简介硬件原理图分析实验程序编写修改设备树文件platform驱动程序编写编写测试APP运行测试编译驱动程序和测试APP运行测试上一章我们详细的讲解了Linux下的驱动分离与分层,以及总线、设备和驱动... 查看详情
设备树下的led驱动实验(代码片段)
目录设备树LED驱动原理硬件原理图分析实验程序编写修改设备树文件LED灯驱动程序编写编写测试APP运行测试编译驱动程序和测试APP运行测试上一章我们详细的讲解了设备树语法以及在驱动开发中常用的OF函数,本章我们就开... 查看详情
platform设备驱动实验(代码片段)
目录Linux驱动的分离与分层驱动的分隔与分离驱动的分层platform平台驱动模型简介platform总线platform驱动platform设备硬件原理图分析试验程序编写platform设备与驱动程序编写测试APP编写54.5运行测试编译驱动程序和测试APP运行测试我... 查看详情
linux的platform驱动(代码片段)
如下内容来自《【正点原子】I.MX6U嵌入式Linux驱动开发指南V1.5.2.pdf》 将设备信息从设备驱动中剥离开来,驱动使用标准方法去获取到设备信息(比如从设备树中获取到设备信息),然后根据获取到的设备信息来初始化设... 查看详情
新字符设备驱动实验(代码片段)
目录新字符设备驱动原理分配和释放设备号新的字符设备注册方法自动创建设备节点mdev机制创建和删除类创建设备参考示例设置文件私有数据硬件原理图分析实验程序编写LED灯驱动程序编写编写测试APP运行测试编译驱动程序和... 查看详情