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

add docs for the rest of the APIs

Browse Source
This commit is contained in:
UpstreamData
2022-07-13 10:11:05 -06:00
parent 0324a21e79
commit d7e9498018
6 changed files with 744 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
docs/api.md
View File

@@ -12,3 +12,31 @@ All API implementations inherit from `BaseMinerAPI`, which implements the basic
options:
show_root_heading: false
heading_level: 4
## BOSMinerAPI
::: pyasic.API.bosminer.BOSMinerAPI
handler: python
options:
show_root_heading: false
heading_level: 4
## BTMinerAPI
::: pyasic.API.btminer.BTMinerAPI
handler: python
options:
show_root_heading: false
heading_level: 4
## CGMinerAPI
::: pyasic.API.cgminer.CGMinerAPI
handler: python
options:
show_root_heading: false
heading_level: 4
## UnknownAPI
::: pyasic.API.unknown.UnknownAPI
handler: python
options:
show_root_heading: false
heading_level: 4

1
mkdocs.yml
View File

@@ -6,3 +6,4 @@ nav:
plugins:
- mkdocstrings
- search

2
pyasic/API/bmminer.py
View File

@@ -182,7 +182,7 @@ class BMMinerAPI(BaseMinerAPI):
<summary>Expand</summary>
Parameters:
n: Pools in order of priority.
*n: Pools in order of priority.
Returns:
A confirmation of setting pool priority.

191
pyasic/API/bosminer.py
View File

