CAT_TOM/pyasic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

removed old tools that will no longer work

Browse Source
This commit is contained in:
UpstreamData
2022-05-19 11:56:12 -06:00
parent ec5563f2f0
commit 3ff43c3ccd
22 changed files with 0 additions and 2514 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
tools/bad_board_util_old/__init__.py
View File

@@ -1,29 +0,0 @@
from tools.bad_board_util_old.ui import ui
import asyncio
import sys
import logging
from logger import logger
logger.info("Initializing logger for Board Util.")
# Fix bug with some whatsminers and asyncio because of a socket not being shut down:
if (
sys.version_info[0] == 3
and sys.version_info[1] >= 8
and sys.platform.startswith("win")
):
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
def main():
logging.info("Starting Board Util.")
loop = asyncio.new_event_loop()
loop.run_until_complete(ui())
logging.info("Closing Board Util.")
if __name__ == "__main__":
main()

0
tools/bad_board_util_old/func/__init__.py
View File

29
tools/bad_board_util_old/func/decorators.py
View File

@@ -1,29 +0,0 @@
from tools.bad_board_util_old.layout import window
def disable_buttons(func):
button_list = [
"scan",
"import_iplist",
"export_iplist",
"select_all_ips",
"refresh_data",
"open_in_web",
"save_report_button",
"light",
]
# handle the inner function that the decorator is wrapping
async def inner(*args, **kwargs):
# disable the buttons
for button in button_list:
window[button].Update(disabled=True)
# call the original wrapped function
await func(*args, **kwargs)
# re-enable the buttons after the wrapped function completes
for button in button_list:
window[button].Update(disabled=False)
return inner

104
tools/bad_board_util_old/func/files.py
View File

@@ -1,104 +0,0 @@
import ipaddress
import os
import re
import xlsxwriter
import aiofiles
from tools.bad_board_util_old.func.ui import update_ui_with_data
from tools.bad_board_util_old.layout import window
from tools.bad_board_util_old.func.decorators import disable_buttons
from miners.miner_factory import MinerFactory
@disable_buttons
async def save_report(file_location):
data = {}
workbook = xlsxwriter.Workbook(file_location)
sheet = workbook.add_worksheet()
for line in window["ip_table"].Values:
data[line[0]] = {
"Model": line[1],
"Total Chips": line[2],
"Left Chips": line[3],
"Center Chips": line[5],
"Right Chips": line[7],
"Nominal": 1,
}
async for miner in MinerFactory().get_miner_generator([key for key in data.keys()]):
if miner:
data[miner.ip]["Nominal"] = miner.nominal
list_data = []
for ip in data.keys():
new_data = data[ip]
new_data["IP"] = ip
list_data.append(new_data)
data = sorted(data, reverse=True, key=lambda x: x["Total Chips"])
headers = [
"IP",
"Miner Model",
"Total Chip Count",
"Left Board Chips",
"Center Board Chips",
"Right Board Chips",
]
print(data)
row = 0
col = 0
for item in headers:
sheet.write(row, col, item)
col += 1
row = 1
for line in data:
col = 0
for point in line:
sheet.write(row, col, point)
col += 1
row += 1
workbook.close()
async def import_iplist(file_location):
await update_ui_with_data("status", "Importing")
if not os.path.exists(file_location):
return
else:
ip_list = []
async with aiofiles.open(file_location, mode="r") as file:
async for line in file:
ips = [
x.group()
for x in re.finditer(
"^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)",
line,
)
]
for ip in ips:
if ip not in ip_list:
ip_list.append(ipaddress.ip_address(ip))
ip_list.sort()
window["ip_table"].update([[str(ip), "", "", "", ""] for ip in ip_list])
await update_ui_with_data("ip_count", str(len(ip_list)))
await update_ui_with_data("status", "")
async def export_iplist(file_location, ip_list_selected):
await update_ui_with_data("status", "Exporting")
if not os.path.exists(file_location):
return
else:
if ip_list_selected is not None and not ip_list_selected == []:
async with aiofiles.open(file_location, mode="w") as file:
for item in ip_list_selected:
await file.write(str(item) + "\n")
else:
async with aiofiles.open(file_location, mode="w") as file:
for item in window["ip_table"].Values:
await file.write(str(item[0]) + "\n")
await update_ui_with_data("status", "")

308
tools/bad_board_util_old/func/miners.py
View File

