IoT Platform:Python 2.7 SDK を使用してクライアントを IoT Platform に接続する

開発環境

この例では Python 2.7 を使用します。

SDK のダウンロード

Python API をカプセル化する Apache Qpid Proton 0.29.0 ライブラリを推奨します。ライブラリのダウンロードと手順については、Qpid Proton 0.29.0 をご参照ください。

Qpid Proton をインストールします。詳細については、「Qpid Proton のインストール」をご参照ください。

Qpid Proton をインストールした後、次の Python コマンドを実行して、SSL ライブラリが利用可能かどうかを確認します。

import proton;print('%s' % 'SSL present' if proton.SSL.present() else 'SSL NOT AVAILABLE')

サンプルコード

# encoding=utf-8
import sys
import logging
import time
from proton.handlers import MessagingHandler
from proton.reactor import Container
import hashlib
import hmac
import base64
import os
reload(sys)
sys.setdefaultencoding('utf-8')
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
console_handler = logging.StreamHandler(sys.stdout)
def current_time_millis():
    return str(int(round(time.time() * 1000)))
def do_sign(secret, sign_content):
    m = hmac.new(secret, sign_content, digestmod=hashlib.sha1)
    return base64.b64encode(m.digest())
class AmqpClient(MessagingHandler):
    def __init__(self):
        super(AmqpClient, self).__init__()
    def on_start(self, event):
        # IoT Platform への接続に使用するエンドポイント。詳細は、AMQP 接続ガイドをご参照ください。
        url = "amqps://${YourHost}:5671"
        # プロジェクトコードに AccessKey ペアをハードコーディングすると、セキュリティリスクが生じます。コードが漏洩した場合、AccessKey ペアが公開され、アカウント内のすべてのリソースが危険に晒されます。次のコードは、ベストプラクティスとして環境変数から AccessKey ペアを取得します。これは参考用です。
        accessKey = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
        accessSecret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        consumerGroupId = "${YourConsumerGroupId}"
        clientId = "${YourClientId}"
        # iotInstanceId：IoT Platform インスタンスの ID。
        iotInstanceId = "${YourIotInstanceId}"
        # 署名アルゴリズム。有効な値：hmacmd5, hmacsha1, hmacsha256。
        signMethod = "hmacsha1"
        timestamp = current_time_millis()
        # userName パラメーターの作成方法については、AMQP 接続ガイドをご参照ください。
        userName = clientId + "|authMode=aksign" + ",signMethod=" + signMethod \
                        + ",timestamp=" + timestamp + ",authId=" + accessKey \
                        + ",iotInstanceId=" + iotInstanceId + ",consumerGroupId=" + consumerGroupId + "|"
        signContent = "authId=" + accessKey + "&timestamp=" + timestamp
        # 署名を計算します。パスワードの作成方法については、AMQP 接続ガイドをご参照ください。
        passWord = do_sign(accessSecret.encode("utf-8"), signContent.encode("utf-8"))
        conn = event.container.connect(url, user=userName, password=passWord, heartbeat=60)
        self.receiver = event.container.create_receiver(conn)
    # 接続が正常に確立されたときに呼び出されます。
    def on_connection_opened(self, event):
        logger.info("Connection established, remoteUrl: %s", event.connection.hostname)
    # 接続が閉じられたときに呼び出されます。
    def on_connection_closed(self, event):
        logger.info("Connection closed: %s", self)
    # エラーによりリモートピアが接続を閉じたときに呼び出されます。
    def on_connection_error(self, event):
        logger.info("Connection error")
    # 認証エラーやソケットエラーなど、AMQP 接続エラーが発生したときに呼び出されます。
    def on_transport_error(self, event):
        if event.transport.condition:
            if event.transport.condition.info:
                logger.error("%s: %s: %s" % (
                    event.transport.condition.name, event.transport.condition.description,
                    event.transport.condition.info))
            else:
                logger.error("%s: %s" % (event.transport.condition.name, event.transport.condition.description))
        else:
            logging.error("Unspecified transport error")
    # メッセージが受信されたときに呼び出されます。
    def on_message(self, event):
        message = event.message
        content = message.body.decode('utf-8')
        topic = message.properties.get("topic")
        message_id = message.properties.get("messageId")
        print("receive message: message_id=%s, topic=%s, content=%s" % (message_id, topic, content))
        event.receiver.flow(1)
Container(AmqpClient()).run()

上記のコードのパラメーターを、次の表の説明に従って設定します。詳細については、「AMQP クライアントの IoT Platform への接続」をご参照ください。

実行結果

• 成功：次のログメッセージは、AMQP クライアントが IoT Platform に接続し、メッセージを受信していることを示します。

  202x-xx-xx xx:xx:xx,490 - __main__ - INFO - Connection established, remoteUrl: xxx.amqp.iothub.aliyuncs.com
  receive message: message_id=13xxx, topic=/xxx/thing/event/property/post, content={"deviceType":"CustomCategory","iotId":"xxx","requestId":"1xxx"}
  receive message: message_id=13xxx, topic=/xxx/thing/event/property/post, content={"deviceType":"CustomCategory","iotId":"xxx","requestId":"1xxx"}

  パラメーター	例	説明
  message_id	13xxx	メッセージの ID。
  topic	/***********/******/thing/event/property/post	デバイスのプロパティを送信するために使用されるトピック。
  content	{"deviceType":"CustomCategory","iotId":"qPi*","requestId":"161*","checkFailedData":{},"productKey":"g4*","gmtCreate":1613635594038,"deviceName":"de*","items":{"Temperature":{"value":24,"time":1613635594036},"Humidity":{"value":26,"time":1613635594036}}}	メッセージの内容。

• 次のような情報が出力された場合、AMQP クライアントは IoT Platform への接続に失敗しています。

  ログに基づいてコードまたはネットワーク環境を確認し、問題を解決してから、コードを再度実行できます。

  2021-02-18 16:19:40,993 - proton - ERROR - Couldn't connect: 10053
  2021-02-18 16:19:40,996 - __main__ - ERROR - proton.pythonio: Connection error: 10053
  2021-02-18 16:19:40,996 - proton - INFO - Disconnected, reconnecting...
  Traceback (most recent call last):
    File "xxx", line 87, in <module>
      Container(AmqpClient()).run()
    File "xxx", line 181, in run
      while self.process(): pass
    File "xxx", line 240, in process
      event.dispatch(self._global_handler)
    File "xxx", line 135, in dispatch
      _dispatch(handler, type.method, self)
    File "xxx", line 117, in _dispatch
      handler.on_unhandled(method, *args)
    File "xxx", line 665, in on_unhandled
      event.dispatch(self.base)
    File "xxx", line 135, in dispatch
      _dispatch(handler, type.method, self)
    File "xxx", line 115, in _dispatch
      m(*args)
    File "xxx", line 892, in on_connection_bound
      addrs = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM)
  socket.gaierror: [Errno 11001] getaddrinfo failed

関連情報

サーバー側サブスクリプション機能に関連するエラーコードの詳細については、「IoT Platform ログ」トピックのメッセージに関連するエラーコードセクションをご参照ください。