Я пытаюсь сделать работу с Azure IOT более удобной и решила, что хочу начать отправку своих собственных данных телеметрии вместо предоставленной ими демонстрационной программы "tempSensor".
Я создал это "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()
И когда я проверяю логи, я вижу это
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
Так что, похоже, все работает правильно?
Тогда я хотел создатьмодуль, который подписывается на сообщения от этого модуля.Модуль довольно прост, то есть
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)
И в настройках маршрутизации у меня есть следующее
"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\")"
},
Но этот получает нет сообщений вообще.
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.
Что мне здесь не хватает?Я могу получать сообщения от модуля tempSensor, если я тоже этого хочу.
Кроме того, в демоверсии можно назвать сообщение как TemperatureOutput , однако в Python API,при создании send_event_async такой опции нет.