@@ -1,308 +0,0 @@
import asyncio
import ipaddress
import warnings
from tools.bad_board_util_old.func.ui import (
update_ui_with_data,
update_prog_bar,
set_progress_bar_len,
)
from tools.bad_board_util_old.layout import window
from miners.miner_factory import MinerFactory
from tools.bad_board_util_old.func.decorators import disable_buttons
@disable_buttons
async def miner_light(ips: list):
await asyncio.gather(*[flip_light(ip) for ip in ips])
async def flip_light(ip):
ip_list = window["ip_table"].Widget
miner = await MinerFactory().get_miner(ip)
index = [item[0] for item in window["ip_table"].Values].index(ip)
index_tags = ip_list.item(index + 1)["tags"]
if "light" not in index_tags and "light+bad" not in index_tags:
tag = "light"
if "bad" in index_tags:
index_tags.remove("bad")
tag = "light+bad"
index_tags.append(tag)
ip_list.item(index + 1, tags=index_tags)
await miner.fault_light_on()
else:
if "light+bad" in index_tags:
index_tags.remove("light+bad")
index_tags.append("bad")
if "light" in index_tags:
index_tags.remove("light")
ip_list.item(index + 1, tags=index_tags)
await miner.fault_light_off()
@disable_buttons
async def scan_network(network):
await update_ui_with_data("status", "Scanning")
await update_ui_with_data("ip_count", "")
window["ip_table"].update([])
network_size = len(network)
miner_generator = network.scan_network_generator()
await set_progress_bar_len(2 * network_size)
progress_bar_len = 0
asyncio.create_task(update_prog_bar(progress_bar_len))
miners = []
async for miner in miner_generator:
if miner:
miners.append(miner)
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
progress_bar_len += network_size - len(miners)
asyncio.create_task(update_prog_bar(progress_bar_len))
get_miner_genenerator = MinerFactory().get_miner_generator(miners)
all_miners = []
async for found_miner in get_miner_genenerator:
all_miners.append(found_miner)
all_miners.sort(key=lambda x: x.ip)
window["ip_table"].update([[str(miner.ip)] for miner in all_miners])
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
await update_ui_with_data("ip_count", str(len(all_miners)))
await update_ui_with_data("status", "")
@disable_buttons
async def refresh_data(ip_list: list):
await update_ui_with_data("status", "Getting Data")
ips = [ipaddress.ip_address(ip) for ip in ip_list]
if len(ips) == 0:
ips = [
ipaddress.ip_address(ip)
for ip in [item[0] for item in window["ip_table"].Values]
]
await set_progress_bar_len(len(ips))
progress_bar_len = 0
asyncio.create_task(update_prog_bar(progress_bar_len))
reset_table_values = []
for item in window["ip_table"].Values:
if item[0] in ip_list:
reset_table_values.append([item[0]])
else:
reset_table_values.append(item)
window["ip_table"].update(reset_table_values)
progress_bar_len = 0
data_gen = asyncio.as_completed([get_formatted_data(miner) for miner in ips])
ip_table_data = window["ip_table"].Values
ordered_all_ips = [item[0] for item in ip_table_data]
row_colors = []
for all_data in data_gen:
data_point = await all_data
if data_point["IP"] in ordered_all_ips:
ip_table_index = ordered_all_ips.index(data_point["IP"])
board_left = ""
board_center = ""
board_right = ""
if data_point["data"]:
if 0 in data_point["data"].keys():
board_left = " ".join(
[chain["chip_status"] for chain in data_point["data"][0]]
).replace("o", "•")
if 1 in data_point["data"].keys():
board_center = " ".join(
[chain["chip_status"] for chain in data_point["data"][1]]
).replace("o", "•")
if 2 in data_point["data"].keys():
board_right = " ".join(
[chain["chip_status"] for chain in data_point["data"][2]]
).replace("o", "•")
data = [
data_point["IP"],
data_point["model"],
len(board_left),
board_left,
len(board_center),
board_center,
len(board_right),
board_right,
]
ip_table_data[ip_table_index] = data
window["ip_table"].update(ip_table_data)
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
await update_ui_with_data("status", "")
@disable_buttons
async def scan_and_get_data(network):
# update status and reset the table
await update_ui_with_data("status", "Scanning")
await update_ui_with_data("ip_count", "")
await update_ui_with_data("ip_table", [])
# set progress bar length to network size
network_size = len(network)
await set_progress_bar_len(3 * network_size)
progress_bar_len = 0
miners = []
# scan the network for miners using a generator
async for miner in network.scan_network_generator():
# the generator will either return None or an IP address
if miner:
miners.append(miner)
# add to the progress bar length after scanning an address
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
# add progress for the miners that we aren't going to identify
progress_bar_len += network_size - len(miners)
asyncio.create_task(update_prog_bar(progress_bar_len))
all_miners = []
# identify different miner instances using the miner factory generator
async for found_miner in MinerFactory().get_miner_generator(miners):
# miner factory generator will always return a miner
all_miners.append(found_miner)
# sort the list of miners by IP address
all_miners.sort(key=lambda x: x.ip)
# add the new miner to the table
window["ip_table"].update([[str(miner.ip)] for miner in all_miners])
# update progress bar
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
# update the count of found miners
await update_ui_with_data("ip_count", str(len(all_miners)))
# update progress bar for miners we wont get data for
progress_bar_len += network_size - len(miners)
asyncio.create_task(update_prog_bar(progress_bar_len))
# get the list of IP addresses from the table
ip_table_data = window["ip_table"].Values
ordered_all_ips = [item[0] for item in ip_table_data]
await update_ui_with_data("status", "Getting Data")
row_colors = []
# create an in place generator for getting data
for all_data in asyncio.as_completed(
[get_formatted_data(miner) for miner in miners]
):
# wait for a generator item to return
data_point = await all_data
# make sure the IP is one we have
# this will likely never fail, but a good failsafe
if data_point["IP"] in ordered_all_ips:
# get the index of the IP in the table
ip_table_index = ordered_all_ips.index(data_point["IP"])
board_left = ""
board_center = ""
board_right = ""
# make sure we have data, some miners don't allow getting board data
if data_point["data"]:
# check if the 0th board (L board) is in the data
if 0 in data_point["data"].keys():
board_left = " ".join(
[chain["chip_status"] for chain in data_point["data"][0]]
).replace("o", "•")
else:
# if the board isn't in data, highlight it red
row_colors.append((ip_table_index, "bad"))
# check if the 1st board (C board) is in the data
if 1 in data_point["data"].keys():
board_center = " ".join(
[chain["chip_status"] for chain in data_point["data"][1]]
).replace("o", "•")
else:
# if the board isn't in data, highlight it red
row_colors.append((ip_table_index, "bad"))
# check if the 2nd board (R board) is in the data
if 2 in data_point["data"].keys():
board_right = " ".join(
[chain["chip_status"] for chain in data_point["data"][2]]
).replace("o", "•")
else:
# if the board isn't in data, highlight it red
row_colors.append((ip_table_index, "bad"))
# check if the miner has all nominal chips
if False in [
# True/False if the miner is nominal
chain["nominal"]
# for each board in the miner
for board in [
data_point["data"][key] for key in data_point["data"].keys()
]
# for each chain in each board in the miner
for chain in board
]:
# if the miner doesn't have all chips, highlight it red
row_colors.append((ip_table_index, "bad"))
else:
# the row is bad if we have no data
row_colors.append((ip_table_index, "bad"))
# split the chip data into thirds
board_left_chips = "\n".join(split_chips(board_left, 3))
board_center_chips = "\n".join(split_chips(board_center, 3))
board_right_chips = "\n".join(split_chips(board_right, 3))
# create data for the table
data = [
data_point["IP"],
data_point["model"],
(len(board_left) + len(board_center) + len(board_right)),
len(board_left),
board_left_chips,
len(board_center),
board_center_chips,
len(board_right),
board_right_chips,
]
# put the data at the index of the IP address
ip_table_data[ip_table_index] = data
window["ip_table"].update(ip_table_data)
# configure "bad" tag to highlight red
table = window["ip_table"].Widget
# set tags on the row if they have been set
for row in row_colors:
table.item(row[0] + 1, tags=row[1])
# add to the progress bar
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
# reset status
await update_ui_with_data("status", "")
def split_chips(string, number_of_splits):
k, m = divmod(len(string), number_of_splits)
return (
string[i * k + min(i, m) : (i + 1) * k + min(i + 1, m)]
for i in range(number_of_splits)
)
async def get_formatted_data(ip: ipaddress.ip_address):
miner = await MinerFactory().get_miner(ip)
model = await miner.get_model()
warnings.filterwarnings("ignore")
board_data = await miner.get_board_info()
data = {"IP": str(ip), "model": str(model), "data": board_data}
return data

392
tools/bad_board_util_old/func/pdf.py
View File

