api call through cloud for i-sight moxa

2026-03-26 08:48:15 +01:00
parent 007f30a786
commit 3dd5974d70
2 changed files with 124 additions and 0 deletions
--- a/Python/azure_iot_hub_thingspro_api_call_generic.py
+++ b/Python/azure_iot_hub_thingspro_api_call_generic.py
@@ -0,0 +1,55 @@
+from azure.iot.hub import IoTHubRegistryManager
+from azure.iot.hub.protocol.models import QuerySpecification, Module
+from azure.iot.hub.models import CloudToDeviceMethod, CloudToDeviceMethodResult
+from dotenv import load_dotenv
+from isight_device import iSightDevice
+
+import json
+import os
+import pandas as pd
+
+load_dotenv()
+
+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!")
+    exit(13)
+
+registry_manager = IoTHubRegistryManager.from_connection_string(CONNECTION_STRING)
+query_spec = QuerySpecification(query="SELECT * FROM devices WHERE IS_DEFINED(tags.site) AND capabilities.iotEdge = true ")
+
+query_result = registry_manager.query_iot_hub(query_spec)
+
+devices = []
+for item in query_result.items:
+    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))
+
+rows = []
+
+for device in devices:
+    if device.getDeviceId() == "TBAIB1114211":
+        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)
+        method_name = "thingspro-api-v1"
+        payload = {
+            "method": "GET",
+            "path": "/device/time"
+        }
+        module_id = "thingspro-agent"
+        try:
+            direct_method = CloudToDeviceMethod(method_name = method_name, payload = payload)
+            response = registry_manager.invoke_device_module_method(device_id = device.deviceId, module_id = module_id, direct_method_request = direct_method)
+            print(json.dumps(response.payload, indent = 2))
+        except Exception as e:
+            print(f"Error: {e}")
+    else:
+        continue
+        print(f"{device.getDeviceId()}")
+
+df = pd.DataFrame(rows)
+df.to_excel("thingspro-version.xlsx", index=False)
--- a/Python/azure_iot_hub_thingspro_api_call_user.py
+++ b/Python/azure_iot_hub_thingspro_api_call_user.py
@@ -0,0 +1,69 @@
+from azure.iot.hub import IoTHubRegistryManager
+from azure.iot.hub.protocol.models import QuerySpecification, Module
+from azure.iot.hub.models import CloudToDeviceMethod, CloudToDeviceMethodResult
+from dotenv import load_dotenv
+from isight_device import iSightDevice
+
+import json
+import os
+import pandas as pd
+
+load_dotenv()
+
+CONNECTION_STRING = str(os.getenv("CONNECTION_STRING_INOX_PROD"))
+NEW_MOXA_PASSWORD = str(os.getenv("NEW_MOXA_PASSWORD"))
+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 IS_DEFINED(tags.site) AND capabilities.iotEdge = true ")
+
+query_result = registry_manager.query_iot_hub(query_spec)
+
+devices = []
+for item in query_result.items:
+    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))
+
+rows = []
+
+for device in devices:
+    current_device_modules = registry_manager.get_modules(device.deviceId)
+    for module in current_device_modules:
+        if (module.module_id == "thingspro-agent"):
+            device.setModule(module)
+    method_name = "thingspro-api-v1"
+    payload = {
+        "method": "GET",
+        "path": "/users"
+    }
+    module_id = "thingspro-agent"
+    try:
+        direct_method = CloudToDeviceMethod(method_name = method_name, payload = payload)
+        response = registry_manager.invoke_device_module_method(device_id = device.deviceId, module_id = module_id, direct_method_request = direct_method)
+        for i in range(int(response.payload['count'])):
+            username = response.payload['data'][i]["name"]
+            userid = int(response.payload['data'][i]["id"])
+            if username == "admin":
+                print(f"Found {username} at ID {str(userid)} on device {device.getDeviceId()} from {device.getSite()} ({device.getNumber()})")
+                try:
+                    payloadpassword = {
+                        "method": "PUT",
+                        "path": f"/users/{str(userid)}/password",
+                        "requestBody": {
+                            "newPassword": f"{NEW_MOXA_PASSWORD}"
+                        }
+                    }
+                    direct_method_password = CloudToDeviceMethod(method_name = method_name, payload = payloadpassword)
+                    response_password = registry_manager.invoke_device_module_method(device_id = device.deviceId, module_id = module_id, direct_method_request = direct_method_password)
+                except Exception as e:
+                    print(f"Error while changing password on device {device.getDeviceId()} from {device.getSite()} ({device.getNumber()})")
+                    continue
+    except Exception as e:
+        print(f"Error while getting users on device {device.getDeviceId()} from {device.getSite()} ({device.getNumber()})")
+        continue
+
+# df = pd.DataFrame(rows)
+# df.to_excel("thingspro-users.xlsx", index=False)