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

Dev (#12)

Browse Source 
* changed over to package format and removed tools, added poetry

* reformat into miner_interface project

* add dist to .gitignore

* update readme and finish reformatting

* Added couple missing imports. (#13)

* change name to pyasic

Co-authored-by: upstreamdata <brett@upstreamdata.ca>
Co-authored-by: Mika Impola <mika@impola.fi>
This commit is contained in:
UpstreamData
2022-07-07 07:57:34 -06:00
committed by GitHub
parent 5261b00aad
commit dcc3e07998
267 changed files with 851 additions and 6693 deletions
Download Patch File Download Diff File Expand all files Collapse all files

214
pyasic/API/__init__.py Normal file
View File

@@ -0,0 +1,214 @@
import asyncio
import json
import ipaddress
import warnings
import logging
class APIError(Exception):
def __init__(self, *args):
if args:
self.message = args[0]
else:
self.message = None
def __str__(self):
if self.message:
return f"{self.message}"
else:
return "Incorrect API parameters."
class APIWarning(Warning):
def __init__(self, *args):
if args:
self.message = args[0]
else:
self.message = None
def __str__(self):
if self.message:
return f"{self.message}"
else:
return "Incorrect API parameters."
class BaseMinerAPI:
def __init__(self, ip: str, port: int = 4028) -> None:
# api port, should be 4028
self.port = port
# ip address of the miner
self.ip = ipaddress.ip_address(ip)
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))
]
]
async def multicommand(
self, *commands: str, ignore_x19_error: bool = False
) -> dict:
"""Creates and sends multiple commands as one command to the miner."""
logging.debug(f"{self.ip}: Sending multicommand: {[*commands]}")
# split the commands into a proper list
user_commands = [*commands]
allowed_commands = self.get_commands()
# 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}
If you are sure you want to use this command please use API.send_command("{item}", ignore_errors=True) instead.""",
APIWarning,
)
# standard multicommand format is "command1+command2"
# doesnt work for S19 which is dealt with in the send command function
command = "+".join(commands)
data = None
try:
data = await self.send_command(command, x19_command=ignore_x19_error)
except APIError:
try:
data = {}
# S19 handler, try again
for cmd in command.split("+"):
data[cmd] = []
data[cmd].append(await self.send_command(cmd))
except APIError as e:
raise APIError(e)
except Exception as e:
logging.warning(f"{self.ip}: API Multicommand Error: {e}")
if data:
logging.debug(f"{self.ip}: Received multicommand data.")
return data
async def send_command(
self,
command: str or bytes,
parameters: str or int or bool = None,
ignore_errors: bool = False,
x19_command: bool = False,
) -> dict:
"""Send an API command to the miner and return the result."""
try:
# get reader and writer streams
reader, writer = await asyncio.open_connection(str(self.ip), self.port)
# handle OSError 121
except OSError as e:
if e.winerror == "121":
logging.warning("Semaphore Timeout has Expired.")
return {}
# create the command
cmd = {"command": command}
if parameters is not None:
cmd["parameter"] = parameters
# send the command
writer.write(json.dumps(cmd).encode("utf-8"))
await writer.drain()
# instantiate data
data = b""
# loop to receive all the data
try:
while True:
d = await reader.read(4096)
if not d:
break
data += d
except Exception as e:
logging.warning(f"{self.ip}: API Command Error: {e}")
data = self.load_api_data(data)
# close the connection
writer.close()
await writer.wait_closed()
# check for if the user wants to allow errors to return
if not ignore_errors:
# validate the command succeeded
validation = self.validate_command_output(data)
if not validation[0]:
if not x19_command:
logging.warning(f"{self.ip}: API Command Error: {validation[1]}")
raise APIError(validation[1])
return data
@staticmethod
def validate_command_output(data: dict) -> tuple:
"""Check if the returned command output is correctly formatted."""
# check if the data returned is correct or an error
# if status isn't a key, it is a multicommand
if "STATUS" not in data.keys():
for key in data.keys():
# make sure not to try to turn id into a dict
if not key == "id":
# make sure they succeeded
if "STATUS" in data[key][0].keys():
if data[key][0]["STATUS"][0]["STATUS"] not in ["S", "I"]:
# this is an error
return False, f"{key}: " + data[key][0]["STATUS"][0]["Msg"]
elif "id" not in data.keys():
if data["STATUS"] not in ["S", "I"]:
return False, data["Msg"]
else:
# make sure the command succeeded
if type(data["STATUS"]) == str:
if data["STATUS"] in ["RESTART"]:
return True, None
elif data["STATUS"][0]["STATUS"] not in ("S", "I"):
# this is an error
if data["STATUS"][0]["STATUS"] not in ("S", "I"):
return False, data["STATUS"][0]["Msg"]
return True, None
@staticmethod
def load_api_data(data: bytes) -> dict:
"""Convert API data from JSON to dict"""
str_data = None
try:
# 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]
else:
# no null byte
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()
str_data = str_data.replace("\n", "")
# fix an error with a bmminer return not having a specific comma that breaks json.loads()
str_data = str_data.replace("}{", "},{")
# fix an error with a bmminer return having a specific comma that breaks json.loads()
str_data = str_data.replace("[,{", "[{")
# fix an error with Avalonminers returning inf and nan
str_data = str_data.replace("inf", "0")
str_data = str_data.replace("nan", "0")
# fix whatever this garbage from avalonminers is `,"id":1}`
if str_data.startswith(","):
str_data = f"{{{str_data[1:]}"
# parse the json
parsed_data = json.loads(str_data)
# handle bad json
except json.decoder.JSONDecodeError as e:
raise APIError(f"Decode Error {e}: {str_data}")
return parsed_data

491
pyasic/API/bmminer.py Normal file
View File

@@ -0,0 +1,491 @@
from pyasic.API import BaseMinerAPI
class BMMinerAPI(BaseMinerAPI):
"""An abstraction of the BMMiner API.
Each method corresponds to an API command in BMMiner.
BMMiner API documentation:
https://github.com/jameshilliard/bmminer/blob/master/API-README
This class abstracts use of the BMMiner 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.
: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)
async def version(self) -> dict:
"""Get miner version info.
:return: Miner version information.
"""
return await self.send_command("version")
async def config(self) -> dict:
"""Get some basic configuration info.
:return: Some miner configuration information:
ASC Count <- the number of ASCs
PGA Count <- the number of PGAs
Pool Count <- the number of Pools
Strategy <- the current pool strategy
Log Interval <- the interval of logging
Device Code <- list of compiled device drivers
OS <- the current operating system
Failover-Only <- failover-only setting
Scan Time <- scan-time setting
Queue <- queue setting
Expiry <- expiry setting
"""
return await self.send_command("config")
async def summary(self) -> dict:
"""Get the status summary of the miner.
:return: The status summary of the miner.
"""
return await self.send_command("summary")
async def pools(self) -> dict:
"""Get pool information.
:return: Miner pool information.
"""
return await self.send_command("pools")
async def devs(self) -> dict:
"""Get data on each PGA/ASC with their details.
:return: Data on each PGA/ASC with their details.
"""
return await self.send_command("devs")
async def edevs(self, old: bool = False) -> dict:
"""Get data on each PGA/ASC with their details, ignoring
blacklisted and zombie devices.
:param old: Include zombie devices that became zombies less
than 'old' seconds ago
:return: Data on each PGA/ASC with their details.
"""
if old:
return await self.send_command("edevs", parameters=old)
else:
return await self.send_command("edevs")
async def pga(self, n: int) -> dict:
"""Get data from PGA n.
:param n: The PGA number to get data from.
:return: Data on the PGA n.
"""
return await self.send_command("pga", parameters=n)
async def pgacount(self) -> dict:
"""Get data fon all PGAs.
:return: Data on the PGAs connected.
"""
return await self.send_command("pgacount")
async def switchpool(self, n: int) -> dict:
"""Switch pools to pool n.
:param n: The pool to switch to.
:return: A confirmation of switching to pool n.
"""
return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int) -> dict:
"""Enable pool n.
:param n: The pool to enable.
:return: A confirmation of enabling pool n.
"""
return await self.send_command("enablepool", parameters=n)
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.
:param username: The users username on the new pool.
:param password: The worker password on the new pool.
:return: A confirmation of adding the pool.
"""
return await self.send_command(
"addpool", parameters=f"{url},{username},{password}"
)
async def poolpriority(self, *n: int) -> dict:
"""Set pool priority.
:param n: Pools in order of priority.
:return: A confirmation of setting pool priority.
"""
pools = f"{','.join([str(item) for item in n])}"
return await self.send_command("poolpriority", parameters=pools)
async def poolquota(self, n: int, q: int) -> dict:
"""Set pool quota.
:param n: Pool number to set quota on.
:param q: Quota to set the pool to.
:return: A confirmation of setting pool quota.
"""
return await self.send_command("poolquota", parameters=f"{n},{q}")
async def disablepool(self, n: int) -> dict:
"""Disable a pool.
:param n: Pool to disable.
:return: A confirmation of diabling the pool.
"""
return await self.send_command("disablepool", parameters=n)
async def removepool(self, n: int) -> dict:
"""Remove a pool.
:param n: Pool to remove.
:return: A confirmation of removing the pool.
"""
return await self.send_command("removepool", parameters=n)
async def save(self, filename: str = None) -> dict:
"""Save the config.
:param filename: Filename to save the config as.
:return: A confirmation of saving the config.
"""
if filename:
return await self.send_command("save", parameters=filename)
else:
return await self.send_command("save")
async def quit(self) -> dict:
"""Quit BMMiner.
:return: A single "BYE" before BMMiner quits.
"""
return await self.send_command("quit")
async def notify(self) -> dict:
"""Notify the user of past errors.
:return: The last status and count of each devices problem(s).
"""
return await self.send_command("notify")
async def privileged(self) -> dict:
"""Check if you have privileged access.
:return: The STATUS section with an error if you have no
privileged access, or success if you have privileged access.
"""
return await self.send_command("privileged")
async def pgaenable(self, n: int) -> dict:
"""Enable PGA n.
:param n: The PGA to enable.
:return: A confirmation of enabling PGA n.
"""
return await self.send_command("pgaenable", parameters=n)
async def pgadisable(self, n: int) -> dict:
"""Disable PGA n.
:param n: The PGA to disable.
:return: A confirmation of disabling PGA n.
"""
return await self.send_command("pgadisable", parameters=n)
async def pgaidentify(self, n: int) -> dict:
"""Identify PGA n.
:param n: The PGA to identify.
:return: A confirmation of identifying PGA n.
"""
return await self.send_command("pgaidentify", parameters=n)
async def devdetails(self) -> dict:
"""Get data on all devices with their static details.
:return: Data on all devices with their static details.
"""
return await self.send_command("devdetails")
async def restart(self) -> dict:
"""Restart BMMiner using the API.
:return: A reply informing of the restart.
"""
return await self.send_command("restart")
async def stats(self) -> dict:
"""Get stats of each device/pool with more than 1 getwork.
:return: Stats of each device/pool with more than 1 getwork.
"""
return await self.send_command("stats")
async def estats(self, old: bool = False) -> dict:
"""Get stats of each device/pool with more than 1 getwork,
ignoring zombie devices.
:param old: Include zombie devices that became zombies less
than 'old' seconds ago.
:return: Stats of each device/pool with more than 1 getwork,
ignoring zombie devices.
"""
if old:
return await self.send_command("estats", parameters=old)
else:
return await self.send_command("estats")
async def check(self, command: str) -> dict:
"""Check if the command command exists in BMMiner.
:param command: The command to check.
:return: Information about a command:
Exists (Y/N) <- the command exists in this version
Access (Y/N) <- you have access to use the command
"""
return await self.send_command("check", parameters=command)
async def failover_only(self, failover: bool) -> dict:
"""Set failover-only.
:param failover: What to set failover-only to.
:return: Confirmation of setting failover-only.
"""
return await self.send_command("failover-only", parameters=failover)
async def coin(self) -> dict:
"""Get information on the current coin.
:return: Information about the current coin being mined:
Hash Method <- the hashing algorithm
Current Block Time <- blocktime as a float, 0 means none
Current Block Hash <- the hash of the current block, blank
means none
LP <- whether LP is in use on at least 1 pool
Network Difficulty: the current network difficulty
"""
return await self.send_command("coin")
async def debug(self, setting: str) -> dict:
"""Set a debug setting.
:param setting: Which setting to switch to. Options are:
Silent,
Quiet,
Verbose,
Debug,
RPCProto,
PerDevice,
WorkTime,
Normal.
:return: Data on which debug setting was enabled or disabled.
"""
return await self.send_command("debug", parameters=setting)
async def setconfig(self, name: str, n: int) -> dict:
"""Set config of name to value n.
:param name: The name of the config setting to set. Options are:
queue,
scantime,
expiry.
:param n: The value to set the 'name' setting to.
:return: The results of setting config of name to n.
"""
return await self.send_command("setconfig", parameters=f"{name},{n}")
async def usbstats(self) -> dict:
"""Get stats of all USB devices except ztex.
:return: The stats of all USB devices except ztex.
"""
return await self.send_command("usbstats")
async def pgaset(self, n: int, opt: str, val: int = None) -> dict:
"""Set PGA option opt to val on PGA n.
Options:
MMQ -
opt: clock
val: 160 - 230 (multiple of 2)
CMR -
opt: clock
val: 100 - 220
:param n: The PGA to set the options on.
:param opt: The option to set. Setting this to 'help'
returns a help message.
:param val: The value to set the option to.
:return: Confirmation of setting PGA n with opt[,val].
"""
if val:
return await self.send_command("pgaset", parameters=f"{n},{opt},{val}")
else:
return await self.send_command("pgaset", parameters=f"{n},{opt}")
async def zero(self, which: str, summary: bool) -> dict:
"""Zero a device.
:param which: Which device to zero.
Setting this to 'all' zeros all devices.
Setting this to 'bestshare' zeros only the bestshare values
for each pool and global.
:param summary: Whether or not to show a full summary.
:return: the STATUS section with info on the zero and optional
summary.
"""
return await self.send_command("zero", parameters=f"{which},{summary}")
async def hotplug(self, n: int) -> dict:
"""Enable hotplug.
:param n: The device number to set hotplug on.
:return: Information on hotplug status.
"""
return await self.send_command("hotplug", parameters=n)
async def asc(self, n: int) -> dict:
"""Get data for ASC device n.
:param n: The device to get data for.
:return: The data for ASC device n.
"""
return await self.send_command("asc", parameters=n)
async def ascenable(self, n: int) -> dict:
"""Enable ASC device n.
:param n: The device to enable.
:return: Confirmation of enabling ASC device n.
"""
return await self.send_command("ascenable", parameters=n)
async def ascdisable(self, n: int) -> dict:
"""Disable ASC device n.
:param n: The device to disable.
:return: Confirmation of disabling ASC device n.
"""
return await self.send_command("ascdisable", parameters=n)
async def ascidentify(self, n: int) -> dict:
"""Identify ASC device n.
:param n: The device to identify.
:return: Confirmation of identifying ASC device n.
"""
return await self.send_command("ascidentify", parameters=n)
async def asccount(self) -> dict:
"""Get data on the number of ASC devices and their info.
:return: Data on all ASC devices.
"""
return await self.send_command("asccount")
async def ascset(self, n: int, opt: str, val: int = None) -> dict:
"""Set ASC n option opt to value val.
Sets an option on the ASC n to a value. Allowed options are:
AVA+BTB -
opt: freq
val: 256 - 1024 (chip frequency)
BTB -
opt: millivolts
val: 1000 - 1400 (core voltage)
MBA -
opt: reset
val: 0 - # of chips (reset a chip)
opt: freq
val: 0 - # of chips, 100 - 1400 (chip frequency)
opt: ledcount
val: 0 - 100 (chip count for LED)
opt: ledlimit
val: 0 - 200 (LED off below GH/s)
opt: spidelay
val: 0 - 9999 (SPI per I/O delay)
opt: spireset
val: i or s, 0 - 9999 (SPI regular reset)
opt: spisleep
val: 0 - 9999 (SPI reset sleep in ms)
BMA -
opt: volt
val: 0 - 9
opt: clock
val: 0 - 15
:param n: The ASC to set the options on.
:param opt: The option to set. Setting this to 'help' returns a
help message.
:param val: The value to set the option to.
:return: Confirmation of setting option opt to value val.
"""
if val:
return await self.send_command("ascset", parameters=f"{n},{opt},{val}")
else:
return await self.send_command("ascset", parameters=f"{n},{opt}")
async def lcd(self) -> dict:
"""Get a general all-in-one status summary of the miner.
:return: An all-in-one status summary of the miner.
"""
return await self.send_command("lcd")
async def lockstats(self) -> dict:
"""Write lockstats to STDERR.
:return: The result of writing the lock stats to STDERR.
"""
return await self.send_command("lockstats")

208
pyasic/API/bosminer.py Normal file
View File

@@ -0,0 +1,208 @@
from pyasic.API import BaseMinerAPI
class BOSMinerAPI(BaseMinerAPI):
"""An abstraction of the BOSMiner API.
Each method corresponds to an API command in BOSMiner.
BOSMiner API documentation:
https://docs.braiins.com/os/plus-en/Development/1_api.html
This class abstracts use of the BOSMiner 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.
: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)
async def asccount(self) -> dict:
"""Get data on the number of ASC devices and their info.
:return: Data on all ASC devices.
"""
return await self.send_command("asccount")
async def asc(self, n: int) -> dict:
"""Get data for ASC device n.
:param n: The device to get data for.
:return: The data for ASC device n.
"""
return await self.send_command("asc", parameters=n)
async def devdetails(self) -> dict:
"""Get data on all devices with their static details.
:return: Data on all devices with their static details.
"""
return await self.send_command("devdetails")
async def devs(self) -> dict:
"""Get data on each PGA/ASC with their details.
:return: Data on each PGA/ASC with their details.
"""
return await self.send_command("devs")
async def edevs(self, old: bool = False) -> dict:
"""Get data on each PGA/ASC with their details, ignoring
blacklisted and zombie devices.
:param old: Include zombie devices that became zombies less
than 'old' seconds ago
:return: Data on each PGA/ASC with their details.
"""
if old:
return await self.send_command("edevs", parameters="old")
else:
return await self.send_command("edevs")
async def pools(self) -> dict:
"""Get pool information.
:return: Miner pool information.
"""
return await self.send_command("pools")
async def summary(self) -> dict:
"""Get the status summary of the miner.
:return: The status summary of the miner.
"""
return await self.send_command("summary")
async def stats(self) -> dict:
"""Get stats of each device/pool with more than 1 getwork.
:return: Stats of each device/pool with more than 1 getwork.
"""
return await self.send_command("stats")
async def version(self) -> dict:
"""Get miner version info.
:return: Miner version information.
"""
return await self.send_command("version")
async def estats(self, old: bool = False) -> dict:
"""Get stats of each device/pool with more than 1 getwork,
ignoring zombie devices.
:param old: Include zombie devices that became zombies less
than 'old' seconds ago.
:return: Stats of each device/pool with more than 1 getwork,
ignoring zombie devices.
"""
if old:
return await self.send_command("estats", parameters=old)
else:
return await self.send_command("estats")
async def check(self, command: str) -> dict:
"""Check if the command command exists in BOSMiner.
:param command: The command to check.
:return: Information about a command:
Exists (Y/N) <- the command exists in this version
Access (Y/N) <- you have access to use the command
"""
return await self.send_command("check", parameters=command)
async def coin(self) -> dict:
"""Get information on the current coin.
:return: Information about the current coin being mined:
Hash Method <- the hashing algorithm
Current Block Time <- blocktime as a float, 0 means none
Current Block Hash <- the hash of the current block, blank
means none
LP <- whether LP is in use on at least 1 pool
Network Difficulty: the current network difficulty
"""
return await self.send_command("coin")
async def lcd(self) -> dict:
"""Get a general all-in-one status summary of the miner.
:return: An all-in-one status summary of the miner.
"""
return await self.send_command("lcd")
async def switchpool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("enablepool", parameters=n)
async def disablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("disablepool", parameters=n)
async def addpool(self, url: str, username: str, password: str) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("addpool", parameters=f"{url},{username},{password}")
async def removepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("removepool", parameters=n)
async def fans(self) -> dict:
"""Get fan data.
:return: Data on the fans of the miner.
"""
return await self.send_command("fans")
async def tempctrl(self) -> dict:
"""Get temperature control data.
:return: Data about the temp control settings of the miner.
"""
return await self.send_command("tempctrl")
async def temps(self) -> dict:
"""Get temperature data.
:return: Data on the temps of the miner.
"""
return await self.send_command("temps")
async def tunerstatus(self) -> dict:
"""Get tuner status data
:return: Data on the status of autotuning.
"""
return await self.send_command("tunerstatus")
async def pause(self) -> dict:
"""Pause mining.
:return: Confirmation of pausing mining.
"""
return await self.send_command("pause")
async def resume(self) -> dict:
"""Resume mining.
:return: Confirmation of resuming mining.
"""
return await self.send_command("resume")

