文档首页> IoT物联网> 开发指南> 应用侧开发> 使用SDK对接> Python SDK使用指南（联通用户专用）
更新时间：2021-07-08 GMT+08:00
查看PDF
分享

Python SDK使用指南（联通用户专用）

非联通用户请查看设备接入服务。

本文以提供的应用侧Python SDK Demo为例说明如何使用Python SDK与物联网平台对接。Demo以Python工程为例，以invokeapiTest下的每个类都包含了main方法，可单独运行，旨在演示如何调用SDK接口。

开发环境要求

开发平台

开发环境

配套要求

推荐的操作系统

IoT
  1. 开发工具：JetBrains PyCharm 2018.1.4 x64
  2. Python Project Interpreter：Python 3.7或Anaconda 3

Python 3.7

Windows7

SDK包为纯Python的egg包，在使用上没有特殊限制，Python在3.7及以上版本即可。

导入Demo工程

  1. 将Python SDK解压至本地Python安装路径的“Lib\site-packages”目录下，如，C:\Program Files\Python37\Lib\site-packages。

  2. 将Python SDK Demo解压到本地。
  3. 打开PyCharm，选择“Open”，再选择Demo解压后的路径，单击“OK”。
  4. 配置Project Interpreter。

    1. 选择“File > Settings > Project: OceanConnect _ Python _SDK_Demo > Project Interpreter”。

    2. 选择SDK所在的解释器，如，C:\Program Files\Python37\python.exe，单击“Apply”，最后单击“OK”即可正确配置工程所需要的SDK 。

  5. Python SDK Demo目录结构如下，Demo工程中已包含SDK库。

配置平台信息及证书

配置文件（application.ini）以及证书都分别放置在工作路径“d:/python_sdk/”下面的cert 和resources文件夹中，用户也可自行设置工作路径。

平台IP、端口、appId和密码都是从配置文件中读取的，因此，当这些信息发生变化时，只要修改配置文件，不用修改应用服务器的代码。

[CLIENT_INFO]
platformIp = 100.100.100.100
platformPort = 8743
appId = FT8EjQ8O****VbW60Qb8xvga
secret = TtuK4Paf*****yAv66himUk8a

在“Constant.py”中，读取证书和CLIENT_INFO。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class Constant(object):
# 工作路径
    workPath = os.path.join('d:/python_sdk/')

# 读取证书
def readCertificate(self):
    certFilePath = os.path.join(Constant.workPath, 'cert/client.crt')
    certFilePath2 = os.path.join(Constant.workPath, 'cert/client.key')
    cert = (certFilePath, certFilePath2)
return cert

# 读取配置文件
def readConfFile(self):
    configFilePath = os.path.join(Constant.workPath, 'resources/application.ini')
    cf.read(configFilePath)
    platformIp = cf.get("CLIENT_INFO", "platformIp")
    platformPort = cf.get("CLIENT_INFO", "platformPort")
    appId = cf.get("CLIENT_INFO", "appId")
    secret = cf.get("CLIENT_INFO", "secret")
return platformIp, platformPort, appId, secret

# 设置平台对接信息
def clientInfo(self):
    clientInfo = ClientInfo()
    clientInfo.setPlatformIp((Constant().readConfFile())[0])
    clientInfo.setPlatformPort((Constant().readConfFile())[1])
    clientInfo.setAppId((Constant().readConfFile())[2])
    clientInfo.setSecret((Constant().readConfFile())[3])
    clientInfo = DictUtil.dto2dict(clientInfo)
return clientInfo

业务接口调用方法

设置好Constant.py后才能调用其他业务接口。所有业务接口的测试都在“invokeapiTest”中。以如下几个接口为例说明如何调用业务接口：

关于哪些参数需要设置，请查看《应用侧Python SDK API参考》对于可选参数，如果业务不需要，可以不设置。

配置日志

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
# 在每个业务接口中都配置了日志，具体日志实现可参考LogUtil.py:
# 开发者可通过修改入参来控制日志的输出，以下三个参数都可自行设置；
# logPath：日志的输出路径，默认在工作路径d:/python_sdk/log/ 
# level：日志等级，默认DEBUG（最低等级，即大于等于该等级的日志级别都会输出），如果level=0没有日
志输出
# logFilename：日志名称

def setLogConfig(self, 
logPath=os.path.join(Constant.workPath, 'log/'), 
level=logging.DEBUG,
		   logFilename="python_sdk.log"):

