jepler/Adafruit_CircuitPython_AzureIoT
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Copying in code from the external repo

Browse source
This commit is contained in:
Jim Bennett 2020-04-11 14:27:32 -07:00
parent 74d5688d82
commit c384d4210f
23 changed files with 999 additions and 344 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -8,4 +8,5 @@ bundles
*.DS_Store
.eggs
dist
**/*.egg-info
**/*.egg-info
.vscode/settings.json

4
.pylintrc
View file

@ -217,7 +217,7 @@ indent-after-paren=4
indent-string=' '
# Maximum number of characters on a single line.
max-line-length=100
max-line-length=140
# Maximum number of lines in a module
max-module-lines=1000
@ -395,7 +395,7 @@ valid-metaclass-classmethod-first-arg=mcs
[DESIGN]
# Maximum number of arguments for function / method
max-args=5
max-args=6
# Maximum number of attributes for a class (see R0902).
# max-attributes=7

240
adafruit_azureiot.py
View file

@ -1,240 +0,0 @@
# The MIT License (MIT)
#
# Copyright (c) 2019 Brent Rubell for Adafruit Industries
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
"""
`adafruit_azureiot`
================================================================================
Microsoft Azure IoT for CircuitPython
* Author(s): Brent Rubell
Implementation Notes
--------------------
**Software and Dependencies:**
* Adafruit CircuitPython firmware for the supported boards:
https://github.com/adafruit/circuitpython/releases
* Adafruit's Bus Device library: https://github.com/adafruit/Adafruit_CircuitPython_BusDevice
* Adafruit's ESP32SPI library: https://github.com/adafruit/Adafruit_CircuitPython_ESP32SPI
"""
__version__ = "0.0.0-auto.0"
__repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_AzureIoT.git"
AZ_API_VER = "2018-06-30" # Azure URI API Version Identifier
AZURE_HTTP_ERROR_CODES = [400, 401, 404, 403, 412, 429, 500] # Azure HTTP Status Codes
class IOT_Hub:
"""
Provides access to a Microsoft Azure IoT Hub.
https://docs.microsoft.com/en-us/rest/api/iothub/
"""
def __init__(self, wifi_manager, iot_hub_name, sas_token, device_id):
""" Creates an instance of an Azure IoT Hub Client.
:param wifi_manager: WiFiManager object from ESPSPI_WiFiManager.
:param str iot_hub_name: Name of your IoT Hub.
:param str sas_token: Azure IoT Hub SAS Token Identifier.
:param str device_id: Unique Azure IoT Device Identifier.
"""
_wifi_type = str(type(wifi_manager))
if "ESPSPI_WiFiManager" in _wifi_type:
self._wifi = wifi_manager
else:
raise TypeError("This library requires a WiFiManager object.")
self._iot_hub_url = "https://{0}.azure-devices.net".format(iot_hub_name)
self._sas_token = sas_token
self._device_id = device_id
self._azure_header = {"Authorization": self._sas_token}
@property
def device_id(self):
"""Returns the current device identifier"""
return self._device_id
@device_id.setter
def device_id(self, device_identifier):
"""Sets the current device identifier
:param str device_identifier: Unique device identifier.
"""
self._device_id = device_identifier
@staticmethod
def _parse_http_status(status_code, status_reason):
"""Parses status code, throws error based on Azure IoT Common Error Codes.
:param int status_code: HTTP status code.
:param str status_reason: Description of HTTP status.
"""
for error in AZURE_HTTP_ERROR_CODES:
if error == status_code:
raise TypeError("Error {0}: {1}".format(status_code, status_reason))
# Cloud-to-Device Messaging
def get_hub_message(self):
"""Returns a message from a Microsoft Azure IoT Hub (Cloud-to-Device).
Returns None if the message queue is empty.
NOTE: HTTP Cloud-to-Device messages are throttled. Poll every 25+ minutes.
"""
reject_message = True
# get a device-bound notification
path = "{0}/devices/{1}/messages/deviceBound?api-version={2}".format(
self._iot_hub_url, self._device_id, AZ_API_VER
)
data = self._get(path, is_c2d=True)
if data == 204: # device's message queue is empty
return None
etag = data[1]["etag"]
if etag: # either complete or nack the message
reject_message = False
path_complete = "{0}/devices/{1}/messages/deviceBound/{2}?api-version={3}".format(
self._iot_hub_url, self._device_id, etag.strip("'\""), AZ_API_VER
)
if reject_message:
path_complete += "&reject"
del_status = self._delete(path_complete)
if del_status == 204:
return data[0]
return None
# Device-to-Cloud Messaging
def send_device_message(self, message):
"""Sends a device-to-cloud message.
:param string message: Message to send to Azure IoT.
"""
path = "{0}/devices/{1}/messages/events?api-version={2}".format(
self._iot_hub_url, self._device_id, AZ_API_VER
)
self._post(path, message, return_response=False)
# Device Twin
def get_device_twin(self):
"""Returns the device's device twin information in JSON format.
"""
path = "{0}/twins/{1}?api-version={2}".format(
self._iot_hub_url, self._device_id, AZ_API_VER
)
return self._get(path)
def update_device_twin(self, properties):
"""Updates tags and desired properties of the device's device twin.
:param str properties: Device Twin Properties
(https://docs.microsoft.com/en-us/rest/api/iothub/service/updatetwin#twinproperties)
"""
path = "{0}/twins/{1}?api-version={2}".format(
self._iot_hub_url, self._device_id, AZ_API_VER
)
return self._patch(path, properties)
def replace_device_twin(self, properties):
"""Replaces tags and desired properties of a device twin.
:param str properties: Device Twin Properties.
"""
path = "{0}/twins/{1}?api-version-{2}".format(
self._iot_hub_url, self._device_id, AZ_API_VER
)
return self._put(path, properties)
# IoT Hub Service
def get_devices(self):
"""Enumerate devices from the identity registry of the IoT Hub.
"""
path = "{0}/devices/?api-version={1}".format(self._iot_hub_url, AZ_API_VER)
return self._get(path)
def get_device(self):
"""Gets device information from the identity
registry of an IoT Hub.
"""
path = "{0}/devices/{1}?api-version={2}".format(
self._iot_hub_url, self._device_id, AZ_API_VER
)
return self._get(path)
# HTTP Helper Methods
def _post(self, path, payload, return_response=True):
"""HTTP POST
:param str path: Formatted Azure IOT Hub Path.
:param str payload: JSON-formatted Data Payload.
"""
response = self._wifi.post(path, json=payload, headers=self._azure_header)
self._parse_http_status(response.status_code, response.reason)
if return_response:
return response.json()
return response.text
def _get(self, path, is_c2d=False):
"""HTTP GET
:param str path: Formatted Azure IOT Hub Path.
:param bool is_c2d: Cloud-to-device get request.
"""
response = self._wifi.get(path, headers=self._azure_header)
status_code = response.status_code
if is_c2d:
if status_code == 200:
data = response.text
headers = response.headers
response.close()
return data, headers
response.close()
return status_code
json = response.json()
response.close()
return json
def _delete(self, path, etag=None):
"""HTTP DELETE
:param str path: Formatted Azure IOT Hub Path.
"""
if etag:
data_headers = {"Authorization": self._sas_token, "If-Match": '"%s"' % etag}
else:
data_headers = self._azure_header
response = self._wifi.delete(path, headers=data_headers)
self._parse_http_status(response.status_code, response.reason)
status_code = response.status_code
response.close()
return status_code
def _patch(self, path, payload):
"""HTTP PATCH
:param str path: Formatted Azure IOT Hub Path.
:param str payload: JSON-formatted payload.
"""
response = self._wifi.patch(path, json=payload, headers=self._azure_header)
self._parse_http_status(response.status_code, response.reason)
json_data = response.json()
response.close()
return json_data
def _put(self, path, payload=None):
"""HTTP PUT
:param str path: Formatted Azure IOT Hub Path.
:param str payload: JSON-formatted payload.
"""
response = self._wifi.put(path, json=payload, headers=self._azure_header)
self._parse_http_status(response.status_code, response.reason)
json_data = response.json()
response.close()
return json_data