705
pyasic/API/btminer.py Normal file
View File

@@ -0,0 +1,705 @@
import asyncio
import re
import json
import hashlib
import binascii
import base64
import logging
from passlib.handlers.md5_crypt import md5_crypt
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from pyasic.API import BaseMinerAPI, APIError
from pyasic.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
# must be changed to something else and set back to admin with this
# or the privileged API will not work using admin as the password. If
# you change the password, you can pass that to the this class as pwd,
# or add it as the Whatsminer_pwd in the settings.toml file.
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
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
if not match:
raise ValueError("Salt format is not correct.")
# save the matched salt in a new variable
new_salt = match.group(2)
# encrypt the word with the salt using md5
result = md5_crypt.hash(word, salt=new_salt)
return result
def _add_to_16(string: str) -> bytes:
"""Add null bytes to a string until the length is a multiple 16
:param string: The string to lengthen to a multiple of 16 and
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: 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
enc_data = data["enc"]
# get the aes key from the token data
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")
)
return ret_msg
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
command["token"] = token_data["host_sign"]
# encode host_passwd data and get 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
aes = Cipher(algorithms.AES(aeskey), modes.ECB())
encryptor = aes.encryptor()
# 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", "")
)
# label the data as being encoded
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")
class BTMinerAPI(BaseMinerAPI):
"""An abstraction of the API for MicroBT Whatsminers, BTMiner.
Each method corresponds to an API command in BMMiner.
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.
:param ip: The IP of the miner to reference the API on.
: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,
**kwargs,
) -> 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 it is a string, put it into the standard command format
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)
# handle OSError 121
except OSError as e:
if e.winerror == "121":
print("Semaphore Timeout has Expired.")
return {}
# send the command
writer.write(command)
await writer.drain()
# instantiate data
data = b""
# loop to receive all the data
try:
while True:
d = await reader.read(4096)
if not d:
break
data += d
except Exception as e:
logging.info(f"{str(self.ip)}: {e}")
data = self.load_api_data(data)
# close the connection
writer.close()
await writer.wait_closed()
# check if the returned data is encoded
if "enc" in data.keys():
# try to parse the encoded data
try:
data = parse_btminer_priviledge_data(self.current_token, data)
except Exception as e:
logging.info(f"{str(self.ip)}: {e}")
if not ignore_errors:
# if it fails to validate, it is likely an error
validation = self.validate_command_output(data)
if not validation[0]:
raise APIError(validation[1])
# return the parsed json as a dict
return data
async def get_token(self):
"""Gets token information from the 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")
# encrypt the admin password with the salt
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("$")
# 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,
}
return self.current_token
#### PRIVILEGED COMMANDS ####
# Please read the top of this file to learn
# 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,
):
"""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
token_data = await self.get_token()
# parse pool data
if not pool_1:
raise APIError("No pools set.")
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,
}
elif pool_2:
command = {
"cmd": "update_pools",
"pool1": pool_1,
"worker1": worker_1,
"passwd1": passwd_1,
"pool2": pool_2,
"worker2": worker_2,
"passwd2": passwd_2,
}
else:
command = {
"cmd": "update_pools",
"pool1": pool_1,
"worker1": worker_1,
"passwd1": passwd_1,
}
# encode the command with the token data
enc_command = create_privileged_cmd(token_data, command)
# send the command
return await self.send_command(enc_command)
async def restart(self):
"""Restart BTMiner using the API.
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"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def power_off(self, respbefore: bool = True):
"""Power off the miner using the API.
Power off the miner using the API, only works after changing
the password of the miner using the Whatsminer tool.
:param respbefore: Whether to respond before powering off.
:return: A reply informing of the status of powering off.
"""
if respbefore:
command = {"cmd": "power_off", "respbefore": "true"}
else:
command = {"cmd": "power_off", "respbefore": "false"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def power_on(self):
"""Power on the miner using the API.
Power on the miner using the API, only works after changing
the password of the miner using the Whatsminer tool.
:return: A reply informing of the status of powering on.
"""
command = {"cmd": "power_on"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def reset_led(self):
"""Reset the LED on the miner using the API.
Reset the LED on the miner using the API, only works after
changing the password of the miner using the Whatsminer tool.
:return: A reply informing of the status of resetting the LED.
"""
command = {"cmd": "set_led", "param": "auto"}
token_data = await self.get_token()
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,
):
"""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.
"""
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)
async def set_low_power(self):
"""Set low power mode on the miner using the API.
Set low power mode on the miner using the API, only works after
changing the password of the miner using the Whatsminer tool.
:return: A reply informing of the status of setting low power
mode.
"""
command = {"cmd": "set_low_power"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def update_firmware(self): # noqa - static
# to be determined if this will be added later
# requires a file stream in bytes
return NotImplementedError
async def reboot(self):
"""Reboot the miner using the API.
:return: A reply informing of the status of the reboot.
"""
command = {"cmd": "reboot"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def factory_reset(self):
"""Reset the miner to factory defaults.
:return: A reply informing of the status of the reset.
"""
command = {"cmd": "factory_reset"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def update_pwd(self, old_pwd: str, new_pwd: str):
"""Update 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.
:param old_pwd: The old admin password.
:param new_pwd: The new password to set.
:return: A reply informing of the status of setting the
password.
"""
# check if password length is greater than 8 bytes
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'))}"
)
command = {"cmd": "update_pwd", "old": old_pwd, "new": new_pwd}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def set_target_freq(self, percent: int):
"""Update the target frequency.
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%.
:param percent: The frequency % to set.
:return: A reply informing of the status of setting the
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"
)
command = {"cmd": "set_target_freq", "percent": str(percent)}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def enable_fast_boot(self):
"""Turn on fast boot.
Turn on fast boot, only works after changing the password of
the miner using the Whatsminer tool.
:return: A reply informing of the status of enabling fast boot.
"""
command = {"cmd": "enable_btminer_fast_boot"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def disable_fast_boot(self):
"""Turn off fast boot.
Turn off fast boot, only works after changing the password of
the miner using the Whatsminer tool.
:return: A reply informing of the status of disabling fast boot.
"""
command = {"cmd": "disable_btminer_fast_boot"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def enable_web_pools(self):
"""Turn on web pool updates.
Turn on web pool updates, only works after changing the
password of the miner using the Whatsminer tool.
:return: A reply informing of the status of enabling web pools.
"""
command = {"cmd": "enable_web_pools"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def disable_web_pools(self):
"""Turn off web pool updates.
Turn off web pool updates, only works after changing the
password of the miner using the Whatsminer tool.
:return: A reply informing of the status of disabling web
pools.
"""
command = {"cmd": "disable_web_pools"}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def set_hostname(self, hostname: str):
"""Set the hostname of the miner.
Set the hostname of the miner, only works after changing the
password of the miner using the Whatsminer tool.
:param hostname: The new hostname to use.
:return: A reply informing of the status of setting the
hostname.
"""
command = {"cmd": "set_hostname", "hostname": hostname}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def set_power_pct(self, percent: int):
"""Set the power percentage of the miner.
Set the power percentage of the miner, only works after changing
the password of the miner using the Whatsminer tool.
:param percent: The power percentage to set.
:return: A reply informing of the status of setting the
power percentage.
"""
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"
)
command = {"cmd": "set_power_pct", "percent": str(percent)}
token_data = await self.get_token()
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def pre_power_on(self, complete: bool, msg: str):
"""Configure or check status of pre power on.
Configure or check status of pre power on, only works after
changing the password of the miner using the Whatsminer tool.
:param complete: check whether pre power on is complete.
:param msg: the message to check.
"wait for adjust temp" or
"adjust complete" or
"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":
return APIError(
"Message is incorrect, please choose one of "
'["wait for adjust temp", '
'"adjust complete", '
'"adjust continue"]'
)
if complete:
complete = "true"
else:
complete = "false"
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)
#### END privileged COMMANDS ####
async def summary(self):
"""Get the summary status from the miner.
:return: Summary status of the miner.
"""
return await self.send_command("summary")
async def pools(self):
"""Get the pool status from the miner.
:return: Pool status of the miner.
"""
return await self.send_command("pools")
async def devs(self):
"""Get data on each PGA/ASC with their details.
:return: Data on each PGA/ASC with their details.
"""
return await self.send_command("devs")
async def edevs(self):
"""Get data on each PGA/ASC with their details, ignoring
blacklisted and zombie devices.
:return: Data on each PGA/ASC with their details.
"""
return await self.send_command("edevs")
async def devdetails(self):
"""Get data on all devices with their static details.
:return: Data on all devices with their static details.
"""
return await self.send_command("devdetails")
async def get_psu(self):
"""Get data on the PSU and power information.
:return: Data on the PSU and power information.
"""
return await self.send_command("get_psu")
async def version(self):
"""Get version data for the miner.
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.
:return: Version data for the miner.
"""
return await self.get_version()
async def get_version(self):
"""Get version data for the miner.
:return: Version data for the miner.
"""
return await self.send_command("get_version")
async def status(self):
"""Get BTMiner status and firmware version.
:return: BTMiner status and firmware version.
"""
return await self.send_command("status")
async def get_miner_info(self):
"""Get general miner info.
:return: General miner info.
"""
return await self.send_command("get_miner_info")

487
pyasic/API/cgminer.py Normal file
View File

@@ -0,0 +1,487 @@
from pyasic.API import BaseMinerAPI
class CGMinerAPI(BaseMinerAPI):
"""An abstraction of the CGMiner API.
Each method corresponds to an API command in GGMiner.
CGMiner API documentation:
https://github.com/ckolivas/cgminer/blob/master/API-README
This class abstracts use of the CGMiner 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.
: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)
async def version(self) -> dict:
"""Get miner version info.
:return: Miner version information.
"""
return await self.send_command("version")
async def config(self) -> dict:
"""Get some basic configuration info.
:return: Some miner configuration information:
ASC Count <- the number of ASCs
PGA Count <- the number of PGAs
Pool Count <- the number of Pools
Strategy <- the current pool strategy
Log Interval <- the interval of logging
Device Code <- list of compiled device drivers
OS <- the current operating system
"""
return await self.send_command("config")
async def summary(self) -> dict:
"""Get the status summary of the miner.
:return: The status summary of the miner.
"""
return await self.send_command("summary")
async def pools(self) -> dict:
"""Get pool information.
:return: Miner pool information.
"""
return await self.send_command("pools")
async def devs(self) -> dict:
"""Get data on each PGA/ASC with their details.
:return: Data on each PGA/ASC with their details.
"""
return await self.send_command("devs")
async def edevs(self, old: bool = False) -> dict:
"""Get data on each PGA/ASC with their details, ignoring
blacklisted and zombie devices.
:param old: Include zombie devices that became zombies less
than 'old' seconds ago
:return: Data on each PGA/ASC with their details.
"""
if old:
return await self.send_command("edevs", parameters="old")
else:
return await self.send_command("edevs")
async def pga(self, n: int) -> dict:
"""Get data from PGA n.
:param n: The PGA number to get data from.
:return: Data on the PGA n.
"""
return await self.send_command("pga", parameters=n)
async def pgacount(self) -> dict:
"""Get data fon all PGAs.
:return: Data on the PGAs connected.
"""
return await self.send_command("pgacount")
async def switchpool(self, n: int) -> dict:
"""Switch pools to pool n.
:param n: The pool to switch to.
:return: A confirmation of switching to pool n.
"""
return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int) -> dict:
"""Enable pool n.
:param n: The pool to enable.
:return: A confirmation of enabling pool n.
"""
return await self.send_command("enablepool", parameters=n)
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.
:param username: The users username on the new pool.
:param password: The worker password on the new pool.
:return: A confirmation of adding the pool.
"""
return await self.send_command(
"addpool", parameters=f"{url},{username},{password}"
)
async def poolpriority(self, *n: int) -> dict:
"""Set pool priority.
:param n: Pools in order of priority.
:return: A confirmation of setting pool priority.
"""
pools = f"{','.join([str(item) for item in n])}"
return await self.send_command("poolpriority", parameters=pools)
async def poolquota(self, n: int, q: int) -> dict:
"""Set pool quota.
:param n: Pool number to set quota on.
:param q: Quota to set the pool to.
:return: A confirmation of setting pool quota.
"""
return await self.send_command("poolquota", parameters=f"{n},{q}")
async def disablepool(self, n: int) -> dict:
"""Disable a pool.
:param n: Pool to disable.
:return: A confirmation of diabling the pool.
"""
return await self.send_command("disablepool", parameters=n)
async def removepool(self, n: int) -> dict:
"""Remove a pool.
:param n: Pool to remove.
:return: A confirmation of removing the pool.
"""
return await self.send_command("removepool", parameters=n)
async def save(self, filename: str = None) -> dict:
"""Save the config.
:param filename: Filename to save the config as.
:return: A confirmation of saving the config.
"""
if filename:
return await self.send_command("save", parameters=filename)
else:
return await self.send_command("save")
async def quit(self) -> dict:
"""Quit CGMiner.
:return: A single "BYE" before CGMiner quits.
"""
return await self.send_command("quit")
async def notify(self) -> dict:
"""Notify the user of past errors.
:return: The last status and count of each devices problem(s).
"""
return await self.send_command("notify")
async def privileged(self) -> dict:
"""Check if you have privileged access.
:return: The STATUS section with an error if you have no
privileged access, or success if you have privileged access.
"""
return await self.send_command("privileged")
async def pgaenable(self, n: int) -> dict:
"""Enable PGA n.
:param n: The PGA to enable.
:return: A confirmation of enabling PGA n.
"""
return await self.send_command("pgaenable", parameters=n)
async def pgadisable(self, n: int) -> dict:
"""Disable PGA n.
:param n: The PGA to disable.
:return: A confirmation of disabling PGA n.
"""
return await self.send_command("pgadisable", parameters=n)
async def pgaidentify(self, n: int) -> dict:
"""Identify PGA n.
:param n: The PGA to identify.
:return: A confirmation of identifying PGA n.
"""
return await self.send_command("pgaidentify", parameters=n)
async def devdetails(self) -> dict:
"""Get data on all devices with their static details.
:return: Data on all devices with their static details.
"""
return await self.send_command("devdetails")
async def restart(self) -> dict:
"""Restart CGMiner using the API.
:return: A reply informing of the restart.
"""
return await self.send_command("restart")
async def stats(self) -> dict:
"""Get stats of each device/pool with more than 1 getwork.
:return: Stats of each device/pool with more than 1 getwork.
"""
return await self.send_command("stats")
async def estats(self, old: bool = False) -> dict:
"""Get stats of each device/pool with more than 1 getwork,
ignoring zombie devices.
:param old: Include zombie devices that became zombies less
than 'old' seconds ago.
:return: Stats of each device/pool with more than 1 getwork,
ignoring zombie devices.
"""
if old:
return await self.send_command("estats", parameters=old)
else:
return await self.send_command("estats")
async def check(self, command: str) -> dict:
"""Check if the command command exists in CGMiner.
:param command: The command to check.
:return: Information about a command:
Exists (Y/N) <- the command exists in this version
Access (Y/N) <- you have access to use the command
"""
return await self.send_command("check", parameters=command)
async def failover_only(self, failover: bool) -> dict:
"""Set failover-only.
:param failover: What to set failover-only to.
:return: Confirmation of setting failover-only.
"""
return await self.send_command("failover-only", parameters=failover)
async def coin(self) -> dict:
"""Get information on the current coin.
:return: Information about the current coin being mined:
Hash Method <- the hashing algorithm
Current Block Time <- blocktime as a float, 0 means none
Current Block Hash <- the hash of the current block, blank
means none
LP <- whether LP is in use on at least 1 pool
Network Difficulty: the current network difficulty
"""
return await self.send_command("coin")
async def debug(self, setting: str) -> dict:
"""Set a debug setting.
:param setting: Which setting to switch to. Options are:
Silent,
Quiet,
Verbose,
Debug,
RPCProto,
PerDevice,
WorkTime,
Normal.
:return: Data on which debug setting was enabled or disabled.
"""
return await self.send_command("debug", parameters=setting)
async def setconfig(self, name: str, n: int) -> dict:
"""Set config of name to value n.
:param name: The name of the config setting to set. Options are:
queue,
scantime,
expiry.
:param n: The value to set the 'name' setting to.
:return: The results of setting config of name to n.
"""
return await self.send_command("setconfig", parameters=f"{name},{n}")
async def usbstats(self) -> dict:
"""Get stats of all USB devices except ztex.
:return: The stats of all USB devices except ztex.
"""
return await self.send_command("usbstats")
async def pgaset(self, n: int, opt: str, val: int = None) -> dict:
"""Set PGA option opt to val on PGA n.
Options:
MMQ -
opt: clock
val: 160 - 230 (multiple of 2)
CMR -
opt: clock
val: 100 - 220
:param n: The PGA to set the options on.
:param opt: The option to set. Setting this to 'help'
returns a help message.
:param val: The value to set the option to.
:return: Confirmation of setting PGA n with opt[,val].
"""
if val:
return await self.send_command("pgaset", parameters=f"{n},{opt},{val}")
else:
return await self.send_command("pgaset", parameters=f"{n},{opt}")
async def zero(self, which: str, summary: bool) -> dict:
"""Zero a device.
:param which: Which device to zero.
Setting this to 'all' zeros all devices.
Setting this to 'bestshare' zeros only the bestshare values
for each pool and global.
:param summary: Whether or not to show a full summary.
:return: the STATUS section with info on the zero and optional
summary.
"""
return await self.send_command("zero", parameters=f"{which},{summary}")
async def hotplug(self, n: int) -> dict:
"""Enable hotplug.
:param n: The device number to set hotplug on.
:return: Information on hotplug status.
"""
return await self.send_command("hotplug", parameters=n)
async def asc(self, n: int) -> dict:
"""Get data for ASC device n.
:param n: The device to get data for.
:return: The data for ASC device n.
"""
return await self.send_command("asc", parameters=n)
async def ascenable(self, n: int) -> dict:
"""Enable ASC device n.
:param n: The device to enable.
:return: Confirmation of enabling ASC device n.
"""
return await self.send_command("ascenable", parameters=n)
async def ascdisable(self, n: int) -> dict:
"""Disable ASC device n.
:param n: The device to disable.
:return: Confirmation of disabling ASC device n.
"""
return await self.send_command("ascdisable", parameters=n)
async def ascidentify(self, n: int) -> dict:
"""Identify ASC device n.
:param n: The device to identify.
:return: Confirmation of identifying ASC device n.
"""
return await self.send_command("ascidentify", parameters=n)
async def asccount(self) -> dict:
"""Get data on the number of ASC devices and their info.
:return: Data on all ASC devices.
"""
return await self.send_command("asccount")
async def ascset(self, n: int, opt: str, val: int = None) -> dict:
"""Set ASC n option opt to value val.
Sets an option on the ASC n to a value. Allowed options are:
AVA+BTB -
opt: freq
val: 256 - 1024 (chip frequency)
BTB -
opt: millivolts
val: 1000 - 1400 (core voltage)
MBA -
opt: reset
val: 0 - # of chips (reset a chip)
opt: freq
val: 0 - # of chips, 100 - 1400 (chip frequency)
opt: ledcount
val: 0 - 100 (chip count for LED)
opt: ledlimit
val: 0 - 200 (LED off below GH/s)
opt: spidelay
val: 0 - 9999 (SPI per I/O delay)
opt: spireset
val: i or s, 0 - 9999 (SPI regular reset)
opt: spisleep
val: 0 - 9999 (SPI reset sleep in ms)
BMA -
opt: volt
val: 0 - 9
opt: clock
val: 0 - 15
:param n: The ASC to set the options on.
:param opt: The option to set. Setting this to 'help' returns a
help message.
:param val: The value to set the option to.
:return: Confirmation of setting option opt to value val.
"""
if val:
return await self.send_command("ascset", parameters=f"{n},{opt},{val}")
else:
return await self.send_command("ascset", parameters=f"{n},{opt}")
async def lcd(self) -> dict:
"""Get a general all-in-one status summary of the miner.
:return: An all-in-one status summary of the miner.
"""
return await self.send_command("lcd")
async def lockstats(self) -> dict:
"""Write lockstats to STDERR.
:return: The result of writing the lock stats to STDERR.
"""
return await self.send_command("lockstats")

80
pyasic/API/unknown.py Normal file
View File

@@ -0,0 +1,80 @@
from pyasic.API import BaseMinerAPI
class UnknownAPI(BaseMinerAPI):
"""An abstraction of an API for a miner which is unknown.
This class is designed to try to be a intersection of as many miner APIs
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)
async def asccount(self) -> dict:
return await self.send_command("asccount")
async def asc(self, n: int) -> dict:
return await self.send_command("asc", parameters=n)
async def devdetails(self) -> dict:
return await self.send_command("devdetails")
async def devs(self) -> dict:
return await self.send_command("devs")
async def edevs(self, old: bool = False) -> dict:
if old:
return await self.send_command("edevs", parameters="old")
else:
return await self.send_command("edevs")
async def pools(self) -> dict:
return await self.send_command("pools")
async def summary(self) -> dict:
return await self.send_command("summary")
async def stats(self) -> dict:
return await self.send_command("stats")
async def version(self) -> dict:
return await self.send_command("version")
async def estats(self) -> dict:
return await self.send_command("estats")
async def check(self) -> dict:
return await self.send_command("check")
async def coin(self) -> dict:
return await self.send_command("coin")
async def lcd(self) -> dict:
return await self.send_command("lcd")
async def switchpool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("enablepool", parameters=n)
async def disablepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("disablepool", parameters=n)
async def addpool(self, url: str, username: str, password: str) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("addpool", parameters=f"{url},{username},{password}")
async def removepool(self, n: int) -> dict:
# BOS has not implemented this yet, they will in the future
raise NotImplementedError
# return await self.send_command("removepool", parameters=n)

