docs: add docstring for antminer and auradine in web (#127)
* docs: add docstring for antminer and auradine in web Signed-off-by: PAINxNAGATO <harshrawat349@gmail.com> * Remove Trailing whitespace from the docstring Signed-off-by: PAINxNAGATO <harshrawat349@gmail.com> * Remove Trailing whitespace from the docstring Signed-off-by: PAINxNAGATO <harshrawat349@gmail.com> * Remove Trailing whitespace from the docstring of the fucntion multicommand Signed-off-by: PAINxNAGATO <harshrawat349@gmail.com> --------- Signed-off-by: PAINxNAGATO <harshrawat349@gmail.com>
This commit is contained in:
committed by
GitHub
parent
6ea26e0e19
commit
8e81e18622
@@ -27,6 +27,11 @@ from pyasic.web.base import BaseWebAPI
|
|||||||
|
|
||||||
class AntminerModernWebAPI(BaseWebAPI):
|
class AntminerModernWebAPI(BaseWebAPI):
|
||||||
def __init__(self, ip: str) -> None:
|
def __init__(self, ip: str) -> None:
|
||||||
|
"""Initialize the modern Antminer API client with a specific IP address.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
ip (str): IP address of the Antminer device.
|
||||||
|
"""
|
||||||
super().__init__(ip)
|
super().__init__(ip)
|
||||||
self.username = "root"
|
self.username = "root"
|
||||||
self.pwd = settings.get("default_antminer_web_password", "root")
|
self.pwd = settings.get("default_antminer_web_password", "root")
|
||||||
@@ -39,6 +44,18 @@ class AntminerModernWebAPI(BaseWebAPI):
|
|||||||
privileged: bool = False,
|
privileged: bool = False,
|
||||||
**parameters: Any,
|
**parameters: Any,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Send a command to the Antminer device using HTTP digest authentication.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
command (str | bytes): The CGI command to send.
|
||||||
|
ignore_errors (bool): If True, ignore any HTTP errors.
|
||||||
|
allow_warning (bool): If True, proceed with warnings.
|
||||||
|
privileged (bool): If set to True, requires elevated privileges.
|
||||||
|
**parameters: Arbitrary keyword arguments to be sent as parameters in the request.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: The JSON response from the device or an empty dictionary if an error occurs.
|
||||||
|
"""
|
||||||
url = f"http://{self.ip}:{self.port}/cgi-bin/{command}.cgi"
|
url = f"http://{self.ip}:{self.port}/cgi-bin/{command}.cgi"
|
||||||
auth = httpx.DigestAuth(self.username, self.pwd)
|
auth = httpx.DigestAuth(self.username, self.pwd)
|
||||||
try:
|
try:
|
||||||
@@ -66,6 +83,16 @@ class AntminerModernWebAPI(BaseWebAPI):
|
|||||||
async def multicommand(
|
async def multicommand(
|
||||||
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
|
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Execute multiple commands simultaneously.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
*commands (str): Multiple command strings to be executed.
|
||||||
|
ignore_errors (bool): If True, ignore any HTTP errors.
|
||||||
|
allow_warning (bool): If True, proceed with warnings.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the results of all commands executed.
|
||||||
|
"""
|
||||||
async with httpx.AsyncClient(transport=settings.transport()) as client:
|
async with httpx.AsyncClient(transport=settings.transport()) as client:
|
||||||
tasks = [
|
tasks = [
|
||||||
asyncio.create_task(self._handle_multicommand(client, command))
|
asyncio.create_task(self._handle_multicommand(client, command))
|
||||||
@@ -83,6 +110,15 @@ class AntminerModernWebAPI(BaseWebAPI):
|
|||||||
async def _handle_multicommand(
|
async def _handle_multicommand(
|
||||||
self, client: httpx.AsyncClient, command: str
|
self, client: httpx.AsyncClient, command: str
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Helper function for handling individual commands in a multicommand execution.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
client (httpx.AsyncClient): The HTTP client to use for the request.
|
||||||
|
command (str): The command to be executed.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the response of the executed command.
|
||||||
|
"""
|
||||||
auth = httpx.DigestAuth(self.username, self.pwd)
|
auth = httpx.DigestAuth(self.username, self.pwd)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -100,29 +136,75 @@ class AntminerModernWebAPI(BaseWebAPI):
|
|||||||
return {command: {}}
|
return {command: {}}
|
||||||
|
|
||||||
async def get_miner_conf(self) -> dict:
|
async def get_miner_conf(self) -> dict:
|
||||||
|
"""Retrieve the miner configuration from the Antminer device.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the current configuration of the miner.
|
||||||
|
"""
|
||||||
return await self.send_command("get_miner_conf")
|
return await self.send_command("get_miner_conf")
|
||||||
|
|
||||||
async def set_miner_conf(self, conf: dict) -> dict:
|
async def set_miner_conf(self, conf: dict) -> dict:
|
||||||
|
"""Set the configuration for the miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
conf (dict): A dictionary of configuration settings to apply to the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary response from the device after setting the configuration.
|
||||||
|
"""
|
||||||
return await self.send_command("set_miner_conf", **conf)
|
return await self.send_command("set_miner_conf", **conf)
|
||||||
|
|
||||||
async def blink(self, blink: bool) -> dict:
|
async def blink(self, blink: bool) -> dict:
|
||||||
|
"""Control the blinking of the LED on the miner device.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
blink (bool): True to start blinking, False to stop.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary response from the device after the command execution.
|
||||||
|
"""
|
||||||
if blink:
|
if blink:
|
||||||
return await self.send_command("blink", blink="true")
|
return await self.send_command("blink", blink="true")
|
||||||
return await self.send_command("blink", blink="false")
|
return await self.send_command("blink", blink="false")
|
||||||
|
|
||||||
async def reboot(self) -> dict:
|
async def reboot(self) -> dict:
|
||||||
|
"""Reboot the miner device.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary response from the device confirming the reboot command.
|
||||||
|
"""
|
||||||
return await self.send_command("reboot")
|
return await self.send_command("reboot")
|
||||||
|
|
||||||
async def get_system_info(self) -> dict:
|
async def get_system_info(self) -> dict:
|
||||||
|
"""Retrieve system information from the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing system information of the miner.
|
||||||
|
"""
|
||||||
return await self.send_command("get_system_info")
|
return await self.send_command("get_system_info")
|
||||||
|
|
||||||
async def get_network_info(self) -> dict:
|
async def get_network_info(self) -> dict:
|
||||||
|
"""Retrieve network configuration information from the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the network configuration of the miner.
|
||||||
|
"""
|
||||||
return await self.send_command("get_network_info")
|
return await self.send_command("get_network_info")
|
||||||
|
|
||||||
async def summary(self) -> dict:
|
async def summary(self) -> dict:
|
||||||
|
"""Get a summary of the miner's status and performance.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A summary of the miner's current operational status.
|
||||||
|
"""
|
||||||
return await self.send_command("summary")
|
return await self.send_command("summary")
|
||||||
|
|
||||||
async def get_blink_status(self) -> dict:
|
async def get_blink_status(self) -> dict:
|
||||||
|
"""Check the status of the LED blinking on the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating whether the LED is currently blinking.
|
||||||
|
"""
|
||||||
return await self.send_command("get_blink_status")
|
return await self.send_command("get_blink_status")
|
||||||
|
|
||||||
async def set_network_conf(
|
async def set_network_conf(
|
||||||
@@ -134,6 +216,19 @@ class AntminerModernWebAPI(BaseWebAPI):
|
|||||||
hostname: str,
|
hostname: str,
|
||||||
protocol: int,
|
protocol: int,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Set the network configuration of the miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
ip (str): IP address of the device.
|
||||||
|
dns (str): DNS server IP address.
|
||||||
|
gateway (str): Gateway IP address.
|
||||||
|
subnet_mask (str): Network subnet mask.
|
||||||
|
hostname (str): Hostname of the device.
|
||||||
|
protocol (int): Network protocol used.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary response from the device after setting the network configuration.
|
||||||
|
"""
|
||||||
return await self.send_command(
|
return await self.send_command(
|
||||||
"set_network_conf",
|
"set_network_conf",
|
||||||
ipAddress=ip,
|
ipAddress=ip,
|
||||||
@@ -147,6 +242,11 @@ class AntminerModernWebAPI(BaseWebAPI):
|
|||||||
|
|
||||||
class AntminerOldWebAPI(BaseWebAPI):
|
class AntminerOldWebAPI(BaseWebAPI):
|
||||||
def __init__(self, ip: str) -> None:
|
def __init__(self, ip: str) -> None:
|
||||||
|
"""Initialize the old Antminer API client with a specific IP address.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
ip (str): IP address of the Antminer device.
|
||||||
|
"""
|
||||||
super().__init__(ip)
|
super().__init__(ip)
|
||||||
self.username = "root"
|
self.username = "root"
|
||||||
self.pwd = settings.get("default_antminer_web_password", "root")
|
self.pwd = settings.get("default_antminer_web_password", "root")
|
||||||
@@ -159,6 +259,18 @@ class AntminerOldWebAPI(BaseWebAPI):
|
|||||||
privileged: bool = False,
|
privileged: bool = False,
|
||||||
**parameters: Any,
|
**parameters: Any,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Send a command to the Antminer device using HTTP digest authentication.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
command (str | bytes): The CGI command to send.
|
||||||
|
ignore_errors (bool): If True, ignore any HTTP errors.
|
||||||
|
allow_warning (bool): If True, proceed with warnings.
|
||||||
|
privileged (bool): If set to True, requires elevated privileges.
|
||||||
|
**parameters: Arbitrary keyword arguments to be sent as parameters in the request.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: The JSON response from the device or an empty dictionary if an error occurs.
|
||||||
|
"""
|
||||||
url = f"http://{self.ip}:{self.port}/cgi-bin/{command}.cgi"
|
url = f"http://{self.ip}:{self.port}/cgi-bin/{command}.cgi"
|
||||||
auth = httpx.DigestAuth(self.username, self.pwd)
|
auth = httpx.DigestAuth(self.username, self.pwd)
|
||||||
try:
|
try:
|
||||||
@@ -184,6 +296,16 @@ class AntminerOldWebAPI(BaseWebAPI):
|
|||||||
async def multicommand(
|
async def multicommand(
|
||||||
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
|
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Execute multiple commands simultaneously.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
*commands (str): Multiple command strings to be executed.
|
||||||
|
ignore_errors (bool): If True, ignore any HTTP errors.
|
||||||
|
allow_warning (bool): If True, proceed with warnings.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the results of all commands executed.
|
||||||
|
"""
|
||||||
data = {k: None for k in commands}
|
data = {k: None for k in commands}
|
||||||
auth = httpx.DigestAuth(self.username, self.pwd)
|
auth = httpx.DigestAuth(self.username, self.pwd)
|
||||||
async with httpx.AsyncClient(transport=settings.transport()) as client:
|
async with httpx.AsyncClient(transport=settings.transport()) as client:
|
||||||
@@ -203,30 +325,81 @@ class AntminerOldWebAPI(BaseWebAPI):
|
|||||||
return data
|
return data
|
||||||
|
|
||||||
async def get_system_info(self) -> dict:
|
async def get_system_info(self) -> dict:
|
||||||
|
"""Retrieve system information from the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing system information of the miner.
|
||||||
|
"""
|
||||||
return await self.send_command("get_system_info")
|
return await self.send_command("get_system_info")
|
||||||
|
|
||||||
async def blink(self, blink: bool) -> dict:
|
async def blink(self, blink: bool) -> dict:
|
||||||
|
"""Control the blinking of the LED on the miner device.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
blink (bool): True to start blinking, False to stop.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary response from the device after the command execution.
|
||||||
|
"""
|
||||||
if blink:
|
if blink:
|
||||||
return await self.send_command("blink", action="startBlink")
|
return await self.send_command("blink", action="startBlink")
|
||||||
return await self.send_command("blink", action="stopBlink")
|
return await self.send_command("blink", action="stopBlink")
|
||||||
|
|
||||||
async def reboot(self) -> dict:
|
async def reboot(self) -> dict:
|
||||||
|
"""Reboot the miner device.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary response from the device confirming the reboot command.
|
||||||
|
"""
|
||||||
return await self.send_command("reboot")
|
return await self.send_command("reboot")
|
||||||
|
|
||||||
async def get_blink_status(self) -> dict:
|
async def get_blink_status(self) -> dict:
|
||||||
|
"""Check the status of the LED blinking on the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating whether the LED is currently blinking.
|
||||||
|
"""
|
||||||
return await self.send_command("blink", action="onPageLoaded")
|
return await self.send_command("blink", action="onPageLoaded")
|
||||||
|
|
||||||
async def get_miner_conf(self) -> dict:
|
async def get_miner_conf(self) -> dict:
|
||||||
|
"""Retrieve the miner configuration from the Antminer device.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the current configuration of the miner.
|
||||||
|
"""
|
||||||
return await self.send_command("get_miner_conf")
|
return await self.send_command("get_miner_conf")
|
||||||
|
|
||||||
async def set_miner_conf(self, conf: dict) -> dict:
|
async def set_miner_conf(self, conf: dict) -> dict:
|
||||||
|
"""Set the configuration for the miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
conf (dict): A dictionary of configuration settings to apply to the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary response from the device after setting the configuration.
|
||||||
|
"""
|
||||||
return await self.send_command("set_miner_conf", **conf)
|
return await self.send_command("set_miner_conf", **conf)
|
||||||
|
|
||||||
async def stats(self) -> dict:
|
async def stats(self) -> dict:
|
||||||
|
"""Retrieve detailed statistical data of the mining operation.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: Detailed statistics of the miner's operation.
|
||||||
|
"""
|
||||||
return await self.send_command("miner_stats")
|
return await self.send_command("miner_stats")
|
||||||
|
|
||||||
async def summary(self) -> dict:
|
async def summary(self) -> dict:
|
||||||
|
"""Get a summary of the miner's status and performance.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A summary of the miner's current operational status.
|
||||||
|
"""
|
||||||
return await self.send_command("miner_summary")
|
return await self.send_command("miner_summary")
|
||||||
|
|
||||||
async def pools(self) -> dict:
|
async def pools(self) -> dict:
|
||||||
|
"""Retrieve current pool information associated with the miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: Information about the mining pools configured in the miner.
|
||||||
|
"""
|
||||||
return await self.send_command("miner_pools")
|
return await self.send_command("miner_pools")
|
||||||
|
|||||||
@@ -30,6 +30,11 @@ from pyasic.web.base import BaseWebAPI
|
|||||||
|
|
||||||
class AuradineWebAPI(BaseWebAPI):
|
class AuradineWebAPI(BaseWebAPI):
|
||||||
def __init__(self, ip: str) -> None:
|
def __init__(self, ip: str) -> None:
|
||||||
|
"""Initializes the API client for interacting with Auradine mining devices.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
ip (str): IP address of the Auradine miner.
|
||||||
|
"""
|
||||||
super().__init__(ip)
|
super().__init__(ip)
|
||||||
self.username = "admin"
|
self.username = "admin"
|
||||||
self.pwd = settings.get("default_auradine_web_password", "admin")
|
self.pwd = settings.get("default_auradine_web_password", "admin")
|
||||||
@@ -37,6 +42,11 @@ class AuradineWebAPI(BaseWebAPI):
|
|||||||
self.token = None
|
self.token = None
|
||||||
|
|
||||||
async def auth(self) -> str | None:
|
async def auth(self) -> str | None:
|
||||||
|
"""Authenticate and retrieve a web token from the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
str | None: A token if authentication is successful, None otherwise.
|
||||||
|
"""
|
||||||
async with httpx.AsyncClient(transport=settings.transport()) as client:
|
async with httpx.AsyncClient(transport=settings.transport()) as client:
|
||||||
try:
|
try:
|
||||||
auth = await client.post(
|
auth = await client.post(
|
||||||
@@ -65,6 +75,18 @@ class AuradineWebAPI(BaseWebAPI):
|
|||||||
privileged: bool = False,
|
privileged: bool = False,
|
||||||
**parameters: Any,
|
**parameters: Any,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Send a command to the Auradine miner, handling authentication and retries.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
command (str | bytes): The specific command to execute.
|
||||||
|
ignore_errors (bool): Whether to ignore HTTP errors.
|
||||||
|
allow_warning (bool): Whether to proceed with warnings.
|
||||||
|
privileged (bool): Whether the command requires privileged access.
|
||||||
|
**parameters: Additional parameters for the command.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: The JSON response from the device.
|
||||||
|
"""
|
||||||
post = privileged or not parameters == {}
|
post = privileged or not parameters == {}
|
||||||
if not parameters == {}:
|
if not parameters == {}:
|
||||||
parameters["command"] = command
|
parameters["command"] = command
|
||||||
@@ -102,6 +124,16 @@ class AuradineWebAPI(BaseWebAPI):
|
|||||||
async def multicommand(
|
async def multicommand(
|
||||||
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
|
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
|
||||||
) -> dict:
|
) -> dict:
|
||||||
|
"""Execute multiple commands simultaneously on the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
*commands (str): Commands to execute.
|
||||||
|
ignore_errors (bool): Whether to ignore errors during command execution.
|
||||||
|
allow_warning (bool): Whether to proceed despite warnings.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing responses for all commands executed.
|
||||||
|
"""
|
||||||
tasks = {}
|
tasks = {}
|
||||||
# send all commands individually
|
# send all commands individually
|
||||||
for cmd in commands:
|
for cmd in commands:
|
||||||
@@ -124,52 +156,157 @@ class AuradineWebAPI(BaseWebAPI):
|
|||||||
return data
|
return data
|
||||||
|
|
||||||
async def factory_reset(self) -> dict:
|
async def factory_reset(self) -> dict:
|
||||||
|
"""Perform a factory reset on the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the reset operation.
|
||||||
|
"""
|
||||||
return await self.send_command("factory-reset", privileged=True)
|
return await self.send_command("factory-reset", privileged=True)
|
||||||
|
|
||||||
async def get_fan(self) -> dict:
|
async def get_fan(self) -> dict:
|
||||||
|
"""Retrieve the current fan status from the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the current fan status.
|
||||||
|
"""
|
||||||
return await self.send_command("fan")
|
return await self.send_command("fan")
|
||||||
|
|
||||||
async def set_fan(self, fan: int, speed_pct: int) -> dict:
|
async def set_fan(self, fan: int, speed_pct: int) -> dict:
|
||||||
|
"""Set the speed of a specific fan on the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
fan (int): The index of the fan to control.
|
||||||
|
speed_pct (int): The speed percentage to set for the fan.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the operation.
|
||||||
|
"""
|
||||||
return await self.send_command("fan", index=fan, percentage=speed_pct)
|
return await self.send_command("fan", index=fan, percentage=speed_pct)
|
||||||
|
|
||||||
async def firmware_upgrade(self, url: str = None, version: str = "latest") -> dict:
|
async def firmware_upgrade(self, url: str = None, version: str = "latest") -> dict:
|
||||||
|
"""Upgrade the firmware of the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
url (str, optional): The URL to download the firmware from.
|
||||||
|
version (str, optional): The version of the firmware to upgrade to, defaults to 'latest'.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the firmware upgrade.
|
||||||
|
"""
|
||||||
if url is not None:
|
if url is not None:
|
||||||
return await self.send_command("firmware-upgrade", url=url)
|
return await self.send_command("firmware-upgrade", url=url)
|
||||||
return await self.send_command("firmware-upgrade", version=version)
|
return await self.send_command("firmware-upgrade", version=version)
|
||||||
|
|
||||||
async def get_frequency(self) -> dict:
|
async def get_frequency(self) -> dict:
|
||||||
|
"""Retrieve the current frequency settings of the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the frequency settings.
|
||||||
|
"""
|
||||||
return await self.send_command("frequency")
|
return await self.send_command("frequency")
|
||||||
|
|
||||||
async def set_frequency(self, board: int, frequency: float) -> dict:
|
async def set_frequency(self, board: int, frequency: float) -> dict:
|
||||||
|
"""Set the frequency for a specific board on the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
board (int): The index of the board to configure.
|
||||||
|
frequency (float): The frequency in MHz to set for the board.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of setting the frequency.
|
||||||
|
"""
|
||||||
return await self.send_command("frequency", board=board, frequency=frequency)
|
return await self.send_command("frequency", board=board, frequency=frequency)
|
||||||
|
|
||||||
async def ipreport(self) -> dict:
|
async def ipreport(self) -> dict:
|
||||||
|
"""Generate an IP report for the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the IP report details.
|
||||||
|
"""
|
||||||
return await self.send_command("ipreport")
|
return await self.send_command("ipreport")
|
||||||
|
|
||||||
async def get_led(self) -> dict:
|
async def get_led(self) -> dict:
|
||||||
|
"""Retrieve the current LED status from the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the current status of the LED settings.
|
||||||
|
"""
|
||||||
return await self.send_command("led")
|
return await self.send_command("led")
|
||||||
|
|
||||||
async def set_led(self, code: int) -> dict:
|
async def set_led(self, code: int) -> dict:
|
||||||
|
"""Set the LED code on the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
code (int): The code that determines the LED behavior.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the operation.
|
||||||
|
"""
|
||||||
return await self.send_command("led", code=code)
|
return await self.send_command("led", code=code)
|
||||||
|
|
||||||
async def set_led_custom(self, code: int, led_1: int, led_2: int, msg: str) -> dict:
|
async def set_led_custom(self, code: int, led_1: int, led_2: int, msg: str) -> dict:
|
||||||
|
"""Set custom LED configurations including messages.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
code (int): The LED code to set.
|
||||||
|
led_1 (int): The first LED indicator number.
|
||||||
|
led_2 (int): The second LED indicator number.
|
||||||
|
msg (str): The message to display or represent with LEDs.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the custom LED configuration.
|
||||||
|
"""
|
||||||
return await self.send_command(
|
return await self.send_command(
|
||||||
"led", code=code, led1=led_1, led2=led_2, msg=msg
|
"led", code=code, led1=led_1, led2=led_2, msg=msg
|
||||||
)
|
)
|
||||||
|
|
||||||
async def get_mode(self) -> dict:
|
async def get_mode(self) -> dict:
|
||||||
|
"""Retrieve the current operational mode of the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the current mode settings.
|
||||||
|
"""
|
||||||
return await self.send_command("mode")
|
return await self.send_command("mode")
|
||||||
|
|
||||||
async def set_mode(self, **kwargs) -> dict:
|
async def set_mode(self, **kwargs) -> dict:
|
||||||
|
"""Set the operational mode of the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
**kwargs: Mode settings specified as keyword arguments.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the mode setting operation.
|
||||||
|
"""
|
||||||
return await self.send_command("mode", **kwargs)
|
return await self.send_command("mode", **kwargs)
|
||||||
|
|
||||||
async def get_network(self) -> dict:
|
async def get_network(self) -> dict:
|
||||||
|
"""Retrieve the network configuration settings of the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the network configuration details.
|
||||||
|
"""
|
||||||
return await self.send_command("network")
|
return await self.send_command("network")
|
||||||
|
|
||||||
async def set_network(self, **kwargs) -> dict:
|
async def set_network(self, **kwargs) -> dict:
|
||||||
|
"""Set the network configuration of the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
**kwargs: Network settings specified as keyword arguments.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the network configuration.
|
||||||
|
"""
|
||||||
return await self.send_command("network", **kwargs)
|
return await self.send_command("network", **kwargs)
|
||||||
|
|
||||||
async def password(self, password: str) -> dict:
|
async def password(self, password: str) -> dict:
|
||||||
|
"""Change the password used for accessing the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
password (str): The new password to set.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the password change operation.
|
||||||
|
"""
|
||||||
res = await self.send_command(
|
res = await self.send_command(
|
||||||
"password", user=self.username, old=self.pwd, new=password
|
"password", user=self.username, old=self.pwd, new=password
|
||||||
)
|
)
|
||||||
@@ -177,43 +314,129 @@ class AuradineWebAPI(BaseWebAPI):
|
|||||||
return res
|
return res
|
||||||
|
|
||||||
async def get_psu(self) -> dict:
|
async def get_psu(self) -> dict:
|
||||||
|
"""Retrieve the status of the power supply unit (PSU) from the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the PSU status.
|
||||||
|
"""
|
||||||
return await self.send_command("psu")
|
return await self.send_command("psu")
|
||||||
|
|
||||||
async def set_psu(self, voltage: float) -> dict:
|
async def set_psu(self, voltage: float) -> dict:
|
||||||
|
"""Set the voltage for the power supply unit of the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
voltage (float): The voltage level to set for the PSU.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of setting the PSU voltage.
|
||||||
|
"""
|
||||||
return await self.send_command("psu", voltage=voltage)
|
return await self.send_command("psu", voltage=voltage)
|
||||||
|
|
||||||
async def get_register(self) -> dict:
|
async def get_register(self) -> dict:
|
||||||
|
"""Retrieve registration information from the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the registration details.
|
||||||
|
"""
|
||||||
return await self.send_command("register")
|
return await self.send_command("register")
|
||||||
|
|
||||||
async def set_register(self, company: str) -> dict:
|
async def set_register(self, company: str) -> dict:
|
||||||
|
"""Set the registration information for the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
company (str): The company name to register the miner under.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the registration operation.
|
||||||
|
"""
|
||||||
return await self.send_command("register", parameter=company)
|
return await self.send_command("register", parameter=company)
|
||||||
|
|
||||||
async def reboot(self) -> dict:
|
async def reboot(self) -> dict:
|
||||||
|
"""Reboot the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the reboot operation.
|
||||||
|
"""
|
||||||
return await self.send_command("restart", privileged=True)
|
return await self.send_command("restart", privileged=True)
|
||||||
|
|
||||||
async def restart_gcminer(self) -> dict:
|
async def restart_gcminer(self) -> dict:
|
||||||
|
"""Restart the GCMiner application on the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the GCMiner restart operation.
|
||||||
|
"""
|
||||||
return await self.send_command("restart", parameter="gcminer")
|
return await self.send_command("restart", parameter="gcminer")
|
||||||
|
|
||||||
async def restart_api_server(self) -> dict:
|
async def restart_api_server(self) -> dict:
|
||||||
|
"""Restart the API server on the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the API server restart operation.
|
||||||
|
"""
|
||||||
return await self.send_command("restart", parameter="api-server")
|
return await self.send_command("restart", parameter="api-server")
|
||||||
|
|
||||||
async def temperature(self) -> dict:
|
async def temperature(self) -> dict:
|
||||||
|
"""Retrieve the current temperature readings from the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing temperature data.
|
||||||
|
"""
|
||||||
return await self.send_command("temperature")
|
return await self.send_command("temperature")
|
||||||
|
|
||||||
async def timedate(self, ntp: str, timezone: str) -> dict:
|
async def timedate(self, ntp: str, timezone: str) -> dict:
|
||||||
|
"""Set the time and date settings for the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
ntp (str): The NTP server to use for time synchronization.
|
||||||
|
timezone (str): The timezone setting.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of setting the time and date.
|
||||||
|
"""
|
||||||
return await self.send_command("timedate", ntp=ntp, timezone=timezone)
|
return await self.send_command("timedate", ntp=ntp, timezone=timezone)
|
||||||
|
|
||||||
async def get_token(self) -> dict:
|
async def get_token(self) -> dict:
|
||||||
|
"""Retrieve the current authentication token for the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the authentication token.
|
||||||
|
"""
|
||||||
return await self.send_command("token", user=self.username, password=self.pwd)
|
return await self.send_command("token", user=self.username, password=self.pwd)
|
||||||
|
|
||||||
async def update_pools(self, pools: list[dict]) -> dict:
|
async def update_pools(self, pools: list[dict]) -> dict:
|
||||||
|
"""Update the mining pools configuration on the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
pools (list[dict]): A list of dictionaries, each representing a pool configuration.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the update operation.
|
||||||
|
"""
|
||||||
return await self.send_command("updatepools", pools=pools)
|
return await self.send_command("updatepools", pools=pools)
|
||||||
|
|
||||||
async def voltage(self) -> dict:
|
async def voltage(self) -> dict:
|
||||||
|
"""Retrieve the voltage settings of the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the voltage details.
|
||||||
|
"""
|
||||||
return await self.send_command("voltage")
|
return await self.send_command("voltage")
|
||||||
|
|
||||||
async def get_ztp(self) -> dict:
|
async def get_ztp(self) -> dict:
|
||||||
|
"""Retrieve the zero-touch provisioning status from the Auradine miner.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary containing the ZTP status.
|
||||||
|
"""
|
||||||
return await self.send_command("ztp")
|
return await self.send_command("ztp")
|
||||||
|
|
||||||
async def set_ztp(self, enable: bool) -> dict:
|
async def set_ztp(self, enable: bool) -> dict:
|
||||||
|
"""Enable or disable zero-touch provisioning (ZTP) on the Auradine miner.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
enable (bool): True to enable ZTP, False to disable.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
dict: A dictionary indicating the result of the ZTP setting operation.
|
||||||
|
"""
|
||||||
return await self.send_command("ztp", parameter="on" if enable else "off")
|
return await self.send_command("ztp", parameter="on" if enable else "off")
|
||||||
|
|||||||
Reference in New Issue
Block a user