pyasic

AntminerModernWebAPI

Bases: BaseWebAPI

Source code in pyasic/web/antminer.py
class AntminerModernWebAPI(BaseWebAPI):
    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)
        self.username = "root"
        self.pwd = settings.get("default_antminer_web_password", "root")

    async def send_command(
        self,
        command: str | bytes,
        ignore_errors: bool = False,
        allow_warning: bool = True,
        privileged: bool = False,
        **parameters: Any,
    ) -> 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"
        auth = httpx.DigestAuth(self.username, self.pwd)
        try:
            async with httpx.AsyncClient(transport=settings.transport()) as client:

                if parameters:
                    data = await client.post(
                        url,
                        auth=auth,
                        timeout=settings.get("api_function_timeout", 3),
                        json=parameters,
                    )
                else:
                    data = await client.get(url, auth=auth)
        except httpx.HTTPError as e:
            return {"success": False, "message": f"HTTP error occurred: {str(e)}"}
        else:
            if data.status_code == 200:
                try:
                    return data.json()
                except json.decoder.JSONDecodeError:
                    return {"success": False, "message": "Failed to decode JSON"}
        return {"success": False, "message": "Unknown error occurred"}

    async def multicommand(
        self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
    ) -> 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:
            tasks = [
                asyncio.create_task(self._handle_multicommand(client, command))
                for command in commands
            ]
            all_data = await asyncio.gather(*tasks)

        data = {}
        for item in all_data:
            data.update(item)

        data["multicommand"] = True
        return data

    async def _handle_multicommand(
        self, client: httpx.AsyncClient, command: str
    ) -> 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)

        try:
            url = f"http://{self.ip}/cgi-bin/{command}.cgi"
            ret = await client.get(url, auth=auth)
        except httpx.HTTPError:
            pass
        else:
            if ret.status_code == 200:
                try:
                    json_data = ret.json()
                    return {command: json_data}
                except json.decoder.JSONDecodeError:
                    pass
        return {command: {}}

    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")

    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)

    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:
            return await self.send_command("blink", blink="true")
        return await self.send_command("blink", blink="false")

    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")

    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")

    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")

    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")

    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")

    async def set_network_conf(
        self,
        ip: str,
        dns: str,
        gateway: str,
        subnet_mask: str,
        hostname: str,
        protocol: int,
    ) -> 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(
            "set_network_conf",
            ipAddress=ip,
            ipDns=dns,
            ipGateway=gateway,
            ipHost=hostname,
            ipPro=protocol,
            ipSub=subnet_mask,
        )

__init__(ip)

Initialize the modern Antminer API client with a specific IP address.

Parameters:

Name Type Description Default
ip str

IP address of the Antminer device.

required
Source code in pyasic/web/antminer.py
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)
    self.username = "root"
    self.pwd = settings.get("default_antminer_web_password", "root")

Control the blinking of the LED on the miner device.

Parameters:

Name Type Description Default
blink bool

True to start blinking, False to stop.

required

Returns:

Name Type Description
dict dict

A dictionary response from the device after the command execution.

Source code in pyasic/web/antminer.py
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:
        return await self.send_command("blink", blink="true")
    return await self.send_command("blink", blink="false")

Check the status of the LED blinking on the miner.

Returns:

Name Type Description
dict dict

A dictionary indicating whether the LED is currently blinking.

Source code in pyasic/web/antminer.py
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")

get_miner_conf() async

Retrieve the miner configuration from the Antminer device.

Returns:

Name Type Description
dict dict

A dictionary containing the current configuration of the miner.

Source code in pyasic/web/antminer.py
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")

get_network_info() async

Retrieve network configuration information from the miner.

Returns:

Name Type Description
dict dict

A dictionary containing the network configuration of the miner.

Source code in pyasic/web/antminer.py
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")

get_system_info() async

Retrieve system information from the miner.

Returns:

Name Type Description
dict dict

A dictionary containing system information of the miner.

Source code in pyasic/web/antminer.py
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")

multicommand(*commands, ignore_errors=False, allow_warning=True) async

Execute multiple commands simultaneously.

Parameters:

Name Type Description Default
*commands str

Multiple command strings to be executed.

()
ignore_errors bool

If True, ignore any HTTP errors.

False
allow_warning bool

If True, proceed with warnings.

True

Returns:

Name Type Description
dict dict

A dictionary containing the results of all commands executed.

Source code in pyasic/web/antminer.py
async def multicommand(
    self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
) -> 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:
        tasks = [
            asyncio.create_task(self._handle_multicommand(client, command))
            for command in commands
        ]
        all_data = await asyncio.gather(*tasks)

    data = {}
    for item in all_data:
        data.update(item)

    data["multicommand"] = True
    return data

reboot() async

Reboot the miner device.

Returns:

Name Type Description
dict dict

A dictionary response from the device confirming the reboot command.

Source code in pyasic/web/antminer.py
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")

send_command(command, ignore_errors=False, allow_warning=True, privileged=False, **parameters) async

Send a command to the Antminer device using HTTP digest authentication.

Parameters:

Name Type Description Default
command str | bytes