0
pyasic/__init__.py Normal file
View File

366
pyasic/config/__init__.py Normal file
View File

@@ -0,0 +1,366 @@
from dataclasses import dataclass, asdict
from typing import List, Literal
import random
import string
import toml
import yaml
import json
import time
@dataclass
class _Pool:
"""A dataclass for pool information.
:param url: URL of the pool.
:param username: Username on the pool.
:param password: Worker password on the pool.
"""
url: str = ""
username: str = ""
password: str = ""
def from_dict(self, data: dict):
"""Convert raw pool data as a dict to usable data and save it to this class.
:param data: The raw config data to convert.
"""
for key in data.keys():
if key == "url":
self.url = data[key]
if key in ["user", "username"]:
self.username = data[key]
if key in ["pass", "password"]:
self.password = data[key]
return self
def as_x19(self, user_suffix: str = None):
"""Convert the data in this class to a dict usable by an X19 device.
:param user_suffix: The suffix to append to username.
"""
username = self.username
if user_suffix:
username = f"{username}{user_suffix}"
pool = {"url": self.url, "user": username, "pass": self.password}
return pool
def as_avalon(self, user_suffix: str = None):
username = self.username
if user_suffix:
username = f"{username}{user_suffix}"
pool = ",".join([self.url, username, self.password])
return pool
def as_bos(self, user_suffix: str = None):
"""Convert the data in this class to a dict usable by an BOSMiner device.
:param user_suffix: The suffix to append to username.
"""
username = self.username
if user_suffix:
username = f"{username}{user_suffix}"
pool = {"url": self.url, "user": username, "password": self.password}
return pool
@dataclass
class _PoolGroup:
"""A dataclass for pool group information.
:param quota: The group quota.
:param group_name: The name of the pool group.
:param pools: A list of pools in this group.
"""
quota: int = 1
group_name: str = None
pools: List[_Pool] = None
def __post_init__(self):
if not self.group_name:
self.group_name = "".join(
random.choice(string.ascii_uppercase + string.digits) for _ in range(6)
) # generate random pool group name in case it isn't set
def from_dict(self, data: dict):
"""Convert raw pool group data as a dict to usable data and save it to this class.
:param data: The raw config data to convert.
"""
pools = []
for key in data.keys():
if key in ["name", "group_name"]:
self.group_name = data[key]
if key == "quota":
self.quota = data[key]
if key in ["pools", "pool"]:
for pool in data[key]:
pools.append(_Pool().from_dict(pool))
self.pools = pools
return self
def as_x19(self, user_suffix: str = None):
"""Convert the data in this class to a dict usable by an X19 device.
:param user_suffix: The suffix to append to username.
"""
pools = []
for pool in self.pools[:3]:
pools.append(pool.as_x19(user_suffix=user_suffix))
return pools
def as_avalon(self, user_suffix: str = None):
pool = self.pools[0].as_avalon(user_suffix=user_suffix)
return pool
def as_bos(self, user_suffix: str = None):
"""Convert the data in this class to a dict usable by an BOSMiner device.
:param user_suffix: The suffix to append to username.
"""
group = {
"name": self.group_name,
"quota": self.quota,
"pool": [pool.as_bos(user_suffix=user_suffix) for pool in self.pools],
}
return group
@dataclass
class MinerConfig:
"""A dataclass for miner configuration information.
:param pool_groups: A list of pool groups in this config.
:param temp_mode: The temperature control mode.
:param temp_target: The target temp.
:param temp_hot: The hot temp (100% fans).
:param temp_dangerous: The dangerous temp (shutdown).
:param minimum_fans: The minimum numbers of fans to run the miner.
:param fan_speed: Manual fan speed to run the fan at (only if temp_mode == "manual").
:param asicboost: Whether or not to enable asicboost.
:param autotuning_enabled: Whether or not to enable autotuning.
:param autotuning_wattage: The wattage to use when autotuning.
:param dps_enabled: Whether or not to enable dynamic power scaling.
:param dps_power_step: The amount of power to reduce autotuning by when the miner reaches dangerous temp.
:param dps_min_power: The minimum power to reduce autotuning to.
:param dps_shutdown_enabled: Whether or not to shutdown the miner when `dps_min_power` is reached.
:param dps_shutdown_duration: The amount of time to shutdown for (in hours).
"""
pool_groups: List[_PoolGroup] = None
temp_mode: Literal["auto", "manual", "disabled"] = "auto"
temp_target: float = 70.0
temp_hot: float = 80.0
temp_dangerous: float = 10.0
minimum_fans: int = None
fan_speed: Literal[tuple(range(101))] = None # noqa - Ignore weird Literal usage
asicboost: bool = None
autotuning_enabled: bool = True
autotuning_wattage: int = 900
dps_enabled: bool = None
dps_power_step: int = None
dps_min_power: int = None
dps_shutdown_enabled: bool = None
dps_shutdown_duration: float = None
def as_dict(self):
"""Convert the data in this class to a dict."""
data_dict = asdict(self)
for key in asdict(self).keys():
if data_dict[key] is None:
del data_dict[key]
return data_dict
def as_toml(self):
"""Convert the data in this class to toml."""
return toml.dumps(self.as_dict())
def as_yaml(self):
"""Convert the data in this class to yaml."""
return yaml.dump(self.as_dict(), sort_keys=False)
def from_raw(self, data: dict):
"""Convert raw config data as a dict to usable data and save it to this class.
:param data: The raw config data to convert.
"""
pool_groups = []
for key in data.keys():
if key == "pools":
pool_groups.append(_PoolGroup().from_dict({"pools": data[key]}))
elif key == "group":
for group in data[key]:
pool_groups.append(_PoolGroup().from_dict(group))
if key == "bitmain-fan-ctrl":
if data[key]:
self.temp_mode = "manual"
if data.get("bitmain-fan-pwm"):
self.fan_speed = int(data["bitmain-fan-pwm"])
elif key == "fan_control":
for _key in data[key].keys():
if _key == "min_fans":
self.minimum_fans = data[key][_key]
elif _key == "speed":
self.fan_speed = data[key][_key]
elif key == "temp_control":
for _key in data[key].keys():
if _key == "mode":
self.temp_mode = data[key][_key]
elif _key == "target_temp":
self.temp_target = data[key][_key]
elif _key == "hot_temp":
self.temp_hot = data[key][_key]
elif _key == "dangerous_temp":
self.temp_dangerous = data[key][_key]
if key == "hash_chain_global":
if data[key].get("asic_boost"):
self.asicboost = data[key]["asic_boost"]
if key == "autotuning":
for _key in data[key].keys():
if _key == "enabled":
self.autotuning_enabled = data[key][_key]
elif _key == "psu_power_limit":
self.autotuning_wattage = data[key][_key]
if key == "power_scaling":
for _key in data[key].keys():
if _key == "enabled":
self.dps_enabled = data[key][_key]
elif _key == "power_step":
self.dps_power_step = data[key][_key]
elif _key == "min_psu_power_limit":
self.dps_min_power = data[key][_key]
elif _key == "shutdown_enabled":
self.dps_shutdown_enabled = data[key][_key]
elif _key == "shutdown_duration":
self.dps_shutdown_duration = data[key][_key]
self.pool_groups = pool_groups
return self
def from_dict(self, data: dict):
"""Convert an output dict of this class back into usable data and save it to this class.
:param data: The raw config data to convert.
"""
pool_groups = []
for group in data["pool_groups"]:
pool_groups.append(_PoolGroup().from_dict(group))
for key in data.keys():
if getattr(self, key) and not key == "pool_groups":
setattr(self, key, data[key])
self.pool_groups = pool_groups
return self
def from_toml(self, data: str):
"""Convert output toml of this class back into usable data and save it to this class.
:param data: The raw config data to convert.
"""
return self.from_dict(toml.loads(data))
def from_yaml(self, data: str):
"""Convert output yaml of this class back into usable data and save it to this class.
:param data: The raw config data to convert.
"""
return self.from_dict(yaml.load(data, Loader=yaml.SafeLoader))
def as_x19(self, user_suffix: str = None) -> str:
"""Convert the data in this class to a config usable by an X19 device.
:param user_suffix: The suffix to append to username.
"""
cfg = {
"pools": self.pool_groups[0].as_x19(user_suffix=user_suffix),
"bitmain-fan-ctrl": False,
"bitmain-fan-pwn": 100,
}
if not self.temp_mode == "auto":
cfg["bitmain-fan-ctrl"] = True
if self.fan_speed:
cfg["bitmain-fan-ctrl"] = str(self.fan_speed)
return json.dumps(cfg)
def as_avalon(self, user_suffix: str = None) -> str:
cfg = self.pool_groups[0].as_avalon()
return cfg
def as_bos(self, model: str = "S9", user_suffix: str = None) -> str:
"""Convert the data in this class to a config usable by an BOSMiner device.
:param model: The model of the miner to be used in the format portion of the config.
:param user_suffix: The suffix to append to username.
"""
cfg = {
"format": {
"version": "1.2+",
"model": f"Antminer {model}",
"generator": "Upstream Config Utility",
"timestamp": int(time.time()),
},
"group": [
group.as_bos(user_suffix=user_suffix) for group in self.pool_groups
],
"temp_control": {
"mode": self.temp_mode,
"target_temp": self.temp_target,
"hot_temp": self.temp_hot,
"dangerous_temp": self.temp_dangerous,
},
}
if self.autotuning_enabled or self.autotuning_wattage:
cfg["autotuning"] = {}
if self.autotuning_enabled:
cfg["autotuning"]["enabled"] = self.autotuning_enabled
if self.autotuning_wattage:
cfg["autotuning"]["psu_power_limit"] = self.autotuning_wattage
if self.asicboost:
cfg["hash_chain_global"] = {}
cfg["hash_chain_global"]["asic_boost"] = self.asicboost
if any(
[
getattr(self, item)
for item in [
"dps_enabled",
"dps_power_step",
"dps_min_power",
"dps_shutdown_enabled",
"dps_shutdown_duration",
]
]
):
cfg["power_scaling"] = {}
if self.dps_enabled:
cfg["power_scaling"]["enabled"] = self.dps_enabled
if self.dps_power_step:
cfg["power_scaling"]["power_step"] = self.dps_power_step
if self.dps_min_power:
cfg["power_scaling"]["min_psu_power_limit"] = self.dps_min_power
if self.dps_shutdown_enabled:
cfg["power_scaling"]["shutdown_enabled"] = self.dps_shutdown_enabled
if self.dps_shutdown_duration:
cfg["power_scaling"]["shutdown_duration"] = self.dps_shutdown_duration
return toml.dumps(cfg)

