qwphr/ess-moxa-configuration-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added Direct Method capabilities

Browse Source 
used for global connectivity check
This commit is contained in:
Quentin WEPHRE
2024-06-21 09:57:00 +02:00
parent 24e471f633
commit 5819418f7c
1 changed files with 27 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
Python/azure_iot_hub_1.py
View File

@@ -1,7 +1,13 @@
from azure.iot.hub import IoTHubRegistryManager
from azure.iot.hub.protocol.models import QueryResult, QuerySpecification
from azure.iot.hub.protocol.models import QuerySpecification
from azure.iot.hub.models import CloudToDeviceMethod, CloudToDeviceMethodResult
import json
module_id = "thingspro-agent"
method_name = "thingspro-api-v1"
payload = '{"method":"GET", "path":"/device/general"}'
# Install the Azure IoT Hub SDK:
# pip install azure-iot-hub
@@ -10,15 +16,30 @@ CONNECTION_STRING = "HostName=IotHub-CUBE-PROD.azure-devices.net;SharedAccessKey
if CONNECTION_STRING == "":
print("Provide a connection string for the Iot Hub before running the script!")
exit(13)
registry_manager = IoTHubRegistryManager.from_connection_string(CONNECTION_STRING)
query_spec = QuerySpecification(query="SELECT * FROM devices WHERE tags.site = 'SASK'")
query_spec = QuerySpecification(query="SELECT * FROM devices WHERE tags.site != 'QWE' AND tags.number != '0' AND capabilities.iotEdge = true")
query_result = registry_manager.query_iot_hub(query_spec)
ordered_devices = []
devices = []
for item in query_result.items:
ordered_devices.append([int(item.tags['number']), item.tags['deviceId']])
devices.append([int(item.tags['number']), item.tags['deviceId'], item.tags['site']])
ordered_devices.sort()
ordered_devices = sorted(devices, key = lambda x: (x[2], x[0]))
for i in ordered_devices:
print(i)
current_device_modules = registry_manager.get_modules(i[1])
for module in current_device_modules:
if module.module_id == module_id:
thingspro_module = module
if thingspro_module:
#print("Found thingspro-agent for " + i[1] + " (" + i[2] + ")")
try:
direct_method = CloudToDeviceMethod(method_name=method_name, payload=json.loads(payload))
response = registry_manager.invoke_device_module_method(device_id=i[1], module_id=module_id, direct_method_request=direct_method)
print(str(i[2]), str(i[0]), str(i[1]), response.payload['data']['description'],sep=";")
except:
print(str(i[2]), str(i[0]), str(i[1]), "UNREACHABLE",sep=";")
else:
print("No thingspro-agent available for " + i[1] + " (" + i[2] + ")")