@@ -1,392 +0,0 @@
import datetime
from base64 import b64decode
from io import BytesIO
from reportlab.lib.pagesizes import letter, inch
from reportlab.lib.utils import ImageReader
from reportlab.lib.styles import ParagraphStyle, TA_CENTER
from reportlab.platypus import (
SimpleDocTemplate,
KeepInFrame,
Table,
Image,
Paragraph,
TableStyle,
PageBreak,
Spacer,
)
from reportlab.lib import colors
import ipaddress
import numpy as np
import matplotlib.dates
import matplotlib.pyplot as plt
from svglib.svglib import svg2rlg
from matplotlib import cm
from matplotlib.ticker import FormatStrFormatter
from miners.miner_factory import MinerFactory
from tools.bad_board_util_old.func.decorators import disable_buttons
from tools.bad_board_util_old.img import IMAGE_SELECTION_MATRIX, LOGO
from tools.bad_board_util_old.layout import window
IP_STYLE = ParagraphStyle(
"IP Style",
alignment=TA_CENTER,
fontSize=7,
fontName="Helvetica-Bold",
)
TITLE_STYLE = ParagraphStyle(
"Title",
alignment=TA_CENTER,
fontSize=20,
spaceAfter=40,
fontName="Helvetica-Bold",
)
def add_first_page_number(canvas, doc):
canvas.saveState()
canvas.drawString(letter[0] - 60, 20, "Page " + str(doc.page))
canvas.restoreState()
def add_page_header(canvas, doc):
canvas.saveState()
canvas.drawCentredString(
(letter[0] / 16) * 14,
letter[1] - 57,
datetime.datetime.now().strftime("%Y-%b-%d"),
)
img_dec = b64decode(LOGO)
img = BytesIO(img_dec)
img.seek(0)
canvas.drawImage(
ImageReader(img),
30,
letter[1] - 65,
150,
35,
)
canvas.drawString(letter[0] - 60, 20, "Page " + str(doc.page))
canvas.restoreState()
@disable_buttons
async def save_report(file_location):
p1_logo, p1_title = create_first_page()
data = {}
for line in window["ip_table"].Values:
data[line[0]] = {
"Model": line[1],
"Total Chips": line[2],
"Left Chips": line[3],
"Center Chips": line[5],
"Right Chips": line[7],
"Nominal": 1,
}
async for miner in MinerFactory().get_miner_generator([key for key in data.keys()]):
if miner:
data[str(miner.ip)]["Nominal"] = miner.nominal_chips
list_data = []
for ip in data.keys():
new_data = data[ip]
new_data["IP"] = ip
list_data.append(new_data)
list_data = sorted(
list_data, reverse=False, key=lambda x: ipaddress.ip_address(x["IP"])
)
image_selection_data = {}
for miner in list_data:
miner_bad_boards = ""
if miner["Left Chips"] < miner["Nominal"]:
miner_bad_boards += "l"
if miner["Center Chips"] < miner["Nominal"]:
miner_bad_boards += "c"
if miner["Right Chips"] < miner["Nominal"]:
miner_bad_boards += "r"
image_selection_data[miner["IP"]] = miner_bad_boards
doc = SimpleDocTemplate(
file_location,
pagesize=letter,
topMargin=1 * inch,
leftMargin=1 * inch,
rightMargin=1 * inch,
bottomMargin=1 * inch,
title=f"Board Report {datetime.datetime.now().strftime('%Y/%b/%d')}",
)
pie_chart, board_table = create_boards_pie_chart(image_selection_data)
table_data = get_table_data(image_selection_data)
miner_img_table = Table(
table_data,
colWidths=0.8 * inch,
# repeatRows=1,
# rowHeights=[4 * inch],
)
miner_img_table.setStyle(
TableStyle(
[
("SPAN", (0, 0), (-1, 0)),
("LEFTPADDING", (0, 0), (-1, -1), 0),
("RIGHTPADDING", (0, 0), (-1, -1), 0),
("BOTTOMPADDING", (0, 1), (-1, -1), 0),
("TOPPADDING", (0, 1), (-1, -1), 0),
("BOTTOMPADDING", (0, 0), (-1, 0), 20),
("TOPPADDING", (0, 0), (-1, 0), 20),
]
)
)
elements = []
elements.append(p1_logo)
elements.append(p1_title)
elements.append(PageBreak())
elements.append(pie_chart)
elements.append(Spacer(0, 60))
elements.append(board_table)
elements.append(PageBreak())
elements.append(miner_img_table)
elements.append(PageBreak())
elements.append(
Paragraph(
"Board Data",
style=TITLE_STYLE,
)
)
elements.append(create_data_table(list_data))
elements.append(PageBreak())
doc.build(
elements,
onFirstPage=add_first_page_number,
onLaterPages=add_page_header,
)
def create_boards_pie_chart(data):
labels = ["All Working", "1 Bad Board", "2 Bad Boards", "3 Bad Boards"]
num_bad_boards = [0, 0, 0, 0]
for item in data.keys():
num_bad_boards[len(data[item])] += 1
cmap = plt.get_cmap("Blues")
cs = cmap(np.linspace(0.2, 0.8, num=len(num_bad_boards)))
fig1, ax = plt.subplots()
ax.pie(
num_bad_boards,
labels=labels,
autopct="%1.2f%%",
shadow=True,
startangle=180,
colors=cs,
pctdistance=0.8,
)
ax.axis("equal")
ax.set_title("Broken Boards", fontsize=24, pad=20)
imgdata = BytesIO()
fig1.savefig(imgdata, format="svg")
imgdata.seek(0) # rewind the data
drawing = svg2rlg(imgdata)
imgdata.close()
plt.close("all")
pie_chart = KeepInFrame(375, 375, [Image(drawing)], hAlign="CENTER")
table_data = [labels, num_bad_boards]
t = Table(table_data)
table_style = TableStyle(
[
# ("FONTSIZE", (0, 0), (-1, -1), 13),
# line for below titles
("LINEBELOW", (0, 0), (-1, 0), 2, colors.black),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
# line for above totals
("LINEABOVE", (0, -1), (-1, -1), 2, colors.black),
# line for beside unit #
("LINEAFTER", (0, 0), (0, -1), 2, colors.black),
# gridlines and outline of table
("INNERGRID", (0, 0), (-1, -1), 0.25, colors.black),
("BOX", (0, 0), (-1, -1), 2, colors.black),
("LEFTPADDING", (0, 0), (-1, -1), 3),
("RIGHTPADDING", (0, 0), (-1, -1), 3),
("BOTTOMPADDING", (0, 0), (-1, -1), 3),
("TOPPADDING", (0, 0), (-1, -1), 3),
]
)
t.setStyle(table_style)
# zebra stripes on table
for each in range(len(table_data)):
if each % 2 == 0:
bg_color = colors.whitesmoke
else:
bg_color = colors.lightgrey
t.setStyle(TableStyle([("BACKGROUND", (0, each), (-1, each), bg_color)]))
return pie_chart, t
def create_first_page():
title_style = ParagraphStyle(
"Title",
alignment=TA_CENTER,
fontSize=50,
spaceAfter=40,
spaceBefore=150,
fontName="Helvetica-Bold",
)
img_dec = b64decode(LOGO)
img = BytesIO(img_dec)
img.seek(0)
logo = KeepInFrame(450, 105, [Image(img)])
title = Paragraph("Board Report", style=title_style)
return logo, title
def create_data_table(data):
left_bad_boards = 0
right_bad_boards = 0
center_bad_boards = 0
table_data = []
for miner in data:
miner_bad_boards = 0
if miner["Left Chips"] < miner["Nominal"]:
miner_bad_boards += 1
left_bad_boards += 1
if miner["Center Chips"] < miner["Nominal"]:
miner_bad_boards += 1
right_bad_boards += 1
if miner["Right Chips"] < miner["Nominal"]:
miner_bad_boards += 1
center_bad_boards += 1
table_data.append(
[
miner["IP"],
miner["Total Chips"],
miner["Left Chips"],
miner["Center Chips"],
miner["Right Chips"],
miner_bad_boards,
]
)
table_data.append(
[
"Total",
sum([miner[1] for miner in table_data]),
sum([miner[2] for miner in table_data]),
sum([miner[3] for miner in table_data]),
sum([miner[4] for miner in table_data]),
sum([miner[5] for miner in table_data]),
]
)
table_data[:0] = (
[
"IP",
"Total Chips",
"Left Board Chips",
"Center Board Chips",
"Right Board Chips",
"Failed Boards",
],
)
# create the table
t = Table(table_data, repeatRows=1)
# generate a basic table style
table_style = TableStyle(
[
("FONTSIZE", (0, 0), (-1, -1), 8),
# line for below titles
("LINEBELOW", (0, 0), (-1, 0), 2, colors.black),
("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
# line for above totals
("LINEABOVE", (0, -1), (-1, -1), 2, colors.black),
("FONTNAME", (0, -1), (-1, -1), "Helvetica-Bold"),
# line for beside unit #
("LINEAFTER", (0, 0), (0, -1), 2, colors.black),
("FONTNAME", (0, 0), (0, -1), "Helvetica-Bold"),
# gridlines and outline of table
("INNERGRID", (0, 0), (-1, -1), 0.25, colors.black),
("BOX", (0, 0), (-1, -1), 2, colors.black),
]
)
for (
row,
values,
) in enumerate(table_data):
if not row == 0 and not row == (len(table_data) - 1):
failed_boards = values[5]
if not failed_boards == 0:
table_style.add("TEXTCOLOR", (5, row), (5, row), colors.red)
# set the styles to the table
t.setStyle(table_style)
# zebra stripes on table
for each in range(len(table_data)):
if each % 2 == 0:
bg_color = colors.whitesmoke
else:
bg_color = colors.lightgrey
t.setStyle(TableStyle([("BACKGROUND", (0, each), (-1, each), bg_color)]))
return t
def get_table_data(data):
table_elems = [[Paragraph("Hashboard Visual Representation", style=TITLE_STYLE)]]
table_row = []
table_style = TableStyle(
[
("LEFTPADDING", (0, 0), (-1, -1), 0),
("RIGHTPADDING", (0, 0), (-1, -1), 0),
("BOTTOMPADDING", (0, 0), (-1, -1), 0),
("BOX", (0, 0), (-1, -1), 2, colors.black),
]
)
table_width = 0.8 * inch
for ip in data.keys():
img_dec = b64decode(IMAGE_SELECTION_MATRIX[data[ip]])
img = BytesIO(img_dec)
img.seek(0)
image = KeepInFrame(table_width, table_width, [Image(img)])
ip_para = Paragraph(ip, style=IP_STYLE)
table_row.append(
Table([[ip_para], [image]], colWidths=table_width, style=table_style)
)
# table_row.append(image)
# table_row_txt.append(ip_para)
if len(table_row) > 7:
# table_elems.append(table_row_txt)
# table_elems.append(table_row)
table_elems.append(table_row)
# table_row_txt = []
table_row = []
if not table_row == []:
table_elems.append(table_row)
return table_elems

96
tools/bad_board_util_old/func/ui.py
View File

@@ -1,96 +0,0 @@
import ipaddress
import re
from tools.bad_board_util_old.layout import window
import pyperclip
def table_select_all():
window["ip_table"].update(
select_rows=([row for row in range(len(window["ip_table"].Values))])
)
def copy_from_table(table):
selection = table.selection()
copy_values = []
for each in selection:
try:
# value = table.item(each)["values"][0]
table_values = table.item(each)["values"]
ip = table_values[0]
model = table_values[1]
total = str(table_values[2])
l_brd_chips = str(table_values[3])
c_brd_chips = str(table_values[5])
r_brd_chips = str(table_values[7])
all_values = [ip, model, total, l_brd_chips, c_brd_chips, r_brd_chips]
value = ", ".join(all_values)
copy_values.append(str(value))
except Exception as e:
print("Copy Error:", e)
copy_string = "\n".join(copy_values)
pyperclip.copy(copy_string)
async def update_ui_with_data(key, message, append=False):
if append:
message = window[key].get_text() + message
window[key].update(message)
async def update_prog_bar(amount):
window["progress"].Update(amount)
percent_done = 100 * (amount / window["progress"].maxlen)
window["progress_percent"].Update(f"{round(percent_done, 2)} %")
if percent_done == 100:
window["progress_percent"].Update("")
async def set_progress_bar_len(amount):
window["progress"].Update(0, max=amount)
window["progress"].maxlen = amount
window["progress_percent"].Update("0.0 %")
async def sort_data(index: int or str):
if window["scan"].Disabled:
return
await update_ui_with_data("status", "Sorting Data")
data_list = window["ip_table"].Values
table = window["ip_table"].Widget
all_data = []
for idx, item in enumerate(data_list):
all_data.append({"data": item, "tags": table.item(int(idx) + 1)["tags"]})
# ip addresses
if re.match(
"^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)",
str(all_data[0]["data"][index]),
):
new_list = sorted(
all_data, key=lambda x: ipaddress.ip_address(x["data"][index])
)
if all_data == new_list:
new_list = sorted(
all_data,
reverse=True,
key=lambda x: ipaddress.ip_address(x["data"][index]),
)
# everything else, model, chips
else:
new_list = sorted(all_data, key=lambda x: x["data"][index])
if all_data == new_list:
new_list = sorted(all_data, reverse=True, key=lambda x: x["data"][index])
new_data = []
for item in new_list:
new_data.append(item["data"])
await update_ui_with_data("ip_table", new_data)
for idx, item in enumerate(new_list):
table.item(idx + 1, tags=item["tags"])
await update_ui_with_data("status", "")

27
tools/bad_board_util_old/img/__init__.py
View File

File diff suppressed because one or more lines are too long

71
tools/bad_board_util_old/layout.py
View File

File diff suppressed because one or more lines are too long

90
tools/bad_board_util_old/ui.py
View File

@@ -1,90 +0,0 @@
import asyncio
import sys
import PySimpleGUI as sg
from tools.bad_board_util_old.layout import window
from tools.bad_board_util_old.func.miners import (
refresh_data,
scan_and_get_data,
miner_light,
)
from tools.bad_board_util_old.func.files import import_iplist, export_iplist
from tools.bad_board_util_old.func.pdf import save_report
from tools.bad_board_util_old.func.ui import (
sort_data,
copy_from_table,
table_select_all,
)
from network import MinerNetwork
import webbrowser
async def ui():
window.read(timeout=0)
table = window["ip_table"].Widget
table.bind("<Control-Key-c>", lambda x: copy_from_table(table))
table.bind("<Control-Key-a>", lambda x: table_select_all())
# light tag shows red row for fault lights
table.tag_configure("bad", foreground="white", background="orange")
table.tag_configure("light", foreground="white", background="red")
table.tag_configure("light+bad", foreground="white", background="red")
while True:
event, value = window.read(timeout=0)
if event in (None, "Close", sg.WIN_CLOSED):
sys.exit()
if isinstance(event, tuple):
if len(window["ip_table"].Values) > 0:
if event[0] == "ip_table":
if event[2][0] == -1:
await sort_data(event[2][1])
if event == "open_in_web":
for row in value["ip_table"]:
webbrowser.open("http://" + window["ip_table"].Values[row][0])
if event == "scan":
if len(value["miner_network"].split("/")) > 1:
network = value["miner_network"].split("/")
miner_network = MinerNetwork(ip_addr=network[0], mask=network[1])
else:
miner_network = MinerNetwork(value["miner_network"])
asyncio.create_task(scan_and_get_data(miner_network))
if event == "save_report":
if not value["save_report"] == "":
asyncio.create_task(save_report(value["save_report"]))
window["save_report"].update("")
if event == "select_all_ips":
if len(value["ip_table"]) == len(window["ip_table"].Values):
window["ip_table"].update(select_rows=())
else:
window["ip_table"].update(
select_rows=([row for row in range(len(window["ip_table"].Values))])
)
if event == "light":
if len(window["ip_table"].Values) > 0:
asyncio.create_task(
miner_light(
[
window["ip_table"].Values[item][0]
for item in value["ip_table"]
]
)
)
if event == "import_iplist":
asyncio.create_task(import_iplist(value["file_iplist"]))
if event == "export_iplist":
asyncio.create_task(
export_iplist(
value["file_iplist"],
[window["ip_table"].Values[item][0] for item in value["ip_table"]],
)
)
if event == "refresh_data":
asyncio.create_task(
refresh_data(
[window["ip_table"].Values[item][0] for item in value["ip_table"]]
)
)
if event == "__TIMEOUT__":
await asyncio.sleep(0)

1
tools/cfg_util_old/__init__.py
View File

@@ -1 +0,0 @@
from tools.cfg_util_old.cfg_util_sg import main

26
tools/cfg_util_old/cfg_util_sg/__init__.py
View File

@@ -1,26 +0,0 @@
import asyncio
import sys
import logging
from tools.cfg_util_old.cfg_util_sg.ui import ui
# initialize logger and get settings
from logger import logger
logger.info("Initializing logger for CFG Util.")
# Fix bug with some whatsminers and asyncio because of a socket not being shut down:
if (
sys.version_info[0] == 3
and sys.version_info[1] >= 8
and sys.platform.startswith("win")
):
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
def main():
logging.info("Starting CFG Util.")
loop = asyncio.new_event_loop()
loop.run_until_complete(ui())
logging.info("Closing CFG Util.")

0
tools/cfg_util_old/cfg_util_sg/func/__init__.py
View File

37
tools/cfg_util_old/cfg_util_sg/func/decorators.py
View File

@@ -1,37 +0,0 @@
from tools.cfg_util_old.cfg_util_sg.layout import window
def disable_buttons(func):
button_list = [
"scan",
"import_file_config",
"export_file_config",
"import_iplist",
"export_iplist",
"export_csv",
"select_all_ips",
"refresh_data",
"open_in_web",
"reboot_miners",
"restart_miner_backend",
"import_config",
"send_config",
"light",
"generate_config",
"send_miner_ssh_command_window",
]
# handle the inner function that the decorator is wrapping
async def inner(*args, **kwargs):
# disable the buttons
for button in button_list:
window[button].Update(disabled=True)
# call the original wrapped function
await func(*args, **kwargs)
# re-enable the buttons after the wrapped function completes
for button in button_list:
window[button].Update(disabled=False)
return inner

96
tools/cfg_util_old/cfg_util_sg/func/files.py
View File

@@ -1,96 +0,0 @@
import ipaddress
import os
import re
import time
import aiofiles
import toml
from tools.cfg_util_old.cfg_util_sg.func.ui import update_ui_with_data
from tools.cfg_util_old.cfg_util_sg.layout import window
from config.bos import bos_config_convert, general_config_convert_bos
async def import_iplist(file_location):
await update_ui_with_data("status", "Importing")
if not os.path.exists(file_location):
return
else:
ip_list = []
async with aiofiles.open(file_location, mode="r") as file:
async for line in file:
ips = [
x.group()
for x in re.finditer(
"^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)",
line,
)
]
for ip in ips:
if ip not in ip_list:
ip_list.append(ipaddress.ip_address(ip))
ip_list.sort()
window["ip_table"].update([[str(ip), "", "", "", ""] for ip in ip_list])
await update_ui_with_data("ip_count", str(len(ip_list)))
await update_ui_with_data("status", "")
async def export_csv(file_location, ip_list_selected):
await update_ui_with_data("status", "Exporting")
if not os.path.exists(file_location):
return
else:
if ip_list_selected is not None and not ip_list_selected == []:
async with aiofiles.open(file_location, mode="w") as file:
for item in ip_list_selected:
await file.write(
str(", ".join([str(part).rstrip().lstrip() for part in item]))
+ "\n"
)
else:
async with aiofiles.open(file_location, mode="w") as file:
for item in window["ip_table"].Values:
await file.write(
str(", ".join([str(part).rstrip().lstrip() for part in item]))
+ "\n"
)
await update_ui_with_data("status", "")
async def export_iplist(file_location, ip_list_selected):
await update_ui_with_data("status", "Exporting")
if not os.path.exists(file_location):
return
else:
if ip_list_selected is not None and not ip_list_selected == []:
async with aiofiles.open(file_location, mode="w") as file:
for item in ip_list_selected:
await file.write(str(item) + "\n")
else:
async with aiofiles.open(file_location, mode="w") as file:
for item in window["ip_table"].Values:
await file.write(str(item[0]) + "\n")
await update_ui_with_data("status", "")
async def import_config_file(file_location):
await update_ui_with_data("status", "Importing")
if not os.path.exists(file_location):
return
else:
async with aiofiles.open(file_location, mode="r") as file:
config = await file.read()
await update_ui_with_data("config", bos_config_convert(toml.loads(config)))
await update_ui_with_data("status", "")
async def export_config_file(file_location, config):
await update_ui_with_data("status", "Exporting")
config = toml.dumps(general_config_convert_bos(config))
config = toml.loads(config)
config["format"]["generator"] = "upstream_config_util"
config["format"]["timestamp"] = int(time.time())
config = toml.dumps(config)
async with aiofiles.open(file_location, mode="w+") as file:
await file.write(config)
await update_ui_with_data("status", "")

583
tools/cfg_util_old/cfg_util_sg/func/miners.py
View File

@@ -1,583 +0,0 @@
import asyncio
import ipaddress
import time
import warnings
import logging
from API import APIError
from tools.cfg_util_old.cfg_util_sg.func.parse_data import safe_parse_api_data
from tools.cfg_util_old.cfg_util_sg.func.ui import (
update_ui_with_data,
update_prog_bar,
set_progress_bar_len,
)
from tools.cfg_util_old.cfg_util_sg.layout import window
from miners.miner_factory import MinerFactory
from config.bos import bos_config_convert
from tools.cfg_util_old.cfg_util_sg.func.decorators import disable_buttons
from settings import (
CFG_UTIL_CONFIG_THREADS as CONFIG_THREADS,
CFG_UTIL_REBOOT_THREADS as REBOOT_THREADS,
)
async def import_config(idx):
await update_ui_with_data("status", "Importing")
miner_ip = window["ip_table"].Values[idx[0]][0]
logging.debug(f"{miner_ip}: Importing config.")
miner = await MinerFactory().get_miner(ipaddress.ip_address(miner_ip))
await miner.get_config()
config = miner.config
await update_ui_with_data("config", str(config))
logging.debug(f"{miner_ip}: Config import completed.")
await update_ui_with_data("status", "")
@disable_buttons
async def scan_network(network):
await update_ui_with_data("status", "Scanning")
await update_ui_with_data("ip_count", "")
await update_ui_with_data("hr_total", "")
window["ip_table"].update([])
network_size = len(network)
miner_generator = network.scan_network_generator()
await set_progress_bar_len(2 * network_size)
progress_bar_len = 0
asyncio.create_task(update_prog_bar(progress_bar_len))
miners = []
async for miner in miner_generator:
if miner:
miners.append(miner)
# can output "Identifying" for each found item, but it gets a bit cluttered
# and could possibly be confusing for the end user because of timing on
# adding the IPs
# window["ip_table"].update([["Identifying...", "", "", "", ""] for miner in miners])
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
progress_bar_len += network_size - len(miners)
asyncio.create_task(update_prog_bar(progress_bar_len))
get_miner_genenerator = MinerFactory().get_miner_generator(miners)
all_miners = []
async for found_miner in get_miner_genenerator:
all_miners.append(found_miner)
all_miners.sort(key=lambda x: x.ip)
window["ip_table"].update([[str(miner.ip)] for miner in all_miners])
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
await update_ui_with_data("ip_count", str(len(all_miners)))
await update_ui_with_data("status", "")
@disable_buttons
async def miner_light(ips: list):
await asyncio.gather(*[flip_light(ip) for ip in ips])
async def flip_light(ip):
ip_list = window["ip_table"].Widget
miner = await MinerFactory().get_miner(ip)
index = [item[0] for item in window["ip_table"].Values].index(ip)
index_tags = ip_list.item(index + 1)["tags"]
if "light" not in index_tags:
index_tags.append("light")
ip_list.item(index + 1, tags=index_tags)
await miner.fault_light_on()
else:
index_tags.remove("light")
ip_list.item(index + 1, tags=index_tags)
await miner.fault_light_off()
async def send_command_generator(miners: list, command: str):
loop = asyncio.get_event_loop()
command_tasks = []
for miner in miners:
if len(command_tasks) >= CONFIG_THREADS:
cmd_sent = asyncio.as_completed(command_tasks)
command_tasks = []
for done in cmd_sent:
yield await done
command_tasks.append(loop.create_task(send_ssh_command(miner, command)))
cmd_sent = asyncio.as_completed(command_tasks)
for done in cmd_sent:
yield await done
async def send_ssh_command(miner, command: str):
proc = await miner.send_ssh_command(command)
return {"IP": miner.ip, "proc": proc}
async def send_miners_ssh_commands(ips: list, command: str, ssh_cmd_window):
get_miner_genenerator = MinerFactory().get_miner_generator(ips)
all_miners = []
async for miner in get_miner_genenerator:
all_miners.append(miner)
data = []
send_cmd_generator = send_command_generator(all_miners, command)
async for command_sent in send_cmd_generator:
data.append(command_sent)
proc_table_data = [[ip, ""] for ip in ips]
for item in data:
if item["proc"].returncode == 0:
return_data = item["proc"].stdout
else:
return_data = item["proc"].stderr
if str(item["IP"]) in ips:
proc_table_index = ips.index(str(item["IP"]))
proc_table_data[proc_table_index] = [
str(item["IP"]),
return_data.replace("\n", " "),
]
ssh_cmd_window["ssh_cmd_table"].update(proc_table_data)
async def reboot_generator(miners: list):
loop = asyncio.get_event_loop()
reboot_tasks = []
for miner in miners:
if len(reboot_tasks) >= REBOOT_THREADS:
rebooted = asyncio.as_completed(reboot_tasks)
reboot_tasks = []
for done in rebooted:
yield await done
reboot_tasks.append(loop.create_task(miner.reboot()))
rebooted = asyncio.as_completed(reboot_tasks)
for done in rebooted:
yield await done
@disable_buttons
async def reboot_miners(ips: list):
await update_ui_with_data("status", "Rebooting")
await set_progress_bar_len(2 * len(ips))
progress_bar_len = 0
get_miner_genenerator = MinerFactory().get_miner_generator(ips)
all_miners = []
async for miner in get_miner_genenerator:
all_miners.append(miner)
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
reboot_miners_generator = reboot_generator(all_miners)
async for _rebooter in reboot_miners_generator:
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
await update_ui_with_data("status", "")
async def restart_backend_generator(miners: list):
loop = asyncio.get_event_loop()
reboot_tasks = []
for miner in miners:
if len(reboot_tasks) >= REBOOT_THREADS:
rebooted = asyncio.as_completed(reboot_tasks)
reboot_tasks = []
for done in rebooted:
yield await done
reboot_tasks.append(loop.create_task(miner.restart_backend()))
rebooted = asyncio.as_completed(reboot_tasks)
for done in rebooted:
yield await done
@disable_buttons
async def restart_miners_backend(ips: list):
await update_ui_with_data("status", "Restarting Backends")
await set_progress_bar_len(2 * len(ips))
progress_bar_len = 0
get_miner_genenerator = MinerFactory().get_miner_generator(ips)
all_miners = []
async for miner in get_miner_genenerator:
all_miners.append(miner)
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
restart_backend_gen = restart_backend_generator(all_miners)
async for _rebooter in restart_backend_gen:
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
await update_ui_with_data("status", "")
async def send_config_generator(miners: list, config, last_octet_ip_user: bool = False):
loop = asyncio.get_event_loop()
config_tasks = []
for miner in miners:
if len(config_tasks) >= CONFIG_THREADS:
configured = asyncio.as_completed(config_tasks)
config_tasks = []
for sent_config in configured:
yield await sent_config
config_tasks.append(
loop.create_task(miner.send_config(config, ip_user=last_octet_ip_user))
)
configured = asyncio.as_completed(config_tasks)
for sent_config in configured:
yield await sent_config
@disable_buttons
async def send_config(ips: list, config, last_octet_ip: bool = False):
await update_ui_with_data("status", "Configuring")
await set_progress_bar_len(2 * len(ips))
progress_bar_len = 0
asyncio.create_task(update_prog_bar(progress_bar_len))
get_miner_genenerator = MinerFactory().get_miner_generator(ips)
all_miners = []
async for miner in get_miner_genenerator:
all_miners.append(miner)
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
config_sender_generator = send_config_generator(
all_miners, config, last_octet_ip_user=last_octet_ip
)
async for _config_sender in config_sender_generator:
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
await update_ui_with_data("status", "Getting Data")
await asyncio.sleep(3)
await refresh_data(ips)
@disable_buttons
async def refresh_data(ip_list: list):
await update_ui_with_data("status", "Getting Data")
await update_ui_with_data("hr_total", "")
ips = [ipaddress.ip_address(ip) for ip in ip_list]
if len(ips) == 0:
ips = [
ipaddress.ip_address(ip)
for ip in [item[0] for item in window["ip_table"].Values]
]
await set_progress_bar_len(len(ips))
progress_bar_len = 0
asyncio.create_task(update_prog_bar(progress_bar_len))
reset_table_values = []
for item in window["ip_table"].Values:
if item[0] in ip_list:
reset_table_values.append([item[0]])
else:
reset_table_values.append(item)
window["ip_table"].update(reset_table_values)
progress_bar_len = 0
data_gen = asyncio.as_completed([get_formatted_data(miner) for miner in ips])
ip_table_data = window["ip_table"].Values
ordered_all_ips = [item[0] for item in ip_table_data]
for all_data in data_gen:
data_point = await all_data
if data_point["IP"] in ordered_all_ips:
ip_table_index = ordered_all_ips.index(data_point["IP"])
ip_table_data[ip_table_index] = [
data_point["IP"],
data_point["model"],
data_point["host"],
str(data_point["TH/s"]) + " TH/s ",
data_point["temp"],
data_point["user"],
str(data_point["wattage"]) + " W",
]
window["ip_table"].update(ip_table_data)
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
hashrate_list = []
hr_idx = 3
for item, _ in enumerate(window["ip_table"].Values):
if len(window["ip_table"].Values[item]) > hr_idx:
if not window["ip_table"].Values[item][hr_idx] == "":
hashrate_list.append(
float(window["ip_table"].Values[item][hr_idx].replace(" TH/s ", ""))
)
else:
hashrate_list.append(0)
else:
hashrate_list.append(0)
total_hr = round(sum(hashrate_list), 2)
window["hr_total"].update(f"{total_hr} TH/s")
await update_ui_with_data("status", "")
@disable_buttons
async def scan_and_get_data(network):
await update_ui_with_data("status", "Scanning")
await update_ui_with_data("hr_total", "")
await update_ui_with_data("ip_count", "")
await update_ui_with_data("ip_table", [])
network_size = len(network)
miner_generator = network.scan_network_generator()
MinerFactory().clear_cached_miners()
logging.info(f"Scanning network: {str(network)}")
await set_progress_bar_len(3 * network_size)
progress_bar_len = 0
miners = []
async for miner in miner_generator:
if miner:
miners.append(miner)
# can output "Identifying" for each found item, but it gets a bit cluttered
# and could possibly be confusing for the end user because of timing on
# adding the IPs
# window["ip_table"].update([["Identifying..."] for miner in miners])
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
logging.info(f"Found {len(miners)} Miners")
logging.debug(f"Found miners: {miners}")
progress_bar_len += network_size - len(miners)
asyncio.create_task(update_prog_bar(progress_bar_len))
get_miner_genenerator = MinerFactory().get_miner_generator(miners)
all_miners = []
async for found_miner in get_miner_genenerator:
all_miners.append(found_miner)
all_miners.sort(key=lambda x: x.ip)
window["ip_table"].update([[str(miner.ip)] for miner in all_miners])
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
logging.info(f"Resolved {len(all_miners)} Miners")
logging.debug(f"Resolved to miner types: {all_miners}")
await update_ui_with_data("ip_count", str(len(all_miners)))
data_gen = asyncio.as_completed([get_formatted_data(miner) for miner in miners])
ip_table_data = window["ip_table"].Values
ordered_all_ips = [item[0] for item in ip_table_data]
progress_bar_len += network_size - len(miners)
asyncio.create_task(update_prog_bar(progress_bar_len))
await update_ui_with_data("status", "Getting Data")
logging.debug("Getting data on miners.")
for all_data in data_gen:
data_point = await all_data
if data_point["IP"] in ordered_all_ips:
ip_table_index = ordered_all_ips.index(data_point["IP"])
ip_table_data[ip_table_index] = [
data_point["IP"],
data_point["model"],
data_point["host"],
str(data_point["TH/s"]) + " TH/s ",
data_point["temp"],
data_point["user"],
str(data_point["wattage"]) + " W",
]
window["ip_table"].update(ip_table_data)
progress_bar_len += 1
asyncio.create_task(update_prog_bar(progress_bar_len))
hashrate_list = [
float(item[3].replace(" TH/s ", ""))
for item in window["ip_table"].Values
if not item[3] == ""
]
total_hr = round(sum(hashrate_list), 2)
await update_ui_with_data("hr_total", f"{total_hr} TH/s")
await update_ui_with_data("status", "")
async def get_formatted_data(ip: ipaddress.ip_address):
miner = await MinerFactory().get_miner(ip)
logging.debug(f"Getting data for miner: {miner.ip}")
warnings.filterwarnings("ignore")
miner_data = None
host = await miner.get_hostname()
try:
model = await miner.get_model()
except APIError:
model = "?"
if not model:
model = "?"
temps = 0
th5s = 0
wattage = 0
user = "?"
try:
miner_data = await miner.api.multicommand(
"summary", "devs", "temps", "tunerstatus", "pools", "stats"
)
except APIError:
try:
# no devs command, it will fail in this case
miner_data = await miner.api.multicommand(
"summary", "temps", "tunerstatus", "pools", "stats"
)
except APIError as e:
logging.warning(f"{str(ip)}: {e}")
return {
"TH/s": 0,
"IP": str(miner.ip),
"model": "Unknown",
"temp": 0,
"host": "Unknown",
"user": "Unknown",
"wattage": 0,
}
if miner_data:
logging.info(f"Received miner data for miner: {miner.ip}")
# get all data from summary
if "summary" in miner_data.keys():
if (
not miner_data["summary"][0].get("SUMMARY") == []
and "SUMMARY" in miner_data["summary"][0].keys()
):
# temperature data, this is the idea spot to get this
if "Temperature" in miner_data["summary"][0]["SUMMARY"][0].keys():
if (
not round(miner_data["summary"][0]["SUMMARY"][0]["Temperature"])
== 0
):
temps = miner_data["summary"][0]["SUMMARY"][0]["Temperature"]
# hashrate data
if "MHS av" in miner_data["summary"][0]["SUMMARY"][0].keys():
th5s = format(
round(
await safe_parse_api_data(
miner_data, "summary", 0, "SUMMARY", 0, "MHS av"
)
/ 1000000,
2,
),
".2f",
).rjust(6, " ")
elif "GHS av" in miner_data["summary"][0]["SUMMARY"][0].keys():
if not miner_data["summary"][0]["SUMMARY"][0]["GHS av"] == "":
th5s = format(
round(
float(
await safe_parse_api_data(
miner_data, "summary", 0, "SUMMARY", 0, "GHS av"
)
)
/ 1000,
2,
),
".2f",
).rjust(6, " ")
# alternate temperature data, for BraiinsOS
if "temps" in miner_data.keys():
if not miner_data["temps"][0].get("TEMPS") == []:
if "Chip" in miner_data["temps"][0]["TEMPS"][0].keys():
for board in miner_data["temps"][0]["TEMPS"]:
if board["Chip"] is not None and not board["Chip"] == 0.0:
temps = board["Chip"]
# alternate temperature data, for Whatsminers
if "devs" in miner_data.keys():
if not miner_data["devs"][0].get("DEVS") == []:
if "Chip Temp Avg" in miner_data["devs"][0]["DEVS"][0].keys():
for board in miner_data["devs"][0]["DEVS"]:
if (
board["Chip Temp Avg"] is not None
and not board["Chip Temp Avg"] == 0.0
):
temps = board["Chip Temp Avg"]
# alternate temperature data
if "stats" in miner_data.keys():
if not miner_data["stats"][0]["STATS"] == []:
for temp in ["temp2", "temp1", "temp3"]:
if temp in miner_data["stats"][0]["STATS"][1].keys():
if (
miner_data["stats"][0]["STATS"][1][temp] is not None
and not miner_data["stats"][0]["STATS"][1][temp] == 0.0
):
temps = miner_data["stats"][0]["STATS"][1][temp]
# alternate temperature data, for Avalonminers
miner_data["stats"][0]["STATS"][0].keys()
if any(
"MM ID" in string
for string in miner_data["stats"][0]["STATS"][0].keys()
):
temp_all = []
for key in [
string
for string in miner_data["stats"][0]["STATS"][0].keys()
if "MM ID" in string
]:
for value in [
string
for string in miner_data["stats"][0]["STATS"][0][key].split(" ")
if "TMax" in string
]:
temp_all.append(int(value.split("[")[1].replace("]", "")))
temps = round(sum(temp_all) / len(temp_all))
# pool information
if "pools" in miner_data.keys():
if not miner_data["pools"][0].get("POOLS") == []:
user = await safe_parse_api_data(
miner_data, "pools", 0, "POOLS", 0, "User"
)
else:
print(miner_data["pools"][0])
user = "Blank"
# braiins tuner status / wattage
if "tunerstatus" in miner_data.keys():
wattage = await safe_parse_api_data(
miner_data, "tunerstatus", 0, "TUNERSTATUS", 0, "PowerLimit"
)
elif "Power" in miner_data["summary"][0]["SUMMARY"][0].keys():
wattage = await safe_parse_api_data(
miner_data, "summary", 0, "SUMMARY", 0, "Power"
)
ret_data = {
"TH/s": th5s,
"IP": str(miner.ip),
"model": model,
"temp": round(temps),
"host": host,
"user": user,
"wattage": wattage,
}
logging.debug(f"{ret_data}")
return ret_data
async def generate_config(username, workername, v2_allowed):
if username and workername:
user = f"{username}.{workername}"
elif username and not workername:
user = username
else:
return
if v2_allowed:
url_1 = "stratum2+tcp://v2.us-east.stratum.slushpool.com/u95GEReVMjK6k5YqiSFNqqTnKU4ypU2Wm8awa6tmbmDmk1bWt"
url_2 = "stratum2+tcp://v2.stratum.slushpool.com/u95GEReVMjK6k5YqiSFNqqTnKU4ypU2Wm8awa6tmbmDmk1bWt"
url_3 = "stratum+tcp://stratum.slushpool.com:3333"
else:
url_1 = "stratum+tcp://ca.stratum.slushpool.com:3333"
url_2 = "stratum+tcp://us-east.stratum.slushpool.com:3333"
url_3 = "stratum+tcp://stratum.slushpool.com:3333"
config = {
"group": [
{
"name": "group",
"quota": 1,
"pool": [
{"url": url_1, "user": user, "password": "123"},
{"url": url_2, "user": user, "password": "123"},
{"url": url_3, "user": user, "password": "123"},
],
}
],
"format": {
"version": "1.2+",
"model": "Antminer S9",
"generator": "upstream_config_util",
"timestamp": int(time.time()),
},
"temp_control": {
"target_temp": 80.0,
"hot_temp": 90.0,
"dangerous_temp": 120.0,
},
"autotuning": {"enabled": True, "psu_power_limit": 900},
}
window["config"].update(bos_config_convert(config))

66
tools/cfg_util_old/cfg_util_sg/func/parse_data.py
View File

@@ -1,66 +0,0 @@
from API import APIError
# noinspection PyPep8
async def safe_parse_api_data(data: dict or list, *path: str or int, idx: int = 0):
path = [*path]
if len(path) == idx + 1:
if isinstance(path[idx], str):
if isinstance(data, dict):
if path[idx] in data.keys():
return data[path[idx]]
elif isinstance(path[idx], int):
if isinstance(data, list):
if len(data) > path[idx]:
return data[path[idx]]
else:
if isinstance(path[idx], str):
if isinstance(data, dict):
if path[idx] in data.keys():
parsed_data = await safe_parse_api_data(
data[path[idx]], idx=idx + 1, *path
)
# has to be == None, or else it fails on 0.0 hashrates
# noinspection PyPep8
if parsed_data == None:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return parsed_data
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False
elif isinstance(path[idx], int):
if isinstance(data, list):
if len(data) > path[idx]:
parsed_data = await safe_parse_api_data(
data[path[idx]], idx=idx + 1, *path
)
# has to be == None, or else it fails on 0.0 hashrates
# noinspection PyPep8
if parsed_data == None:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return parsed_data
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False

124
tools/cfg_util_old/cfg_util_sg/func/ui.py
View File

@@ -1,124 +0,0 @@
import ipaddress
import re
from tools.cfg_util_old.cfg_util_sg.layout import window
import pyperclip
def table_select_all():
window["ip_table"].update(
select_rows=([row for row in range(len(window["ip_table"].Values))])
)
def copy_from_table(table):
selection = table.selection()
copy_values = []
for each in selection:
try:
value = table.item(each)["values"][0]
copy_values.append(str(value))
except:
pass
copy_string = "\n".join(copy_values)
pyperclip.copy(copy_string)
def copy_from_ssh_table(table):
selection = table.selection()
copy_values = []
for each in selection:
try:
value = ", ".join(table.item(each)["values"])
copy_values.append(str(value))
except:
pass
copy_string = "\n".join(copy_values)
pyperclip.copy(copy_string)
async def update_ui_with_data(key, message, append=False):
if append:
message = window[key].get_text() + message
window[key].update(message)
async def update_prog_bar(amount):
window["progress"].Update(amount)
percent_done = 100 * (amount / window["progress"].maxlen)
window["progress_percent"].Update(f"{round(percent_done, 2)} %")
if percent_done == 100:
window["progress_percent"].Update("")
async def set_progress_bar_len(amount):
window["progress"].Update(0, max=amount)
window["progress"].maxlen = amount
window["progress_percent"].Update("0.0 %")
async def sort_data(index: int or str):
if window["scan"].Disabled:
return
await update_ui_with_data("status", "Sorting Data")
data_list = window["ip_table"].Values
table = window["ip_table"].Widget
all_data = []
for idx, item in enumerate(data_list):
all_data.append({"data": item, "tags": table.item(int(idx) + 1)["tags"]})
# wattage
if re.match("[0-9]* W", str(all_data[0]["data"][index])):
new_list = sorted(
all_data, key=lambda x: int(x["data"][index].replace(" W", ""))
)
if all_data == new_list:
new_list = sorted(
all_data,
reverse=True,
key=lambda x: int(x["data"][index].replace(" W", "")),
)
# hashrate
elif re.match("[0-9]*\.?[0-9]* TH\/s", str(all_data[0]["data"][index])):
new_list = sorted(
all_data, key=lambda x: float(x["data"][index].replace(" TH/s", ""))
)
if all_data == new_list:
new_list = sorted(
all_data,
reverse=True,
key=lambda x: float(x["data"][index].replace(" TH/s", "")),
)
# ip addresses
elif re.match(
"^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)",
str(all_data[0]["data"][index]),
):
new_list = sorted(
all_data, key=lambda x: ipaddress.ip_address(x["data"][index])
)
if all_data == new_list:
new_list = sorted(
all_data,
reverse=True,
key=lambda x: ipaddress.ip_address(x["data"][index]),
)
# everything else, hostname, temp, and user
else:
new_list = sorted(all_data, key=lambda x: x["data"][index])
if all_data == new_list:
new_list = sorted(all_data, reverse=True, key=lambda x: x["data"][index])
new_data = []
for item in new_list:
new_data.append(item["data"])
await update_ui_with_data("ip_table", new_data)
for idx, item in enumerate(new_list):
table.item(idx + 1, tags=item["tags"])
await update_ui_with_data("status", "")

163
tools/cfg_util_old/cfg_util_sg/layout.py
View File

File diff suppressed because one or more lines are too long

206
tools/cfg_util_old/cfg_util_sg/ui.py
View File

@@ -1,206 +0,0 @@
import asyncio
import sys
import PySimpleGUI as sg
import tkinter as tk
from tools.cfg_util_old.cfg_util_sg.layout import (
window,
generate_config_layout,
send_ssh_cmd_layout,
)
from tools.cfg_util_old.cfg_util_sg.func.miners import (
send_config,
miner_light,
refresh_data,
generate_config,
import_config,
scan_and_get_data,
restart_miners_backend,
reboot_miners,
send_miners_ssh_commands,
)
from tools.cfg_util_old.cfg_util_sg.func.files import (
import_iplist,
import_config_file,
export_iplist,
export_config_file,
export_csv,
)
from tools.cfg_util_old.cfg_util_sg.func.decorators import disable_buttons
from tools.cfg_util_old.cfg_util_sg.func.ui import (
sort_data,
copy_from_table,
table_select_all,
copy_from_ssh_table,
)
from network import MinerNetwork
import webbrowser
async def ui():
window.read(timeout=0)
table = window["ip_table"].Widget
table.bind("<Control-Key-c>", lambda x: copy_from_table(table))
table.bind("<Control-Key-a>", lambda x: table_select_all())
# light tag shows red row for fault lights
table.tag_configure("light", foreground="white", background="red")
# left justify the hostnames
table.column(2, anchor=tk.W)
while True:
event, value = window.read(timeout=0)
if event in (None, "Close", sg.WIN_CLOSED):
sys.exit()
if isinstance(event, tuple):
if len(window["ip_table"].Values) > 0:
if event[0] == "ip_table":
if event[2][0] == -1:
await sort_data(event[2][1])
if event == "open_in_web":
for row in value["ip_table"]:
webbrowser.open("http://" + window["ip_table"].Values[row][0])
if event == "scan":
if len(value["miner_network"].split("/")) > 1:
network = value["miner_network"].split("/")
miner_network = MinerNetwork(ip_addr=network[0], mask=network[1])
else:
miner_network = MinerNetwork(value["miner_network"])
asyncio.create_task(scan_and_get_data(miner_network))
if event == "select_all_ips":
if len(value["ip_table"]) == len(window["ip_table"].Values):
window["ip_table"].update(select_rows=())
else:
window["ip_table"].update(
select_rows=([row for row in range(len(window["ip_table"].Values))])
)
if event == "import_config":
if 2 > len(value["ip_table"]) > 0:
asyncio.create_task(import_config(value["ip_table"]))
if event == "restart_miner_backend":
if len(window["ip_table"].Values) > 0:
asyncio.create_task(
restart_miners_backend(
[
window["ip_table"].Values[item][0]
for item in value["ip_table"]
]
)
)
if event == "reboot_miners":
if len(window["ip_table"].Values) > 0:
asyncio.create_task(
reboot_miners(
[
window["ip_table"].Values[item][0]
for item in value["ip_table"]
]
)
)
if event == "send_miner_ssh_command_window":
ips = [window["ip_table"].Values[item][0] for item in value["ip_table"]]
if len(ips) == 0:
ips = [item[0] for item in window["ip_table"].Values]
if not len(ips) == 0:
await generate_ssh_cmd_ui(ips)
if event == "light":
if len(window["ip_table"].Values) > 0:
asyncio.create_task(
miner_light(
[
window["ip_table"].Values[item][0]
for item in value["ip_table"]
]
)
)
if event == "import_iplist":
asyncio.create_task(import_iplist(value["file_iplist"]))
if event == "export_iplist":
asyncio.create_task(
export_iplist(
value["file_iplist"],
[window["ip_table"].Values[item][0] for item in value["ip_table"]],
)
)
if event == "export_csv":
asyncio.create_task(
export_csv(
value["file_iplist"],
[window["ip_table"].Values[item] for item in value["ip_table"]],
)
)
if event == "send_config":
if len(window["ip_table"].Values) > 0:
asyncio.create_task(
send_config(
[
window["ip_table"].Values[item][0]
for item in value["ip_table"]
],
value["config"],
last_octet_ip=value["last_octet_user"],
)
)
if event == "import_file_config":
asyncio.create_task(import_config_file(value["file_config"]))
if event == "export_file_config":
asyncio.create_task(
export_config_file(value["file_config"], value["config"])
)
if event == "refresh_data":
if len(window["ip_table"].Values) > 0:
asyncio.create_task(
refresh_data(
[
window["ip_table"].Values[item][0]
for item in value["ip_table"]
]
)
)
if event == "generate_config":
await generate_config_ui()
if event == "__TIMEOUT__":
await asyncio.sleep(0)
async def generate_config_ui():
generate_config_window = sg.Window(
"Generate Config", generate_config_layout(), modal=True
)
while True:
event, values = generate_config_window.read()
if event in (None, "Close", sg.WIN_CLOSED):
break
if event == "generate_config_window_generate":
if values["generate_config_window_username"]:
await generate_config(
values["generate_config_window_username"],
values["generate_config_window_workername"],
values["generate_config_window_allow_v2"],
)
generate_config_window.close()
break
@disable_buttons
async def generate_ssh_cmd_ui(selected_miners: list):
ssh_cmd_window = sg.Window(
"Send Command", send_ssh_cmd_layout(selected_miners), modal=True
)
ssh_cmd_window.read(timeout=0)
table = ssh_cmd_window["ssh_cmd_table"].Widget
table.bind("<Control-Key-c>", lambda x: copy_from_ssh_table(table))
# left justify the results
table.column(1, anchor=tk.W)
while True:
event, values = ssh_cmd_window.read(timeout=0)
if event in (None, "Close", sg.WIN_CLOSED):
break
if event == "ssh_command_window_send_cmd":
asyncio.create_task(
send_miners_ssh_commands(
selected_miners, values["ssh_command_window_cmd"], ssh_cmd_window
)
)
if event == "__TIMEOUT__":
await asyncio.sleep(0)

0
tools/cfg_util_old/func/__init__.py
View File

66
tools/cfg_util_old/func/parse_data.py
View File

@@ -1,66 +0,0 @@
from API import APIError
# noinspection PyPep8
async def safe_parse_api_data(data: dict or list, *path: str or int, idx: int = 0):
path = [*path]
if len(path) == idx + 1:
if isinstance(path[idx], str):
if isinstance(data, dict):
if path[idx] in data.keys():
return data[path[idx]]
elif isinstance(path[idx], int):
if isinstance(data, list):
if len(data) > path[idx]:
return data[path[idx]]
else:
if isinstance(path[idx], str):
if isinstance(data, dict):
if path[idx] in data.keys():
parsed_data = await safe_parse_api_data(
data[path[idx]], idx=idx + 1, *path
)
# has to be == None, or else it fails on 0.0 hashrates
# noinspection PyPep8
if parsed_data == None:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return parsed_data
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False
elif isinstance(path[idx], int):
if isinstance(data, list):
if len(data) > path[idx]:
parsed_data = await safe_parse_api_data(
data[path[idx]], idx=idx + 1, *path
)
# has to be == None, or else it fails on 0.0 hashrates
# noinspection PyPep8
if parsed_data == None:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return parsed_data
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False
else:
if idx == 0:
raise APIError(
f"Data parsing failed on path index {idx} - \nKey: {path[idx]} \nData: {data}"
)
return False