I'm trying to get more comfortable working with Azure IOT and decided that I wanted to start sending my own telemetry data instead of the "tempSensor" demo that they have provided.
I created this "basic_device" module.
import time
import json
import iothub_client
# pylint: disable=E0611
from iothub_client import IoTHubClient, IoTHubTransportProvider, IoTHubMessage, IoTHubError
# Connection string and protocol set
CONNECTION_STRING = "REMOVED-FOR-POST"
PROTOCOL = IoTHubTransportProvider.MQTT
def set_sample_rate(sample_rate=0.1):
"""Creates a decorator that has the given sample_rate."""
def decorate_sample_rate(func):
"""The actual decorator."""
def wrapper(*args, **kwargs):
"""Wrapper method."""
fname = func.__name__
# If first time, use the last time
if fname not in args[0]._time:
args[0]._time[fname] = args[0]._time_last
# Check if it's time to add this message
if args[0]._time_last - args[0]._time[fname] >= sample_rate:
component_msg = func(*args, **kwargs)
for comp in component_msg:
args[1][comp] = component_msg[comp]
args[0]._time[fname] = args[0]._time_last
return args[1]
return wrapper
return decorate_sample_rate
def send_confirmation_callback(message, result, user_context):
"""Send confirmation upon sucessful message."""
print ( "IoT Hub responded to message with status: %s" % (result) )
class SimulateDevice():
def __init__(self, msg_max=100):
self._client = IoTHubClient(CONNECTION_STRING, PROTOCOL)
self._time_start = time.time()
self._time_last = time.time()
self._time = {}
self._msg_count = 0
self._msg_max = msg_max
@set_sample_rate(2)
def _noisy_wave_message(self, msg):
"""Create a message that is noisy."""
dt = time.time() - self._time_start
component_msg = {
'noisy_sinus': np.sin(dt / 100.0) + np.random.normal(0, 0.25, 1)[0],
'noisy_cosine': np.cos(dt / 100.0) + np.random.normal(0, 0.25, 1)[0],
}
return component_msg
def send_message(self):
"""Send a message."""
msg = self._noisy_wave_message({})
if msg and self._msg_count <= self._msg_max:
msg['timeStamp'] = self._time_last
msg_body = json.dumps(msg)
print("Sending message: %s" % msg_body)
iot_msg = IoTHubMessage(msg_body)
iot_msg.message_id = "message_%d" % self._msg_count
self._client.send_event_async(iot_msg, send_confirmation_callback, None)
self._msg_count +=1
self._time_last = time.time()
def start_device():
try:
device = SimulateDevice()
print ( "IoT Hub device sending periodic messages, press Ctrl-C to exit" )
while True:
# Send the message.
device.send_message()
except IoTHubError as iothub_error:
print ( "Unexpected error %s from IoTHub" % iothub_error )
return
except KeyboardInterrupt:
print ( "IoTHubClient sample stopped" )
if __name__ == '__main__':
print ( "IoT Hub Quickstart #1 - Simulated device" )
print ( "Press Ctrl-C to exit" )
start_device()
And when I check the logs, I see this
azureuser@EdgeVM:~$ sudo iotedge logs basic_device
IoT Hub Quickstart #1 - Simulated device
Press Ctrl-C to exit
IoT Hub device sending periodic messages, press Ctrl-C to exit
Sending message: {"noisy_sinus": -0.12927878622262406, "noisy_cosine": 0.5951663552778992, "timeStamp": 1542717185.0867708}
IoT Hub responded to message with status: OK
So it seems like it works right?
Then I wanted to create a module that subscribes to messages from this module. The module is rather simple, ie
import random
import time
import sys
import iothub_client
# pylint: disable=E0611
from iothub_client import IoTHubModuleClient, IoTHubTransportProvider
from iothub_client import IoTHubMessage, IoTHubMessageDispositionResult, IoTHubError
MESSAGE_TIMEOUT = 10000
PROTOCOL = IoTHubTransportProvider.MQTT
def noisy_sinus(message, user_context):
print("Received a message")
print(message)
return IoTHubMessageDispositionResult.ACCEPTED
class AnalysisManager():
"""A class that manages different analysis for differnet signals."""
def __init__(self, protocol=IoTHubTransportProvider.MQTT):
self.client_protocol = protocol
self.client = IoTHubModuleClient()
self.client.create_from_environment(protocol)
# set the time until a message times out
self.client.set_option("messageTimeout", MESSAGE_TIMEOUT)
# sets the callback when a message arrives on "input1" queue.
self.client.set_message_callback("input1", noisy_sinus, None)
def main(protocol):
try:
print ( "\nPython %s\n" % sys.version )
print ( "IoT Hub Client for Python" )
hub_manager = AnalysisManager(protocol)
print ( "Starting the IoT Hub Python sample using protocol %s..." % hub_manager.client_protocol )
print ( "The sample is now waiting for messages and will indefinitely. Press Ctrl-C to exit. ")
while True:
time.sleep(1)
except IoTHubError as iothub_error:
print ( "Unexpected error %s from IoTHub" % iothub_error )
return
except KeyboardInterrupt:
print ( "IoTHubModuleClient sample stopped" )
if __name__ == '__main__':
main(PROTOCOL)
And in the routing settings, I have the following
"routes": {
"basic_deviceToIoTHub": "FROM /messages/modules/basic_device/outputs/* INTO $upstream",
"basic_analysisToIoTHub": "FROM /messages/modules/basic_analysis/outputs/* INTO $upstream",
"sensorTobasic_analysis": "FROM /messages/modules/basic_device/outputs/* INTO BrokeredEndpoint(\"/modules/basic_analysis/inputs/input1\")"
},
But this one receives no messages at all.
azureuser@EdgeVM:~$ sudo iotedge logs basic_analysis
Python 3.5.2 (default, Nov 23 2017, 16:37:01)
[GCC 5.4.0 20160609]
IoT Hub Client for Python
Starting the IoT Hub Python sample using protocol MQTT...
The sample is now waiting for messages and will indefinitely. Press Ctrl-C to exit.
What am I missing here? I am able to receive messages from the tempSensor module if I want too.
Furthermore, in the demo, it is possible to name the message as temperatureOutput, however, in the Python API, there is no such option when creating a send_event_async.
Well, after some digging around I found a solution that worked.
In the basic_device module, I now use IoTHubModuleClient instead of IoTHubClient.
class SimulateDevice():
def __init__(self, protocol, msg_max=100):
self._client_protocol = protocol
self._client = IoTHubModuleClient()
self._client.create_from_environment(self._client_protocol)
Which then allows me to send a message like so
self._client.send_event_async("allmessages", iot_msg, send_confirmation_callback, None)
Then in the routing, I can use
"routes": {
"basic_deviceToIoTHub": "FROM /messages/modules/basic_device/outputs/* INTO $upstream",
"basic_analysisToIoTHub": "FROM /messages/modules/basic_analysis/outputs/* INTO $upstream",
"sensorTobasic_analysis": "FROM /messages/modules/basic_device/outputs/allmessages INTO BrokeredEndpoint(\"/modules/basic_analysis/inputs/input1\")"
},
Then the analysis module is able to receive the message. I'm unsure why using "*" did not work in the previous case.
Edit: Fix grammar