46
adafruit_azureiot/__init__.py Normal file
View file

@ -0,0 +1,46 @@
# The MIT License (MIT)
#
# Copyright (c) 2019 Jim Bennett
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
"""
`adafruit_azureiot`
================================================================================
Microsoft Azure IoT for CircuitPython
* Author(s): Jim Bennett, Elena Horton
Implementation Notes
--------------------
**Software and Dependencies:**
* Adafruit CircuitPython firmware for the supported boards:
https://github.com/adafruit/circuitpython/releases
* Adafruit's Bus Device library: https://github.com/adafruit/Adafruit_CircuitPython_BusDevice
* Adafruit's ESP32SPI library: https://github.com/adafruit/Adafruit_CircuitPython_ESP32SPI
* Community HMAC library: https://github.com/jimbobbennett/CircuitPython_HMAC
* Community base64 library: https://github.com/jimbobbennett/CircuitPython_Base64
* Community Parse library: https://github.com/jimbobbennett/CircuitPython_Parse
"""
__version__ = "0.0.0-auto.0"
__repo__ = "https://github.com/adafruit/Adafruit_CircuitPython_AzureIoT.git"

6
adafruit_azureiot/constants.py Normal file
View file

@ -0,0 +1,6 @@
"""This file is for maintaining constants that could be changed or added to over time for different scenarios
"""
DPS_API_VERSION = "2018-11-01"
IOTC_API_VERSION = "2016-11-14"
DPS_END_POINT = "global.azure-devices-provisioning.net"

225
adafruit_azureiot/device_registration.py Normal file
View file

