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

Merge branch 'main' of github.com:TotalEnergiesCode/ess-moxa-configuration-tools

Browse Source
This commit is contained in:
Quentin WEPHRE
2025-09-17 14:43:52 +02:00
parent 07e234e901 a277c70fda
commit 56fceb7542
4 changed files with 8 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Python/azure_iot_hub_get_connection_strings.py
View File

@@ -15,7 +15,7 @@ if CONNECTION_STRING == "":
print("Provide a connection string for the Iot Hub before running the script!")
exit(13)
SITE_NAME = "COTTONWOOD"
SITE_NAME = "LIBERTY"
registry_manager = IoTHubRegistryManager.from_connection_string(CONNECTION_STRING)

26
Python/azure_iot_hub_list_devices.py
View File

@@ -13,11 +13,6 @@ 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
# Authenticate to your Azure account
# CONNECTION_STRING = str(os.getenv("CONNECTION_STRING_SAFT_PROD"))
CONNECTION_STRING = str(os.getenv("CONNECTION_STRING_INOX_PROD"))
if CONNECTION_STRING == "":
print("Provide a connection string for the Iot Hub before running the script!")
@@ -25,27 +20,10 @@ if CONNECTION_STRING == "":
registry_manager = IoTHubRegistryManager.from_connection_string(CONNECTION_STRING)
query_spec = QuerySpecification(query="SELECT * FROM devices WHERE IS_DEFINED(tags.site) AND tags.site = 'DANISH' AND capabilities.iotEdge = true ")
query_spec = QuerySpecification(query="SELECT * FROM devices")
query_result = registry_manager.query_iot_hub(query_spec)
devices = []
for item in query_result.items:
# currentDevice = iSightDevice(str(item.device_id), str(item.tags['site']), int(item.tags['number']), str(item.tags['version']))
devices.append(iSightDevice(str(item.device_id), str(item.tags['site']), int(item.tags['number']), str(item.tags['version'])))
devices.sort(key = lambda d: (d.site, d.number))
for device in devices:
print(device, end="\t")
current_device_modules = registry_manager.get_modules(device.deviceId)
for module in current_device_modules:
if (module.module_id == "thingspro-agent"):
device.setModule(module)
try:
direct_method = CloudToDeviceMethod(method_name=method_name, payload=json.loads(payload))
response = registry_manager.invoke_device_module_method(device_id=device.deviceId, module_id=module_id, direct_method_request=direct_method)
device_version = str(response.payload['data']['firmwareVersion'])
print(device_version)
except:
print("ERROR")
print(item)