The CGI command to send.

required
ignore_errors bool

If True, ignore any HTTP errors.

False
allow_warning bool

If True, proceed with warnings.

True
privileged bool

If set to True, requires elevated privileges.

False
**parameters Any

Arbitrary keyword arguments to be sent as parameters in the request.

{}

Returns:

Name Type Description
dict dict

The JSON response from the device or an empty dictionary if an error occurs.

Source code in pyasic/web/antminer.py
async def send_command(
    self,
    command: str | bytes,
    ignore_errors: bool = False,
    allow_warning: bool = True,
    privileged: bool = False,
    **parameters: Any,
) -> 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"
    auth = httpx.DigestAuth(self.username, self.pwd)
    try:
        async with httpx.AsyncClient(transport=settings.transport()) as client:

            if parameters:
                data = await client.post(
                    url,
                    auth=auth,
                    timeout=settings.get("api_function_timeout", 3),
                    json=parameters,
                )
            else:
                data = await client.get(url, auth=auth)
    except httpx.HTTPError as e:
        return {"success": False, "message": f"HTTP error occurred: {str(e)}"}
    else:
        if data.status_code == 200:
            try:
                return data.json()
            except json.decoder.JSONDecodeError:
                return {"success": False, "message": "Failed to decode JSON"}
    return {"success": False, "message": "Unknown error occurred"}

set_miner_conf(conf) async

Set the configuration for the miner.

Parameters:

Name Type Description Default
conf dict

A dictionary of configuration settings to apply to the miner.

required

Returns:

Name Type Description
dict dict

A dictionary response from the device after setting the configuration.

Source code in pyasic/web/antminer.py
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)

set_network_conf(ip, dns, gateway, subnet_mask, hostname, protocol) async

Set the network configuration of the miner.

Parameters:

Name Type Description Default
ip str

IP address of the device.

required
dns str

DNS server IP address.

required
gateway str

Gateway IP address.

required
subnet_mask str

Network subnet mask.

required
hostname str

Hostname of the device.

required
protocol int

Network protocol used.

required

Returns:

Name Type Description
dict dict

A dictionary response from the device after setting the network configuration.

Source code in pyasic/web/antminer.py
async def set_network_conf(
    self,
    ip: str,
    dns: str,
    gateway: str,
    subnet_mask: str,
    hostname: str,
    protocol: int,
) -> 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(
        "set_network_conf",
        ipAddress=ip,
        ipDns=dns,
        ipGateway=gateway,
        ipHost=hostname,
        ipPro=protocol,
        ipSub=subnet_mask,
    )

summary() async

Get a summary of the miner's status and performance.

Returns:

Name Type Description
dict dict

A summary of the miner's current operational status.

Source code in pyasic/web/antminer.py
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")

AntminerOldWebAPI

Bases: BaseWebAPI

Source code in pyasic/web/antminer.py
class AntminerOldWebAPI(BaseWebAPI):
    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)
        self.username = "root"
        self.pwd = settings.get("default_antminer_web_password", "root")

    async def send_command(
        self,
        command: str | bytes,
        ignore_errors: bool = False,
        allow_warning: bool = True,
        privileged: bool = False,
        **parameters: Any,
    ) -> 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"
        auth = httpx.DigestAuth(self.username, self.pwd)
        try:
            async with httpx.AsyncClient(transport=settings.transport()) as client:
                if parameters:
                    data = await client.post(
                        url,
                        data=parameters,
                        auth=auth,
                        timeout=settings.get("api_function_timeout", 3),
                    )
                else:
                    data = await client.get(url, auth=auth)
        except httpx.HTTPError:
            pass
        else:
            if data.status_code == 200:
                try:
                    return data.json()
                except json.decoder.JSONDecodeError:
                    pass

    async def multicommand(
        self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
    ) -> 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}
        auth = httpx.DigestAuth(self.username, self.pwd)
        async with httpx.AsyncClient(transport=settings.transport()) as client:
            for command in commands:
                try:
                    url = f"http://{self.ip}/cgi-bin/{command}.cgi"
                    ret = await client.get(url, auth=auth)
                except httpx.HTTPError:
                    pass
                else:
                    if ret.status_code == 200:
                        try:
                            json_data = ret.json()
                            data[command] = json_data
                        except json.decoder.JSONDecodeError:
                            pass
        return data

    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")

    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:
            return await self.send_command("blink", action="startBlink")
        return await self.send_command("blink", action="stopBlink")

    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")

    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")

    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")

    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)

    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")

    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")

    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")

    async def update_firmware(self, file: Path, keep_settings: bool = True) -> dict:
        """Perform a system update by uploading a firmware file and sending a command to initiate the update."""

        async with aiofiles.open(file, "rb") as firmware:
            file_content = await firmware.read()

        parameters = {
            "file": (file.name, file_content, "application/octet-stream"),
            "filename": file.name,
            "keep_settings": keep_settings,
        }

        return await self.send_command(command="upgrade", **parameters)

__init__(ip)

Initialize the old Antminer API client with a specific IP address.

Parameters:

Name Type Description Default
ip str

IP address of the Antminer device.

