336 lines
13 KiB
Python
336 lines
13 KiB
Python
import pandas as pd
|
|
import requests
|
|
from urllib3.exceptions import InsecureRequestWarning
|
|
import jq
|
|
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"
|
|
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 update connection string through PATCH request
|
|
def update_connection_string(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"
|
|
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 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 = {
|
|
"timezone": "America/Chicago"
|
|
}
|
|
patch_url = f"https://{device_ip}:8443/api/v1/device/time"
|
|
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 create an upgrade job
|
|
def create_upgrade_job(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"
|
|
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 get upgrade job
|
|
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}"
|
|
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"
|
|
try:
|
|
response = requests.get(patch_url, json=payload, headers=headers, verify=False)
|
|
if response.status_code == 200:
|
|
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']
|
|
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']}%")
|
|
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:
|
|
raise Exception("Failed getting last upgrade job!") from e
|
|
|
|
# Function to start upgrade job
|
|
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"
|
|
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 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"
|
|
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_version(device_ip, token):
|
|
headers = {
|
|
"mx-api-token": token
|
|
}
|
|
payload = {
|
|
}
|
|
patch_url = f"https://{device_ip}:8443/api/v1/device/general"
|
|
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 = {
|
|
"mx-api-token": token
|
|
}
|
|
payload = {
|
|
}
|
|
patch_url = f"https://{device_ip}:8443/api/v1/device/time"
|
|
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 = {
|
|
"mx-api-token": token
|
|
}
|
|
payload = {
|
|
}
|
|
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
|
|
|
|
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)
|
|
|
|
default_user = str(os.getenv("DEFAULT_MOXA_USER"))
|
|
default_password = str(os.getenv("DEFAULT_MOXA_PASSWORD"))
|
|
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)
|
|
|
|
payload_auth = {
|
|
"acceptEULA": True,
|
|
"name": default_user,
|
|
"password": default_password
|
|
}
|
|
|
|
upgrade_url = "http://10.84.157.137:8080/1.8.1.yaml"
|
|
|
|
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 |