Embedded/pikapython
1
0
Fork 0
mirror of https://gitee.com/Lyon1998/pikapython.git synced 2025-01-29 17:22:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

add onenet

Browse Source
This commit is contained in:
dreamcmi 2022-12-18 00:48:37 +08:00
parent 4ccaee7b2e
commit 4e8e08c6f6
2 changed files with 138 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
package/iotcloud/iotcloud.py
View File

@ -4,6 +4,9 @@ import random
import mqtt
from PikaStdDevice import Time
ONENET_API = 0
ONENET_DEVICE = 1
class IOT:
def __init__(self):
@ -16,7 +19,29 @@ class IOT:
return a
def getTimeStamp(self, t):
return int(Time.time()*1000) + t # todo 时间戳64位整形实现
return str(int(Time.time()*1000) + t) # todo 时间戳64位整形实现
def urlEncode(self, s):
rs = ""
for i in s:
if i == "+":
i = "%2B"
elif i == " ":
i = "%20"
elif i == "/":
i = "%2F"
elif i == "?":
i = "%3F"
elif i == "%":
i = "%25"
elif i == "#":
i = "%23"
elif i == "&":
i = "%26"
elif i == "=":
i = "%3D"
rs = rs + i
return rs
def aliyun(self, clientId: str, productKey: str, deviceName: str, deviceSecret: str,
signMethod="hmac-md5", regionID="cn-shanghai", ssl=False):
@ -56,14 +81,55 @@ class IOT:
self.address = productId + ".iotcloud.tencentdevices.com"
self.port = int(1883)
self.clientid = productId + deviceName
self.username = self.clientid + ";12010126;" + connid + ";" + str(expiry)
self.username = self.clientid + ";12010126;" + connid + ";" + expiry
token = hmac.new(base64.b64decode(deviceSecret.encode()),
msg=self.username.encode(),
digestmod=signMethod).hexdigest()
self.password = token + ";" + signMethod.replace("-", "")
return True
def onenet(self): ...
def onenet(self, productId, deviceId, accessKey, mode=ONENET_DEVICE, signMethod="hmac-md5", expiryTime=3600, ssl=False):
if productId == None or deviceId == None or accessKey == None:
print("[Error]input param is None")
return False
if signMethod not in self._signMethodTable:
print("[Error]not support signMethod")
return False
if ssl:
self.address = "mqttstls.heclouds.com"
self.port = int(8883)
else:
self.address = "mqtts.heclouds.com"
self.port = int(1883)
self.clientid = deviceId
self.username = productId
method = signMethod.replace("hmac-", "")
expiry = self.getTimeStamp(expiryTime)
if mode == ONENET_DEVICE:
res = "products/" + productId + "/devices/" + deviceId
elif mode == ONENET_API:
res = "products/" + productId
else:
print("[Error]onenet not support mode")
return False
org = expiry + '\n' + method + '\n' + res + '\n' + "2018-10-31"
k = base64.b64decode(accessKey.encode())
h = hmac.new(k, msg=org.encode(), digestmod=signMethod)
sign = base64.b64encode(h.digest()).decode()
res = self.urlEncode(res)
sign = self.urlEncode(sign)
self.password = "version=2018-10-31&res=%s&et=%s&method=%s&sign=%s" % (res, expiry, method, sign)
print(self.clientid)
print(self.username)
print(self.password)
return True
def onenetMulti(self, productId, deviceId, apiKey):
self.address = "mqtt.heclouds.com"
self.port = int(6002)
self.clientid = deviceId
self.username = productId
self.password = apiKey
def connect(self, keepalive=600):
self._client = mqtt.MQTT(self.address, port=self.port, clinetID=self.clientid,

72
port/linux/package/pikascript/iotcloud.py
View File

@ -4,6 +4,9 @@ import random
import mqtt
from PikaStdDevice import Time
ONENET_API = 0
ONENET_DEVICE = 1
class IOT:
def __init__(self):
@ -16,7 +19,29 @@ class IOT:
return a
def getTimeStamp(self, t):
return int(Time.time()*1000) + t # todo 时间戳64位整形实现
return str(int(Time.time()*1000) + t) # todo 时间戳64位整形实现
def urlEncode(self, s):
rs = ""
for i in s:
if i == "+":
i = "%2B"
elif i == " ":
i = "%20"
elif i == "/":
i = "%2F"
elif i == "?":
i = "%3F"
elif i == "%":
i = "%25"
elif i == "#":
i = "%23"
elif i == "&":
i = "%26"
elif i == "=":
i = "%3D"
rs = rs + i
return rs
def aliyun(self, clientId: str, productKey: str, deviceName: str, deviceSecret: str,
signMethod="hmac-md5", regionID="cn-shanghai", ssl=False):
@ -56,14 +81,55 @@ class IOT:
self.address = productId + ".iotcloud.tencentdevices.com"
self.port = int(1883)
self.clientid = productId + deviceName
self.username = self.clientid + ";12010126;" + connid + ";" + str(expiry)
self.username = self.clientid + ";12010126;" + connid + ";" + expiry
token = hmac.new(base64.b64decode(deviceSecret.encode()),
msg=self.username.encode(),
digestmod=signMethod).hexdigest()
self.password = token + ";" + signMethod.replace("-", "")
return True
def onenet(self): ...
def onenet(self, productId, deviceId, accessKey, mode=ONENET_DEVICE, signMethod="hmac-md5", expiryTime=3600, ssl=False):
if productId == None or deviceId == None or accessKey == None:
print("[Error]input param is None")
return False
if signMethod not in self._signMethodTable:
print("[Error]not support signMethod")
return False
if ssl:
self.address = "mqttstls.heclouds.com"
self.port = int(8883)
else:
self.address = "mqtts.heclouds.com"
self.port = int(1883)
self.clientid = deviceId
self.username = productId
method = signMethod.replace("hmac-", "")
expiry = self.getTimeStamp(expiryTime)
if mode == ONENET_DEVICE:
res = "products/" + productId + "/devices/" + deviceId
elif mode == ONENET_API:
res = "products/" + productId
else:
print("[Error]onenet not support mode")
return False
org = expiry + '\n' + method + '\n' + res + '\n' + "2018-10-31"
k = base64.b64decode(accessKey.encode())
h = hmac.new(k, msg=org.encode(), digestmod=signMethod)
sign = base64.b64encode(h.digest()).decode()
res = self.urlEncode(res)
sign = self.urlEncode(sign)
self.password = "version=2018-10-31&res=%s&et=%s&method=%s&sign=%s" % (res, expiry, method, sign)
print(self.clientid)
print(self.username)
print(self.password)
return True
def onenetMulti(self, productId, deviceId, apiKey):
self.address = "mqtt.heclouds.com"
self.port = int(6002)
self.clientid = deviceId
self.username = productId
self.password = apiKey
def connect(self, keepalive=600):
self._client = mqtt.MQTT(self.address, port=self.port, clinetID=self.clientid,