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

danish moxa update

Browse Source
This commit is contained in:
Quentin WEPHRE
2025-11-19 12:26:12 +01:00
parent 520dc2f371
commit 3bebce0f4b
2 changed files with 435 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
Python/danish_batch_api.py
View File

@@ -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!")
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!")