鉴权

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# 在AuthenticationTest.py中对鉴权和刷新token接口进行测试:
if __name__ == "__main__":
    from com.huawei.iotplatform.utils.LogUtil import Log

    # 实例化Authentication()
    authentication = Authentication()

    # 调用鉴权类实例authentication提供的业务接口，如getAuthToken
    ag = authentication.getAuthToken(Constant().clientInfo())
    print("====== get access token ======")
    print("result:", ag + "\n")

# 从返回的ag中获取需要的参数，如accessToken 
authOutDTO = AuthOutDTO()
authOutDTO.setAccessToken(json.loads(ag)['accessToken'])
print("token", authOutDTO.getAccessToken())

订阅

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
# 在SubscriptionManagementTest.py中对订阅订阅平台业务数据接口进行测试:
class SubscriptionManagementTest(object):
    def subDeviceBusinessData(self):
        sdbdInDTO = SubDeviceBusinessDataInDTO()
        sdbdInDTO.notifyType = "bindDevice"  
        sdbdInDTO.callbackUrl = "https://XXX.XXX.XXX.XXX:443/callbackurltest"
        return sdbdInDTO

if __name__ == "__main__":
    from com.huawei.iotplatform.utils.LogUtil import Log

    # 实例化
    smTest = SubscriptionManagementTest()
subscriptionManagement = SubscriptionManagement()

    # get accessToken at first
    result = Authentication().getAuthToken(Constant().clientInfo())
    authOutDTO = AuthOutDTO()
    authOutDTO.setAccessToken(json.loads(result)['accessToken'])
    accessToken = authOutDTO.getAccessToken()

    # 调用订阅类实例subscriptionManagement提供的业务接口，如subDeviceData
    ss = subscriptionManagement.subDeviceBusinessData(smTest.subDeviceBusinessData(), accessToken)
    print("====== subscribe to device business data notification ======")
    print("result:", ss + "\n")

注册设备

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# 在DeviceManagementTest.py中对注册设备接口进行测试:
class DeviceManagementTest(object):
    def regDirectDeviceInfo(self):
        rddInDto = RegDirectDeviceInDTO()
        rddInDto.nodeId = "AAA" + str(random.randint(0, 9999))
        return rddInDto

if __name__ == "__main__":
    from com.huawei.iotplatform.utils.LogUtil import Log

    # 实例化
    dmTest = DeviceManagementTest()
    deviceManagement = DeviceManagement()

    # get accessToken at first
    result = Authentication().getAuthToken(Constant().clientInfo())
    authOutDTO = AuthOutDTO()
    authOutDTO.setAccessToken(json.loads(result)['accessToken'])
    accessToken = authOutDTO.getAccessToken()

    # 调用设备管理类实例deviceManagement提供的业务接口，如regDirectDevice
    dr = deviceManagement.regDirectDevice(dmTest.regDirectDeviceInfo(), None, accessToken)
    print("====== register a new device ======")
    print("result:", dr + "\n")

# 从返回的dr中获取需要的参数，如deviceId
rddod = RegDirectDeviceOutDTO()
rddod.setDeviceId(json.loads(dr)['deviceId'])
deviceId = rddod.getDeviceId()
print("deviceId==", deviceId + "\n")

回调接口实现及证书制作

回调接口实现

回调接口的启动在“PushMessageReceiverTest.py”中，配置请看以下说明。
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
if __name__ == '__main__':

# 回调地址和端口
    callbackUrl = "XXX.XXX.XXX.XXX"
    port = 8099

    # 这里使用了pyOpenSSL自带证书，注意ssl_context的值必须是adhoc
    app.run(host=callbackUrl, port=port, ssl_context='adhoc')

    # 使用自己配置的证书，ssl_context的值配置如下：此处放置的是第8章节步骤3生成的server证书
    app.run(host=callbackUrl, port=port, ssl_context=('d:/python_sdk/cert/server.crt', 'd:/python_sdk/cert/server.key'))

具体业务实现在“PushMessageReceiver.py”中的“PushMessageReceiver”类中，可以参考Demo中的“PushMessageReceiver”类，需要接收哪一类消息就改写对应的方法，如：

1
2
3
4
5
class PushMessageReceiver(object):
    def handleDeviceAdded(self):
        print("deviceAdded ==> ", request.json)
#TODO deal with deviceAdded notification
        pass

接收到平台推送的消息后，开发者需要根据业务进行处理，但不建议进行复杂计算、I/O操作或者可能长时间等待的动作，可以先写数据库，应用进入相应界面或者刷新界面再从数据库取数据并进行数据处理。