@ -0,0 +1,225 @@
"""
Device Registration
=====================
Handles registration of IoT Central devices, and gets the hostname to use when connecting
to IoT Central over MQTT
"""
import gc
import json
import time
import circuitpython_base64 as base64
import circuitpython_hmac as hmac
import circuitpython_parse as parse
from adafruit_esp32spi.adafruit_esp32spi_wifimanager import ESPSPI_WiFiManager
import adafruit_logging as logging
from adafruit_logging import Logger
import adafruit_hashlib as hashlib
from . import constants
AZURE_HTTP_ERROR_CODES = [400, 401, 404, 403, 412, 429, 500] # Azure HTTP Status Codes
class DeviceRegistrationError(Exception):
"""
An error from the device registration
"""
def __init__(self, message):
super(DeviceRegistrationError, self).__init__(message)
self.message = message
class DeviceRegistration:
"""
Handles registration of IoT Central devices, and gets the hostname to use when connecting
to IoT Central over MQTT
"""
_dps_endpoint = constants.DPS_END_POINT
_dps_api_version = constants.DPS_API_VERSION
_loop_interval = 2
@staticmethod
def _parse_http_status(status_code, status_reason):
"""Parses status code, throws error based on Azure IoT Common Error Codes.
:param int status_code: HTTP status code.
:param str status_reason: Description of HTTP status.
"""
for error in AZURE_HTTP_ERROR_CODES:
if error == status_code:
raise TypeError("Error {0}: {1}".format(status_code, status_reason))
def __init__(self, wifi_manager: ESPSPI_WiFiManager, id_scope: str, device_id: str, key: str, logger: Logger = None):
"""Creates an instance of the device registration
:param wifi_manager: WiFiManager object from ESPSPI_WiFiManager.
:param str id_scope: The ID scope of the device to register
:param str device_id: The device ID of the device to register
:param str key: The primary or secondary key of the device to register
:param adafruit_logging.Logger key: The primary or secondary key of the device to register
"""
wifi_type = str(type(wifi_manager))
if "ESPSPI_WiFiManager" not in wifi_type:
raise TypeError("This library requires a WiFiManager object.")
self._wifi_manager = wifi_manager
self._id_scope = id_scope
self._device_id = device_id
self._key = key
self._logger = logger if logger is not None else logging.getLogger("log")
@staticmethod
def compute_derived_symmetric_key(secret, reg_id):
"""Computes a derived symmetric key from a secret and a message
"""
secret = base64.b64decode(secret)
return base64.b64encode(hmac.new(secret, msg=reg_id.encode("utf8"), digestmod=hashlib.sha256).digest())
def _loop_assign(self, operation_id, headers) -> str:
uri = "https://%s/%s/registrations/%s/operations/%s?api-version=%s" % (
self._dps_endpoint,
self._id_scope,
self._device_id,
operation_id,
self._dps_api_version,
)
self._logger.info("- iotc :: _loop_assign :: " + uri)
target = parse.urlparse(uri)
response = self.__run_get_request_with_retry(target.geturl(), headers)
try:
data = response.json()
except Exception as error:
err = "ERROR: " + str(error) + " => " + str(response)
self._logger.error(err)
raise DeviceRegistrationError(err)
loop_try = 0
if data is not None and "status" in data:
if data["status"] == "assigning":
time.sleep(self._loop_interval)
if loop_try < 20:
loop_try = loop_try + 1
return self._loop_assign(operation_id, headers)
err = "ERROR: Unable to provision the device."
self._logger.error(err)
raise DeviceRegistrationError(err)
if data["status"] == "assigned":
state = data["registrationState"]
return state["assignedHub"]
else:
data = str(data)
err = "DPS L => " + str(data)
self._logger.error(err)
raise DeviceRegistrationError(err)
def __run_put_request_with_retry(self, url, body, headers):
retry = 0
response = None
while True:
gc.collect()
try:
self._logger.debug("Trying to send...")
response = self._wifi_manager.put(url, json=body, headers=headers)
self._logger.debug("Sent!")
break
except RuntimeError as runtime_error:
self._logger.info("Could not send data, retrying after 0.5 seconds: " + str(runtime_error))
retry = retry + 1
if retry >= 10:
self._logger.error("Failed to send data")
raise
time.sleep(0.5)
continue
gc.collect()
return response
def __run_get_request_with_retry(self, url, headers):
retry = 0
response = None
while True:
gc.collect()
try:
self._logger.debug("Trying to send...")
response = self._wifi_manager.get(url, headers=headers)
self._logger.debug("Sent!")
break
except RuntimeError as runtime_error:
self._logger.info("Could not send data, retrying after 0.5 seconds: " + str(runtime_error))
retry = retry + 1
if retry >= 10:
self._logger.error("Failed to send data")
raise
time.sleep(0.5)
continue
gc.collect()
return response
def register_device(self, expiry: int) -> str:
"""
Registers the device with the IoT Central device registration service.
Returns the hostname of the IoT hub to use over MQTT
:param str expiry: The expiry time
"""
# pylint: disable=c0103
sr = self._id_scope + "%2Fregistrations%2F" + self._device_id
sig_no_encode = DeviceRegistration.compute_derived_symmetric_key(self._key, sr + "\n" + str(expiry))
sig_encoded = parse.quote(sig_no_encode, "~()*!.'")
auth_string = "SharedAccessSignature sr=" + sr + "&sig=" + sig_encoded + "&se=" + str(expiry) + "&skn=registration"
headers = {
"content-type": "application/json; charset=utf-8",
"user-agent": "iot-central-client/1.0",
"Accept": "*/*",
}
if auth_string is not None:
headers["authorization"] = auth_string
body = {"registrationId": self._device_id}
uri = "https://%s/%s/registrations/%s/register?api-version=%s" % (
self._dps_endpoint,
self._id_scope,
self._device_id,
self._dps_api_version,
)
target = parse.urlparse(uri)
self._logger.info("Connecting...")
self._logger.info("URL: " + target.geturl())
self._logger.info("body: " + json.dumps(body))
print("headers: " + json.dumps(headers))
response = self.__run_put_request_with_retry(target.geturl(), body, headers)
data = None
try:
data = response.json()
except Exception as e:
err = "ERROR: non JSON is received from " + self._dps_endpoint + " => " + str(response) + " .. message : " + str(e)
self._logger.error(err)
raise DeviceRegistrationError(err)
if "errorCode" in data:
err = "DPS => " + str(data)
self._logger.error(err)
raise DeviceRegistrationError(err)
time.sleep(1)
return self._loop_assign(data["operationId"], headers)

13
adafruit_azureiot/iot_error.py Normal file
View file

