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

reformatted all files to use the Black formatting style

Browse Source
This commit is contained in:
UpstreamData
2022-03-31 11:27:57 -06:00
parent e1383f2002
commit c57a523553
54 changed files with 1375 additions and 964 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
API/__init__.py
View File

@@ -42,17 +42,22 @@ class BaseMinerAPI:
def get_commands(self) -> list:
"""Get a list of command accessible to a specific type of API on the miner."""
return [func for func in
# each function in self
dir(self) if callable(getattr(self, func)) and
# no __ methods
not func.startswith("__") and
# remove all functions that are in this base class
func not in
[func for func in
dir(BaseMinerAPI) if callable(getattr(BaseMinerAPI, func))
]
]
return [
func
for func in
# each function in self
dir(self)
if callable(getattr(self, func)) and
# no __ methods
not func.startswith("__") and
# remove all functions that are in this base class
func
not in [
func
for func in dir(BaseMinerAPI)
if callable(getattr(BaseMinerAPI, func))
]
]
async def multicommand(self, *commands: str) -> dict:
"""Creates and sends multiple commands as one command to the miner."""
@@ -63,9 +68,11 @@ class BaseMinerAPI:
# make sure we can actually run the command, otherwise it will fail
commands = [command for command in user_commands if command in allowed_commands]
for item in list(set(user_commands) - set(commands)):
warnings.warn(f"""Removing incorrect command: {item}
warnings.warn(
f"""Removing incorrect command: {item}
If you are sure you want to use this command please use API.send_command("{item}", ignore_errors=True) instead.""",
APIWarning)
APIWarning,
)
# standard multicommand format is "command1+command2"
# doesnt work for S19 which is dealt with in the send command function
command = "+".join(commands)
@@ -87,7 +94,12 @@ If you are sure you want to use this command please use API.send_command("{item}
logging.debug(f"{self.ip}: Received multicommand data.")
return data
async def send_command(self, command: str, parameters: str or int or bool = None, ignore_errors: bool = False) -> dict:
async def send_command(
self,
command: str,
parameters: str or int or bool = None,
ignore_errors: bool = False,
) -> dict:
"""Send an API command to the miner and return the result."""
try:
# get reader and writer streams
@@ -104,7 +116,7 @@ If you are sure you want to use this command please use API.send_command("{item}
cmd["parameter"] = parameters
# send the command
writer.write(json.dumps(cmd).encode('utf-8'))
writer.write(json.dumps(cmd).encode("utf-8"))
await writer.drain()
# instantiate data
@@ -169,10 +181,10 @@ If you are sure you want to use this command please use API.send_command("{item}
# some json from the API returns with a null byte (\x00) on the end
if data.endswith(b"\x00"):
# handle the null byte
str_data = data.decode('utf-8')[:-1]
str_data = data.decode("utf-8")[:-1]
else:
# no null byte
str_data = data.decode('utf-8')
str_data = data.decode("utf-8")
# fix an error with a btminer return having an extra comma that breaks json.loads()
str_data = str_data.replace(",}", "}")
# fix an error with a btminer return having a newline that breaks json.loads()

44
API/bmminer.py
View File

@@ -18,6 +18,7 @@ class BMMinerAPI(BaseMinerAPI):
:param ip: The IP of the miner to reference the API on.
:param port: The port to reference the API on. Default is 4028.
"""
def __init__(self, ip: str, port: int = 4028) -> None:
super().__init__(ip, port)
@@ -115,11 +116,7 @@ class BMMinerAPI(BaseMinerAPI):
"""
return await self.send_command("enablepool", parameters=n)
async def addpool(self,
url: str,
username: str,
password: str
) -> dict:
async def addpool(self, url: str, username: str, password: str) -> dict:
"""Add a pool to the miner.
:param url: The URL of the new pool to add.
@@ -128,11 +125,9 @@ class BMMinerAPI(BaseMinerAPI):
:return: A confirmation of adding the pool.
"""
return await self.send_command("addpool",
parameters=f"{url}, "
f"{username}, "
f"{password}"
)
return await self.send_command(
"addpool", parameters=f"{url}, " f"{username}, " f"{password}"
)
async def poolpriority(self, *n: int) -> dict:
"""Set pool priority.
@@ -142,8 +137,7 @@ class BMMinerAPI(BaseMinerAPI):
:return: A confirmation of setting pool priority.
"""
pools = f"{','.join([str(item) for item in n])}"
return await self.send_command("poolpriority",
parameters=pools)
return await self.send_command("poolpriority", parameters=pools)
async def poolquota(self, n: int, q: int) -> dict:
"""Set pool quota.
@@ -153,10 +147,7 @@ class BMMinerAPI(BaseMinerAPI):
:return: A confirmation of setting pool quota.
"""
return await self.send_command("poolquota",
parameters=f"{n}, "
f"{q}"
)
return await self.send_command("poolquota", parameters=f"{n}, " f"{q}")
async def disablepool(self, n: int) -> dict:
"""Disable a pool.
@@ -292,9 +283,7 @@ class BMMinerAPI(BaseMinerAPI):
:return: Confirmation of setting failover-only.
"""
return await self.send_command("failover-only",
parameters=failover
)
return await self.send_command("failover-only", parameters=failover)
async def coin(self) -> dict:
"""Get information on the current coin.
@@ -337,10 +326,7 @@ class BMMinerAPI(BaseMinerAPI):
:return: The results of setting config of name to n.
"""
return await self.send_command("setconfig",
parameters=f"{name}, "
f"{n}"
)
return await self.send_command("setconfig", parameters=f"{name}, " f"{n}")
async def usbstats(self) -> dict:
"""Get stats of all USB devices except ztex.
@@ -368,15 +354,11 @@ class BMMinerAPI(BaseMinerAPI):
:return: Confirmation of setting PGA n with opt[,val].
"""
if val:
return await self.send_command("pgaset",
parameters=f"{n}, "
f"{opt}, "
f"{val}"
)
return await self.send_command(
"pgaset", parameters=f"{n}, " f"{opt}, " f"{val}"
)
else:
return await self.send_command("pgaset",
parameters=f"{n}, "
f"{opt}")
return await self.send_command("pgaset", parameters=f"{n}, " f"{opt}")
async def zero(self, which: str, summary: bool) -> dict:
"""Zero a device.

1
API/bosminer.py
View File

@@ -18,6 +18,7 @@ class BOSMinerAPI(BaseMinerAPI):
:param ip: The IP of the miner to reference the API on.
:param port: The port to reference the API on. Default is 4028.
"""
def __init__(self, ip, port=4028):
super().__init__(ip, port)

175
API/btminer.py
View File

@@ -6,14 +6,12 @@ import binascii
import base64
from passlib.handlers.md5_crypt import md5_crypt
from cryptography.hazmat.primitives.ciphers import \
Cipher, algorithms, modes
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from API import BaseMinerAPI, APIError
from settings import WHATSMINER_PWD
### IMPORTANT ###
# you need to change the password of the miners using the Whatsminer
# tool, then you can set them back to admin with this tool, but they
@@ -35,7 +33,7 @@ def _crypt(word: str, salt: str) -> str:
:return: An MD5 hash of the word with 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
match = standard_salt.match(salt)
# if the matching fails, the salt is incorrect
@@ -58,7 +56,7 @@ def _add_to_16(string: str) -> bytes:
length.
"""
while len(string) % 16 != 0:
string += '\0'
string += "\0"
return str.encode(string) # return bytes
@@ -74,20 +72,20 @@ def parse_btminer_priviledge_data(token_data: dict, data: dict):
:return: A decoded dict version of the privileged command output.
"""
# get the encoded data from the dict
enc_data = data['enc']
enc_data = data["enc"]
# 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
aeskey = binascii.unhexlify(aeskey.encode())
# create the required decryptor
aes = Cipher(algorithms.AES(aeskey), modes.ECB())
decryptor = aes.decryptor()
# decode the message with the decryptor
ret_msg = json.loads(decryptor.update(
base64.decodebytes(bytes(enc_data, encoding='utf8'))
).rstrip(b'\0').decode("utf8"))
ret_msg = json.loads(
decryptor.update(base64.decodebytes(bytes(enc_data, encoding="utf8")))
.rstrip(b"\0")
.decode("utf8")
)
return ret_msg
@@ -104,11 +102,9 @@ def create_privileged_cmd(token_data: dict, command: dict) -> bytes:
:return: The encrypted privileged command to be sent to the miner.
"""
# add token to command
command['token'] = token_data['host_sign']
command["token"] = token_data["host_sign"]
# 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
aeskey = binascii.unhexlify(aeskey.encode())
# create a new AES key
@@ -117,18 +113,16 @@ def create_privileged_cmd(token_data: dict, command: dict) -> bytes:
# dump the command to json
api_json_str = json.dumps(command)
# 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
data_enc = {'enc': 1, 'data': api_json_str_enc}
data_enc = {"enc": 1, "data": api_json_str_enc}
# dump the labeled data to json
api_packet_str = json.dumps(data_enc)
return api_packet_str.encode('utf-8')
return api_packet_str.encode("utf-8")
class BTMinerAPI(BaseMinerAPI):
@@ -157,16 +151,18 @@ class BTMinerAPI(BaseMinerAPI):
:param port: The port to reference the API on. Default is 4028.
:param pwd: The admin password of the miner. Default is admin.
"""
def __init__(self, ip, port=4028, pwd: str = WHATSMINER_PWD):
super().__init__(ip, port)
self.admin_pwd = pwd
self.current_token = None
async def send_command(self,
command: str or bytes,
parameters: str or int or bool = None,
ignore_errors: bool = False
) -> dict:
async def send_command(
self,
command: str or bytes,
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,
@@ -187,10 +183,7 @@ class BTMinerAPI(BaseMinerAPI):
command = json.dumps({"command": command}).encode("utf-8")
try:
# 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
except OSError as e:
if e.winerror == "121":
@@ -221,13 +214,10 @@ class BTMinerAPI(BaseMinerAPI):
await writer.wait_closed()
# check if the returned data is encoded
if 'enc' in data.keys():
if "enc" in data.keys():
# try to parse the encoded data
try:
data = parse_btminer_priviledge_data(
self.current_token,
data
)
data = parse_btminer_priviledge_data(self.current_token, data)
except Exception as e:
print(e)
@@ -250,25 +240,24 @@ class BTMinerAPI(BaseMinerAPI):
data = await self.send_command("get_token")
# encrypt the admin password with the salt
pwd = _crypt(self.admin_pwd, "$1$" + data["Msg"]["salt"] + '$')
pwd = pwd.split('$')
pwd = _crypt(self.admin_pwd, "$1$" + data["Msg"]["salt"] + "$")
pwd = pwd.split("$")
# take the 4th item from the pwd split
host_passwd_md5 = pwd[3]
# encrypt the pwd with the time and new salt
tmp = _crypt(pwd[3] + data["Msg"]["time"],
"$1$" + data["Msg"]["newsalt"] + '$'
)
tmp = tmp.split('$')
tmp = _crypt(pwd[3] + data["Msg"]["time"], "$1$" + data["Msg"]["newsalt"] + "$")
tmp = tmp.split("$")
# take the 4th item from the encrypted pwd split
host_sign = tmp[3]
# set the current token
self.current_token = {'host_sign': host_sign,
'host_passwd_md5': host_passwd_md5
}
self.current_token = {
"host_sign": host_sign,
"host_passwd_md5": host_passwd_md5,
}
return self.current_token
#### PRIVILEGED COMMANDS ####
@@ -276,19 +265,18 @@ class BTMinerAPI(BaseMinerAPI):
# how to configure the Whatsminer API to
# use these commands.
async def update_pools(self,
pool_1: str,
worker_1: str,
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
):
async def update_pools(
self,
pool_1: str,
worker_1: str,
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
@@ -314,15 +302,12 @@ class BTMinerAPI(BaseMinerAPI):
elif pool_2 and pool_3:
command = {
"cmd": "update_pools",
"pool1": pool_1,
"worker1": worker_1,
"passwd1": passwd_1,
"pool2": pool_2,
"worker2": worker_2,
"passwd2": passwd_2,
"pool3": pool_3,
"worker3": worker_3,
"passwd3": passwd_3,
@@ -333,10 +318,9 @@ class BTMinerAPI(BaseMinerAPI):
"pool1": pool_1,
"worker1": worker_1,
"passwd1": passwd_1,
"pool2": pool_2,
"worker2": worker_2,
"passwd2": passwd_2
"passwd2": passwd_2,
}
else:
command = {
@@ -406,12 +390,13 @@ class BTMinerAPI(BaseMinerAPI):
enc_command = create_privileged_cmd(token_data, 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
@@ -423,12 +408,13 @@ class BTMinerAPI(BaseMinerAPI):
:param start: LED on time offset in the cycle in ms.
:return: A reply informing of the status of setting the LED.
"""
command = {"cmd": "set_led",
"color": color,
"period": period,
"duration": duration,
"start": start
}
command = {
"cmd": "set_led",
"color": color,
"period": period,
"duration": duration,
"start": start,
}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
@@ -486,10 +472,11 @@ class BTMinerAPI(BaseMinerAPI):
password.
"""
# 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(
f"New password too long, the max length is 8. "
f"Password size: {len(new_pwd.encode('utf-8'))}")
f"Password size: {len(new_pwd.encode('utf-8'))}"
)
command = {"cmd": "update_pwd", "old": old_pwd, "new": new_pwd}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
@@ -507,9 +494,11 @@ class BTMinerAPI(BaseMinerAPI):
frequency.
"""
if not -10 < percent < 100:
return APIError(f"Frequency % is outside of the allowed "
f"range. Please set a % between -10 and "
f"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)}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
@@ -596,9 +585,11 @@ class BTMinerAPI(BaseMinerAPI):
"""
if not 0 < percent < 100:
return APIError(f"Power PCT % is outside of the allowed "
f"range. Please set a % between 0 and "
f"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)}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
@@ -618,12 +609,9 @@ class BTMinerAPI(BaseMinerAPI):
: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(
'Message is incorrect, please choose one of '
"Message is incorrect, please choose one of "
'["wait for adjust temp", '
'"adjust complete", '
'"adjust continue"]'
@@ -632,10 +620,7 @@ class BTMinerAPI(BaseMinerAPI):
complete = "true"
else:
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()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)

53
API/cgminer.py
View File

@@ -18,6 +18,7 @@ class CGMinerAPI(BaseMinerAPI):
:param ip: The IP of the miner to reference the API on.
:param port: The port to reference the API on. Default is 4028.
"""
def __init__(self, ip, port=4028):
super().__init__(ip, port)
@@ -111,11 +112,7 @@ class CGMinerAPI(BaseMinerAPI):
"""
return await self.send_command("enablepool", parameters=n)
async def addpool(self,
url: str,
username: str,
password: str
) -> dict:
async def addpool(self, url: str, username: str, password: str) -> dict:
"""Add a pool to the miner.
:param url: The URL of the new pool to add.
@@ -124,11 +121,9 @@ class CGMinerAPI(BaseMinerAPI):
:return: A confirmation of adding the pool.
"""
return await self.send_command("addpool",
parameters=f"{url}, "
f"{username}, "
f"{password}"
)
return await self.send_command(
"addpool", parameters=f"{url}, " f"{username}, " f"{password}"
)
async def poolpriority(self, *n: int) -> dict:
"""Set pool priority.
@@ -138,8 +133,7 @@ class CGMinerAPI(BaseMinerAPI):
:return: A confirmation of setting pool priority.
"""
pools = f"{','.join([str(item) for item in n])}"
return await self.send_command("poolpriority",
parameters=pools)
return await self.send_command("poolpriority", parameters=pools)
async def poolquota(self, n: int, q: int) -> dict:
"""Set pool quota.
@@ -149,10 +143,7 @@ class CGMinerAPI(BaseMinerAPI):
:return: A confirmation of setting pool quota.
"""
return await self.send_command("poolquota",
parameters=f"{n}, "
f"{q}"
)
return await self.send_command("poolquota", parameters=f"{n}, " f"{q}")
async def disablepool(self, n: int) -> dict:
"""Disable a pool.
@@ -288,9 +279,7 @@ class CGMinerAPI(BaseMinerAPI):
:return: Confirmation of setting failover-only.
"""
return await self.send_command("failover-only",
parameters=failover
)
return await self.send_command("failover-only", parameters=failover)
async def coin(self) -> dict:
"""Get information on the current coin.
@@ -333,10 +322,7 @@ class CGMinerAPI(BaseMinerAPI):
:return: The results of setting config of name to n.
"""
return await self.send_command("setconfig",
parameters=f"{name}, "
f"{n}"
)
return await self.send_command("setconfig", parameters=f"{name}, " f"{n}")
async def usbstats(self) -> dict:
"""Get stats of all USB devices except ztex.
@@ -364,12 +350,11 @@ class CGMinerAPI(BaseMinerAPI):
:return: Confirmation of setting PGA n with opt[,val].
"""
if val:
return await self.send_command("pgaset", parameters=f"{n}, "
f"{opt}, "
f"{val}")
return await self.send_command(
"pgaset", parameters=f"{n}, " f"{opt}, " f"{val}"
)
else:
return await self.send_command("pgaset", parameters=f"{n}, "
f"{opt}")
return await self.send_command("pgaset", parameters=f"{n}, " f"{opt}")
async def zero(self, which: str, summary: bool) -> dict:
"""Zero a device.
@@ -384,8 +369,7 @@ class CGMinerAPI(BaseMinerAPI):
:return: the STATUS section with info on the zero and optional
summary.
"""
return await self.send_command("zero", parameters=f"{which}, "
f"{summary}")
return await self.send_command("zero", parameters=f"{which}, " f"{summary}")
async def hotplug(self, n: int) -> dict:
"""Enable hotplug.
@@ -486,12 +470,11 @@ class CGMinerAPI(BaseMinerAPI):
:return: Confirmation of setting option opt to value val.
"""
if val:
return await self.send_command("ascset", parameters=f"{n}, "
f"{opt}, "
f"{val}")
return await self.send_command(
"ascset", parameters=f"{n}, " f"{opt}, " f"{val}"
)
else:
return await self.send_command("ascset", parameters=f"{n}, "
f"{opt}")
return await self.send_command("ascset", parameters=f"{n}, " f"{opt}")
async def lcd(self) -> dict:
"""Get a general all-in-one status summary of the miner.

1
API/unknown.py
View File

@@ -8,6 +8,7 @@ class UnknownAPI(BaseMinerAPI):
and API commands as possible (API ⋂ API), to ensure that it can be used
with as many APIs as possible.
"""
def __init__(self, ip, port=4028):
super().__init__(ip, port)