回调路径已在SDK中设置好了，所以在订阅时要注意设置对应的回调地址。回调的IP地址则是服务器的地址，需要是公网地址，端口自行配置。

回调证书制作

本章节以自签名证书为例。如果是使用商用证书，请直接向CA机构申请。

注：每个步骤的创建证书请求中的配置，请自行设置。“Common Name (e.g. server FQDN or YOUR name) []:”要输入应用服务器的IP或域名。

  1. 打开windows命令行窗口，输入where openssl，查询是否安装openssl。

    1
    where openssl

  2. 创建测试CA证书。

    1. 创建私钥。
      1
      openssl genrsa -out ca-key.pem 1024
    2. 创建证书请求。
      1
      openssl req -new -out ca-req.csr -key ca-key.pem
    3. 自签署证书。
      1
      openssl x509 -req -in ca-req.csr -out ca-cert.pem -signkey ca-key.pem -days 3650

  3. 生成server证书。

    1. 创建私钥。
      1
      openssl genrsa -out server.key 1024
    2. 创建证书请求。
      1
      openssl req -new -out server-req.csr -key server.key
    3. 自签署证书。
      1
      openssl x509 -req -in server-req.csr -out server.cert -signkey server.key -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -days 3650

  4. 生成client证书。

    1. 创建私钥。
      1
      openssl genrsa -out client.key 1024
    2. 创建证书请求。
      1
      openssl req -new -out client-req.csr -key client.key
    3. 自签署证书。
      1
      openssl x509 -req -in client-req.csr -out client.cert -signkey client.key -CA ca-cert.pem -CAkey ca-key.pem -CAcreateserial -days 3650

  5. 在Demo工程中，已经配置好生成的server证书。
  6. 单击“启动”PushMessageReceiverTest，选择“Run” “PushMessageReceiverTest”。

    当有数据推送到应用服务器时，就会进入相应的回调函数中。

回调证书导出

  1. 使用浏览器打开回调地址https://server:8099/v1.0.0/messageReceiver，以Google为例，并查看证书。

    server是应用服务器的地址（即本机地址），8099是在application.properties中配置的端口。

  2. 系统将弹出证书窗口，选择“详细信息”，单击“复制到文件”。

  3. 单击“下一步”，进入“导出文件格式”界面，选择“Base64编码”，然后单击“下一步”。

  4. 指定证书的保存路径，完成证书导出。

    如果应用服务器最后部署到云上，可能会有多级证书，建议在部署完成后再导出证书。此时需要将证书链上面几级的证书一一导出。

    1. 在“要导出的文件”界面，单击“浏览”，选择一个路径，输入文件名，单击“保存”，回到证书导出向导，单击“下一步”。

    2. 单击“完成”，完成证书导出。

  5. 若存在多级证书，需一一导出。

    1. 在“证书”窗口，选择“证书路径”，查看多级证书，选中证书路径中的某个证书，单击“查看证书”。

    2. 系统将弹出证书窗口，选择“详细信息”，然后重复上述导出证书的步骤，导出已选证书。

  6. 使用文本编辑器，将所有导出的证书以首尾相连的方式，合并为一个.pem格式的文件。该文件需要上传到物联网平台相应的应用下。

回调证书上传

  1. 登录开发中心，进入相关项目。
  2. 选择“应用 > 对接信息”，单击“证书管理”。
  3. 单击“添加”，上传证书。

业务接口调用流程及注意事项

  • 请按如下流程调用业务接口。

  • Demo中使用的Profile如下图所示，只有一个Brightness服务，Brightness服务下有一个brightness属性和一个PUT命令。在调用创建设备命令或设备服务调用等接口时，如果不是使用以下Profile内容，请将相关服务、属性或者命令名称修改为相应的名称。

  • 创建新的Profile方法：

    登录开发者中心，选择“产品 > 产品开发 > 添加 > 自定义产品”，点击“自定义产品”，进入“设置产品信息”页面。填写“产品名称”、“型号”、“厂商ID”、“所属行业”、“设备类型”、“接入应用层协议类型”等产品信息，点击“创建”。然后点击“+新建服务”（根据设备功能添加属性和命令）最后点击“保存”。

  • 修改设备信息接口使用到的字段值如“设备类型”、“厂商名”、“厂商ID”、“设备型号”要与Profile定义的保持一致。
  • accessToken可以由SDK管理，也可由第三方应用自行管理，具体信息可参考《Python SDK API参考文档》中“应用安全接入 > 定时刷新token”章节的说明。
父主题： 使用SDK对接
分享：

    相关文档

    相关产品

close