@ -0,0 +1,13 @@
"""
An error from the IoT service
"""
class IoTError(Exception):
"""
An error from the IoT service
"""
def __init__(self, message):
super(IoTError, self).__init__(message)
self.message = message

403
adafruit_azureiot/iot_mqtt.py Normal file
View file

@ -0,0 +1,403 @@
"""MQTT client for Azure IoT
"""
import gc
import json
import time
from adafruit_minimqtt import MQTT
import circuitpython_parse as parse
from device_registration import DeviceRegistration
import adafruit_logging as logging
from . import constants
class IoTResponse:
"""A response from a direct method call
"""
def __init__(self, code, message):
self._code = code
self._message = message
def get_response_code(self):
"""Gets the method response code
"""
return self._code
def get_response_message(self):
"""Gets the method response message
"""
return self._message
class IoTMQTTCallback:
"""An interface for classes that can be called by MQTT events
"""
def message_sent(self, data) -> None:
"""Called when a message is sent to the cloud
"""
def connection_status_change(self, connected: bool) -> None:
"""Called when the connection status changes
"""
# pylint: disable=W0613, R0201
def direct_method_called(self, method_name: str, data) -> IoTResponse:
"""Called when a direct method is invoked
"""
return IoTResponse("", "")
# pylint: disable=C0103
def cloud_to_device_message_received(self, body: str, properties: dict) -> None:
"""Called when a cloud to device message is received
"""
def device_twin_desired_updated(self, desired_property_name: str, desired_property_value, desired_version: int) -> None:
"""Called when the device twin desired properties are updated
"""
def device_twin_reported_updated(self, reported_property_name: str, reported_property_value, reported_version: int) -> None:
"""Called when the device twin reported values are updated
"""
def settings_updated(self) -> None:
"""Called when the settings are updated
"""
# pylint: disable=R0902
class IoTMQTT:
"""MQTT client for Azure IoT
"""
_iotc_api_version = constants.IOTC_API_VERSION
def _gen_sas_token(self):
token_expiry = int(time.time() + self._token_expires)
uri = self._hostname + "%2Fdevices%2F" + self._device_id
signed_hmac_sha256 = DeviceRegistration.compute_derived_symmetric_key(self._key, uri + "\n" + str(token_expiry))
signature = parse.quote(signed_hmac_sha256, "~()*!.'")
if signature.endswith("\n"): # somewhere along the crypto chain a newline is inserted
signature = signature[:-1]
token = "SharedAccessSignature sr={}&sig={}&se={}".format(uri, signature, token_expiry)
return token
# Workaround for https://github.com/adafruit/Adafruit_CircuitPython_MiniMQTT/issues/25
def _try_create_mqtt_client(self, hostname):
self._mqtts = MQTT(
broker=hostname,
username=self._username,
password=self._passwd,
port=8883,
keep_alive=120,
is_ssl=True,
client_id=self._device_id,
log=True,
)
self._mqtts.logger.setLevel(logging.INFO)
# set actions to take throughout connection lifecycle
self._mqtts.on_connect = self._on_connect
self._mqtts.on_message = self._on_message
self._mqtts.on_log = self._on_log
self._mqtts.on_publish = self._on_publish
self._mqtts.on_disconnect = self._on_disconnect
# initiate the connection using the adafruit_minimqtt library
self._mqtts.last_will()
self._mqtts.connect()
def _create_mqtt_client(self):
try:
self._try_create_mqtt_client(self._hostname)
except ValueError:
# Workaround for https://github.com/adafruit/Adafruit_CircuitPython_MiniMQTT/issues/25
self._try_create_mqtt_client("https://" + self._hostname)
# pylint: disable=C0103, W0613
def _on_connect(self, client, userdata, _, rc):
self._logger.info("- iot_mqtt :: _on_connect :: rc = " + str(rc) + ", userdata = " + str(userdata))
if rc == 0:
self._mqtt_connected = True
self._auth_response_received = True
self._callback.connection_status_change(True)
# pylint: disable=C0103, W0613
def _on_log(self, client, userdata, level, buf):
self._logger.info("mqtt-log : " + buf)
if level <= 8:
self._logger.error("mqtt-log : " + buf)
def _on_disconnect(self, client, userdata, rc):
self._logger.info("- iot_mqtt :: _on_disconnect :: rc = " + str(rc))
self._auth_response_received = True
if rc == 5:
self._logger.error("on(disconnect) : Not authorized")
self.disconnect()
if rc == 1:
self._mqtt_connected = False
if rc != 5:
self._callback.connection_status_change(False)
def _on_publish(self, client, data, topic, msg_id):
self._logger.info("- iot_mqtt :: _on_publish :: " + str(data) + " on topic " + str(topic))
# pylint: disable=W0703
def _handle_device_twin_update(self, msg: str, topic: str):
self._logger.debug("- iot_mqtt :: _echo_desired :: " + topic)
twin = None
desired = None
print(msg)
try:
twin = json.loads(msg)
except Exception as e:
self._logger.error("ERROR: JSON parse for Device Twin message object has failed. => " + msg + " => " + str(e))
return
if "reported" in twin:
reported = twin["reported"]
if "$version" in reported:
reported_version = reported["$version"]
reported.pop("$version")
else:
self._logger.error("ERROR: Unexpected payload for reported twin update => " + msg)
return
for property_name, value in reported.items():
self._callback.device_twin_reported_updated(property_name, value, reported_version)
is_patch = "desired" not in twin
if is_patch:
desired = twin
else:
desired = twin["desired"]
if "$version" in desired:
desired_version = desired["$version"]
desired.pop("$version")
else:
self._logger.error("ERROR: Unexpected payload for desired twin update => " + msg)
return
for property_name, value in desired.items():
self._callback.device_twin_desired_updated(property_name, value, desired_version)
def _handle_direct_method(self, msg: str, topic: str):
index = topic.find("$rid=")
method_id = 1
method_name = "None"
if index == -1:
self._logger.error("ERROR: C2D doesn't include topic id")
else:
method_id = topic[index + 5 :]
topic_template = "$iothub/methods/POST/"
len_temp = len(topic_template)
method_name = topic[len_temp : topic.find("/", len_temp + 1)]
ret = self._callback.direct_method_called(method_name, msg)
ret_code = 200
ret_message = "{}"
if ret.get_response_code() is not None:
ret_code = ret.get_response_code()
if ret.get_response_message() is not None:
ret_message = ret.get_response_message()
# ret message must be JSON
if not ret_message.startswith("{") or not ret_message.endswith("}"):
ret_json = {"Value": ret_message}
ret_message = json.dumps(ret_json)
next_topic = "$iothub/methods/res/{}/?$rid={}".format(ret_code, method_id)
self._logger.info("C2D: => " + next_topic + " with data " + ret_message + " and name => " + method_name)
self._send_common(next_topic, ret_message)
def _handle_cloud_to_device_message(self, msg: str, topic: str):
parts = topic.split("&")[1:]
properties = {}
for part in parts:
key_value = part.split("=")
properties[key_value[0]] = key_value[1]
self._callback.cloud_to_device_message_received(msg, properties)
# pylint: disable=W0702, R0912
def _on_message(self, client, msg_topic, payload):
topic = ""
msg = None
print("Topic: ", str(msg_topic))
self._logger.info("- iot_mqtt :: _on_message :: payload(" + str(payload) + ")")
if payload is not None:
try:
msg = payload.decode("utf-8")
except:
msg = str(payload)
if msg_topic is not None:
try:
topic = msg_topic.decode("utf-8")
except:
topic = str(msg_topic)
if topic.startswith("$iothub/"):
if topic.startswith("$iothub/twin/PATCH/properties/desired/") or topic.startswith("$iothub/twin/res/200/?$rid="):
self._handle_device_twin_update(str(msg), topic)
elif topic.startswith("$iothub/methods"):
self._handle_direct_method(str(msg), topic)
else:
if not topic.startswith("$iothub/twin/res/"): # not twin response
self._logger.error("ERROR: unknown twin! - {}".format(msg))
elif topic.startswith("devices/{}/messages/devicebound".format(self._device_id)):
self._handle_cloud_to_device_message(str(msg), topic)
else:
self._logger.error("ERROR: (unknown message) - {}".format(msg))
def _send_common(self, topic, data) -> None:
self._logger.debug("Sending message on topic: " + topic)
self._logger.debug("Sending message: " + str(data))
retry = 0
while True:
gc.collect()
try:
self._logger.debug("Trying to send...")
self._mqtts.publish(topic, data)
self._logger.debug("Data sent")
break
except RuntimeError as runtime_error:
self._logger.info("Could not send data, retrying after 0.5 seconds: " + str(runtime_error))
retry = retry + 1
if retry >= 10:
self._logger.error("Failed to send data")
raise
time.sleep(0.5)
continue
print("finished _send_common")
gc.collect()
def _get_device_settings(self) -> None:
self._logger.info("- iot_mqtt :: _get_device_settings :: ")
self.loop()
self._send_common("$iothub/twin/GET/?$rid=0", " ")
# pylint: disable=R0913
def __init__(
self, callback: IoTMQTTCallback, hostname: str, device_id: str, key: str, token_expires: int = 21600, logger: logging = None
):
"""Create the Azure IoT MQTT client
:param IoTMQTTCallback callback: A callback class
:param str hostname: The hostname of the MQTT broker to connect to, get this by registering the device
:param str device_id: The device ID of the device to register
:param str key: The primary or secondary key of the device to register
:param int token_expires: The number of seconds till the token expires, defaults to 6 hours
:param adafruit_logging logger: The logger
"""
self._callback = callback
self._mqtt_connected = False
self._auth_response_received = False
self._mqtts = None
self._device_id = device_id
self._hostname = hostname
self._key = key
self._token_expires = token_expires
self._username = "{}/{}/api-version={}".format(self._hostname, device_id, self._iotc_api_version)
self._passwd = self._gen_sas_token()
self._logger = logger if logger is not None else logging.getLogger("log")
def connect(self):
"""Connects to the MQTT broker
"""
self._logger.info("- iot_mqtt :: connect :: " + self._hostname)
self._create_mqtt_client()
self._logger.info(" - iot_mqtt :: connect :: created mqtt client. connecting..")
while self._auth_response_received is None:
self.loop()
self._logger.info(" - iot_mqtt :: connect :: on_connect must be fired. Connected ? " + str(self.is_connected()))
if not self.is_connected():
return 1
self._mqtt_connected = True
self._auth_response_received = True
self._mqtts.subscribe("devices/{}/messages/events/#".format(self._device_id))
self._mqtts.subscribe("devices/{}/messages/devicebound/#".format(self._device_id))
self._mqtts.subscribe("$iothub/twin/PATCH/properties/desired/#") # twin desired property changes
self._mqtts.subscribe("$iothub/twin/res/#") # twin properties response
self._mqtts.subscribe("$iothub/methods/#")
if self._get_device_settings() == 0:
self._callback.settings_updated()
else:
return 1
return 0
def disconnect(self):
"""Disconnects from the MQTT broker
"""
if not self.is_connected():
return
self._logger.info("- iot_mqtt :: disconnect :: ")
self._mqtt_connected = False
self._mqtts.disconnect()
def is_connected(self):
"""Gets if there is an open connection to the MQTT broker
"""
return self._mqtt_connected
def loop(self):
"""Listens for MQTT messages
"""
if not self.is_connected():
return
self._mqtts.loop()
def _send_common(self, topic, data):
self._mqtts.publish(topic, data)
def send_device_to_cloud_message(self, data, system_properties=None) -> None:
"""Send a device to cloud message from this device to Azure IoT Hub
"""
self._logger.info("- iot_mqtt :: send_device_to_cloud_message :: " + data)
topic = "devices/{}/messages/events/".format(self._device_id)
if system_properties is not None:
firstProp = True
for prop in system_properties:
if not firstProp:
topic += "&"
else:
firstProp = False
topic += prop + "=" + str(system_properties[prop])
self._send_common(topic, data)
self._callback.message_sent(data)
def send_twin_patch(self, data):
"""Send a patch for the reported properties of the device twin
"""
self._logger.info("- iot_mqtt :: sendProperty :: " + data)
topic = "$iothub/twin/PATCH/properties/reported/?$rid={}".format(int(time.time()))
return self._send_common(topic, data)