@@ -6,60 +6,80 @@ class BOSMinerAPI(BaseMinerAPI):
Each method corresponds to an API command in BOSMiner.
BOSMiner API documentation:
https://docs.braiins.com/os/plus-en/Development/1_api.html
[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()
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.
Parameters:
ip: The IP of the miner to reference the API on.
port: The port to reference the API on. Default is 4028.
"""
def __init__(self, ip, port=4028):
def __init__(self, ip: str, port: int = 4028):
super().__init__(ip, port)
async def asccount(self) -> dict:
"""Get data on the number of ASC devices and their info.
<details>
<summary>Expand</summary>
:return: Data on all ASC devices.
Returns:
Data on all ASC devices.
</details>
"""
return await self.send_command("asccount")
async def asc(self, n: int) -> dict:
"""Get data for ASC device n.
<details>
<summary>Expand</summary>
:param n: The device to get data for.
Parameters:
n: The device to get data for.
:return: The data for ASC device n.
Returns:
The data for ASC device n.
</details>
"""
return await self.send_command("asc", parameters=n)
async def devdetails(self) -> dict:
"""Get data on all devices with their static details.
<details>
<summary>Expand</summary>
:return: Data on all devices with their static details.
Returns:
Data on all devices with their static details.
</details>
"""
return await self.send_command("devdetails")
async def devs(self) -> dict:
"""Get data on each PGA/ASC with their details.
<details>
<summary>Expand</summary>
:return: Data on each PGA/ASC with their details.
Returns:
Data on each PGA/ASC with their details.
</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.
"""Get data on each PGA/ASC with their details, ignoring blacklisted and zombie devices.
<details>
<summary>Expand</summary>
:param old: Include zombie devices that became zombies less
than 'old' seconds ago
Parameters:
old: Include zombie devices that became zombies less than 'old' seconds ago
:return: Data on each PGA/ASC with their details.
Returns:
Data on each PGA/ASC with their details.
</details>
"""
if old:
return await self.send_command("edevs", parameters="old")
@@ -69,40 +89,62 @@ class BOSMinerAPI(BaseMinerAPI):
async def pools(self) -> dict:
"""Get pool information.
:return: Miner pool information.
<details>
<summary>Expand</summary>
Returns:
Miner pool information.
</details>
"""
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.
<details>
<summary>Expand</summary>
Returns:
The status summary of the miner.
</details>
"""
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.
<details>
<summary>Expand</summary>
Returns:
Stats of each device/pool with more than 1 getwork.
</details>
"""
return await self.send_command("stats")
async def version(self) -> dict:
"""Get miner version info.
:return: Miner version information.
<details>
<summary>Expand</summary>
Returns:
Miner version information.
</details>
"""
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.
"""Get stats of each device/pool with more than 1 getwork, ignoring zombie devices.
<details>
<summary>Expand</summary>
:param old: Include zombie devices that became zombies less
than 'old' seconds ago.
Parameters:
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.
Returns:
Stats of each device/pool with more than 1 getwork, ignoring zombie devices.
</details>
"""
if old:
return await self.send_command("estats", parameters=old)
@@ -111,98 +153,109 @@ class BOSMinerAPI(BaseMinerAPI):
async def check(self, command: str) -> dict:
"""Check if the command command exists in BOSMiner.
<details>
<summary>Expand</summary>
:param command: The command to check.
Parameters:
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
Returns:
## Information about a command:
* Exists (Y/N) <- the command exists in this version
* Access (Y/N) <- you have access to use the command
</details>
"""
return await self.send_command("check", parameters=command)
async def coin(self) -> dict:
"""Get information on the current coin.
<details>
<summary>Expand</summary>
: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
Returns:
## 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
</details>
"""
return await self.send_command("coin")
async def lcd(self) -> dict:
"""Get a general all-in-one status summary of the miner.
<details>
<summary>Expand</summary>
:return: An all-in-one status summary of the miner.
Returns:
An all-in-one status summary of the miner.
</details>
"""
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.
<details>
<summary>Expand</summary>
:return: Data on the fans of the miner.
Returns:
Data on the fans of the miner.
</details>
"""
return await self.send_command("fans")
async def tempctrl(self) -> dict:
"""Get temperature control data.
<details>
<summary>Expand</summary>
:return: Data about the temp control settings of the miner.
Returns:
Data about the temp control settings of the miner.
</details>
"""
return await self.send_command("tempctrl")
async def temps(self) -> dict:
"""Get temperature data.
<details>
<summary>Expand</summary>
:return: Data on the temps of the miner.
Returns:
Data on the temps of the miner.
</details>
"""
return await self.send_command("temps")
async def tunerstatus(self) -> dict:
"""Get tuner status data
<details>
<summary>Expand</summary>
:return: Data on the status of autotuning.
Returns:
Data on the status of autotuning.
</details>
"""
return await self.send_command("tunerstatus")
async def pause(self) -> dict:
"""Pause mining.
<details>
<summary>Expand</summary>
:return: Confirmation of pausing mining.
Returns:
Confirmation of pausing mining.
</details>
"""
return await self.send_command("pause")
async def resume(self) -> dict:
"""Resume mining.
<details>
<summary>Expand</summary>
:return: Confirmation of resuming mining.
Returns:
Confirmation of resuming mining.
</details>
"""
return await self.send_command("resume")

399
pyasic/API/btminer.py
View File

@@ -29,9 +29,12 @@ def _crypt(word: str, salt: str) -> str:
'\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.
Parameters:
word: The word to be encrypted.
salt: The salt to encrypt the word.
Returns:
An MD5 hash of the word with the salt.
"""
# compile a standard format for the salt
standard_salt = re.compile("\s*\$(\d+)\$([\w\./]*)\$")
@@ -50,11 +53,11 @@ def _crypt(word: str, salt: str) -> str:
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.
Parameters:
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.
Returns:
The input string as bytes with a multiple of 16 as the length.
"""
while len(string) % 16 != 0:
string += "\0"
@@ -67,10 +70,12 @@ def parse_btminer_priviledge_data(token_data: dict, data: dict):
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.
Parameters:
token_data: The token information from self.get_token().
data: The data to parse, returned from the API.
:return: A decoded dict version of the privileged command output.
Returns:
A decoded dict version of the privileged command output.
"""
# get the encoded data from the dict
enc_data = data["enc"]
@@ -97,10 +102,12 @@ def create_privileged_cmd(token_data: dict, command: dict) -> bytes:
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.
Parameters:
token_data: The token information from self.get_token().
command: The command to turn into a privileged command.
:return: The encrypted privileged command to be sent to the miner.
Returns:
The encrypted privileged command to be sent to the miner.
"""
# add token to command
command["token"] = token_data["host_sign"]
@@ -132,7 +139,7 @@ class BTMinerAPI(BaseMinerAPI):
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()
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.
@@ -148,12 +155,13 @@ class BTMinerAPI(BaseMinerAPI):
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.
Parameters:
ip: The IP of the miner to reference the API on.
port: The port to reference the API on. Default is 4028.
pwd: The admin password of the miner. Default is admin.
"""
def __init__(self, ip, port=4028, pwd: str = WHATSMINER_PWD):
def __init__(self, ip: str, port: int = 4028, pwd: str = WHATSMINER_PWD):
super().__init__(ip, port)
self.admin_pwd = pwd
self.current_token = None
@@ -165,19 +173,6 @@ class BTMinerAPI(BaseMinerAPI):
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):
@@ -232,11 +227,14 @@ class BTMinerAPI(BaseMinerAPI):
# return the parsed json as a dict
return data
async def get_token(self):
async def get_token(self) -> dict:
"""Gets token information from the API.
<details>
<summary>Expand</summary>
:return: An encoded token and md5 password, which are used
for the privileged API.
Returns:
An encoded token and md5 password, which are used for the privileged API.
</details>
"""
# get the token
data = await self.send_command("get_token")
@@ -278,22 +276,28 @@ class BTMinerAPI(BaseMinerAPI):
pool_3: str = None,
worker_3: str = None,
passwd_3: str = None,
):
) -> dict:
"""Update the pools of the miner using the API.
<details>
<summary>Expand</summary>
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.
Parameters:
pool_1: The URL to update pool 1 to.
worker_1: The worker name for pool 1 to update to.
passwd_1: The password for pool 1 to update to.
pool_2: The URL to update pool 2 to.
worker_2: The worker name for pool 2 to update to.
passwd_2: The password for pool 2 to update to.
pool_3: The URL to update pool 3 to.
worker_3: The worker name for pool 3 to update to.
passwd_3: The password for pool 3 to update to.
Returns:
A dict from the API to confirm the pools were updated.
</details>
"""
# get the token and password from the miner
token_data = await self.get_token()
@@ -336,27 +340,36 @@ class BTMinerAPI(BaseMinerAPI):
# send the command
return await self.send_command(enc_command)
async def restart(self):
async def restart(self) -> dict:
"""Restart BTMiner using the API.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the restart.
</details>
"""
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):
async def power_off(self, respbefore: bool = True) -> dict:
"""Power off the miner using the API.
<details>
<summary>Expand</summary>
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.
Parameters:
respbefore: Whether to respond before powering off.
Returns:
A reply informing of the status of powering off.
</details>
"""
if respbefore:
command = {"cmd": "power_off", "respbefore": "true"}
@@ -366,26 +379,36 @@ class BTMinerAPI(BaseMinerAPI):
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def power_on(self):
async def power_on(self) -> dict:
"""Power on the miner using the API.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the status of powering on.
</details>
"""
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):
async def reset_led(self) -> dict:
"""Reset the LED on the miner using the API.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the status of resetting the LED.
</details>
"""
command = {"cmd": "set_led", "param": "auto"}
token_data = await self.get_token()
@@ -398,17 +421,23 @@ class BTMinerAPI(BaseMinerAPI):
period: int = 2000,
duration: int = 1000,
start: int = 0,
):
) -> dict:
"""Set the LED on the miner using the API.
<details>
<summary>Expand</summary>
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.
Parameters:
color: The LED color to set, either 'red' or 'green'.
period: The flash cycle in ms.
duration: LED on time in the cycle in ms.
start: LED on time offset in the cycle in ms.
Returns:
A reply informing of the status of setting the LED.
</details>
"""
command = {
"cmd": "set_led",
@@ -421,14 +450,18 @@ class BTMinerAPI(BaseMinerAPI):
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def set_low_power(self):
async def set_low_power(self) -> dict:
"""Set low power mode on the miner using the API.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the status of setting low power mode.
</details>
"""
command = {"cmd": "set_low_power"}
token_data = await self.get_token()
@@ -436,46 +469,62 @@ class BTMinerAPI(BaseMinerAPI):
return await self.send_command(enc_command)
async def update_firmware(self): # noqa - static
"""Not implemented."""
# to be determined if this will be added later
# requires a file stream in bytes
return NotImplementedError
async def reboot(self):
async def reboot(self) -> dict:
"""Reboot the miner using the API.
<details>
<summary>Expand</summary>
:return: A reply informing of the status of the reboot.
Returns:
A reply informing of the status of the reboot.
</details>
"""
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):
async def factory_reset(self) -> dict:
"""Reset the miner to factory defaults.
<details>
<summary>Expand</summary>
:return: A reply informing of the status of the reset.
Returns:
A reply informing of the status of the reset.
</details>
"""
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):
async def update_pwd(self, old_pwd: str, new_pwd: str) -> dict:
"""Update the admin user's password.
<details>
<summary>Expand</summary>
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.
Parameters:
old_pwd: The old admin password.
new_pwd: The new password to set.
Returns:
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(
raise APIError(
f"New password too long, the max length is 8. "
f"Password size: {len(new_pwd.encode('utf-8'))}"
)
@@ -484,19 +533,25 @@ class BTMinerAPI(BaseMinerAPI):
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def set_target_freq(self, percent: int):
async def set_target_freq(self, percent: int) -> dict:
"""Update the target frequency.
<details>
<summary>Expand</summary>
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.
Parameters:
percent: The frequency % to set.
Returns:
A reply informing of the status of setting the frequency.
</details>
"""
if not -10 < percent < 100:
return APIError(
raise APIError(
f"Frequency % is outside of the allowed "
f"range. Please set a % between -10 and "
f"100"
@@ -506,88 +561,122 @@ class BTMinerAPI(BaseMinerAPI):
enc_command = create_privileged_cmd(token_data, command)
return await self.send_command(enc_command)
async def enable_fast_boot(self):
async def enable_fast_boot(self) -> dict:
"""Turn on fast boot.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the status of enabling fast boot.
</details>
"""
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):
async def disable_fast_boot(self) -> dict:
"""Turn off fast boot.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the status of disabling fast boot.
</details>
"""
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):
async def enable_web_pools(self) -> dict:
"""Turn on web pool updates.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the status of enabling web pools.
</details>
"""
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):
async def disable_web_pools(self) -> dict:
"""Turn off web pool updates.
<details>
<summary>Expand</summary>
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.
Returns:
A reply informing of the status of disabling web pools.
</details>
"""
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):
async def set_hostname(self, hostname: str) -> dict:
"""Set the hostname of the miner.
<details>
<summary>Expand</summary>
Set the hostname of the miner, only works after changing the
password of the miner using the Whatsminer tool.
Parameters:
hostname: The new hostname to use.
Returns:
:param hostname: The new hostname to use.
:return: A reply informing of the status of setting the
hostname.
A reply informing of the status of setting the hostname.
</details>
"""
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):
async def set_power_pct(self, percent: int) -> dict:
"""Set the power percentage of the miner.
<details>
<summary>Expand</summary>
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.
Parameters:
percent: The power percentage to set.
Returns:
A reply informing of the status of setting the power percentage.
</details>
"""
if not 0 < percent < 100:
return APIError(
raise APIError(
f"Power PCT % is outside of the allowed "
f"range. Please set a % between 0 and "
f"100"
@@ -597,22 +686,29 @@ class BTMinerAPI(BaseMinerAPI):
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):
async def pre_power_on(self, complete: bool, msg: str) -> dict:
"""Configure or check status of pre power on.
<details>
<summary>Expand</summary>
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.
Parameters:
complete: check whether pre power on is complete.
msg: ## the message to check.
* `wait for adjust temp`
* `adjust complete`
* `adjust continue`
Returns:
A reply informing of the status of pre power on.
</details>
"""
if not msg == "wait for adjust temp" or "adjust complete" or "adjust continue":
return APIError(
raise APIError(
"Message is incorrect, please choose one of "
'["wait for adjust temp", '
'"adjust complete", '
@@ -629,77 +725,126 @@ class BTMinerAPI(BaseMinerAPI):
#### END privileged COMMANDS ####
async def summary(self):
async def summary(self) -> dict:
"""Get the summary status from the miner.
<details>
<summary>Expand</summary>
:return: Summary status of the miner.
Returns:
Summary status of the miner.
</details>
"""
return await self.send_command("summary")
async def pools(self):
async def pools(self) -> dict:
"""Get the pool status from the miner.
<details>
<summary>Expand</summary>
:return: Pool status of the miner.
Returns:
Pool status of the miner.
</details>
"""
return await self.send_command("pools")
async def devs(self):
async def devs(self) -> dict:
"""Get data on each PGA/ASC with their details.
<details>
<summary>Expand</summary>
:return: Data on each PGA/ASC with their details.
Returns:
Data on each PGA/ASC with their details.
</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.
async def edevs(self) -> dict:
"""Get data on each PGA/ASC with their details, ignoring blacklisted and zombie devices.
<details>
<summary>Expand</summary>
:return: Data on each PGA/ASC with their details.
Returns:
Data on each PGA/ASC with their details.
</details>
"""
return await self.send_command("edevs")
async def devdetails(self):
async def devdetails(self) -> dict:
"""Get data on all devices with their static details.
<details>
<summary>Expand</summary>
:return: Data on all devices with their static details.
Returns:
Data on all devices with their static details.
</details>
"""
return await self.send_command("devdetails")
async def get_psu(self):
async def get_psu(self) -> dict:
"""Get data on the PSU and power information.
<details>
<summary>Expand</summary>
:return: Data on the PSU and power information.
Returns:
Data on the PSU and power information.
</details>
"""
return await self.send_command("get_psu")
async def version(self):
"""Get version data for the miner.
async def version(self) -> dict:
"""Get version data for the miner. Wraps `self.get_version()`.
<details>
<summary>Expand</summary>
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.
Returns:
Version data for the miner.
</details>
"""
return await self.get_version()
async def get_version(self):
async def get_version(self) -> dict:
"""Get version data for the miner.
<details>
<summary>Expand</summary>
:return: Version data for the miner.
Returns:
Version data for the miner.
</details>
"""
return await self.send_command("get_version")
async def status(self):
async def status(self) -> dict:
"""Get BTMiner status and firmware version.
<details>
<summary>Expand</summary>
:return: BTMiner status and firmware version.
Returns:
BTMiner status and firmware version.
</details>
"""
return await self.send_command("status")
async def get_miner_info(self):
async def get_miner_info(self) -> dict:
"""Get general miner info.
<details>
<summary>Expand</summary>
:return: General miner info.
Returns:
General miner info.
</details>
"""
return await self.send_command("get_miner_info")

445
pyasic/API/cgminer.py
View File

@@ -6,8 +6,7 @@ class CGMinerAPI(BaseMinerAPI):
Each method corresponds to an API command in GGMiner.
CGMiner API documentation:
https://github.com/ckolivas/cgminer/blob/master/API-README
[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()
@@ -15,111 +14,163 @@ class CGMinerAPI(BaseMinerAPI):
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.
Parameters:
ip: The IP of the miner to reference the API on.
port: The port to reference the API on. Default is 4028.
"""
def __init__(self, ip, port=4028):
def __init__(self, ip: str, port: int = 4028):
super().__init__(ip, port)
async def version(self) -> dict:
"""Get miner version info.
<details>
<summary>Expand</summary>
:return: Miner version information.
Returns:
Miner version information.
</details>
"""
return await self.send_command("version")
async def config(self) -> dict:
"""Get some basic configuration info.
<details>
<summary>Expand</summary>
: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
Returns:
## 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
</details>
"""
return await self.send_command("config")
async def summary(self) -> dict:
"""Get the status summary of the miner.
<details>
<summary>Expand</summary>
:return: The status summary of the miner.
Returns:
The status summary of the miner.
</details>
"""
return await self.send_command("summary")
async def pools(self) -> dict:
"""Get pool information.
<details>
<summary>Expand</summary>
:return: Miner pool information.
Returns:
Miner pool information.
</details>
"""
return await self.send_command("pools")
async def devs(self) -> dict:
"""Get data on each PGA/ASC with their details.
<details>
<summary>Expand</summary>
:return: Data on each PGA/ASC with their details.
Returns:
Data on each PGA/ASC with their details.
</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.
"""Get data on each PGA/ASC with their details, ignoring blacklisted and zombie devices.
<details>
<summary>Expand</summary>
:param old: Include zombie devices that became zombies less
than 'old' seconds ago
Parameters:
old: Include zombie devices that became zombies less than 'old' seconds ago
:return: Data on each PGA/ASC with their details.
Returns:
Data on each PGA/ASC with their details.
</details>
"""
if old:
return await self.send_command("edevs", parameters="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.
<details>
<summary>Expand</summary>
:param n: The PGA number to get data from.
Parameters:
n: The PGA number to get data from.
:return: Data on the PGA n.
Returns:
Data on the PGA n.
</details>
"""
return await self.send_command("pga", parameters=n)
async def pgacount(self) -> dict:
"""Get data fon all PGAs.
<details>
<summary>Expand</summary>
:return: Data on the PGAs connected.
Returns:
Data on the PGAs connected.
</details>
"""
return await self.send_command("pgacount")
async def switchpool(self, n: int) -> dict:
"""Switch pools to pool n.
<details>
<summary>Expand</summary>
:param n: The pool to switch to.
Parameters:
n: The pool to switch to.
:return: A confirmation of switching to pool n.
Returns:
A confirmation of switching to pool n.
</details>
"""
return await self.send_command("switchpool", parameters=n)
async def enablepool(self, n: int) -> dict:
"""Enable pool n.
<details>
<summary>Expand</summary>
:param n: The pool to enable.
Parameters:
n: The pool to enable.
:return: A confirmation of enabling pool n.
Returns:
A confirmation of enabling pool n.
</details>
"""
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.
<details>
<summary>Expand</summary>
: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.
Parameters:
url: The URL of the new pool to add.
username: The users username on the new pool.
password: The worker password on the new pool.
:return: A confirmation of adding the pool.
Returns:
A confirmation of adding the pool.
</details>
"""
return await self.send_command(
"addpool", parameters=f"{url},{username},{password}"
@@ -127,48 +178,73 @@ class CGMinerAPI(BaseMinerAPI):
async def poolpriority(self, *n: int) -> dict:
"""Set pool priority.
<details>
<summary>Expand</summary>
:param n: Pools in order of priority.
Parameters:
*n: Pools in order of priority.
:return: A confirmation of setting pool priority.
Returns:
A confirmation of setting pool priority.
</details>
"""
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.
<details>
<summary>Expand</summary>
:param n: Pool number to set quota on.
:param q: Quota to set the pool to.
Parameters:
n: Pool number to set quota on.
q: Quota to set the pool to.
:return: A confirmation of setting pool quota.
Returns:
A confirmation of setting pool quota.
</details>
"""
return await self.send_command("poolquota", parameters=f"{n},{q}")
async def disablepool(self, n: int) -> dict:
"""Disable a pool.
<details>
<summary>Expand</summary>
:param n: Pool to disable.
Parameters:
n: Pool to disable.
:return: A confirmation of diabling the pool.
Returns:
A confirmation of diabling the pool.
</details>
"""
return await self.send_command("disablepool", parameters=n)
async def removepool(self, n: int) -> dict:
"""Remove a pool.
<details>
<summary>Expand</summary>
:param n: Pool to remove.
Parameters:
n: Pool to remove.
:return: A confirmation of removing the pool.
Returns:
A confirmation of removing the pool.
</details>
"""
return await self.send_command("removepool", parameters=n)
async def save(self, filename: str = None) -> dict:
"""Save the config.
<details>
<summary>Expand</summary>
:param filename: Filename to save the config as.
Parameters:
filename: Filename to save the config as.
:return: A confirmation of saving the config.
Returns:
A confirmation of saving the config.
</details>
"""
if filename:
return await self.send_command("save", parameters=filename)
@@ -177,83 +253,123 @@ class CGMinerAPI(BaseMinerAPI):
async def quit(self) -> dict:
"""Quit CGMiner.
<details>
<summary>Expand</summary>
:return: A single "BYE" before CGMiner quits.
Returns:
A single "BYE" before CGMiner quits.
</details>
"""
return await self.send_command("quit")
async def notify(self) -> dict:
"""Notify the user of past errors.
<details>
<summary>Expand</summary>
:return: The last status and count of each devices problem(s).
Returns:
The last status and count of each devices problem(s).
</details>
"""
return await self.send_command("notify")
async def privileged(self) -> dict:
"""Check if you have privileged access.
<details>
<summary>Expand</summary>
:return: The STATUS section with an error if you have no
privileged access, or success if you have privileged access.
Returns:
The STATUS section with an error if you have no privileged access, or success if you have privileged access.
</details>
"""
return await self.send_command("privileged")
async def pgaenable(self, n: int) -> dict:
"""Enable PGA n.
<details>
<summary>Expand</summary>
:param n: The PGA to enable.
Parameters:
n: The PGA to enable.
:return: A confirmation of enabling PGA n.
Returns:
A confirmation of enabling PGA n.
</details>
"""
return await self.send_command("pgaenable", parameters=n)
async def pgadisable(self, n: int) -> dict:
"""Disable PGA n.
<details>
<summary>Expand</summary>
:param n: The PGA to disable.
Parameters:
n: The PGA to disable.
:return: A confirmation of disabling PGA n.
Returns:
A confirmation of disabling PGA n.
</details>
"""
return await self.send_command("pgadisable", parameters=n)
async def pgaidentify(self, n: int) -> dict:
"""Identify PGA n.
<details>
<summary>Expand</summary>
:param n: The PGA to identify.
Parameters:
n: The PGA to identify.
:return: A confirmation of identifying PGA n.
Returns:
A confirmation of identifying PGA n.
</details>
"""
return await self.send_command("pgaidentify", parameters=n)
async def devdetails(self) -> dict:
"""Get data on all devices with their static details.
<details>
<summary>Expand</summary>
:return: Data on all devices with their static details.
Returns:
Data on all devices with their static details.
</details>
"""
return await self.send_command("devdetails")
async def restart(self) -> dict:
"""Restart CGMiner using the API.
<details>
<summary>Expand</summary>
:return: A reply informing of the restart.
Returns:
A reply informing of the restart.
</details>
"""
return await self.send_command("restart")
async def stats(self) -> dict:
"""Get stats of each device/pool with more than 1 getwork.
<details>
<summary>Expand</summary>
:return: Stats of each device/pool with more than 1 getwork.
Returns:
Stats of each device/pool with more than 1 getwork.
</details>
"""
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.
"""Get stats of each device/pool with more than 1 getwork, ignoring zombie devices.
<details>
<summary>Expand</summary>
:param old: Include zombie devices that became zombies less
than 'old' seconds ago.
Parameters:
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.
Returns:
Stats of each device/pool with more than 1 getwork, ignoring zombie devices.
</details>
"""
if old:
return await self.send_command("estats", parameters=old)
@@ -262,92 +378,126 @@ class CGMinerAPI(BaseMinerAPI):
async def check(self, command: str) -> dict:
"""Check if the command command exists in CGMiner.
<details>
<summary>Expand</summary>
:param command: The command to check.
Parameters:
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
Returns:
## Information about a command:
* Exists (Y/N) <- the command exists in this version
* Access (Y/N) <- you have access to use the command
</details>
"""
return await self.send_command("check", parameters=command)
async def failover_only(self, failover: bool) -> dict:
"""Set failover-only.
<details>
<summary>Expand</summary>
Parameters:
failover: What to set failover-only to.
:param failover: What to set failover-only to.
:return: Confirmation of setting failover-only.
Returns:
Confirmation of setting failover-only.
</details>
"""
return await self.send_command("failover-only", parameters=failover)
async def coin(self) -> dict:
"""Get information on the current coin.
<details>
<summary>Expand</summary>
: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
Returns:
## 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
</details>
"""
return await self.send_command("coin")
async def debug(self, setting: str) -> dict:
"""Set a debug setting.
<details>
<summary>Expand</summary>
:param setting: Which setting to switch to. Options are:
Silent,
Quiet,
Verbose,
Debug,
RPCProto,
PerDevice,
WorkTime,
Normal.
Parameters:
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.
Returns:
Data on which debug setting was enabled or disabled.
</details>
"""
return await self.send_command("debug", parameters=setting)
async def setconfig(self, name: str, n: int) -> dict:
"""Set config of name to value n.
<details>
<summary>Expand</summary>
: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.
Parameters:
name: The name of the config setting to set.
## Options are:
* queue
* scantime
* expiry
n: The value to set the 'name' setting to.
:return: The results of setting config of name to n.
Returns:
The results of setting config of name to n.
</details>
"""
return await self.send_command("setconfig", parameters=f"{name},{n}")
async def usbstats(self) -> dict:
"""Get stats of all USB devices except ztex.
<details>
<summary>Expand</summary>
:return: The stats of all USB devices except ztex.
Returns:
The stats of all USB devices except ztex.
</details>
"""
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.
<details>
<summary>Expand</summary>
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.
Parameters:
n: The PGA to set the options on.
opt: The option to set. Setting this to 'help' returns a help message.
val: The value to set the option to.
:return: Confirmation of setting PGA n with opt[,val].
Returns:
Confirmation of setting PGA n with opt[,val].
</details>
"""
if val:
return await self.send_command("pgaset", parameters=f"{n},{opt},{val}")
@@ -356,75 +506,108 @@ class CGMinerAPI(BaseMinerAPI):
async def zero(self, which: str, summary: bool) -> dict:
"""Zero a device.
<details>
<summary>Expand</summary>
: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.
Parameters:
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.
summary: Whether or not to show a full summary.
:return: the STATUS section with info on the zero and optional
summary.
Returns:
the STATUS section with info on the zero and optional summary.
</details>
"""
return await self.send_command("zero", parameters=f"{which},{summary}")
async def hotplug(self, n: int) -> dict:
"""Enable hotplug.
<details>
<summary>Expand</summary>
:param n: The device number to set hotplug on.
Parameters:
n: The device number to set hotplug on.
:return: Information on hotplug status.
Returns:
Information on hotplug status.
</details>
"""
return await self.send_command("hotplug", parameters=n)
async def asc(self, n: int) -> dict:
"""Get data for ASC device n.
<details>
<summary>Expand</summary>
:param n: The device to get data for.
Parameters:
n: The device to get data for.
:return: The data for ASC device n.
Returns:
The data for ASC device n.
</details>
"""
return await self.send_command("asc", parameters=n)
async def ascenable(self, n: int) -> dict:
"""Enable ASC device n.
<details>
<summary>Expand</summary>
:param n: The device to enable.
Parameters:
n: The device to enable.
:return: Confirmation of enabling ASC device n.
Returns:
Confirmation of enabling ASC device n.
</details>
"""
return await self.send_command("ascenable", parameters=n)
async def ascdisable(self, n: int) -> dict:
"""Disable ASC device n.
<details>
<summary>Expand</summary>
:param n: The device to disable.
Parameters:
n: The device to disable.
:return: Confirmation of disabling ASC device n.
Returns:
Confirmation of disabling ASC device n.
</details>
"""
return await self.send_command("ascdisable", parameters=n)
async def ascidentify(self, n: int) -> dict:
"""Identify ASC device n.
<details>
<summary>Expand</summary>
:param n: The device to identify.
Parameters:
n: The device to identify.
:return: Confirmation of identifying ASC device n.
Returns:
Confirmation of identifying ASC device n.
</details>
"""
return await self.send_command("ascidentify", parameters=n)
async def asccount(self) -> dict:
"""Get data on the number of ASC devices and their info.
<details>
<summary>Expand</summary>
:return: Data on all ASC devices.
Returns:
Data on all ASC devices.
</details>
"""
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.
<details>
<summary>Expand</summary>
Sets an option on the ASC n to a value. Allowed options are:
```
AVA+BTB -
opt: freq
val: 256 - 1024 (chip frequency)
@@ -458,14 +641,16 @@ class CGMinerAPI(BaseMinerAPI):
opt: clock
val: 0 - 15
```
Parameters:
n: The ASC to set the options on.
opt: The option to set. Setting this to 'help' returns a help message.
val: The value to set the option to.
: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.
Returns:
Confirmation of setting option opt to value val.
</details>
"""
if val:
return await self.send_command("ascset", parameters=f"{n},{opt},{val}")
@@ -474,14 +659,22 @@ class CGMinerAPI(BaseMinerAPI):
async def lcd(self) -> dict:
"""Get a general all-in-one status summary of the miner.
<details>
<summary>Expand</summary>
:return: An all-in-one status summary of the miner.
Returns:
An all-in-one status summary of the miner.
</details>
"""
return await self.send_command("lcd")
async def lockstats(self) -> dict:
"""Write lockstats to STDERR.
<details>
<summary>Expand</summary>
:return: The result of writing the lock stats to STDERR.
Returns:
The result of writing the lock stats to STDERR.
</details>
"""
return await self.send_command("lockstats")