From 3bebce0f4b558d54bc4d65ab48dab0e5ae7953d8 Mon Sep 17 00:00:00 2001 From: Quentin WEPHRE Date: Wed, 19 Nov 2025 12:26:12 +0100 Subject: [PATCH] danish moxa update --- Python/danish_batch_api copy.py | 366 ++++++++++++++++++++++++++++++++ Python/danish_batch_api.py | 85 ++++++-- 2 files changed, 435 insertions(+), 16 deletions(-) create mode 100644 Python/danish_batch_api copy.py diff --git a/Python/danish_batch_api copy.py b/Python/danish_batch_api copy.py new file mode 100644 index 0000000..dd60f83 --- /dev/null +++ b/Python/danish_batch_api copy.py @@ -0,0 +1,366 @@ +import pandas as pd +import requests +from urllib3.exceptions import InsecureRequestWarning +import jq +import json +import os +from dotenv import load_dotenv +import time + +load_dotenv() + +# Function to authenticate and get token +def authenticate(device_ip, payload): + auth_url = f"https://{device_ip}:8443/api/v1/auth" + response = requests.post(auth_url, json=payload, verify=False) + if response.status_code == 200: + token = response.json()["data"]["token"] + #print(f"Authentication successful. Token received: {token}") + #print(" authenticated!") + return token + else: + print(f"Authentication failed. Status code: {response.status_code}") + return None + +# Function to send PATCH request +def send_patch_request(device_ip, token, connection_string): + headers = { + "mx-api-token": token + } + payload = { + "provisioning": { + "source": "manual", + "connectionString": connection_string, + "enable": True + } + } + patch_url = f"https://{device_ip}:8443/api/v1/azure-iotedge" + response = requests.patch(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + print(f"PATCH request successful for device {device_ip}") + else: + print(f"Failed to send PATCH request to device {device_ip}. Status code: {response.status_code}") + + +# Function to send PATCH request +def patch_time(device_ip, token): + headers = { + "mx-api-token": token + } + # payload = { + # "ntp": { + # "source": "timeserver", + # "server": "10.84.171.254", + # "enable": True + # } + # } + payload = { + "timezone": "America/Chicago" + } + patch_url = f"https://{device_ip}:8443/api/v1/device/time" + response = requests.patch(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + json_data = json.loads(response.content.decode()) + time = json_data['data']['time'] + timezone = json_data['data']['timezone'] + last = json_data['data']['lastUpdateTime'] + server = json_data['data']['ntp']['server'] + enabled = json_data['data']['ntp']['enable'] + print(time + " " + timezone + " " + last + " " + server + " " + str(enabled)) + else: + json_data = json.loads(response.content.decode()) + #print(json.dumps(json_data, indent=2)) + + +# Function to send UPGRADE request +def send_upgrade_request(device_ip, token, upgrade_url): + headers = { + "mx-api-token": token + } + payload = { + "download": True, + "install": True, + "url": upgrade_url, + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades" + response = requests.post(patch_url, json=payload, headers=headers, verify=False) + json_data = json.loads(response.content.decode()) + #print(json.dumps(json_data, indent=4, sort_keys=True)) + id = json_data['data']['id'] + return id + +# Function to send UPGRADE request +def get_upgrade_job(device_ip, token, id): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades/{id}" + response = requests.get(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + #print(f"GET request successful for device {device_ip}") + #print(json_str) + #print(json_data['data'][json_data['count'] - 1]['parameter']['url'], json_data['data'][json_data['count'] - 1]['state'], json_data['count']) + json_data = json.loads(response.content.decode()) + getid = json_data['data']['id'] + created = json_data['data']['createdAt'] + cur_status = json_data['data']['state'] + current_tasks = json_data['data']['completedTask'] + total_tasks = json_data['data']['totalTask'] + print("JOB #" + str(getid) + " " + str(current_tasks) + "/" + str(total_tasks)) + print("CREATED ON: " + str(created)) + print("CURRENT STATUS: " + str(cur_status)) + print(json.dumps(json_data, indent=2)) + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content) + +def get_upgrade_jobs(device_ip, token): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades" + response = requests.get(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + #print(f"GET request successful for device {device_ip}") + #print(json_str) + #print(json_data['data'][json_data['count'] - 1]['parameter']['url'], json_data['data'][json_data['count'] - 1]['state'], json_data['count']) + json_data = json.loads(response.content.decode()) + count = json_data['count'] + print(str(count)) + for i in range(count): + getid = json_data['data'][i]['id'] + created = json_data['data'][i]['createdAt'] + cur_status = json_data['data'][i]['state'] + current_tasks = json_data['data'][i]['completedTask'] + total_tasks = json_data['data'][i]['totalTask'] + print("JOB #" + str(getid) + " " + str(current_tasks) + "/" + str(total_tasks)) + print("CREATED ON: " + str(created)) + print("CURRENT STATUS: " + str(cur_status)) + print(json.dumps(json_data, indent=4, sort_keys=True)) + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content) + +def get_last_job(device_ip, token): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades" + try: + response = requests.get(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + #print(f"GET request successful for device {device_ip}") + #print(json_str) + #print(json_data['data'][json_data['count'] - 1]['parameter']['url'], json_data['data'][json_data['count'] - 1]['state'], json_data['count']) + json_data = json.loads(response.content.decode()) + last_job = int(json_data['count'] - 1) + getid = json_data['data'][last_job]['id'] + created = json_data['data'][last_job]['createdAt'] + cur_status = json_data['data'][last_job]['state'] + current_tasks = json_data['data'][last_job]['completedTask'] + total_tasks = json_data['data'][last_job]['totalTask'] + print("JOB #" + str(getid) + " " + str(current_tasks) + "/" + str(total_tasks)) + print("CREATED ON: " + str(created)) + print("CURRENT STATUS: " + str(cur_status)) + for task in range(total_tasks): + if json_data['data'][last_job]['tasks'][task]['type'] == "download": + print(f"Downloaded: {json_data['data'][last_job]['tasks'][task]['progress']}%") + #print(json.dumps(json_data, indent=4, sort_keys=True)) + return getid + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content) + except Exception as e: + print(f"Connection failed {e}") + +# Function to send UPGRADE request +def start_upgrade_job(device_ip, token, id): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades/{id}/start" + response = requests.put(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + json_data = json.loads(response.content.decode()) + curid = json_data['data']['id'] + startedat = json_data['data']['startedAt'] + print("Job #" + str(curid) + " started on " + str(startedat)) + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content.decode()) + +# Function to send UPGRADE request +def put_API(device_ip, token): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades/3/start" + response = requests.put(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + #print(f"GET request successful for device {device_ip}") + #json_data = json.loads(response.content.decode()) + #json_str = json.dumps(json_data) + #print(jq.compile('.data.completedAt').input(json.loads(json_str)).first()) + json_data = json.loads(response.content.decode()) + print(json_data['data']['firmwareVersion']) + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content.decode()) + +def get_API(device_ip, token): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/device/general" + response = requests.get(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + #print(f"GET request successful for device {device_ip}") + #json_data = json.loads(response.content.decode()) + #json_str = json.dumps(json_data) + #print(jq.compile('.data.completedAt').input(json.loads(json_str)).first()) + json_data = json.loads(response.content.decode()) + hostname = json_data['data']['hostName'] + serial = json_data['data']['serialNumber'] + version = json_data['data']['firmwareVersion'] + description = json_data['data']['description'] + print(hostname + " " + serial + " " + version + " " + description) + #json_str = json.dumps(json_data, indent=2) + #print(json_str) + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content.decode()) + +def get_time(device_ip, token): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/device/time" + response = requests.get(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + #print(f"GET request successful for device {device_ip}") + #json_data = json.loads(response.content.decode()) + #json_str = json.dumps(json_data) + #print(jq.compile('.data.completedAt').input(json.loads(json_str)).first()) + json_data = json.loads(response.content.decode()) + time = json_data['data']['time'] + timezone = json_data['data']['timezone'] + last = json_data['data']['lastUpdateTime'] + print(time + " " + timezone + " " + last) + #json_str = json.dumps(json_data, indent=2) + #print(json_str) + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content.decode()) + +def delete_API(device_ip, token): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades/5" + response = requests.delete(patch_url, json=payload, headers=headers, verify=False) + print(response.status_code) + json_data = json.loads(response.content.decode()) + json_str = json.dumps(json_data, indent=2) + print(json_str) + +# Read the Excel file +# excel_file_path = "" +# if excel_file_path == "": +# print("Provide Excel file path before running the script!") +# exit(11) +# df = pd.read_excel(excel_file_path) +# df = df[df["device_name"].notnull()] + +# Iterate over each row in the DataFrame + +requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) + +# for index, row in df.iterrows(): +# device_name = row['device_name'] +# device_ip_address_https = row['device_ip_address_http'] +# connection_string = row['connection_string'] +# upgrade_url = "" #https://files.thingsprocloud.com/package/moxa-aig-301-series-includes-security-patch-firmware-v1.0.deb.yaml +# if upgrade_url == "": +# print("Provide upgrade URL before running the script!") +# exit(12) + + +# # Authenticate and get token +# payload_auth = { +# "acceptEULA": True, +# "name": "", +# "password": "" +# } +# if payload_auth["name"] == "" or payload_auth["password"] == "": +# print("Provide the credentials before running the script!") +# exit(10) +# print(device_name, end="") +# token = authenticate(device_ip_address_https, payload_auth) +# if token: +# get_API(device_ip_address_https, token) +# print("\n") + +default_user = str(os.getenv("DEFAULT_MOXA_USER")) +default_password = str(os.getenv("DEFAULT_MOXA_PASSWORD")) +moxa_range = [i for i in range(131, 132) if i not in (136, 137)] +for i in moxa_range: + device_ip_address = str("10.84.157." + str(i)) + print(device_ip_address, end=" ") + + payload_auth = { + "acceptEULA": True, + "name": default_user, + "password": default_password + } + token = authenticate(device_ip_address, payload_auth) + print(hash(token)) + + upgrade_url = "http://10.84.157.137:8080/1.7.0.yaml" + + if token: + #send_upgrade_request(device_ip_address, token, upgrade_url) + #time.sleep(10) + id = get_last_job(device_ip_address, token) + #time.sleep(10) + #start_upgrade_job(device_ip_address, token, id) + #time.sleep(120) + #input("Continue?") + else: + raise(Exception("Authentication failed!")) + + # upgrade_url = "https://10.84.157.137/Upgrade_AIG-301_2.5.0-4404_IMG_1.5_to_1.6.0.yaml" + + + # device_ip_address = str("10.84.157." + str(i)) + # print(f"{device_ip_address}") + + # #print(device_ip_address, end="") + # token = authenticate(device_ip_address, payload_auth) + # if token: + # #id = send_upgrade_request(device_ip_address,token,upgrade_url) + # #print(id) + # get_last_job(device_ip_address, token) + # #start_upgrade_job(device_ip_address, token, id) + # #put_API(device_ip_address, token) + # #patch_time(device_ip_address,token) + # #get_time(device_ip_address, token) + # #delete_API(device_ip_address,token) + # #get_API(device_ip_address, token) + # else: + # print("Authentication failed!") \ No newline at end of file diff --git a/Python/danish_batch_api.py b/Python/danish_batch_api.py index 291c417..c8eedd3 100644 --- a/Python/danish_batch_api.py +++ b/Python/danish_batch_api.py @@ -5,6 +5,7 @@ import jq import json import os from dotenv import load_dotenv +import time load_dotenv() @@ -84,6 +85,7 @@ def send_upgrade_request(device_ip, token, upgrade_url): patch_url = f"https://{device_ip}:8443/api/v1/upgrades" response = requests.post(patch_url, json=payload, headers=headers, verify=False) json_data = json.loads(response.content.decode()) + #print(json.dumps(json_data, indent=4, sort_keys=True)) id = json_data['data']['id'] return id @@ -138,6 +140,38 @@ def get_upgrade_jobs(device_ip, token): print("JOB #" + str(getid) + " " + str(current_tasks) + "/" + str(total_tasks)) print("CREATED ON: " + str(created)) print("CURRENT STATUS: " + str(cur_status)) + print(json.dumps(json_data, indent=4, sort_keys=True)) + else: + print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") + print(response.content) + +def get_last_job(device_ip, token): + headers = { + "mx-api-token": token + } + payload = { + } + patch_url = f"https://{device_ip}:8443/api/v1/upgrades" + response = requests.get(patch_url, json=payload, headers=headers, verify=False) + if response.status_code == 200: + #print(f"GET request successful for device {device_ip}") + #print(json_str) + #print(json_data['data'][json_data['count'] - 1]['parameter']['url'], json_data['data'][json_data['count'] - 1]['state'], json_data['count']) + json_data = json.loads(response.content.decode()) + last_job = int(json_data['count'] - 1) + getid = json_data['data'][last_job]['id'] + created = json_data['data'][last_job]['createdAt'] + cur_status = json_data['data'][last_job]['state'] + current_tasks = json_data['data'][last_job]['completedTask'] + total_tasks = json_data['data'][last_job]['totalTask'] + print("JOB #" + str(getid) + " " + str(current_tasks) + "/" + str(total_tasks)) + print("CREATED ON: " + str(created)) + print("CURRENT STATUS: " + str(cur_status)) + for task in range(total_tasks): + if json_data['data'][last_job]['tasks'][task]['type'] == "download": + print(f"Downloaded: {json_data['data'][last_job]['tasks'][task]['progress']}%") + #print(json.dumps(json_data, indent=4, sort_keys=True)) + return getid else: print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}") print(response.content) @@ -281,28 +315,47 @@ requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning) default_user = str(os.getenv("DEFAULT_MOXA_USER")) default_password = str(os.getenv("DEFAULT_MOXA_PASSWORD")) +moxa_range = [i for i in range(148, 159) if i not in (136, 137)] +for i in moxa_range: + device_ip_address = str("10.84.157." + str(i)) + print(device_ip_address, end=" ") -for i in range(193, 222): - upgrade_url = "https://files.thingsprocloud.com/package/Upgrade_AIG-301_2.4.0-4020_IMG_1.4_to_1.5.deb.yaml" payload_auth = { "acceptEULA": True, "name": default_user, "password": default_password } - - device_ip_address = str("10.84.171." + str(i)) - - #print(device_ip_address, end="") token = authenticate(device_ip_address, payload_auth) + print(hash(token)) + + upgrade_url = "http://10.84.157.137:8080/1.6.0.yaml" + if token: - #id = send_upgrade_request(device_ip_address,token,upgrade_url) - #print(id) - #get_upgrade_job(device_ip_address, token, 6) - #start_upgrade_job(device_ip_address, token, id) - #put_API(device_ip_address, token) - #patch_time(device_ip_address,token) - #get_time(device_ip_address, token) - #delete_API(device_ip_address,token) - get_API(device_ip_address, token) + #send_upgrade_request(device_ip_address, token, upgrade_url) + id = get_last_job(device_ip_address, token) + start_upgrade_job(device_ip_address, token, id) + time.sleep(300) + #input("Continue?") else: - print("Authentication failed!") \ No newline at end of file + raise(Exception("Authentication failed!")) + + # upgrade_url = "https://10.84.157.137/Upgrade_AIG-301_2.5.0-4404_IMG_1.5_to_1.6.0.yaml" + + + # device_ip_address = str("10.84.157." + str(i)) + # print(f"{device_ip_address}") + + # #print(device_ip_address, end="") + # token = authenticate(device_ip_address, payload_auth) + # if token: + # #id = send_upgrade_request(device_ip_address,token,upgrade_url) + # #print(id) + # get_last_job(device_ip_address, token) + # #start_upgrade_job(device_ip_address, token, id) + # #put_API(device_ip_address, token) + # #patch_time(device_ip_address,token) + # #get_time(device_ip_address, token) + # #delete_API(device_ip_address,token) + # #get_API(device_ip_address, token) + # else: + # print("Authentication failed!") \ No newline at end of file