123
adafruit_azureiot/iotcentral_device.py Normal file
View file

@ -0,0 +1,123 @@
"""Connectivity to Azure IoT Central
"""
import json
import time
from adafruit_esp32spi.adafruit_esp32spi_wifimanager import ESPSPI_WiFiManager
from device_registration import DeviceRegistration
from iot_error import IoTError
from iot_mqtt import IoTMQTT, IoTMQTTCallback, IoTResponse
import adafruit_logging as logging
class IoTCentralDevice(IoTMQTTCallback):
"""A device client for the Azure IoT Central service
"""
def connection_status_change(self, connected: bool) -> None:
"""Called when the connection status changes
"""
if self.on_connection_status_changed is not None:
# pylint: disable=E1102
self.on_connection_status_changed(connected)
# pylint: disable=W0613, R0201
def direct_method_called(self, method_name: str, data) -> IoTResponse:
"""Called when a direct method is invoked
"""
if self.on_command_executed is not None:
# pylint: disable=E1102
return self.on_command_executed(method_name, data)
raise IoTError("on_command_executed not set")
def device_twin_desired_updated(self, desired_property_name: str, desired_property_value, desired_version: int) -> None:
"""Called when the device twin is updated
"""
if self.on_property_changed is not None:
# pylint: disable=E1102
self.on_property_changed(desired_property_name, desired_property_value, desired_version)
# when a desired property changes, update the reported to match to keep them in sync
self.send_property(desired_property_name, desired_property_value)
def device_twin_reported_updated(self, reported_property_name: str, reported_property_value, reported_version: int) -> None:
"""Called when the device twin is updated
"""
if self.on_property_changed is not None:
# pylint: disable=E1102
self.on_property_changed(reported_property_name, reported_property_value, reported_version)
# pylint: disable=R0913
def __init__(
self, wifi_manager: ESPSPI_WiFiManager, id_scope: str, device_id: str, key: str, token_expires: int = 21600, logger: logging = None
):
super(IoTCentralDevice, self).__init__()
self._wifi_manager = wifi_manager
self._id_scope = id_scope
self._device_id = device_id
self._key = key
self._token_expires = token_expires
self._logger = logger
self._device_registration = None
self._mqtt = None
self.on_connection_status_changed = None
self.on_command_executed = None
self.on_property_changed = None
def connect(self):
"""Connects to Azure IoT Central
"""
self._device_registration = DeviceRegistration(self._wifi_manager, self._id_scope, self._device_id, self._key, self._logger)
token_expiry = int(time.time() + self._token_expires)
hostname = self._device_registration.register_device(token_expiry)
self._mqtt = IoTMQTT(self, hostname, self._device_id, self._key, self._token_expires, self._logger)
self._mqtt.connect()
def disconnect(self):
"""Disconnects from the MQTT broker
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
self._mqtt.disconnect()
def is_connected(self) -> bool:
"""Gets if there is an open connection to the MQTT broker
"""
if self._mqtt is not None:
return self._mqtt.is_connected()
return False
def loop(self):
"""Listens for MQTT messages
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
self._mqtt.loop()
def send_property(self, property_name, data):
"""Updates the value of a writable property
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
patch_json = {property_name: data}
patch = json.dumps(patch_json)
self._mqtt.send_twin_patch(patch)
def send_telemetry(self, data):
"""Sends telemetry to the IoT Central app
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
if isinstance(data, dict):
data = json.dumps(data)
self._mqtt.send_device_to_cloud_message(data)