116
pyasic/data/__init__.py Normal file
View File

@@ -0,0 +1,116 @@
from dataclasses import dataclass, field, asdict
from datetime import datetime
@dataclass
class MinerData:
"""A Dataclass to standardize data returned from miners (specifically AnyMiner().get_data())
:param ip: The IP of the miner as a str.
:param datetime: The time and date this data was generated.
:param model: The model of the miner as a str.
:param hostname: The network hostname of the miner as a str.
:param hashrate: The hashrate of the miner in TH/s as a int.
:param left_board_temp: The temp of the left PCB as an int.
:param left_board_chip_temp: The temp of the left board chips as an int.
:param center_board_temp: The temp of the center PCB as an int.
:param center_board_chip_temp: The temp of the center board chips as an int.
:param right_board_temp: The temp of the right PCB as an int.
:param right_board_chip_temp: The temp of the right board chips as an int.
:param wattage: Wattage of the miner as an int.
:param fan_1: The speed of the first fan as an int.
:param fan_2: The speed of the second fan as an int.
:param fan_3: The speed of the third fan as an int.
:param fan_4: The speed of the fourth fan as an int.
:param left_chips: The number of chips online in the left board as an int.
:param center_chips: The number of chips online in the left board as an int.
:param right_chips: The number of chips online in the left board as an int.
:param ideal_chips: The ideal number of chips in the miner as an int.
:param pool_split: The pool split as a str.
:param pool_1_url: The first pool url on the miner as a str.
:param pool_1_user: The first pool user on the miner as a str.
:param pool_2_url: The second pool url on the miner as a str.
:param pool_2_user: The second pool user on the miner as a str.
"""
ip: str
datetime: datetime = None
mac: str = "00:00:00:00:00:00"
model: str = "Unknown"
hostname: str = "Unknown"
hashrate: float = 0
temperature_avg: int = field(init=False)
env_temp: float = 0
left_board_temp: int = 0
left_board_chip_temp: int = 0
center_board_temp: int = 0
center_board_chip_temp: int = 0
right_board_temp: int = 0
right_board_chip_temp: int = 0
wattage: int = 0
fan_1: int = -1
fan_2: int = -1
fan_3: int = -1
fan_4: int = -1
left_chips: int = 0
center_chips: int = 0
right_chips: int = 0
total_chips: int = field(init=False)
ideal_chips: int = 1
percent_ideal: float = field(init=False)
nominal: int = field(init=False)
pool_split: str = "0"
pool_1_url: str = "Unknown"
pool_1_user: str = "Unknown"
pool_2_url: str = ""
pool_2_user: str = ""
def __post_init__(self):
self.datetime = datetime.now()
@property
def total_chips(self): # noqa - Skip PyCharm inspection
return self.right_chips + self.center_chips + self.left_chips
@total_chips.setter
def total_chips(self, val):
pass
@property
def nominal(self): # noqa - Skip PyCharm inspection
return self.ideal_chips == self.total_chips
@nominal.setter
def nominal(self, val):
pass
@property
def percent_ideal(self): # noqa - Skip PyCharm inspection
return round((self.total_chips / self.ideal_chips) * 100)
@percent_ideal.setter
def percent_ideal(self, val):
pass
@property
def temperature_avg(self): # noqa - Skip PyCharm inspection
total_temp = 0
temp_count = 0
for temp in [
self.left_board_chip_temp,
self.center_board_chip_temp,
self.right_board_chip_temp,
]:
if temp and not temp == 0:
total_temp += temp
temp_count += 1
if not temp_count > 0:
return 0
return round(total_temp / temp_count)
@temperature_avg.setter
def temperature_avg(self, val):
pass
def asdict(self):
return asdict(self)

