fixed some issues in CFG-Util-README.md, and started reformatting docstrings and cleaning API code.

This commit is contained in:
UpstreamData
2022-01-31 15:35:08 -07:00
parent ca47f2817f
commit b50da98322
3 changed files with 352 additions and 177 deletions

View File

@@ -7,29 +7,38 @@ import re
import json import json
import hashlib import hashlib
import binascii import binascii
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.primitives.ciphers import \
Cipher, algorithms, modes
import base64 import base64
### IMPORTANT ### ### IMPORTANT ###
# you need to change the password of the miners using # you need to change the password of the miners using the whatsminer
# the whatsminer tool, then you can set them back to # tool, then you can set them back to admin with this tool, but they
# admin with this tool, but they must be changed to # must be changed to something else and set back to admin with this
# something else and set back to admin with this or # or the privileged API will not work using admin as the password. If
# the privileged API will not work using admin as # you change the password, you can pass that to the this class as pwd,
# the password. If you change the password, you can # or add it as the whatsminer_pwd in the settings.toml file.
# pass that to the this class as pwd, or added as
# whatsminer_pwd in the settings.toml file.
def _crypt(word: str, salt: str) -> str: def _crypt(word: str, salt: str) -> str:
"""Encrypts a word with a salt, using a standard salt format.
Encrypts a word using a salt with the format
'\s*\$(\d+)\$([\w\./]*)\$'. If this format is not used, a
ValueError is raised.
:param word: The word to be encrypted.
:param salt: The salt to encrypt the word.
:return: An MD5 hash of the word with the salt.
"""
# compile a standard format for the salt # compile a standard format for the salt
standard_salt = re.compile('\s*\$(\d+)\$([\w\./]*)\$') standard_salt = re.compile('\s*\$(\d+)\$([\w\./]*)\$')
# check if the salt matches # check if the salt matches
match = standard_salt.match(salt) match = standard_salt.match(salt)
# if the matching fails, the salt is incorrect # if the matching fails, the salt is incorrect
if not match: if not match:
raise ValueError("salt format is not correct") raise ValueError("Salt format is not correct.")
# save the matched salt in a new variable # save the matched salt in a new variable
new_salt = match.group(2) new_salt = match.group(2)
# encrypt the word with the salt using md5 # encrypt the word with the salt using md5
@@ -37,18 +46,37 @@ def _crypt(word: str, salt: str) -> str:
return result return result
def _add_to_16(s: str) -> bytes: def _add_to_16(string: str) -> bytes:
"""Add null bytes to a string until the length is 16""" """Add null bytes to a string until the length is a multiple 16
while len(s) % 16 != 0:
s += '\0' :param string: The string to lengthen to a multiple of 16 and
return str.encode(s) # return bytes encode.
:return: The input string as bytes with a multiple of 16 as the
length.
"""
while len(string) % 16 != 0:
string += '\0'
return str.encode(string) # return bytes
def parse_btminer_priviledge_data(token_data, data): def parse_btminer_priviledge_data(token_data: dict, data: dict):
"""Parses data returned from the BTMiner privileged API.
Parses data from the BTMiner privileged API using the the token
from the API in an AES format.
:param token_data: The token information from self.get_token().
:param data: The data to parse, returned from the API.
:return: A decoded dict version of the privileged command output.
"""
# get the encoded data from the dict # get the encoded data from the dict
enc_data = data['enc'] enc_data = data['enc']
# get the aes key from the token data # get the aes key from the token data
aeskey = hashlib.sha256(token_data['host_passwd_md5'].encode()).hexdigest() aeskey = hashlib.sha256(
token_data['host_passwd_md5'].encode()
).hexdigest()
# unhexlify the aes key # unhexlify the aes key
aeskey = binascii.unhexlify(aeskey.encode()) aeskey = binascii.unhexlify(aeskey.encode())
# create the required decryptor # create the required decryptor
@@ -62,10 +90,23 @@ def parse_btminer_priviledge_data(token_data, data):
def create_privileged_cmd(token_data: dict, command: dict) -> bytes: def create_privileged_cmd(token_data: dict, command: dict) -> bytes:
"""Create a privileged command to send to the BTMiner API.
Creates a privileged command using the token from the API and the
command as a dict of {'command': cmd}, with cmd being any command
that the miner API accepts.
:param token_data: The token information from self.get_token().
:param command: The command to turn into a privileged command.
:return: The encrypted privileged command to be sent to the miner.
"""
# add token to command # add token to command
command['token'] = token_data['host_sign'] command['token'] = token_data['host_sign']
# encode host_passwd data and get hexdigest # encode host_passwd data and get hexdigest
aeskey = hashlib.sha256(token_data['host_passwd_md5'].encode()).hexdigest() aeskey = hashlib.sha256(
token_data['host_passwd_md5'].encode()
).hexdigest()
# unhexlify the encoded host_passwd # unhexlify the encoded host_passwd
aeskey = binascii.unhexlify(aeskey.encode()) aeskey = binascii.unhexlify(aeskey.encode())
# create a new AES key # create a new AES key
@@ -74,7 +115,13 @@ def create_privileged_cmd(token_data: dict, command: dict) -> bytes:
# dump the command to json # dump the command to json
api_json_str = json.dumps(command) api_json_str = json.dumps(command)
# encode the json command with the aes key # encode the json command with the aes key
api_json_str_enc = base64.encodebytes(encryptor.update(_add_to_16(api_json_str))).decode("utf-8").replace("\n", "") api_json_str_enc = base64.encodebytes(
encryptor.update(
_add_to_16(
api_json_str
)
)
).decode("utf-8").replace("\n", "")
# label the data as being encoded # label the data as being encoded
data_enc = {'enc': 1, 'data': api_json_str_enc} data_enc = {'enc': 1, 'data': api_json_str_enc}
# dump the labeled data to json # dump the labeled data to json
@@ -83,23 +130,59 @@ def create_privileged_cmd(token_data: dict, command: dict) -> bytes:
class BTMinerAPI(BaseMinerAPI): class BTMinerAPI(BaseMinerAPI):
def __init__(self, ip, port=4028, pwd: str = "admin"): """An abstraction of the API for MicroBT Whatsminers, BTMiner.
This class abstracts use of the BTMiner API, as well as the
methods for sending commands to it. The self.send_command()
function handles sending a command to the miner asynchronously, and
as such is the base for many of the functions in this class, which
rely on it to send the command for them.
All privileged commands for BTMiner's API require that you change
the password of the miners using the whatsminer tool, and it can be
changed back to admin with this tool after. Set the new password
either by passing it to the __init__ method, or changing it in
settings.toml.
Additionally, the API commands for the privileged API must be
encoded using a token from the miner, all privileged commands do
this automatically for you and will decode the output to look like
a normal output from a miner API.
"""
def __init__(self, ip, port=4028, pwd: str = WHATSMINER_PWD):
super().__init__(ip, port) super().__init__(ip, port)
if pwd: self.admin_pwd = pwd
self.admin_pwd = pwd
else:
self.admin_pwd = WHATSMINER_PWD
self.current_token = None self.current_token = None
async def send_command(self, command: str | bytes, ignore_errors: bool = False, **kwargs) -> dict: async def send_command(self,
"""Send an API command to the miner and return the result.""" command: str | bytes,
# check if command is a string, if its bytes its encoded and needs to be send raw parameters: str or int or bool = None,
ignore_errors: bool = False
) -> dict:
"""Send a command to the miner API.
Send a command using an asynchronous connection, load the data,
parse encoded data if needed, and return the result.
:param command: The command to send to the miner.
:param parameters: Parameters to pass to the command.
:param ignore_errors: Ignore the E (Error) status code from the
API.
:return: The data received from the API after sending the
command.
"""
# check if command is a string
# if its bytes its encoded and needs to be sent raw
if isinstance(command, str): if isinstance(command, str):
# if it is a string, put it into the standard command format # if it is a string, put it into the standard command format
command = json.dumps({"command": command}).encode("utf-8") command = json.dumps({"command": command}).encode("utf-8")
try: try:
# get reader and writer streams # get reader and writer streams
reader, writer = await asyncio.open_connection(str(self.ip), self.port) reader, writer = await asyncio.open_connection(
str(self.ip),
self.port
)
# handle OSError 121 # handle OSError 121
except OSError as e: except OSError as e:
if e.winerror == "121": if e.winerror == "121":
@@ -129,11 +212,14 @@ class BTMinerAPI(BaseMinerAPI):
writer.close() writer.close()
await writer.wait_closed() await writer.wait_closed()
# check if th returned data is encoded # check if the returned data is encoded
if 'enc' in data.keys(): if 'enc' in data.keys():
# try to parse the encoded data # try to parse the encoded data
try: try:
data = parse_btminer_priviledge_data(self.current_token, data) data = parse_btminer_priviledge_data(
self.current_token,
data
)
except Exception as e: except Exception as e:
print(e) print(e)
@@ -147,20 +233,35 @@ class BTMinerAPI(BaseMinerAPI):
return data return data
async def get_token(self): async def get_token(self):
""" """Gets token information from the API.
API 'get_token' command.
Returns an encoded token and md5 password, which are used for the privileged API. :return: An encoded token and md5 password, which are used
for the privileged API.
""" """
# get the token
data = await self.send_command("get_token") data = await self.send_command("get_token")
# encrypt the admin password with the salt
pwd = _crypt(self.admin_pwd, "$1$" + data["Msg"]["salt"] + '$') pwd = _crypt(self.admin_pwd, "$1$" + data["Msg"]["salt"] + '$')
pwd = pwd.split('$') pwd = pwd.split('$')
# take the 4th item from the pwd split
host_passwd_md5 = pwd[3] host_passwd_md5 = pwd[3]
tmp = _crypt(pwd[3] + data["Msg"]["time"], "$1$" + data["Msg"]["newsalt"] + '$')
# encrypt the pwd with the time and new salt
tmp = _crypt(pwd[3] + data["Msg"]["time"],
"$1$" + data["Msg"]["newsalt"] + '$'
)
tmp = tmp.split('$') tmp = tmp.split('$')
# take the 4th item from the encrypted pwd split
host_sign = tmp[3] host_sign = tmp[3]
self.current_token = {'host_sign': host_sign, 'host_passwd_md5': host_passwd_md5}
return {'host_sign': host_sign, 'host_passwd_md5': host_passwd_md5} # set the current token
self.current_token = {'host_sign': host_sign,
'host_passwd_md5': host_passwd_md5
}
return self.current_token
#### PRIVILEGED COMMANDS #### #### PRIVILEGED COMMANDS ####
# Please read the top of this file to learn # Please read the top of this file to learn
@@ -168,9 +269,34 @@ class BTMinerAPI(BaseMinerAPI):
# use these commands. # use these commands.
async def update_pools(self, async def update_pools(self,
pool_1: str, worker_1: str, passwd_1: str, pool_1: str,
pool_2: str = None, worker_2: str = None, passwd_2: str = None, worker_1: str,
pool_3: str = None, worker_3: str = None, passwd_3: str = None): passwd_1: str,
pool_2: str = None,
worker_2: str = None,
passwd_2: str = None,
pool_3: str = None,
worker_3: str = None,
passwd_3: str = None
):
"""Update the pools of the miner using the API.
Update the pools of the miner using the API, only works after
changing the password of the miner using the whatsminer tool.
:param pool_1: The URL to update pool 1 to.
:param worker_1: The worker name for pool 1 to update to.
:param passwd_1: The password for pool 1 to update to.
:param pool_2: The URL to update pool 2 to.
:param worker_2: The worker name for pool 2 to update to.
:param passwd_2: The password for pool 2 to update to.
:param pool_3: The URL to update pool 3 to.
:param worker_3: The worker name for pool 3 to update to.
:param passwd_3: The password for pool 3 to update to.
:return: A dict from the API to confirm the pools were updated.
"""
# get the token and password from the miner # get the token and password from the miner
token_data = await self.get_token() token_data = await self.get_token()
@@ -180,20 +306,36 @@ class BTMinerAPI(BaseMinerAPI):
elif pool_2 and pool_3: elif pool_2 and pool_3:
command = { command = {
"cmd": "update_pools", "cmd": "update_pools",
"pool1": pool_1, "worker1": worker_1, "passwd1": passwd_1,
"pool2": pool_2, "worker2": worker_2, "passwd2": passwd_2, "pool1": pool_1,
"pool3": pool_3, "worker3": worker_3, "passwd3": passwd_3, "worker1": worker_1,
"passwd1": passwd_1,
"pool2": pool_2,
"worker2": worker_2,
"passwd2": passwd_2,
"pool3": pool_3,
"worker3": worker_3,
"passwd3": passwd_3,
} }
elif pool_2: elif pool_2:
command = { command = {
"cmd": "update_pools", "cmd": "update_pools",
"pool1": pool_1, "worker1": worker_1, "passwd1": passwd_1, "pool1": pool_1,
"pool2": pool_2, "worker2": worker_2, "passwd2": passwd_2 "worker1": worker_1,
"passwd1": passwd_1,
"pool2": pool_2,
"worker2": worker_2,
"passwd2": passwd_2
} }
else: else:
command = { command = {
"cmd": "update_pools", "cmd": "update_pools",
"pool1": pool_1, "worker1": worker_1, "passwd1": passwd_1, "pool1": pool_1,
"worker1": worker_1,
"passwd1": passwd_1,
} }
# encode the command with the token data # encode the command with the token data
enc_command = create_privileged_cmd(token_data, command) enc_command = create_privileged_cmd(token_data, command)
@@ -201,10 +343,12 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def restart(self): async def restart(self):
""" """Restart btminer using the API.
API 'restart_btminer' command
Returns a reply informing of the restart and restarts BTMiner. Restart btminer using the API, only works after changing
the password of the miner using the whatsminer tool.
:return: A reply informing of the restart.
""" """
command = {"cmd": "restart_btminer"} command = {"cmd": "restart_btminer"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -212,15 +356,13 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def power_off(self, respbefore: bool = True): async def power_off(self, respbefore: bool = True):
""" """Power off the miner using the API.
API 'power_off' command.
Powers off the mining of the miner. Power off the miner using the API, only works after changing
the password of the miner using the whatsminer tool.
Returns info on the power off. :param respbefore: Whether to respond before powering off.
:return: A reply informing of the status of powering off.
Parameters:
respbefore (optional): respond before powering off.
""" """
if respbefore: if respbefore:
command = {"cmd": "power_off", "respbefore": "true"} command = {"cmd": "power_off", "respbefore": "true"}
@@ -231,12 +373,12 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def power_on(self): async def power_on(self):
""" """Power on the miner using the API.
API 'power_on' command.
Powers on the mining of the miner. Power on the miner using the API, only works after changing
the password of the miner using the whatsminer tool.
Returns info on the power on. :return: A reply informing of the status of powering on.
""" """
command = {"cmd": "power_on"} command = {"cmd": "power_on"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -244,44 +386,53 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def reset_led(self): async def reset_led(self):
""" """Reset the LED on the miner using the API.
API 'reset_led' command.
Resets the LED flashing to normal. Reset the LED on the miner using the API, only works after
changing the password of the miner using the whatsminer tool.
Returns a confirmation of resetting the LED. :return: A reply informing of the status of resetting the LED.
""" """
command = {"cmd": "set_led", "param": "auto"} command = {"cmd": "set_led", "param": "auto"}
token_data = await self.get_token() token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command) enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def set_led(self, color: str = "red", period: int = 2000, duration: int = 1000, start: int = 0): async def set_led(self,
color: str = "red",
period: int = 2000,
duration: int = 1000,
start: int = 0
):
"""Set the LED on the miner using the API.
Set the LED on the miner using the API, only works after
changing the password of the miner using the whatsminer tool.
:param color: The LED color to set, either 'red' or 'green'.
:param period: The flash cycle in ms.
:param duration: LED on time in the cycle in ms.
:param start: LED on time offset in the cycle in ms.
:return: A reply informing of the status of setting the LED.
""" """
API 'set_led' command. command = {"cmd": "set_led",
"color": color,
Sets the LED to do some pattern set with parameters. "period": period,
"duration": duration,
Returns a confirmation of setting the LED. "start": start
}
Parameters:
color: 'red' or 'green'
period: flash cycle in ms
duration: led on time in the cycle in ms
start: led on time offset in the cycle in ms
"""
command = {"cmd": "set_led", "color": color, "period": period, "duration": duration, "start": start}
token_data = await self.get_token() token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command) enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def set_low_power(self): async def set_low_power(self):
""" """Set low power mode on the miner using the API.
API 'set_low_power' command.
Sets the miner to low power mode. Set low power mode on the miner using the API, only works after
changing the password of the miner using the whatsminer tool.
Returns the status of setting the miner to low power mode. :return: A reply informing of the status of setting low power
mode.
""" """
command = {"cmd": "set_low_power"} command = {"cmd": "set_low_power"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -294,12 +445,9 @@ class BTMinerAPI(BaseMinerAPI):
return NotImplementedError return NotImplementedError
async def reboot(self): async def reboot(self):
""" """Reboot the miner using the API.
API 'reboot' command.
Reboots the miner. :return: A reply informing of the status of the reboot.
Returns the status of the command then reboots.
""" """
command = {"cmd": "reboot"} command = {"cmd": "reboot"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -307,12 +455,9 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def factory_reset(self): async def factory_reset(self):
""" """Reset the miner to factory defaults.
API 'factory_reset' command.
Resets the miner to factory defaults. :return: A reply informing of the status of the reset.
Returns the status of the command then resets.
""" """
command = {"cmd": "factory_reset"} command = {"cmd": "factory_reset"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -320,48 +465,55 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def update_pwd(self, old_pwd: str, new_pwd: str): async def update_pwd(self, old_pwd: str, new_pwd: str):
""" """Update the admin user's password.
API 'update_pwd' command.
Updates the admin user's password. Update the admin user's password, only works after changing the
password of the miner using the whatsminer tool. New password
has a max length of 8 bytes, using letters, numbers, and
underscores.
Returns the status of setting the password to the new password. :param old_pwd: The old admin password.
:param new_pwd: The new password to set.
Parameters: :return: A reply informing of the status of setting the
old_pwd: the old admin password. password.
new_pwd: the new password to set. Max length of 8 bytes, using letters, numbers, and underscores.
""" """
# check if password length is greater than 8 bytes # check if password length is greater than 8 bytes
if len(new_pwd.encode('utf-8')) > 8: if len(new_pwd.encode('utf-8')) > 8:
return APIError( return APIError(
f"New password too long, the max length is 8. Password size: {len(new_pwd.encode('utf-8'))}") f"New password too long, the max length is 8. "
f"Password size: {len(new_pwd.encode('utf-8'))}")
command = {"cmd": "update_pwd", "old": old_pwd, "new": new_pwd} command = {"cmd": "update_pwd", "old": old_pwd, "new": new_pwd}
token_data = await self.get_token() token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command) enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def set_target_freq(self, percent: int): async def set_target_freq(self, percent: int):
""" """Update the target frequency.
API 'set_target_freq' command.
Sets the frequency for the miner ot use. Update the target frequency, only works after changing the
password of the miner using the whatsminer tool. The new
frequency must be between -10% and 100%.
Returns the status of setting the frequency. :param percent: The frequency % to set.
:return: A reply informing of the status of setting the
frequency.
""" """
if not -10 < percent < 100: if not -10 < percent < 100:
return APIError(f"Frequency % is outside of the allowed range. Please set a % between -10 and 100") return APIError(f"Frequency % is outside of the allowed "
f"range. Please set a % between -10 and "
f"100")
command = {"cmd": "set_target_freq", "percent": str(percent)} command = {"cmd": "set_target_freq", "percent": str(percent)}
token_data = await self.get_token() token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command) enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def enable_fast_boot(self): async def enable_fast_boot(self):
""" """Turn on fast boot.
API 'enable_fast_boot' command.
Turns on the fast boot feature on the miner. Turn on fast boot, only works after changing the password of
the miner using the whatsminer tool.
Returns the status of setting the fast boot to on. :return: A reply informing of the status of enabling fast boot.
""" """
command = {"cmd": "enable_btminer_fast_boot"} command = {"cmd": "enable_btminer_fast_boot"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -369,12 +521,12 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def disable_fast_boot(self): async def disable_fast_boot(self):
""" """Turn off fast boot.
API 'disable'_fast_boot' command.
Turns off the fast boot feature on the miner. Turn off fast boot, only works after changing the password of
the miner using the whatsminer tool.
Returns the status of setting the fast boot to off. :return: A reply informing of the status of disabling fast boot.
""" """
command = {"cmd": "disable_btminer_fast_boot"} command = {"cmd": "disable_btminer_fast_boot"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -382,12 +534,12 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def enable_web_pools(self): async def enable_web_pools(self):
""" """Turn on web pool updates.
API 'enable_web_pools' command.
Turns on the ability to change the pools through the web interface. Turn on web pool updates, only works after changing the
password of the miner using the whatsminer tool.
Returns the status of setting the web pools to enabled. :return: A reply informing of the status of enabling web pools.
""" """
command = {"cmd": "enable_web_pools"} command = {"cmd": "enable_web_pools"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -395,12 +547,13 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def disable_web_pools(self): async def disable_web_pools(self):
""" """Turn off web pool updates.
API 'disable_web_pools' command.
Turns off the ability to change the pools through the web interface. Turn off web pool updates, only works after changing the
password of the miner using the whatsminer tool.
Returns the status of setting the web pools to disabled. :return: A reply informing of the status of disabling web
pools.
""" """
command = {"cmd": "disable_web_pools"} command = {"cmd": "disable_web_pools"}
token_data = await self.get_token() token_data = await self.get_token()
@@ -408,12 +561,15 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def set_hostname(self, hostname: str): async def set_hostname(self, hostname: str):
""" """Set the hostname of the miner.
API 'set_hostname' command.
Sets the hostname of the miner. Set the hostname of the miner, only works after changing the
password of the miner using the whatsminer tool.
Returns the status of setting the hostname.
:param hostname: The new hostname to use.
:return: A reply informing of the status of setting the
hostname.
""" """
command = {"cmd": "set_hostname", "hostname": hostname} command = {"cmd": "set_hostname", "hostname": hostname}
token_data = await self.get_token() token_data = await self.get_token()
@@ -421,47 +577,57 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def set_power_pct(self, percent: int): async def set_power_pct(self, percent: int):
""" """Set the power percentage of the miner.
API 'set_power_pct' command.
Sets the percent of power the miner should use. Set the power percentage of the miner, only works after changing
the password of the miner using the whatsminer tool.
Returns the status of setting the power usage to this percent. :param percent: The power percentage to set.
:return: A reply informing of the status of setting the
Parameters: power percentage.
percent: the percent to set the power usage to, between 0 and 100.
""" """
if not 0 < percent < 100: if not 0 < percent < 100:
return APIError(f"Power PCT % is outside of the allowed range. Please set a % between 0 and 100") return APIError(f"Power PCT % is outside of the allowed "
f"range. Please set a % between 0 and "
f"100")
command = {"cmd": "set_power_pct", "percent": str(percent)} command = {"cmd": "set_power_pct", "percent": str(percent)}
token_data = await self.get_token() token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command) enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command) return await self.send_command(enc_command)
async def pre_power_on(self, complete: bool, msg: str): async def pre_power_on(self, complete: bool, msg: str):
""" """Configure or check status of pre power on.
API 'pre_power_on' command.
Preheats the miner for the 'power_on' command. Can also be used to query the status of pre powering on. Configure or check status of pre power on, only works after
changing the password of the miner using the whatsminer tool.
Returns status of pre powering on. :param complete: check whether pre power on is complete.
:param msg: the message to check.
Parameters: "wait for adjust temp" or
complete: check whether or not it is complete. "adjust complete" or
msg: the message to check. "wait for adjust temp" or "adjust complete" or "adjust continue" "adjust continue"
:return: A reply informing of the status of pre power on.
""" """
if not msg == "wait for adjust temp" or "adjust complete" or "adjust continue": if not msg == \
"wait for adjust temp" or \
"adjust complete" or \
"adjust continue":
return APIError( return APIError(
'Message is incorrect, please choose one of ' 'Message is incorrect, please choose one of '
'["wait for adjust temp", "adjust complete", "adjust continue"]' '["wait for adjust temp", '
'"adjust complete", '
'"adjust continue"]'
) )
if complete: if complete:
complete = "true" complete = "true"
else: else:
complete = "false" complete = "false"
command = {"cmd": "pre_power_on", "complete": complete, "msg": msg} command = {"cmd": "pre_power_on",
"complete": complete,
"msg": msg
}
token_data = await self.get_token() token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command) enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command) return await self.send_command(enc_command)
@@ -469,82 +635,76 @@ class BTMinerAPI(BaseMinerAPI):
#### END privileged COMMANDS #### #### END privileged COMMANDS ####
async def summary(self): async def summary(self):
""" """Get the summary status from the miner.
API 'summary' command.
Returns a dict containing the status summary of the miner. :return: Summary status of the miner.
""" """
return await self.send_command("summary") return await self.send_command("summary")
async def pools(self): async def pools(self):
""" """Get the pool status from the miner.
API 'pools' command.
Returns a dict containing the status of each pool. :return: Pool status of the miner.
""" """
return await self.send_command("pools") return await self.send_command("pools")
async def devs(self): async def devs(self):
""" """Get data on each PGA/ASC with their details.
API 'devs' command.
Returns a dict containing each PGA/ASC with their details. :returns: Data on each PGA/ASC with their details.
""" """
return await self.send_command("devs") return await self.send_command("devs")
async def edevs(self): async def edevs(self):
""" """Get data on each PGA/ASC with their details, ignoring
API 'edevs' command. blacklisted and zombie devices.
Returns a dict containing each PGA/ASC with their details, :returns: Data on each PGA/ASC with their details.
ignoring blacklisted devices and zombie devices.
""" """
return await self.send_command("edevs") return await self.send_command("edevs")
async def devdetails(self): async def devdetails(self):
""" """Get data on all devices with their static details.
API 'devdetails' command.
Returns a dict containing all devices with their static details. :returns: Data on all devices with their static details.
""" """
return await self.send_command("devdetails") return await self.send_command("devdetails")
async def get_psu(self): async def get_psu(self):
""" """Get data on the PSU and power information.
API 'get_psu' command.
Returns a dict containing PSU and power information. :returns: Data on the PSU and power information.
""" """
return await self.send_command("get_psu") return await self.send_command("get_psu")
async def version(self): async def version(self):
""" """Get version data for the miner.
API 'get_version' command.
Returns a dict containing version information. Get version data for the miner. This calls another function,
self.get_version(), but is named version to stay consistent
with the other miner APIs.
:returns: Version data for the miner.
""" """
return await self.get_version() return await self.get_version()
async def get_version(self): async def get_version(self):
""" """Get version data for the miner.
API 'get_version' command.
Returns a dict containing version information. :returns: Version data for the miner.
""" """
return await self.send_command("get_version") return await self.send_command("get_version")
async def status(self): async def status(self):
""" """Get BTMiner status and firmware version.
API 'status' command.
Returns a dict containing BTMiner status and firmware version. :returns: BTMiner status and firmware version.
""" """
return await self.send_command("status") return await self.send_command("status")
async def get_miner_info(self): async def get_miner_info(self):
""" """Get general miner info.
API 'get_miner_info' command.
Returns a dict containing general miner info. :returns: General miner info.
""" """
return await self.send_command("get_miner_info") return await self.send_command("get_miner_info")

View File

@@ -5,4 +5,10 @@ scan_threads = 300
config_threads = 300 config_threads = 300
reboot_threads = 300 reboot_threads = 300
### IMPORTANT ###
# You need to change the password of the miners using the whatsminer
# tool or the privileged API will not work using admin as the password.
# If you change the password, you can pass that password here.
whatsminer_pwd = "admin" whatsminer_pwd = "admin"

View File

@@ -1,3 +1,12 @@
[//]: # (If you can read this, you are viewing this document incorrectly.)
[//]: # (This is a Markdown document. Use an online Markdown viewer to)
[//]: # (view this file, such as https://dillinger.io/)
# CFG-Util # CFG-Util
## Interact with bitcoin mining ASICs using a simple GUI. ## Interact with bitcoin mining ASICs using a simple GUI.