171
adafruit_azureiot/iothub_device.py Executable file
View file

@ -0,0 +1,171 @@
"""Connectivity to Azure IoT Hub
"""
import json
from iot_error import IoTError
from iot_mqtt import IoTMQTT, IoTMQTTCallback, IoTResponse
import adafruit_logging as logging
def _validate_keys(connection_string_parts):
"""Raise ValueError if incorrect combination of keys
"""
host_name = connection_string_parts.get(HOST_NAME)
shared_access_key_name = connection_string_parts.get(SHARED_ACCESS_KEY_NAME)
shared_access_key = connection_string_parts.get(SHARED_ACCESS_KEY)
device_id = connection_string_parts.get(DEVICE_ID)
if host_name and device_id and shared_access_key:
pass
elif host_name and shared_access_key and shared_access_key_name:
pass
else:
raise ValueError("Invalid Connection String - Incomplete")
DELIMITER = ";"
VALUE_SEPARATOR = "="
HOST_NAME = "HostName"
SHARED_ACCESS_KEY_NAME = "SharedAccessKeyName"
SHARED_ACCESS_KEY = "SharedAccessKey"
SHARED_ACCESS_SIGNATURE = "SharedAccessSignature"
DEVICE_ID = "DeviceId"
MODULE_ID = "ModuleId"
GATEWAY_HOST_NAME = "GatewayHostName"
VALID_KEYS = [
HOST_NAME,
SHARED_ACCESS_KEY_NAME,
SHARED_ACCESS_KEY,
SHARED_ACCESS_SIGNATURE,
DEVICE_ID,
MODULE_ID,
GATEWAY_HOST_NAME,
]
class IoTHubDevice(IoTMQTTCallback):
"""A device client for the Azure IoT Hub service
"""
def connection_status_change(self, connected: bool) -> None:
"""Called when the connection status changes
"""
if self.on_connection_status_changed is not None:
# pylint: disable=E1102
self.on_connection_status_changed(connected)
# pylint: disable=W0613, R0201
def direct_method_called(self, method_name: str, data) -> IoTResponse:
"""Called when a direct method is invoked
"""
if self.on_direct_method_called is not None:
# pylint: disable=E1102
return self.on_direct_method_called(method_name, data)
raise IoTError("on_direct_method_called not set")
# pylint: disable=C0103
def cloud_to_device_message_received(self, body: str, properties: dict):
"""Called when a cloud to device message is received
"""
if self.on_cloud_to_device_message_received is not None:
# pylint: disable=E1102
self.on_cloud_to_device_message_received(body, properties)
def device_twin_desired_updated(self, desired_property_name: str, desired_property_value, desired_version: int) -> None:
"""Called when the device twin is updated
"""
if self.on_device_twin_desired_updated is not None:
# pylint: disable=E1102
self.on_device_twin_desired_updated(desired_property_name, desired_property_value, desired_version)
def device_twin_reported_updated(self, reported_property_name: str, reported_property_value, reported_version: int) -> None:
"""Called when the device twin is updated
"""
if self.on_device_twin_reported_updated is not None:
# pylint: disable=E1102
self.on_device_twin_reported_updated(reported_property_name, reported_property_value, reported_version)
def __init__(self, device_connection_string: str, token_expires: int = 21600, logger: logging = None):
self._token_expires = token_expires
self._logger = logger if logger is not None else logging.getLogger("log")
connection_string_values = {}
try:
cs_args = device_connection_string.split(DELIMITER)
connection_string_values = dict(arg.split(VALUE_SEPARATOR, 1) for arg in cs_args)
except (ValueError, AttributeError):
raise ValueError("Connection string is required and should not be empty or blank and must be supplied as a string")
if len(cs_args) != len(connection_string_values):
raise ValueError("Invalid Connection String - Unable to parse")
_validate_keys(connection_string_values)
self._hostname = connection_string_values[HOST_NAME]
self._device_id = connection_string_values[DEVICE_ID]
self._shared_access_key = connection_string_values[SHARED_ACCESS_KEY]
self._logger.debug("Hostname: " + self._hostname)
self._logger.debug("Device Id: " + self._device_id)
self._logger.debug("Shared Access Key: " + self._shared_access_key)
self.on_connection_status_changed = None
self.on_direct_method_called = None
self.on_cloud_to_device_message_received = None
self.on_device_twin_desired_updated = None
self.on_device_twin_reported_updated = None
self._mqtt = None
def connect(self):
"""Connects to Azure IoT Central
"""
self._mqtt = IoTMQTT(self, self._hostname, self._device_id, self._shared_access_key, self._token_expires, self._logger)
self._mqtt.connect()
def disconnect(self):
"""Disconnects from the MQTT broker
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
self._mqtt.disconnect()
def is_connected(self) -> bool:
"""Gets if there is an open connection to the MQTT broker
"""
if self._mqtt is not None:
return self._mqtt.is_connected()
return False
def loop(self):
"""Listens for MQTT messages
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
self._mqtt.loop()
def send_device_to_cloud_message(self, message, system_properties=None):
"""Sends a device to cloud message to the IoT Hub
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
self._mqtt.send_device_to_cloud_message(message, system_properties)
def update_twin(self, patch):
"""Updates the reported properties in the devices device twin
"""
if self._mqtt is None:
raise IoTError("You are not connected to IoT Central")
if isinstance(patch, dict):
patch = json.dumps(patch)
self._mqtt.send_twin_patch(patch)

2
docs/conf.py
View file

@ -21,7 +21,7 @@ extensions = [
# Uncomment the below if you use native CircuitPython modules such as
# digitalio, micropython and busio. List the modules you use. Without it, the
# autodoc module docs will fail to generate with a warning.
# autodoc_mock_imports = ["digitalio", "busio"]
autodoc_mock_imports = ["adafruit_loging"]
intersphinx_mapping = {

44
examples/azureiot_device_management.py
View file

@ -1,44 +0,0 @@
import board
import busio
from digitalio import DigitalInOut
from adafruit_esp32spi import adafruit_esp32spi, adafruit_esp32spi_wifimanager
import neopixel
from adafruit_azureiot import IOT_Hub
# Get wifi details and more from a secrets.py file
try:
from secrets import secrets
except ImportError:
print("WiFi secrets are kept in secrets.py, please add them there!")
raise
# ESP32 Setup
try:
esp32_cs = DigitalInOut(board.ESP_CS)
esp32_ready = DigitalInOut(board.ESP_BUSY)
esp32_reset = DigitalInOut(board.ESP_RESET)
except AttributeError:
esp32_cs = DigitalInOut(board.D9)
esp32_ready = DigitalInOut(board.D10)
esp32_reset = DigitalInOut(board.D5)
spi = busio.SPI(board.SCK, board.MOSI, board.MISO)
esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
status_light = neopixel.NeoPixel(
board.NEOPIXEL, 1, brightness=0.2
) # Uncomment for Most Boards
"""Uncomment below for ItsyBitsy M4"""
# status_light = dotstar.DotStar(board.APA102_SCK, board.APA102_MOSI, 1, brightness=0.2)
wifi = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(esp, secrets, status_light)
# Create an instance of the Azure IoT Hub
hub = IOT_Hub(
wifi, secrets["azure_iot_hub"], secrets["azure_iot_sas"], secrets["device_id"]
)
# Enumerate all devices on an Azure IoT Hub
all_hub_devices = hub.get_devices()
print(all_hub_devices)
# Get a specified device on an Azure IoT Hub
device_data = hub.get_device()
print(device_data)

53
examples/azureiot_devicetwin.py
View file

@ -1,53 +0,0 @@
import board
import busio
from digitalio import DigitalInOut
from adafruit_esp32spi import adafruit_esp32spi, adafruit_esp32spi_wifimanager
import neopixel
from adafruit_azureiot import IOT_Hub
# Get wifi details and more from a secrets.py file
try:
from secrets import secrets
except ImportError:
print("WiFi secrets are kept in secrets.py, please add them there!")
raise
# ESP32 Setup
try:
esp32_cs = DigitalInOut(board.ESP_CS)
esp32_ready = DigitalInOut(board.ESP_BUSY)
esp32_reset = DigitalInOut(board.ESP_RESET)
except AttributeError:
esp32_cs = DigitalInOut(board.D9)
esp32_ready = DigitalInOut(board.D10)
esp32_reset = DigitalInOut(board.D5)
spi = busio.SPI(board.SCK, board.MOSI, board.MISO)
esp = adafruit_esp32spi.ESP_SPIcontrol(spi, esp32_cs, esp32_ready, esp32_reset)
status_light = neopixel.NeoPixel(
board.NEOPIXEL, 1, brightness=0.2
) # Uncomment for Most Boards
"""Uncomment below for ItsyBitsy M4"""
# status_light = dotstar.DotStar(board.APA102_SCK, board.APA102_MOSI, 1, brightness=0.2)
wifi = adafruit_esp32spi_wifimanager.ESPSPI_WiFiManager(esp, secrets, status_light)
# Create an instance of the Azure IoT Hub
hub = IOT_Hub(
wifi, secrets["azure_iot_hub"], secrets["azure_iot_sas"], secrets["device_id"]
)
# Get a Device Twin
device_twin = hub.get_device_twin()
# Filter out the device's name from the twin's properties
device_name = device_twin["properties"]["desired"]["deviceName"]
print(device_name)
# Update a Device Twin's Properties
data = {
"properties": {"desired": {"deviceName": "{{BasementTemperatureLoggerFeather}}"}}
}
hub.update_device_twin(data)
# And read the updated device twin information
device_twin = hub.get_device_twin()
device_name = device_twin["properties"]["desired"]["deviceName"]
print(device_name)

0
examples/iotcentral_commands.py Normal file
View file

0
examples/iotcentral_properties.py Normal file
View file

0
examples/azureiot_simpletest.py → examples/iotcentral_simpletest.py
View file

0
examples/iotcentral_telemetry.py Normal file
View file

0
examples/iothub_directmethods.py Normal file
View file

0
examples/iothub_messages.py Normal file
View file

0
examples/iothub_simpletest.py Normal file
View file

0
examples/iothub_twin_operations.py Normal file
View file

6
requirements.txt
View file

@ -1,2 +1,6 @@
Adafruit-Blinka
Adafruit_CircuitPython_ESP32SPI
Adafruit_CircuitPython_ESP32SPI
Adafruit-CircuitPython-miniMQTT
CircuitPython-HMAC
CircuitPython-Base64
CircuitPython-Parse

4
setup.py
View file

@ -29,7 +29,7 @@ setup(
# Author details
author="Adafruit Industries",
author_email="circuitpython@adafruit.com",
install_requires=["Adafruit-Blinka", "Adafruit_CircuitPython_ESP32SPI"],
install_requires=["Adafruit-Blinka", "Adafruit_CircuitPython_ESP32SPI", "Adafruit-CircuitPython-miniMQTT", "CircuitPython-HMAC", "CircuitPython-Base64", "CircuitPython-Parse"],
# Choose your license
license="MIT",
# See https://pypi.python.org/pypi?%3Aaction=list_classifiers
@ -44,7 +44,7 @@ setup(
"Programming Language :: Python :: 3.5",
],
# What does your project relate to?
keywords="adafruit blinka circuitpython micropython azureiot azure, iot, device, services",
keywords="adafruit blinka circuitpython micropython azureiot azure iot device services, iothub, iotcentral",
# You can just specify the packages manually here if your project is
# simple. Or you can use find_packages().
# TODO: IF LIBRARY FILES ARE A PACKAGE FOLDER,