31
pyasic/logger/__init__.py Normal file
View File

@@ -0,0 +1,31 @@
import logging
from pyasic.settings import DEBUG, LOGFILE
def init_logger():
if LOGFILE:
logging.basicConfig(
filename="logfile.txt",
filemode="a",
format="%(pathname)s:%(lineno)d in %(funcName)s\n[%(levelname)s][%(asctime)s](%(name)s) - %(message)s",
datefmt="%x %X",
)
else:
logging.basicConfig(
format="%(pathname)s:%(lineno)d in %(funcName)s\n[%(levelname)s][%(asctime)s](%(name)s) - %(message)s",
datefmt="%x %X",
)
_logger = logging.getLogger()
if DEBUG:
_logger.setLevel(logging.DEBUG)
logging.getLogger("asyncssh").setLevel(logging.DEBUG)
else:
_logger.setLevel(logging.WARNING)
logging.getLogger("asyncssh").setLevel(logging.WARNING)
return _logger
logger = init_logger()

104
pyasic/miners/__init__.py Normal file
View File

@@ -0,0 +1,104 @@
import asyncssh
import logging
import ipaddress
from pyasic.data import MinerData
class BaseMiner:
def __init__(self, *args) -> None:
self.ip = None
self.uname = "root"
self.pwd = "admin"
self.api = None
self.api_type = None
self.model = None
self.light = None
self.hostname = None
self.nominal_chips = 1
self.version = None
self.fan_count = 2
self.config = None
def __repr__(self):
return f"{'' if not self.api_type else self.api_type} {'' if not self.model else self.model}: {str(self.ip)}"
def __lt__(self, other):
return ipaddress.ip_address(self.ip) < ipaddress.ip_address(other.ip)
def __gt__(self, other):
return ipaddress.ip_address(self.ip) > ipaddress.ip_address(other.ip)
def __eq__(self, other):
return ipaddress.ip_address(self.ip) == ipaddress.ip_address(other.ip)
async def _get_ssh_connection(self) -> asyncssh.connect:
"""Create a new asyncssh connection"""
try:
conn = await asyncssh.connect(
str(self.ip),
known_hosts=None,
username=self.uname,
password=self.pwd,
server_host_key_algs=["ssh-rsa"],
)
return conn
except asyncssh.misc.PermissionDenied:
try:
conn = await asyncssh.connect(
str(self.ip),
known_hosts=None,
username="root",
password="admin",
server_host_key_algs=["ssh-rsa"],
)
return conn
except Exception as e:
# logging.warning(f"{self} raised an exception: {e}")
raise e
except OSError as e:
logging.warning(f"Connection refused: {self}")
raise e
except Exception as e:
# logging.warning(f"{self} raised an exception: {e}")
raise e
async def fault_light_on(self) -> bool:
return False
async def fault_light_off(self) -> bool:
return False
async def send_file(self, src, dest):
async with (await self._get_ssh_connection()) as conn:
await asyncssh.scp(src, (conn, dest))
async def check_light(self):
return self.light
async def get_board_info(self):
return None
async def get_config(self):
return None
async def get_hostname(self):
return None
async def get_model(self):
return None
async def reboot(self):
return False
async def restart_backend(self):
return False
async def send_config(self, *args, **kwargs):
return None
async def get_mac(self):
return None
async def get_data(self) -> MinerData:
return MinerData(ip=str(self.ip))

5
pyasic/miners/_backends/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from .bmminer import BMMiner
from .bosminer import BOSMiner
from .btminer import BTMiner
from .cgminer import CGMiner
from .hiveon import Hiveon

264
pyasic/miners/_backends/bmminer.py Normal file
View File