required
Source code in pyasic/web/antminer.py
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)
    self.username = "root"
    self.pwd = settings.get("default_antminer_web_password", "root")

Control the blinking of the LED on the miner device.

Parameters:

Name Type Description Default
blink bool

True to start blinking, False to stop.

required

Returns:

Name Type Description
dict dict

A dictionary response from the device after the command execution.

Source code in pyasic/web/antminer.py
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:
        return await self.send_command("blink", action="startBlink")
    return await self.send_command("blink", action="stopBlink")

Check the status of the LED blinking on the miner.

Returns:

Name Type Description
dict dict

A dictionary indicating whether the LED is currently blinking.

Source code in pyasic/web/antminer.py
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")

get_miner_conf() async

Retrieve the miner configuration from the Antminer device.

Returns:

Name Type Description
dict dict

A dictionary containing the current configuration of the miner.

Source code in pyasic/web/antminer.py
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")

get_system_info() async

Retrieve system information from the miner.

Returns:

Name Type Description
dict dict

A dictionary containing system information of the miner.

Source code in pyasic/web/antminer.py
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")

multicommand(*commands, ignore_errors=False, allow_warning=True) async

Execute multiple commands simultaneously.

Parameters:

Name Type Description Default
*commands str

Multiple command strings to be executed.

()
ignore_errors bool

If True, ignore any HTTP errors.

False
allow_warning bool

If True, proceed with warnings.

True

Returns:

Name Type Description
dict dict

A dictionary containing the results of all commands executed.

Source code in pyasic/web/antminer.py
async def multicommand(
    self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
) -> 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}
    auth = httpx.DigestAuth(self.username, self.pwd)
    async with httpx.AsyncClient(transport=settings.transport()) as client:
        for command in commands:
            try:
                url = f"http://{self.ip}/cgi-bin/{command}.cgi"
                ret = await client.get(url, auth=auth)
            except httpx.HTTPError:
                pass
            else:
                if ret.status_code == 200:
                    try:
                        json_data = ret.json()
                        data[command] = json_data
                    except json.decoder.JSONDecodeError:
                        pass
    return data

pools() async

Retrieve current pool information associated with the miner.

Returns:

Name Type Description
dict dict

Information about the mining pools configured in the miner.

Source code in pyasic/web/antminer.py
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")

reboot() async

Reboot the miner device.

Returns:

Name Type Description
dict dict

A dictionary response from the device confirming the reboot command.

Source code in pyasic/web/antminer.py
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")

send_command(command, ignore_errors=False, allow_warning=True, privileged=False, **parameters) async

Send a command to the Antminer device using HTTP digest authentication.

Parameters:

Name Type Description Default
command str | bytes

The CGI command to send.

required
ignore_errors bool

If True, ignore any HTTP errors.

False
allow_warning bool

If True, proceed with warnings.

True
privileged bool

If set to True, requires elevated privileges.

False
**parameters Any

Arbitrary keyword arguments to be sent as parameters in the request.

{}

Returns:

Name Type Description
dict dict

The JSON response from the device or an empty dictionary if an error occurs.

Source code in pyasic/web/antminer.py
async def send_command(
    self,
    command: str | bytes,
    ignore_errors: bool = False,
    allow_warning: bool = True,
    privileged: bool = False,
    **parameters: Any,
) -> 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"
    auth = httpx.DigestAuth(self.username, self.pwd)
    try:
        async with httpx.AsyncClient(transport=settings.transport()) as client:
            if parameters:
                data = await client.post(
                    url,
                    data=parameters,
                    auth=auth,
                    timeout=settings.get("api_function_timeout", 3),
                )
            else:
                data = await client.get(url, auth=auth)
    except httpx.HTTPError:
        pass
    else:
        if data.status_code == 200:
            try:
                return data.json()
            except json.decoder.JSONDecodeError:
                pass

set_miner_conf(conf) async

Set the configuration for the miner.

Parameters:

Name Type Description Default
conf dict

A dictionary of configuration settings to apply to the miner.

required

Returns:

Name Type Description
dict dict

A dictionary response from the device after setting the configuration.

Source code in pyasic/web/antminer.py
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)

stats() async

Retrieve detailed statistical data of the mining operation.

Returns:

Name Type Description
dict dict

Detailed statistics of the miner's operation.

Source code in pyasic/web/antminer.py
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")

summary() async

Get a summary of the miner's status and performance.

Returns:

Name Type Description
dict dict

A summary of the miner's current operational status.

Source code in pyasic/web/antminer.py
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")

update_firmware(file, keep_settings=True) async

Perform a system update by uploading a firmware file and sending a command to initiate the update.

Source code in pyasic/web/antminer.py
async def update_firmware(self, file: Path, keep_settings: bool = True) -> dict:
    """Perform a system update by uploading a firmware file and sending a command to initiate the update."""

    async with aiofiles.open(file, "rb") as firmware:
        file_content = await firmware.read()

    parameters = {
        "file": (file.name, file_content, "application/octet-stream"),
        "filename": file.name,
        "keep_settings": keep_settings,
    }

    return await self.send_command(command="upgrade", **parameters)