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

multiple evolutions

Browse Source
This commit is contained in:
Quentin WEPHRE
2025-12-04 08:11:00 +01:00
parent e9715dc239
commit abd00f9e04
9 changed files with 403 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files

430
Python/danish_batch_api copy.py
View File

@@ -6,24 +6,28 @@ import json
import os
from dotenv import load_dotenv
import time
from tqdm import tqdm
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
try:
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
except Exception as e:
raise Exception("Authentication failed!") from e
# Function to send PATCH request
def send_patch_request(device_ip, token, connection_string):
# Function to update connection string through PATCH request
def update_connection_string(device_ip, token, connection_string):
headers = {
"mx-api-token": token
}
@@ -35,45 +39,50 @@ def send_patch_request(device_ip, token, connection_string):
}
}
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}")
try:
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}")
except Exception as e:
raise Exception("Update connection string failed!") from e
# Function to send PATCH request
def patch_time(device_ip, token):
# Function to update NTP
def update_ntp(device_ip, token):
headers = {
"mx-api-token": token
}
# payload = {
# "ntp": {
# "source": "timeserver",
# "server": "10.84.171.254",
# "enable": True
# }
# }
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))
try:
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())
except Exception as e:
raise Exception("Update NTP failed!") from e
# Function to send UPGRADE request
def send_upgrade_request(device_ip, token, upgrade_url):
# Function to create an upgrade job
def create_upgrade_job(device_ip, token, upgrade_url):
headers = {
"mx-api-token": token
}
@@ -83,13 +92,15 @@ def send_upgrade_request(device_ip, token, upgrade_url):
"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
try:
response = requests.post(patch_url, json=payload, headers=headers, verify=False)
json_data = json.loads(response.content.decode())
id = json_data['data']['id']
return id
except Exception as e:
raise Exception("Create upgrade job failed") from e
# Function to send UPGRADE request
# Function to get upgrade job
def get_upgrade_job(device_ip, token, id):
headers = {
"mx-api-token": token
@@ -97,55 +108,24 @@ def get_upgrade_job(device_ip, token, id):
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)
try:
response = requests.get(patch_url, json=payload, headers=headers, verify=False)
if response.status_code == 200:
json_data = json.loads(response.content.decode())
getid = json_data['data']['id']
created = json_data['data']['createdAt']
started = json_data['data']['startedAt']
cur_status = json_data['data']['state']
current_tasks = json_data['data']['completedTask']
total_tasks = json_data['data']['totalTask']
print(f"Upgrade job #{str(getid)} ({str(current_tasks)}/{str(total_tasks)}) {str(cur_status)}")
else:
print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}")
print(response.content)
except Exception as e:
raise Exception("Failed getting upgrade job!") from e
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
}
@@ -155,9 +135,34 @@ def get_last_job(device_ip, token):
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())
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))
else:
print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}")
print(response.content)
except Exception as e:
raise Exception("Failed getting all upgrade jobs!") from e
def get_last_upgrade_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:
json_data = json.loads(response.content.decode())
last_job = int(json_data['count'] - 1)
getid = json_data['data'][last_job]['id']
@@ -171,15 +176,14 @@ def get_last_job(device_ip, token):
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}")
raise Exception("Failed getting last upgrade job!") from e
# Function to send UPGRADE request
# Function to start upgrade job
def start_upgrade_job(device_ip, token, id):
headers = {
"mx-api-token": token
@@ -187,60 +191,61 @@ def start_upgrade_job(device_ip, token, id):
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())
try:
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())
except Exception as e:
raise Exception("Failed starting upgrade job!") from e
# Function to send UPGRADE request
# Function to send a PUT 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())
patch_url = "" #f"https://{device_ip}:8443/api/v1/upgrades/3/start"
try:
if patch_url == "":
raise Exception("Empty URL!")
response = requests.put(patch_url, json=payload, headers=headers, verify=False)
if response.status_code == 200:
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())
except Exception as e:
raise Exception("Failed sending PUT request!") from e
def get_API(device_ip, token):
def get_version(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())
try:
response = requests.get(patch_url, json=payload, headers=headers, verify=False)
if response.status_code == 200:
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)
else:
print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}")
print(response.content.decode())
except Exception as e:
raise Exception("Failed getting version!") from e
def get_time(device_ip, token):
headers = {
@@ -249,22 +254,19 @@ def get_time(device_ip, 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())
try:
response = requests.get(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']
print(time + " " + timezone + " " + last)
else:
print(f"Failed to send GET request to device {device_ip}. Status code: {response.status_code}")
print(response.content.decode())
except Exception as e:
raise Exception("Failed getting time!") from e
def delete_API(device_ip, token):
headers = {
@@ -272,95 +274,63 @@ def delete_API(device_ip, 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)
patch_url = "" #f"https://{device_ip}:8443/api/v1/upgrades/5"
try:
if patch_url == "":
raise Exception("Empty URL!")
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)
except Exception as e:
raise Exception("Failed sending DELETE request!") from e
# 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
def visual_wait(total_seconds):
# "total=total_seconds" sets the bar max value
# "bar_format" removes the default stats to keep it clean
with tqdm(total=total_seconds, bar_format="{desc} [{bar}]") as pbar:
# Loop backwards from 200 down to 1
for remaining in range(total_seconds, 0, -1):
# Manually update the text description to show the countdown
pbar.set_description_str(f"Waiting {remaining}s")
# Advance the visual bar by 1 step
pbar.update(1)
time.sleep(1)
# Final update to show 0s at the very end
pbar.set_description_str(f"Finished waiting!")
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)]
moxa_range = [i for i in range(132, 159) ]
moxa_range.remove(137)
for i in moxa_range:
device_ip_address = str("10.84.157." + str(i))
print(device_ip_address, end=" ")
print(device_ip_address)
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"
upgrade_url = "http://10.84.157.137:8080/1.8.1.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!")
try:
token = authenticate(device_ip_address, payload_auth)
if token:
get_version(device_ip_address, token)
get_last_upgrade_job(device_ip_address, token)
# id = create_upgrade_job(device_ip_address, token, upgrade_url)
# visual_wait(3)
# start_upgrade_job(device_ip_address, token, id)
# visual_wait(3)
# get_upgrade_job(device_ip_address, token, id)
# visual_wait(30)
except Exception as e:
print(f"Exception for {device_ip_address}: {e}")
continue