@@ -0,0 +1,264 @@
import ipaddress
import logging
from pyasic.API.bmminer import BMMinerAPI
from pyasic.miners import BaseMiner
from pyasic.data import MinerData
from pyasic.settings import MINER_FACTORY_GET_VERSION_RETRIES as DATA_RETRIES
class BMMiner(BaseMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ipaddress.ip_address(ip)
self.api = BMMinerAPI(ip)
self.api_type = "BMMiner"
self.uname = "root"
self.pwd = "admin"
async def get_model(self) -> str or None:
"""Get miner model.
:return: Miner model or None.
"""
# check if model is cached
if self.model:
logging.debug(f"Found model for {self.ip}: {self.model}")
return self.model
# get devdetails data
version_data = await self.api.devdetails()
# if we get data back, parse it for model
if version_data:
# handle Antminer BMMiner as a base
self.model = version_data["DEVDETAILS"][0]["Model"].replace("Antminer ", "")
logging.debug(f"Found model for {self.ip}: {self.model}")
return self.model
# if we don't get devdetails, log a failed attempt
logging.warning(f"Failed to get model for miner: {self}")
return None
async def get_hostname(self) -> str:
"""Get miner hostname.
:return: The hostname of the miner as a string or "?"
"""
if self.hostname:
return self.hostname
try:
# open an ssh connection
async with (await self._get_ssh_connection()) as conn:
# if we get the connection, check hostname
if conn is not None:
# get output of the hostname file
data = await conn.run("cat /proc/sys/kernel/hostname")
host = data.stdout.strip()
# return hostname data
logging.debug(f"Found hostname for {self.ip}: {host}")
self.hostname = host
return self.hostname
else:
# return ? if we fail to get hostname with no ssh connection
logging.warning(f"Failed to get hostname for miner: {self}")
return "?"
except Exception:
# return ? if we fail to get hostname with an exception
logging.warning(f"Failed to get hostname for miner: {self}")
return "?"
async def send_ssh_command(self, cmd: str) -> str or None:
"""Send a command to the miner over ssh.
:param cmd: The command to run.
:return: Result of the command or None.
"""
result = None
# open an ssh connection
async with (await self._get_ssh_connection()) as conn:
# 3 retries
for i in range(3):
try:
# run the command and get the result
result = await conn.run(cmd)
result = result.stdout
except Exception as e:
# if the command fails, log it
logging.warning(f"{self} command {cmd} error: {e}")
# on the 3rd retry, return None
if i == 3:
return
continue
# return the result, either command output or None
return result
async def get_config(self) -> list or None:
"""Get the pool configuration of the miner.
:return: Pool config data or None.
"""
# get pool data
pools = await self.api.pools()
pool_data = []
# ensure we got pool data
if not pools:
return
# parse all the pools
for pool in pools["POOLS"]:
pool_data.append({"url": pool["URL"], "user": pool["User"], "pwd": "123"})
return pool_data
async def reboot(self) -> bool:
logging.debug(f"{self}: Sending reboot command.")
_ret = await self.send_ssh_command("reboot")
logging.debug(f"{self}: Reboot command completed.")
if isinstance(_ret, str):
return True
return False
async def get_data(self) -> MinerData:
data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
board_offset = -1
fan_offset = -1
model = await self.get_model()
hostname = await self.get_hostname()
mac = await self.get_mac()
if model:
data.model = model
if hostname:
data.hostname = hostname
if mac:
data.mac = mac
miner_data = None
for i in range(DATA_RETRIES):
miner_data = await self.api.multicommand(
"summary", "pools", "stats", ignore_x19_error=True
)
if miner_data:
break
if not miner_data:
return data
summary = miner_data.get("summary")[0]
pools = miner_data.get("pools")[0]
stats = miner_data.get("stats")[0]
if summary:
hr = summary.get("SUMMARY")
if hr:
if len(hr) > 0:
hr = hr[0].get("GHS av")
if hr:
data.hashrate = round(hr / 1000, 2)
if stats:
boards = stats.get("STATS")
if boards:
if len(boards) > 0:
for board_num in range(1, 16, 5):
for _b_num in range(5):
b = boards[1].get(f"chain_acn{board_num + _b_num}")
if b and not b == 0 and board_offset == -1:
board_offset = board_num
if board_offset == -1:
board_offset = 1
data.left_chips = boards[1].get(f"chain_acn{board_offset}")
data.center_chips = boards[1].get(f"chain_acn{board_offset+1}")
data.right_chips = boards[1].get(f"chain_acn{board_offset+2}")
if stats:
temp = stats.get("STATS")
if temp:
if len(temp) > 1:
for fan_num in range(1, 8, 4):
for _f_num in range(4):
f = temp[1].get(f"fan{fan_num + _f_num}")
if f and not f == 0 and fan_offset == -1:
fan_offset = fan_num
if fan_offset == -1:
fan_offset = 1
for fan in range(self.fan_count):
setattr(
data, f"fan_{fan + 1}", temp[1].get(f"fan{fan_offset+fan}")
)
board_map = {0: "left_board", 1: "center_board", 2: "right_board"}
env_temp_list = []
for item in range(3):
board_temp = temp[1].get(f"temp{item + board_offset}")
chip_temp = temp[1].get(f"temp2_{item + board_offset}")
setattr(data, f"{board_map[item]}_chip_temp", chip_temp)
setattr(data, f"{board_map[item]}_temp", board_temp)
if f"temp_pcb{item}" in temp[1].keys():
env_temp = temp[1][f"temp_pcb{item}"].split("-")[0]
if not env_temp == 0:
env_temp_list.append(int(env_temp))
data.env_temp = sum(env_temp_list) / len(env_temp_list)
if pools:
pool_1 = None
pool_2 = None
pool_1_user = None
pool_2_user = None
pool_1_quota = 1
pool_2_quota = 1
quota = 0
for pool in pools.get("POOLS"):
if not pool_1_user:
pool_1_user = pool.get("User")
pool_1 = pool["URL"]
pool_1_quota = pool["Quota"]
elif not pool_2_user:
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if not pool.get("User") == pool_1_user:
if not pool_2_user == pool.get("User"):
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if pool_2_user and not pool_2_user == pool_1_user:
quota = f"{pool_1_quota}/{pool_2_quota}"
if pool_1:
pool_1 = pool_1.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_1_url = pool_1
if pool_1_user:
data.pool_1_user = pool_1_user
if pool_2:
pool_2 = pool_2.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_2_url = pool_2
if pool_2_user:
data.pool_2_user = pool_2_user
if quota:
data.pool_split = str(quota)
return data

403
pyasic/miners/_backends/bosminer.py Normal file
View File

@@ -0,0 +1,403 @@
import ipaddress
import logging
import json
import toml
from pyasic.miners import BaseMiner
from pyasic.API.bosminer import BOSMinerAPI
from pyasic.API import APIError
from pyasic.data import MinerData
from pyasic.config import MinerConfig
from pyasic.settings import MINER_FACTORY_GET_VERSION_RETRIES as DATA_RETRIES
class BOSMiner(BaseMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ipaddress.ip_address(ip)
self.api = BOSMinerAPI(ip)
self.api_type = "BOSMiner"
self.uname = "root"
self.pwd = "admin"
self.config = None
async def send_ssh_command(self, cmd: str) -> str or None:
"""Send a command to the miner over ssh.
:return: Result of the command or None.
"""
result = None
# open an ssh connection
async with (await self._get_ssh_connection()) as conn:
# 3 retries
for i in range(3):
try:
# run the command and get the result
result = await conn.run(cmd)
result = result.stdout
except Exception as e:
# if the command fails, log it
logging.warning(f"{self} command {cmd} error: {e}")
# on the 3rd retry, return None
if i == 3:
return
continue
# return the result, either command output or None
return str(result)
async def fault_light_on(self) -> bool:
"""Sends command to turn on fault light on the miner."""
logging.debug(f"{self}: Sending fault_light on command.")
self.light = True
_ret = await self.send_ssh_command("miner fault_light on")
logging.debug(f"{self}: fault_light on command completed.")
if isinstance(_ret, str):
return True
return False
async def fault_light_off(self) -> bool:
"""Sends command to turn off fault light on the miner."""
logging.debug(f"{self}: Sending fault_light off command.")
self.light = False
_ret = await self.send_ssh_command("miner fault_light off")
logging.debug(f"{self}: fault_light off command completed.")
if isinstance(_ret, str):
return True
return False
async def restart_backend(self) -> bool:
return await self.restart_bosminer()
async def restart_bosminer(self) -> bool:
"""Restart bosminer hashing process."""
logging.debug(f"{self}: Sending bosminer restart command.")
_ret = await self.send_ssh_command("/etc/init.d/bosminer restart")
logging.debug(f"{self}: bosminer restart command completed.")
if isinstance(_ret, str):
return True
return False
async def reboot(self) -> bool:
"""Reboots power to the physical miner."""
logging.debug(f"{self}: Sending reboot command.")
_ret = await self.send_ssh_command("/sbin/reboot")
logging.debug(f"{self}: Reboot command completed.")
if isinstance(_ret, str):
return True
return False
async def get_config(self) -> None:
logging.debug(f"{self}: Getting config.")
async with (await self._get_ssh_connection()) as conn:
logging.debug(f"{self}: Opening SFTP connection.")
async with conn.start_sftp_client() as sftp:
logging.debug(f"{self}: Reading config file.")
async with sftp.open("/etc/bosminer.toml") as file:
toml_data = toml.loads(await file.read())
logging.debug(f"{self}: Converting config file.")
cfg = MinerConfig().from_raw(toml_data)
self.config = cfg
return self.config
async def get_hostname(self) -> str:
"""Get miner hostname.
:return: The hostname of the miner as a string or "?"
"""
if self.hostname:
return self.hostname
try:
async with (await self._get_ssh_connection()) as conn:
if conn is not None:
data = await conn.run("cat /proc/sys/kernel/hostname")
host = data.stdout.strip()
logging.debug(f"Found hostname for {self.ip}: {host}")
self.hostname = host
return self.hostname
else:
logging.warning(f"Failed to get hostname for miner: {self}")
return "?"
except Exception:
logging.warning(f"Failed to get hostname for miner: {self}")
return "?"
async def get_model(self) -> str or None:
"""Get miner model.
:return: Miner model or None.
"""
# check if model is cached
if self.model:
logging.debug(f"Found model for {self.ip}: {self.model} (BOS)")
return self.model + " (BOS)"
# get devdetails data
try:
version_data = await self.api.devdetails()
except APIError as e:
version_data = None
if e.message == "Not ready":
cfg = json.loads(await self.send_ssh_command("bosminer config --data"))
model = cfg.get("data").get("format").get("model")
if model:
model = model.replace("Antminer ", "")
self.model = model
return self.model + " (BOS)"
# if we get data back, parse it for model
if version_data:
if not version_data["DEVDETAILS"] == []:
# handle Antminer BOSMiner as a base
self.model = version_data["DEVDETAILS"][0]["Model"].replace(
"Antminer ", ""
)
logging.debug(f"Found model for {self.ip}: {self.model} (BOS)")
return self.model + " (BOS)"
# if we don't get devdetails, log a failed attempt
logging.warning(f"Failed to get model for miner: {self}")
return None
async def get_version(self):
"""Get miner firmware version.
:return: Miner firmware version or None.
"""
# check if version is cached
if self.version:
logging.debug(f"Found version for {self.ip}: {self.version}")
return self.version
# get output of bos version file
version_data = await self.send_ssh_command("cat /etc/bos_version")
# if we get the version data, parse it
if version_data:
self.version = version_data.split("-")[5]
logging.debug(f"Found version for {self.ip}: {self.version}")
return self.version
# if we fail to get version, log a failed attempt
logging.warning(f"Failed to get model for miner: {self}")
return None
async def send_config(self, yaml_config, ip_user: bool = False) -> None:
"""Configures miner with yaml config."""
logging.debug(f"{self}: Sending config.")
if ip_user:
suffix = str(self.ip).split(".")[-1]
toml_conf = (
MinerConfig()
.from_yaml(yaml_config)
.as_bos(model=self.model.replace(" (BOS)", ""), user_suffix=suffix)
)
else:
toml_conf = (
MinerConfig()
.from_yaml(yaml_config)
.as_bos(model=self.model.replace(" (BOS)", ""))
)
async with (await self._get_ssh_connection()) as conn:
logging.debug(f"{self}: Opening SFTP connection.")
async with conn.start_sftp_client() as sftp:
logging.debug(f"{self}: Opening config file.")
async with sftp.open("/etc/bosminer.toml", "w+") as file:
await file.write(toml_conf)
logging.debug(f"{self}: Restarting BOSMiner")
await conn.run("/etc/init.d/bosminer restart")
async def get_board_info(self) -> dict:
"""Gets data on each board and chain in the miner."""
logging.debug(f"{self}: Getting board info.")
devdetails = await self.api.devdetails()
if not devdetails.get("DEVDETAILS"):
print("devdetails error", devdetails)
return {0: [], 1: [], 2: []}
devs = devdetails["DEVDETAILS"]
boards = {}
offset = devs[0]["ID"]
for board in devs:
boards[board["ID"] - offset] = []
if not board["Chips"] == self.nominal_chips:
nominal = False
else:
nominal = True
boards[board["ID"] - offset].append(
{
"chain": board["ID"] - offset,
"chip_count": board["Chips"],
"chip_status": "o" * board["Chips"],
"nominal": nominal,
}
)
logging.debug(f"Found board data for {self}: {boards}")
return boards
async def get_bad_boards(self) -> dict:
"""Checks for and provides list of non working boards."""
boards = await self.get_board_info()
bad_boards = {}
for board in boards.keys():
for chain in boards[board]:
if not chain["chip_count"] == 63:
if board not in bad_boards.keys():
bad_boards[board] = []
bad_boards[board].append(chain)
return bad_boards
async def check_good_boards(self) -> str:
"""Checks for and provides list for working boards."""
devs = await self.api.devdetails()
bad = 0
chains = devs["DEVDETAILS"]
for chain in chains:
if chain["Chips"] == 0:
bad += 1
if not bad > 0:
return str(self.ip)
async def get_data(self) -> MinerData:
data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
board_offset = -1
fan_offset = -1
model = await self.get_model()
hostname = await self.get_hostname()
mac = await self.get_mac()
if model:
data.model = model
if hostname:
data.hostname = hostname
if mac:
data.mac = mac
miner_data = None
for i in range(DATA_RETRIES):
try:
miner_data = await self.api.multicommand(
"summary", "temps", "tunerstatus", "pools", "devdetails", "fans"
)
except APIError as e:
if str(e.message) == "Not ready":
miner_data = await self.api.multicommand(
"summary", "tunerstatus", "pools", "fans"
)
if miner_data:
break
if not miner_data:
return data
summary = miner_data.get("summary")
temps = miner_data.get("temps")
tunerstatus = miner_data.get("tunerstatus")
pools = miner_data.get("pools")
devdetails = miner_data.get("devdetails")
fans = miner_data.get("fans")
if summary:
hr = summary[0].get("SUMMARY")
if hr:
if len(hr) > 0:
hr = hr[0].get("MHS 1m")
if hr:
data.hashrate = round(hr / 1000000, 2)
if temps:
temp = temps[0].get("TEMPS")
if temp:
if len(temp) > 0:
board_map = {0: "left_board", 1: "center_board", 2: "right_board"}
offset = 6 if temp[0]["ID"] in [6, 7, 8] else temp[0]["ID"]
for board in temp:
_id = board["ID"] - offset
chip_temp = round(board["Chip"])
board_temp = round(board["Board"])
setattr(data, f"{board_map[_id]}_chip_temp", chip_temp)
setattr(data, f"{board_map[_id]}_temp", board_temp)
if fans:
fan_data = fans[0].get("FANS")
if fan_data:
for fan in range(self.fan_count):
setattr(data, f"fan_{fan+1}", fan_data[fan]["RPM"])
if pools:
pool_1 = None
pool_2 = None
pool_1_user = None
pool_2_user = None
pool_1_quota = 1
pool_2_quota = 1
quota = 0
for pool in pools[0].get("POOLS"):
if not pool_1_user:
pool_1_user = pool.get("User")
pool_1 = pool["URL"]
pool_1_quota = pool["Quota"]
elif not pool_2_user:
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if not pool.get("User") == pool_1_user:
if not pool_2_user == pool.get("User"):
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if pool_2_user and not pool_2_user == pool_1_user:
quota = f"{pool_1_quota}/{pool_2_quota}"
if pool_1:
pool_1 = pool_1.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_1_url = pool_1
if pool_1_user:
data.pool_1_user = pool_1_user
if pool_2:
pool_2 = pool_2.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_2_url = pool_2
if pool_2_user:
data.pool_2_user = pool_2_user
if quota:
data.pool_split = str(quota)
if tunerstatus:
tuner = tunerstatus[0].get("TUNERSTATUS")
if tuner:
if len(tuner) > 0:
wattage = tuner[0].get("PowerLimit")
if wattage:
data.wattage = wattage
if devdetails:
boards = devdetails[0].get("DEVDETAILS")
if boards:
if len(boards) > 0:
board_map = {0: "left_chips", 1: "center_chips", 2: "right_chips"}
offset = 6 if boards[0]["ID"] in [6, 7, 8] else boards[0]["ID"]
for board in boards:
_id = board["ID"] - offset
chips = board["Chips"]
setattr(data, board_map[_id], chips)
return data
async def get_mac(self):
result = await self.send_ssh_command("cat /sys/class/net/eth0/address")
return result.upper().strip()

50
pyasic/miners/_backends/bosminer_old.py Normal file
View File

@@ -0,0 +1,50 @@
import logging
import ipaddress
from pyasic.API.bosminer import BOSMinerAPI
from pyasic.miners import BaseMiner
class BOSMinerOld(BaseMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ipaddress.ip_address(ip)
self.api = BOSMinerAPI(ip)
self.api_type = "BOSMiner"
self.uname = "root"
self.pwd = "admin"
async def send_ssh_command(self, cmd: str) -> str or None:
"""Send a command to the miner over ssh.
:return: Result of the command or None.
"""
result = None
# open an ssh connection
async with (await self._get_ssh_connection()) as conn:
# 3 retries
for i in range(3):
try:
# run the command and get the result
result = await conn.run(cmd)
if result.stdout:
result = result.stdout
except Exception as e:
if e == "SSH connection closed":
return "Update completed."
# if the command fails, log it
logging.warning(f"{self} command {cmd} error: {e}")
# on the 3rd retry, return None
if i == 3:
return
continue
# return the result, either command output or None
return str(result)
async def update_to_plus(self):
result = await self.send_ssh_command("opkg update && opkg install bos_plus")
return result

246
pyasic/miners/_backends/btminer.py Normal file
View File

@@ -0,0 +1,246 @@
import ipaddress
import logging
from pyasic.API.btminer import BTMinerAPI
from pyasic.miners import BaseMiner
from pyasic.API import APIError
from pyasic.data import MinerData
from pyasic.settings import MINER_FACTORY_GET_VERSION_RETRIES as DATA_RETRIES
class BTMiner(BaseMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ipaddress.ip_address(ip)
self.api = BTMinerAPI(ip)
self.api_type = "BTMiner"
async def get_model(self):
if self.model:
logging.debug(f"Found model for {self.ip}: {self.model}")
return self.model
version_data = await self.api.devdetails()
if version_data:
self.model = version_data["DEVDETAILS"][0]["Model"].split("V")[0]
logging.debug(f"Found model for {self.ip}: {self.model}")
return self.model
logging.warning(f"Failed to get model for miner: {self}")
return None
async def get_hostname(self) -> str or None:
if self.hostname:
return self.hostname
try:
host_data = await self.api.get_miner_info()
if host_data:
host = host_data["Msg"]["hostname"]
logging.debug(f"Found hostname for {self.ip}: {host}")
self.hostname = host
return self.hostname
except APIError:
logging.info(f"Failed to get hostname for miner: {self}")
return None
except Exception:
logging.warning(f"Failed to get hostname for miner: {self}")
return None
async def get_board_info(self) -> dict:
"""Gets data on each board and chain in the miner."""
logging.debug(f"{self}: Getting board info.")
devs = await self.api.devs()
if not devs.get("DEVS"):
print("devs error", devs)
return {0: [], 1: [], 2: []}
devs = devs["DEVS"]
boards = {}
offset = devs[0]["ID"]
for board in devs:
boards[board["ID"] - offset] = []
if "Effective Chips" in board.keys():
if not board["Effective Chips"] in self.nominal_chips:
nominal = False
else:
nominal = True
boards[board["ID"] - offset].append(
{
"chain": board["ID"] - offset,
"chip_count": board["Effective Chips"],
"chip_status": "o" * board["Effective Chips"],
"nominal": nominal,
}
)
else:
logging.warning(f"Incorrect board data from {self}: {board}")
print(board)
logging.debug(f"Found board data for {self}: {boards}")
return boards
async def get_mac(self):
mac = ""
data = await self.api.summary()
if data:
if data.get("SUMMARY"):
if len(data["SUMMARY"]) > 0:
_mac = data["SUMMARY"][0].get("MAC")
if _mac:
mac = _mac
if mac == "":
try:
data = await self.api.get_miner_info()
if data:
if "Msg" in data.keys():
if "mac" in data["Msg"].keys():
mac = data["Msg"]["mac"]
except APIError:
pass
return str(mac).upper()
async def get_data(self):
data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
mac = None
try:
model = await self.get_model()
except APIError:
logging.info(f"Failed to get model: {self}")
model = None
data.model = "Whatsminer"
try:
hostname = await self.get_hostname()
except APIError:
logging.info(f"Failed to get hostname: {self}")
hostname = None
data.hostname = "Whatsminer"
if model:
data.model = model
if hostname:
data.hostname = hostname
miner_data = None
for i in range(DATA_RETRIES):
try:
miner_data = await self.api.multicommand("summary", "devs", "pools")
if miner_data:
break
except APIError:
pass
if not miner_data:
return data
summary = miner_data.get("summary")[0]
devs = miner_data.get("devs")[0]
pools = miner_data.get("pools")[0]
if summary:
summary_data = summary.get("SUMMARY")
if summary_data:
if len(summary_data) > 0:
if summary_data[0].get("MAC"):
mac = summary_data[0]["MAC"]
if summary_data[0].get("Env Temp"):
data.env_temp = summary_data[0]["Env Temp"]
data.fan_1 = summary_data[0]["Fan Speed In"]
data.fan_2 = summary_data[0]["Fan Speed Out"]
hr = summary_data[0].get("MHS 1m")
if hr:
data.hashrate = round(hr / 1000000, 2)
wattage = summary_data[0].get("Power")
if wattage:
data.wattage = round(wattage)
if devs:
temp_data = devs.get("DEVS")
if temp_data:
board_map = {0: "left_board", 1: "center_board", 2: "right_board"}
for board in temp_data:
_id = board["ASC"]
chip_temp = round(board["Chip Temp Avg"])
board_temp = round(board["Temperature"])
setattr(data, f"{board_map[_id]}_chip_temp", chip_temp)
setattr(data, f"{board_map[_id]}_temp", board_temp)
if devs:
boards = devs.get("DEVS")
if boards:
if len(boards) > 0:
board_map = {0: "left_chips", 1: "center_chips", 2: "right_chips"}
if "ID" in boards[0].keys():
id_key = "ID"
else:
id_key = "ASC"
offset = boards[0][id_key]
for board in boards:
_id = board[id_key] - offset
chips = board["Effective Chips"]
setattr(data, board_map[_id], chips)
if pools:
pool_1 = None
pool_2 = None
pool_1_user = None
pool_2_user = None
pool_1_quota = 1
pool_2_quota = 1
quota = 0
for pool in pools.get("POOLS"):
if not pool_1_user:
pool_1_user = pool.get("User")
pool_1 = pool["URL"]
pool_1_quota = pool["Quota"]
elif not pool_2_user:
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if not pool.get("User") == pool_1_user:
if not pool_2_user == pool.get("User"):
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if pool_2_user and not pool_2_user == pool_1_user:
quota = f"{pool_1_quota}/{pool_2_quota}"
if pool_1:
pool_1 = pool_1.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_1_url = pool_1
if pool_1_user:
data.pool_1_user = pool_1_user
if pool_2:
pool_2 = pool_2.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_2_url = pool_2
if pool_2_user:
data.pool_2_user = pool_2_user
if quota:
data.pool_split = str(quota)
if not mac:
try:
mac = await self.get_mac()
except APIError:
logging.info(f"Failed to get mac: {self}")
mac = None
if mac:
data.mac = mac
return data

210
pyasic/miners/_backends/cgminer.py Normal file
View File

@@ -0,0 +1,210 @@
import ipaddress
import logging
from pyasic.API.cgminer import CGMinerAPI
from pyasic.miners import BaseMiner
from pyasic.API import APIError
from pyasic.data import MinerData
from pyasic.settings import MINER_FACTORY_GET_VERSION_RETRIES as DATA_RETRIES
class CGMiner(BaseMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ipaddress.ip_address(ip)
self.api = CGMinerAPI(ip)
self.api_type = "CGMiner"
self.uname = "root"
self.pwd = "admin"
self.config = None
async def get_model(self):
if self.model:
return self.model
try:
version_data = await self.api.devdetails()
except APIError:
return None
if version_data:
self.model = version_data["DEVDETAILS"][0]["Model"].replace("Antminer ", "")
return self.model
return None
async def get_hostname(self) -> str or None:
if self.hostname:
return self.hostname
try:
async with (await self._get_ssh_connection()) as conn:
if conn is not None:
data = await conn.run("cat /proc/sys/kernel/hostname")
host = data.stdout.strip()
self.hostname = host
return self.hostname
else:
return None
except Exception:
return None
async def send_ssh_command(self, cmd):
result = None
async with (await self._get_ssh_connection()) as conn:
for i in range(3):
try:
result = await conn.run(cmd)
result = result.stdout
except Exception as e:
print(f"{cmd} error: {e}")
if i == 3:
return
continue
return result
async def restart_backend(self) -> bool:
return await self.restart_cgminer()
async def restart_cgminer(self) -> bool:
commands = ["cgminer-api restart", "/usr/bin/cgminer-monitor >/dev/null 2>&1"]
commands = ";".join(commands)
_ret = await self.send_ssh_command(commands)
if isinstance(_ret, str):
return True
return False
async def reboot(self) -> bool:
logging.debug(f"{self}: Sending reboot command.")
_ret = await self.send_ssh_command("reboot")
logging.debug(f"{self}: Reboot command completed.")
if isinstance(_ret, str):
return True
return False
async def start_cgminer(self) -> None:
commands = [
"mkdir -p /etc/tmp/",
'echo "*/3 * * * * /usr/bin/cgminer-monitor" > /etc/tmp/root',
"crontab -u root /etc/tmp/root",
"/usr/bin/cgminer-monitor >/dev/null 2>&1",
]
commands = ";".join(commands)
await self.send_ssh_command(commands)
async def stop_cgminer(self) -> None:
commands = [
"mkdir -p /etc/tmp/",
'echo "" > /etc/tmp/root',
"crontab -u root /etc/tmp/root",
"killall cgminer",
]
commands = ";".join(commands)
await self.send_ssh_command(commands)
async def get_config(self) -> None:
async with (await self._get_ssh_connection()) as conn:
command = "cat /etc/config/cgminer"
result = await conn.run(command, check=True)
self.config = result.stdout
print(str(self.config))
async def get_data(self):
data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
model = await self.get_model()
hostname = await self.get_hostname()
mac = await self.get_mac()
if model:
data.model = model
if hostname:
data.hostname = hostname
if mac:
data.mac = mac
miner_data = None
for i in range(DATA_RETRIES):
miner_data = await self.api.multicommand("summary", "pools", "stats")
if miner_data:
break
if not miner_data:
return data
summary = miner_data.get("summary")[0]
pools = miner_data.get("pools")[0]
stats = miner_data.get("stats")[0]
if summary:
hr = summary.get("SUMMARY")
if hr:
if len(hr) > 0:
hr = hr[0].get("GHS 1m")
if hr:
data.hashrate = round(hr / 1000, 2)
if stats:
temp = stats.get("STATS")
if temp:
if len(temp) > 1:
data.fan_1 = temp[1].get("fan1")
data.fan_2 = temp[1].get("fan2")
data.fan_3 = temp[1].get("fan3")
data.fan_4 = temp[1].get("fan4")
board_map = {1: "left_board", 2: "center_board", 3: "right_board"}
for item in range(1, 4):
board_temp = temp[1].get(f"temp{item}")
chip_temp = temp[1].get(f"temp2_{item}")
setattr(data, f"{board_map[item]}_chip_temp", chip_temp)
setattr(data, f"{board_map[item]}_temp", board_temp)
if pools:
pool_1 = None
pool_2 = None
pool_1_user = None
pool_2_user = None
pool_1_quota = 1
pool_2_quota = 1
quota = 0
for pool in pools.get("POOLS"):
if not pool_1_user:
pool_1_user = pool.get("User")
pool_1 = pool["URL"]
pool_1_quota = pool["Quota"]
elif not pool_2_user:
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if not pool.get("User") == pool_1_user:
if not pool_2_user == pool.get("User"):
pool_2_user = pool.get("User")
pool_2 = pool["URL"]
pool_2_quota = pool["Quota"]
if pool_2_user and not pool_2_user == pool_1_user:
quota = f"{pool_1_quota}/{pool_2_quota}"
if pool_1:
pool_1 = pool_1.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_1_url = pool_1
if pool_1_user:
data.pool_1_user = pool_1_user
if pool_2:
pool_2 = pool_2.replace("stratum+tcp://", "").replace(
"stratum2+tcp://", ""
)
data.pool_2_url = pool_2
if pool_2_user:
data.pool_2_user = pool_2_user
if quota:
data.pool_split = str(quota)
return data

48
pyasic/miners/_backends/hiveon.py Normal file
View File

@@ -0,0 +1,48 @@
from pyasic.miners._backends import BMMiner
import ipaddress
class Hiveon(BMMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ipaddress.ip_address(ip)
self.api_type = "Hiveon"
self.uname = "root"
self.pwd = "admin"
async def get_board_info(self) -> dict:
"""Gets data on each board and chain in the miner."""
board_stats = await self.api.stats()
stats = board_stats["STATS"][1]
boards = {}
board_chains = {0: [2, 9, 10], 1: [3, 11, 12], 2: [4, 13, 14]}
for idx, board in enumerate(board_chains):
boards[board] = []
for chain in board_chains[board]:
count = stats[f"chain_acn{chain}"]
chips = stats[f"chain_acs{chain}"].replace(" ", "")
if not count == 18 or "x" in chips:
nominal = False
else:
nominal = True
boards[board].append(
{
"chain": chain,
"chip_count": count,
"chip_status": chips,
"nominal": nominal,
}
)
return boards
async def get_bad_boards(self) -> dict:
"""Checks for and provides list of non working boards."""
boards = await self.get_board_info()
bad_boards = {}
for board in boards.keys():
for chain in boards[board]:
if not chain["chip_count"] == 18 or "x" in chain["chip_status"]:
if board not in bad_boards.keys():
bad_boards[board] = []
bad_boards[board].append(chain)
return bad_boards

3
pyasic/miners/_types/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .antminer import *
from .avalonminer import *
from .whatsminer import *

10
pyasic/miners/_types/antminer/X17/S17.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S17(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S17"
self.nominal_chips = 48
self.fan_count = 4

10
pyasic/miners/_types/antminer/X17/S17_Plus.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S17Plus(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S17+"
self.nominal_chips = 65
self.fan_count = 4

10
pyasic/miners/_types/antminer/X17/S17_Pro.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S17Pro(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S17 Pro"
self.nominal_chips = 48
self.fan_count = 4

10
pyasic/miners/_types/antminer/X17/S17e.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S17e(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S17e"
self.nominal_chips = 135
self.fan_count = 4

10
pyasic/miners/_types/antminer/X17/T17.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class T17(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "T17"
self.nominal_chips = 30
self.fan_count = 4

10
pyasic/miners/_types/antminer/X17/T17_Plus.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class T17Plus(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "T17+"
self.nominal_chips = 44
self.fan_count = 4

10
pyasic/miners/_types/antminer/X17/T17e.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class T17e(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "T17e"
self.nominal_chips = 78
self.fan_count = 4

8
pyasic/miners/_types/antminer/X17/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
from .S17 import S17
from .S17_Plus import S17Plus
from .S17_Pro import S17Pro
from .S17e import S17e
from .T17 import T17
from .T17_Plus import T17Plus
from .T17e import T17e

10
pyasic/miners/_types/antminer/X19/S19.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S19(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S19"
self.nominal_chips = 76
self.fan_count = 4

10
pyasic/miners/_types/antminer/X19/S19_Pro.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S19Pro(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S19 Pro"
self.nominal_chips = 114
self.fan_count = 4

10
pyasic/miners/_types/antminer/X19/S19a.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S19a(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S19a"
self.nominal_chips = 72
self.fan_count = 4

10
pyasic/miners/_types/antminer/X19/S19j.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S19j(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S19j"
self.nominal_chips = 114
self.fan_count = 4

10
pyasic/miners/_types/antminer/X19/S19j_Pro.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S19jPro(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S19j Pro"
self.nominal_chips = 126
self.fan_count = 4

10
pyasic/miners/_types/antminer/X19/T19.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class T19(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "T19"
self.nominal_chips = 76
self.fan_count = 4

9
pyasic/miners/_types/antminer/X19/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
from .S19 import S19
from .S19_Pro import S19Pro
from .S19j import S19j
from .S19j_Pro import S19jPro
from .S19a import S19a
from .T19 import T19

10
pyasic/miners/_types/antminer/X9/S9.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S9(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S9"
self.nominal_chips = 63
self.fan_count = 2

10
pyasic/miners/_types/antminer/X9/S9i.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class S9i(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "S9i"
self.nominal_chips = 63
self.fan_count = 2

10
pyasic/miners/_types/antminer/X9/T9.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class T9(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "T9"
self.nominal_chips = 57
self.fan_count = 2

3
pyasic/miners/_types/antminer/X9/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .S9 import S9
from .S9i import S9i
from .T9 import T9

3
pyasic/miners/_types/antminer/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .X9 import *
from .X17 import *
from .X19 import *

10
pyasic/miners/_types/avalonminer/A10X/A1026.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon1026(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 1026"
self.nominal_chips = 80
self.fan_count = 2

10
pyasic/miners/_types/avalonminer/A10X/A1047.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon1047(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 1047"
self.nominal_chips = 80
self.fan_count = 2

10
pyasic/miners/_types/avalonminer/A10X/A1066.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon1066(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 1066"
self.nominal_chips = 114
self.fan_count = 4

3
pyasic/miners/_types/avalonminer/A10X/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .A1026 import Avalon1026
from .A1047 import Avalon1047
from .A1066 import Avalon1066

10
pyasic/miners/_types/avalonminer/A7X/A721.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon721(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 721"
self.chip_count = 18 # This miner has 4 boards totaling 72
self.fan_count = 1 # also only 1 fan

10
pyasic/miners/_types/avalonminer/A7X/A741.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon741(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 741"
self.chip_count = 22 # This miner has 4 boards totaling 88
self.fan_count = 1 # also only 1 fan

10
pyasic/miners/_types/avalonminer/A7X/A761.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon761(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 761"
self.chip_count = 18 # This miner has 4 boards totaling 72
self.fan_count = 1 # also only 1 fan

3
pyasic/miners/_types/avalonminer/A7X/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .A721 import Avalon721
from .A741 import Avalon741
from .A761 import Avalon761

10
pyasic/miners/_types/avalonminer/A8X/A821.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon821(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 821"
self.chip_count = 26 # This miner has 4 boards totaling 104
self.fan_count = 1 # also only 1 fan

10
pyasic/miners/_types/avalonminer/A8X/A841.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon841(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 841"
self.chip_count = 26 # This miner has 4 boards totaling 104
self.fan_count = 1 # also only 1 fan

10
pyasic/miners/_types/avalonminer/A8X/A851.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon851(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 851"
self.chip_count = 26 # This miner has 4 boards totaling 104
self.fan_count = 1 # also only 1 fan

3
pyasic/miners/_types/avalonminer/A8X/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .A821 import Avalon821
from .A841 import Avalon841
from .A851 import Avalon851

10
pyasic/miners/_types/avalonminer/A9X/A921.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class Avalon921(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "Avalon 921"
self.chip_count = 26 # This miner has 4 boards totaling 104
self.fan_count = 1 # also only 1 fan

1
pyasic/miners/_types/avalonminer/A9X/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .A921 import Avalon921

4
pyasic/miners/_types/avalonminer/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
from .A7X import *
from .A8X import *
from .A9X import *
from .A10X import *

10
pyasic/miners/_types/whatsminer/M2X/M20S.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class M20S(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M20S"
self.nominal_chips = 66
self.fan_count = 2

10
pyasic/miners/_types/whatsminer/M2X/M20S_Plus.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class M20SPlus(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M20S+"
self.nominal_chips = 66
self.fan_count = 2

10
pyasic/miners/_types/whatsminer/M2X/M21.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class M21(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M21"
self.nominal_chips = 105
self.fan_count = 2

28
pyasic/miners/_types/whatsminer/M2X/M21S.py Normal file
View File

@@ -0,0 +1,28 @@
from pyasic.miners import BaseMiner
class M21S(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M21S"
self.nominal_chips = 66
self.fan_count = 2
class M21SV60(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M21S V60"
self.nominal_chips = 105
self.fan_count = 2
class M21SV20(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M21S V20"
self.nominal_chips = 66
self.fan_count = 2

10
pyasic/miners/_types/whatsminer/M2X/M21S_Plus.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class M21SPlus(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M21S+"
self.nominal_chips = 105
self.fan_count = 2

6
pyasic/miners/_types/whatsminer/M2X/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
from .M20S import M20S
from .M20S_Plus import M20SPlus
from .M21 import M21
from .M21S import M21S, M21SV20, M21SV60
from .M21S_Plus import M21SPlus

46
pyasic/miners/_types/whatsminer/M3X/M30S.py Normal file
View File

@@ -0,0 +1,46 @@
from pyasic.miners import BaseMiner
class M30S(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S"
self.nominal_chips = 148
self.fan_count = 2
class M30SV50(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S V50"
self.nominal_chips = 156
self.fan_count = 2
class M30SVG20(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S VG20"
self.nominal_chips = 70
self.fan_count = 2
class M30SVE20(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S VE20"
self.nominal_chips = 111
self.fan_count = 2
class M30SVE10(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S VE10"
self.nominal_chips = 105
self.fan_count = 2

37
pyasic/miners/_types/whatsminer/M3X/M30S_Plus.py Normal file
View File

@@ -0,0 +1,37 @@
from pyasic.miners import BaseMiner
class M30SPlus(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S+"
self.nominal_chips = 156
self.fan_count = 2
class M30SPlusVG60(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S+ VG60"
self.nominal_chips = 86
self.fan_count = 2
class M30SPlusVE40(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S+ VE40"
self.nominal_chips = 156
self.fan_count = 2
class M30SPlusVF20(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S+ VF20"
self.nominal_chips = 111
self.fan_count = 2

28
pyasic/miners/_types/whatsminer/M3X/M30S_Plus_Plus.py Normal file
View File

@@ -0,0 +1,28 @@
from pyasic.miners import BaseMiner
class M30SPlusPlus(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S++"
self.nominal_chips = 111
self.fan_count = 2
class M30SPlusPlusVG30(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S++ V30"
self.nominal_chips = 111
self.fan_count = 2
class M30SPlusPlusVG40(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M30S++ V40"
self.nominal_chips = 117
self.fan_count = 2

10
pyasic/miners/_types/whatsminer/M3X/M31S.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class M31S(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M31S"
# TODO: Add chip count for this miner (per board) - self.nominal_chips
self.fan_count = 2

19
pyasic/miners/_types/whatsminer/M3X/M31S_Plus.py Normal file
View File

@@ -0,0 +1,19 @@
from pyasic.miners import BaseMiner
class M31SPlus(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M31S+"
self.nominal_chips = 78
self.fan_count = 2
class M31SPlusVE20(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M31S+ VE20"
self.nominal_chips = 78
self.fan_count = 2

10
pyasic/miners/_types/whatsminer/M3X/M32S.py Normal file
View File

@@ -0,0 +1,10 @@
from pyasic.miners import BaseMiner
class M32S(BaseMiner):
def __init__(self, ip: str):
super().__init__()
self.ip = ip
self.model = "M32S"
self.nominal_chips = 78
self.fan_count = 2

8
pyasic/miners/_types/whatsminer/M3X/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
from .M30S import M30S, M30SVE10, M30SVE20, M30SVG20, M30SV50
from .M30S_Plus import M30SPlus, M30SPlusVG60, M30SPlusVE40, M30SPlusVF20
from .M30S_Plus_Plus import M30SPlusPlus, M30SPlusPlusVG30, M30SPlusPlusVG40
from .M31S import M31S
from .M31S_Plus import M31SPlus, M31SPlusVE20
from .M32S import M32S

2
pyasic/miners/_types/whatsminer/__init__.py Normal file
View File

@@ -0,0 +1,2 @@
from .M2X import *
from .M3X import *

4
pyasic/miners/antminer/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
from .bmminer import *
from .bosminer import *
from .cgminer import *
from .hiveon import *

8
pyasic/miners/antminer/bmminer/X17/S17.py Normal file
View File

@@ -0,0 +1,8 @@
from .X17 import BMMinerX17
from pyasic.miners._types import S17 # noqa - Ignore access to _module
class BMMinerS17(BMMinerX17, S17):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X17/S17_Plus.py Normal file
View File

@@ -0,0 +1,8 @@
from .X17 import BMMinerX17
from pyasic.miners._types import S17Plus # noqa - Ignore access to _module
class BMMinerS17Plus(BMMinerX17, S17Plus):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X17/S17_Pro.py Normal file
View File

@@ -0,0 +1,8 @@
from .X17 import BMMinerX17
from pyasic.miners._types import S17Pro # noqa - Ignore access to _module
class BMMinerS17Pro(BMMinerX17, S17Pro):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X17/S17e.py Normal file
View File

@@ -0,0 +1,8 @@
from .X17 import BMMinerX17
from pyasic.miners._types import S17e # noqa - Ignore access to _module
class BMMinerS17e(BMMinerX17, S17e):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X17/T17.py Normal file
View File

@@ -0,0 +1,8 @@
from .X17 import BMMinerX17
from pyasic.miners._types import T17 # noqa - Ignore access to _module
class BMMinerT17(BMMinerX17, T17):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X17/T17_Plus.py Normal file
View File

@@ -0,0 +1,8 @@
from .X17 import BMMinerX17
from pyasic.miners._types import T17Plus # noqa - Ignore access to _module
class BMMinerT17Plus(BMMinerX17, T17Plus):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X17/T17e.py Normal file
View File

@@ -0,0 +1,8 @@
from .X17 import BMMinerX17
from pyasic.miners._types import T17e # noqa - Ignore access to _module
class BMMinerT17e(BMMinerX17, T17e):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

83
pyasic/miners/antminer/bmminer/X17/X17.py Normal file
View File

@@ -0,0 +1,83 @@
from pyasic.miners._backends import BMMiner # noqa - Ignore access to _module
import httpx
class BMMinerX17(BMMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip
async def get_hostname(self) -> str or None:
hostname = None
url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.get(url, auth=auth)
if data.status_code == 200:
data = data.json()
if len(data.keys()) > 0:
if "hostname" in data.keys():
hostname = data["hostname"]
return hostname
async def get_mac(self):
mac = None
url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.get(url, auth=auth)
if data.status_code == 200:
data = data.json()
if len(data.keys()) > 0:
if "macaddr" in data.keys():
mac = data["macaddr"]
return mac
async def fault_light_on(self) -> bool:
url = f"http://{self.ip}/cgi-bin/blink.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
try:
await client.post(url, data={"action": "startBlink"}, auth=auth)
except httpx.ReadTimeout:
# Expected behaviour
pass
data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
if data.status_code == 200:
data = data.json()
if data["isBlinking"]:
return True
return False
async def fault_light_off(self) -> bool:
url = f"http://{self.ip}/cgi-bin/blink.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
await client.post(url, data={"action": "stopBlink"}, auth=auth)
data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
if data.status_code == 200:
data = data.json()
if not data["isBlinking"]:
return True
return False
async def check_light(self):
url = f"http://{self.ip}/cgi-bin/blink.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
if data.status_code == 200:
data = data.json()
if data["isBlinking"]:
return True
return False
async def reboot(self) -> bool:
url = f"http://{self.ip}/cgi-bin/reboot.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.get(url, auth=auth)
if data.status_code == 200:
return True
return False

8
pyasic/miners/antminer/bmminer/X17/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
from .S17 import BMMinerS17
from .S17_Plus import BMMinerS17Plus
from .S17_Pro import BMMinerS17Pro
from .S17e import BMMinerS17e
from .T17 import BMMinerT17
from .T17_Plus import BMMinerT17Plus
from .T17e import BMMinerT17e

8
pyasic/miners/antminer/bmminer/X19/S19.py Normal file
View File

@@ -0,0 +1,8 @@
from .X19 import BMMinerX19
from pyasic.miners._types import S19 # noqa - Ignore access to _module
class BMMinerS19(BMMinerX19, S19):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X19/S19_Pro.py Normal file
View File

@@ -0,0 +1,8 @@
from .X19 import BMMinerX19
from pyasic.miners._types import S19Pro # noqa - Ignore access to _module
class BMMinerS19Pro(BMMinerX19, S19Pro):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X19/S19a.py Normal file
View File

@@ -0,0 +1,8 @@
from .X19 import BMMinerX19
from pyasic.miners._types import S19a # noqa - Ignore access to _module
class BMMinerS19a(BMMinerX19, S19a):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X19/S19j.py Normal file
View File

@@ -0,0 +1,8 @@
from .X19 import BMMinerX19
from pyasic.miners._types import S19j # noqa - Ignore access to _module
class BMMinerS19j(BMMinerX19, S19j):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X19/S19j_Pro.py Normal file
View File

@@ -0,0 +1,8 @@
from .X19 import BMMinerX19
from pyasic.miners._types import S19jPro # noqa - Ignore access to _module
class BMMinerS19jPro(BMMinerX19, S19jPro):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X19/T19.py Normal file
View File

@@ -0,0 +1,8 @@
from .X19 import BMMinerX19
from pyasic.miners._types import T19 # noqa - Ignore access to _module
class BMMinerT19(BMMinerX19, T19):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

102
pyasic/miners/antminer/bmminer/X19/X19.py Normal file
View File

@@ -0,0 +1,102 @@
from pyasic.miners._backends import BMMiner # noqa - Ignore access to _module
from pyasic.config import MinerConfig
import httpx
import json
import asyncio
class BMMinerX19(BMMiner):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip
async def get_config(self) -> MinerConfig:
url = f"http://{self.ip}/cgi-bin/get_miner_conf.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.get(url, auth=auth)
if data.status_code == 200:
data = data.json()
self.config = MinerConfig().from_raw(data)
return self.config
async def send_config(self, yaml_config, ip_user: bool = False) -> None:
url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
auth = httpx.DigestAuth("root", "root")
if ip_user:
suffix = str(self.ip).split(".")[-1]
conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
else:
conf = MinerConfig().from_yaml(yaml_config).as_x19()
try:
async with httpx.AsyncClient() as client:
await client.post(url, data=conf, auth=auth)
except httpx.ReadTimeout:
pass
for i in range(7):
data = await self.get_config()
if data.as_x19() == conf:
break
await asyncio.sleep(1)
async def get_hostname(self) -> str or None:
hostname = None
url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.get(url, auth=auth)
if data.status_code == 200:
data = data.json()
if len(data.keys()) > 0:
if "hostname" in data.keys():
hostname = data["hostname"]
return hostname
async def get_mac(self):
mac = None
url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.get(url, auth=auth)
if data.status_code == 200:
data = data.json()
if len(data.keys()) > 0:
if "macaddr" in data.keys():
mac = data["macaddr"]
return mac
async def fault_light_on(self) -> bool:
url = f"http://{self.ip}/cgi-bin/blink.cgi"
auth = httpx.DigestAuth("root", "root")
data = json.dumps({"blink": "true"})
async with httpx.AsyncClient() as client:
data = await client.post(url, data=data, auth=auth)
if data.status_code == 200:
data = data.json()
if data.get("code") == "B000":
return True
return False
async def fault_light_off(self) -> bool:
url = f"http://{self.ip}/cgi-bin/blink.cgi"
auth = httpx.DigestAuth("root", "root")
data = json.dumps({"blink": "false"})
async with httpx.AsyncClient() as client:
data = await client.post(url, data=data, auth=auth)
if data.status_code == 200:
data = data.json()
if data.get("code") == "B100":
return True
return False
async def reboot(self) -> bool:
url = f"http://{self.ip}/cgi-bin/reboot.cgi"
auth = httpx.DigestAuth("root", "root")
async with httpx.AsyncClient() as client:
data = await client.get(url, auth=auth)
if data.status_code == 200:
return True
return False

9
pyasic/miners/antminer/bmminer/X19/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
from .S19 import BMMinerS19
from .S19_Pro import BMMinerS19Pro
from .S19j import BMMinerS19j
from .S19j_Pro import BMMinerS19jPro
from .S19a import BMMinerS19a
from .T19 import BMMinerT19

8
pyasic/miners/antminer/bmminer/X9/S9.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BMMiner # noqa - Ignore access to _module
from pyasic.miners._types import S9 # noqa - Ignore access to _module
class BMMinerS9(BMMiner, S9):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X9/S9i.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BMMiner # noqa - Ignore access to _module
from pyasic.miners._types import S9i # noqa - Ignore access to _module
class BMMinerS9i(BMMiner, S9i):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bmminer/X9/T9.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BMMiner # noqa - Ignore access to _module
from pyasic.miners._types import T9 # noqa - Ignore access to _module
class BMMinerT9(BMMiner, T9):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

3
pyasic/miners/antminer/bmminer/X9/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .S9 import BMMinerS9
from .S9i import BMMinerS9i
from .T9 import BMMinerT9

3
pyasic/miners/antminer/bmminer/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .X9 import *
from .X17 import *
from .X19 import *

8
pyasic/miners/antminer/bosminer/X17/S17.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import S17 # noqa - Ignore access to _module
class BOSMinerS17(BOSMiner, S17):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bosminer/X17/S17_Plus.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import S17Plus # noqa - Ignore access to _module
class BOSMinerS17Plus(BOSMiner, S17Plus):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bosminer/X17/S17_Pro.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import S17Pro # noqa - Ignore access to _module
class BOSMinerS17Pro(BOSMiner, S17Pro):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bosminer/X17/S17e.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import S17e # noqa - Ignore access to _module
class BOSMinerS17e(BOSMiner, S17e):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bosminer/X17/T17.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import T17 # noqa - Ignore access to _module
class BOSMinerT17(BOSMiner, T17):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bosminer/X17/T17_Plus.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import T17Plus # noqa - Ignore access to _module
class BOSMinerT17Plus(BOSMiner, T17Plus):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bosminer/X17/T17e.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import T17e # noqa - Ignore access to _module
class BOSMinerT17e(BOSMiner, T17e):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

8
pyasic/miners/antminer/bosminer/X17/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
from .S17 import BOSMinerS17
from .S17_Plus import BOSMinerS17Plus
from .S17_Pro import BOSMinerS17Pro
from .S17e import BOSMinerS17e
from .T17 import BOSMinerT17
from .T17_Plus import BOSMinerT17Plus
from .T17e import BOSMinerT17e

8
pyasic/miners/antminer/bosminer/X19/S19.py Normal file
View File

@@ -0,0 +1,8 @@
from pyasic.miners._backends import BOSMiner # noqa - Ignore access to _module
from pyasic.miners._types import S19 # noqa - Ignore access to _module
class BOSMinerS19(BOSMiner, S19):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.ip = ip

Some files were not shown because too many files have changed in this diff Show More