pyasic

BTMinerRPCAPI

Bases: BaseMinerRPCAPI

An abstraction of the API for MicroBT Whatsminers, BTMiner.

Each method corresponds to an API command in BMMiner.

This class abstracts use of the BTMiner API, as well as the methods for sending commands to it. The self.send_command() function handles sending a command to the miner asynchronously, and as such is the base for many of the functions in this class, which rely on it to send the command for them.

All privileged commands for BTMiner's API require that you change the password of the miners using the Whatsminer tool, and it can be changed back to admin with this tool after. Set the new password either by passing it to the init method, or changing it in settings.toml.

Additionally, the API commands for the privileged API must be encoded using a token from the miner, all privileged commands do this automatically for you and will decode the output to look like a normal output from a miner API.

Source code in pyasic/rpc/btminer.py

class BTMinerRPCAPI(BaseMinerRPCAPI):
    """An abstraction of the API for MicroBT Whatsminers, BTMiner.

    Each method corresponds to an API command in BMMiner.

    This class abstracts use of the BTMiner API, as well as the
    methods for sending commands to it.  The `self.send_command()`
    function handles sending a command to the miner asynchronously, and
    as such is the base for many of the functions in this class, which
    rely on it to send the command for them.

    All privileged commands for BTMiner's API require that you change
    the password of the miners using the Whatsminer tool, and it can be
    changed back to admin with this tool after.  Set the new password
    either by passing it to the __init__ method, or changing it in
    settings.toml.

    Additionally, the API commands for the privileged API must be
    encoded using a token from the miner, all privileged commands do
    this automatically for you and will decode the output to look like
    a normal output from a miner API.
    """

    def __init__(self, ip: str, port: int = 4028, api_ver: str = "0.0.0") -> None:
        super().__init__(ip, port, api_ver)
        self.pwd = settings.get("default_whatsminer_rpc_password", "admin")
        self.token = None

    async def multicommand(self, *commands: str, allow_warning: bool = True) -> dict:
        """Creates and sends multiple commands as one command to the miner.

        Parameters:
            *commands: The commands to send as a multicommand to the miner.
            allow_warning: A boolean to supress APIWarnings.
        """
        # make sure we can actually run each command, otherwise they will fail
        commands = self._check_commands(*commands)
        # standard multicommand format is "command1+command2"
        # commands starting with "get_" and the "status" command aren't supported, but we can fake that

        split_commands = []

        for command in list(commands):
            if command.startswith("get_") or command == "status":
                commands.remove(command)
                # send seperately and append later
                split_commands.append(command)

        command = "+".join(commands)

        tasks = []
        if len(split_commands) > 0:
            tasks.append(
                asyncio.create_task(
                    self._send_split_multicommand(
                        *split_commands, allow_warning=allow_warning
                    )
                )
            )
        tasks.append(
            asyncio.create_task(self.send_command(command, allow_warning=allow_warning))
        )

        try:
            all_data = await asyncio.gather(*tasks)
        except APIError:
            return {}

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

        data["multicommand"] = True
        return data

    async def send_privileged_command(
        self,
        command: Union[str, bytes],
        ignore_errors: bool = False,
        timeout: int = 10,
        **kwargs,
    ) -> dict:
        try:
            return await self._send_privileged_command(
                command=command, ignore_errors=ignore_errors, timeout=timeout, **kwargs
            )
        except APIError as e:
            if not e.message == "can't access write cmd":
                raise
        try:
            await self.open_api()
        except Exception as e:
            raise APIError("Failed to open whatsminer API.") from e
        return await self._send_privileged_command(
            command=command, ignore_errors=ignore_errors, timeout=timeout, **kwargs
        )

    async def _send_privileged_command(
        self,
        command: Union[str, bytes],
        ignore_errors: bool = False,
        timeout: int = 10,
        **kwargs,
    ) -> dict:
        logging.debug(
            f"{self} - (Send Privileged Command) - {command} " + f"with args {kwargs}"
            if len(kwargs) > 0
            else ""
        )
        command = {"cmd": command, **kwargs}

        token_data = await self.get_token()
        enc_command = create_privileged_cmd(token_data, command)

        logging.debug(f"{self} - (Send Privileged Command) - Sending")
        try:
            data = await self._send_bytes(enc_command, timeout=timeout)
        except (asyncio.CancelledError, asyncio.TimeoutError):
            if ignore_errors:
                return {}
            raise APIError("No data was returned from the API.")

        if not data:
            if ignore_errors:
                return {}
            raise APIError("No data was returned from the API.")
        data = self._load_api_data(data)

        try:
            data = parse_btminer_priviledge_data(self.token, data)
        except Exception as e:
            logging.info(f"{str(self.ip)}: {e}")

        if not ignore_errors:
            # if it fails to validate, it is likely an error
            validation = validate_command_output(data)
            if not validation[0]:
                raise APIError(validation[1])

        # return the parsed json as a dict
        return data

    async def get_token(self) -> dict:
        """Gets token information from the API.
        <details>
            <summary>Expand</summary>

        Returns:
            An encoded token and md5 password, which are used for the privileged API.
        </details>
        """
        logging.debug(f"{self} - (Get Token) - Getting token")
        if self.token:
            if self.token["timestamp"] > datetime.datetime.now() - datetime.timedelta(
                minutes=30
            ):
                return self.token

        # get the token
        data = await self.send_command("get_token")

        # encrypt the admin password with the salt
        pwd = _crypt(self.pwd, "$1$" + data["Msg"]["salt"] + "$")
        pwd = pwd.split("$")

        # take the 4th item from the pwd split
        host_passwd_md5 = pwd[3]

        # encrypt the pwd with the time and new salt
        tmp = _crypt(pwd[3] + data["Msg"]["time"], "$1$" + data["Msg"]["newsalt"] + "$")
        tmp = tmp.split("$")

        # take the 4th item from the encrypted pwd split
        host_sign = tmp[3]

        # set the current token
        self.token = {
            "host_sign": host_sign,
            "host_passwd_md5": host_passwd_md5,
            "timestamp": datetime.datetime.now(),
        }
        logging.debug(f"{self} - (Get Token) - Gathered token data: {self.token}")
        return self.token

    async def open_api(self):
        async with httpx.AsyncClient() as c:
            stage1_req = (
                await c.post(
                    "https://wmt.pyasic.org/v1/stage1",
                    json={"ip": str(self.ip)},
                    follow_redirects=True,
                )
            ).json()
            stage1_res = binascii.hexlify(
                await self._send_bytes(binascii.unhexlify(stage1_req), port=8889)
            )
            stage2_req = (
                await c.post(
                    "https://wmt.pyasic.org/v1/stage2",
                    json={
                        "ip": str(self.ip),
                        "stage1_result": stage1_res.decode("utf-8"),
                    },
                )
            ).json()
        for command in stage2_req:
            try:
                await self._send_bytes(
                    binascii.unhexlify(command), timeout=3, port=8889
                )
            except asyncio.TimeoutError:
                pass
        return True

    #### PRIVILEGED COMMANDS ####
    # Please read the top of this file to learn
    # how to configure the Whatsminer API to
    # use these commands.

    async def update_pools(
        self,
        pool_1: str,
        worker_1: str,
        passwd_1: str,
        pool_2: str = None,
        worker_2: str = None,
        passwd_2: str = None,
        pool_3: str = None,
        worker_3: str = None,
        passwd_3: str = None,
    ) -> 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.

        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>
        """
        return await self.send_privileged_command(
            "update_pools",
            pool1=pool_1,
            worker1=worker_1,
            passwd1=passwd_1,
            pool2=pool_2,
            worker2=worker_2,
            passwd2=passwd_2,
            pool3=pool_3,
            worker3=worker_3,
            passwd3=passwd_3,
        )

    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.

        Returns:
            A reply informing of the restart.
        </details>
        """
        return await self.send_privileged_command("restart_btminer")

    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.

        Parameters:
            respbefore: Whether to respond before powering off.
        Returns:
            A reply informing of the status of powering off.
        </details>
        """
        if respbefore:
            return await self.send_privileged_command("power_off", respbefore="true")
        return await self.send_privileged_command("power_off", respbefore="false")

    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.

        Returns:
            A reply informing of the status of powering on.
        </details>
        """
        return await self.send_privileged_command("power_on")

    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.

        Returns:
            A reply informing of the status of resetting the LED.
        </details>
        """
        return await self.set_led(auto=True)

    async def set_led(
        self,
        auto: bool = True,
        color: str = "red",
        period: int = 60,
        duration: int = 20,
        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.

        Parameters:
            auto: Whether or not to reset the LED to auto mode.
            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>
        """
        if auto:
            return await self.send_privileged_command("set_led", param="auto")
        return await self.send_privileged_command(
            "set_led", color=color, period=period, duration=duration, start=start
        )

    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.

        Returns:
            A reply informing of the status of setting low power mode.
        </details>
        """
        return await self.send_privileged_command("set_low_power")

    async def set_high_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.

        Returns:
            A reply informing of the status of setting low power mode.
        </details>
        """
        return await self.send_privileged_command("set_high_power")

    async def set_normal_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.

        Returns:
            A reply informing of the status of setting low power mode.
        </details>
        """
        return await self.send_privileged_command("set_normal_power")

    async def update_firmware(self, firmware: bytes) -> bool:
        """Upgrade the firmware running on the miner and using the firmware passed in bytes.
        <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.

        Parameters:
            firmware (bytes): The firmware binary data to be uploaded.
        Returns:
            A boolean indicating the success of the firmware upgrade.
        Raises:
            APIError: If the miner is not ready for firmware update.
        </details>
        """
        ready = await self.send_privileged_command("upgrade_firmware")
        if not ready.get("Msg") == "ready":
            raise APIError(f"Not ready for firmware update: {self}")
        file_size = struct.pack("<I", len(firmware))
        await self._send_bytes(file_size + firmware)
        return True

    async def reboot(self, timeout: int = 10) -> dict:
        """Reboot the miner using the API.
        <details>
            <summary>Expand</summary>

        Returns:
            A reply informing of the status of the reboot.
        </details>
        """
        try:
            d = await asyncio.wait_for(
                self.send_privileged_command("reboot"), timeout=timeout
            )
        except (asyncio.CancelledError, asyncio.TimeoutError):
            return {}
        else:
            return d

    async def factory_reset(self) -> dict:
        """Reset the miner to factory defaults.
        <details>
            <summary>Expand</summary>

        Returns:
            A reply informing of the status of the reset.
        </details>
        """
        return await self.send_privileged_command("factory_reset")

    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.

        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.
        """
        self.pwd = old_pwd
        # check if password length is greater than 8 bytes
        if len(new_pwd.encode("utf-8")) > 8:
            raise APIError(
                f"New password too long, the max length is 8.  "
                f"Password size: {len(new_pwd.encode('utf-8'))}"
            )
        try:
            data = await self.send_privileged_command(
                "update_pwd", old=old_pwd, new=new_pwd
            )
        except APIError as e:
            raise e
        self.pwd = new_pwd
        return data

    async def net_config(
        self,
        ip: str = None,
        mask: str = None,
        gate: str = None,
        dns: str = None,
        host: str = None,
        dhcp: bool = True,
    ):
        if dhcp:
            return await self.send_privileged_command("net_config", param="dhcp")
        if None in [ip, mask, gate, dns, host]:
            raise APIError("Incorrect parameters.")
        return await self.send_privileged_command(
            "net_config", ip=ip, mask=mask, gate=gate, dns=dns, host=host
        )

    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%.

        Parameters:
            percent: The frequency % to set.
        Returns:
            A reply informing of the status of setting the frequency.
        </details>
        """
        if not -100 < percent < 100:
            raise APIError(
                f"Frequency % is outside of the allowed "
                f"range.  Please set a % between -100 and "
                f"100"
            )
        return await self.send_privileged_command(
            "set_target_freq", percent=str(percent)
        )

    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.

        Returns:
            A reply informing of the status of enabling fast boot.
        </details>
        """
        return await self.send_privileged_command("enable_btminer_fast_boot")

    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.

        Returns:
            A reply informing of the status of disabling fast boot.
        </details>
        """
        return await self.send_privileged_command("disable_btminer_fast_boot")

    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.

        Returns:
            A reply informing of the status of enabling web pools.
        </details>
        """
        return await self.send_privileged_command("enable_web_pools")

    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.

        Returns:
            A reply informing of the status of disabling web pools.
        </details>
        """
        return await self.send_privileged_command("disable_web_pools")

    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:
            A reply informing of the status of setting the hostname.
        </details>
        """
        return await self.send_privileged_command("set_hostname", hostname=hostname)

    async def set_power_pct(self, percent: int) -> dict:
        """Set the power percentage of the miner based on current power.  Used for temporary adjustment.

        <details>
            <summary>Expand</summary>

        Set the power percentage of the miner, only works after changing
        the password of the miner using the Whatsminer tool.

        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:
            raise APIError(
                f"Power PCT % is outside of the allowed "
                f"range.  Please set a % between 0 and "
                f"100"
            )
        return await self.send_privileged_command("set_power_pct", percent=str(percent))

    async def pre_power_on(self, complete: bool, msg: PrePowerOnMessage) -> 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.

        Parameters:
            complete: check whether pre power on is complete.
            msg: the message to check.

        Returns:
            A reply informing of the status of pre power on.
        </details>
        """

        if msg not in ("wait for adjust temp", "adjust complete", "adjust continue"):
            raise APIError(
                "Message is incorrect, please choose one of "
                '["wait for adjust temp", '
                '"adjust complete", '
                '"adjust continue"]'
            )
        if complete:
            return await self.send_privileged_command(
                "pre_power_on", complete="true", msg=msg
            )
        return await self.send_privileged_command(
            "pre_power_on", complete="false", msg=msg
        )

    ### ADDED IN V2.0.5 Whatsminer API ###

    @api_min_version("2.0.5")
    async def set_power_pct_v2(self, percent: int) -> dict:
        """Set the power percentage of the miner based on current power.  Used for temporary adjustment.  Added in API v2.0.5.

        <details>
            <summary>Expand</summary>

        Set the power percentage of the miner, only works after changing
        the password of the miner using the Whatsminer tool.

        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:
            raise APIError(
                f"Power PCT % is outside of the allowed "
                f"range.  Please set a % between 0 and "
                f"100"
            )
        return await self.send_privileged_command(
            "set_power_pct_v2", percent=str(percent)
        )

    @api_min_version("2.0.5")
    async def set_temp_offset(self, temp_offset: int) -> dict:
        """Set the offset of miner hash board target temperature.

        <details>
            <summary>Expand</summary>

        Set the offset of miner hash board target temperature, only works after
        changing the password of the miner using the Whatsminer tool.

        Parameters:
            temp_offset: Target temperature offset.
        Returns:
            A reply informing of the status of setting temp offset.
        </details>

        """
        if not -30 < temp_offset < 0:
            raise APIError(
                f"Temp offset is outside of the allowed "
                f"range.  Please set a number between -30 and "
                f"0."
            )

        return await self.send_privileged_command(
            "set_temp_offset", temp_offset=temp_offset
        )

    @api_min_version("2.0.5")
    async def adjust_power_limit(self, power_limit: int) -> dict:
        """Set the upper limit of the miner's power. Cannot be higher than the ordinary power of the machine.

        <details>
            <summary>Expand</summary>

        Set the upper limit of the miner's power, only works after
        changing the password of the miner using the Whatsminer tool.
        The miner will reboot after this is set.

        Parameters:
            power_limit: New power limit.
        Returns:
            A reply informing of the status of setting power limit.
        </details>

        """
        return await self.send_privileged_command(
            "adjust_power_limit", power_limit=str(power_limit)
        )

    @api_min_version("2.0.5")
    async def adjust_upfreq_speed(self, upfreq_speed: int) -> dict:
        """Set the upfreq speed, 0 is the normal speed, 9 is the fastest speed.

        <details>
            <summary>Expand</summary>

        Set the upfreq speed, 0 is the normal speed, 9 is the fastest speed, only works after
        changing the password of the miner using the Whatsminer tool.
        The faster the speed, the greater the final hash rate and power deviation, and the stability
        may be impacted. Fast boot mode cannot be used at the same time.

        Parameters:
            upfreq_speed: New upfreq speed.
        Returns:
            A reply informing of the status of setting upfreq speed.
        </details>
        """
        if not 0 < upfreq_speed < 9:
            raise APIError(
                f"Upfreq speed is outside of the allowed "
                f"range.  Please set a number between 0 (Normal) and "
                f"9 (Fastest)."
            )
        return await self.send_privileged_command(
            "adjust_upfreq_speed", upfreq_speed=upfreq_speed
        )

    @api_min_version("2.0.5")
    async def set_poweroff_cool(self, poweroff_cool: bool) -> dict:
        """Set whether to cool the machine when mining is stopped.

        <details>
            <summary>Expand</summary>

        Set whether to cool the machine when mining is stopped, only works after
        changing the password of the miner using the Whatsminer tool.

        Parameters:
            poweroff_cool: Whether to cool the miner during power off mode.
        Returns:
            A reply informing of the status of setting power off cooling mode.
        </details>
        """

        return await self.send_privileged_command(
            "set_poweroff_cool", poweroff_cool=int(poweroff_cool)
        )

    @api_min_version("2.0.5")
    async def set_fan_zero_speed(self, fan_zero_speed: bool) -> dict:
        """Sets whether the fan speed supports the lowest 0 speed.

        <details>
            <summary>Expand</summary>

        Sets whether the fan speed supports the lowest 0 speed, only works after
        changing the password of the miner using the Whatsminer tool.

        Parameters:
            fan_zero_speed: Whether the fan is allowed to support 0 speed.
        Returns:
            A reply informing of the status of setting fan minimum speed.
        </details>

        """
        return await self.send_privileged_command(
            "set_fan_zero_speed", fan_zero_speed=int(fan_zero_speed)
        )

    #### END privileged COMMANDS ####

    async def summary(self) -> dict:
        """Get the summary status from the miner.
        <details>
            <summary>Expand</summary>

        Returns:
            Summary status of the miner.
        </details>
        """
        return await self.send_command("summary")

    async def pools(self) -> dict:
        """Get the pool status from the miner.
        <details>
            <summary>Expand</summary>

        Returns:
            Pool status of the miner.
        </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>

        Returns:
            Data on each PGA/ASC with their details.
        </details>
        """
        return await self.send_command("devs")

    async def edevs(self) -> dict:
        """Get data on each PGA/ASC with their details, ignoring blacklisted and zombie devices.
        <details>
            <summary>Expand</summary>

        Returns:
            Data on each PGA/ASC with their details.
        </details>
        """
        return await self.send_command("edevs")

    async def devdetails(self) -> dict:
        """Get data on all devices with their static details.
        <details>
            <summary>Expand</summary>

        Returns:
            Data on all devices with their static details.
        </details>
        """
        return await self.send_command("devdetails")

    async def get_psu(self) -> dict:
        """Get data on the PSU and power information.
        <details>
            <summary>Expand</summary>

        Returns:
            Data on the PSU and power information.
        </details>
        """
        return await self.send_command("get_psu")

    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.

        Returns:
            Version data for the miner.
        </details>
        """
        return await self.get_version()

    async def get_version(self) -> dict:
        """Get version data for the miner.
        <details>
            <summary>Expand</summary>

        Returns:
            Version data for the miner.
        </details>
        """
        return await self.send_command("get_version")

    async def status(self) -> dict:
        """Get BTMiner status and firmware version.
        <details>
            <summary>Expand</summary>

        Returns:
            BTMiner status and firmware version.
        </details>
        """
        return await self.send_command("status")

    async def get_miner_info(self) -> dict:
        """Get general miner info.
        <details>
            <summary>Expand</summary>

        Returns:
            General miner info.
        </details>
        """
        return await self.send_command("get_miner_info", allow_warning=False)

    @api_min_version("2.0.1")
    async def get_error_code(self) -> dict:
        """Get a list of error codes from the miner.

        <details>
            <summary>Expand</summary>
        Get a list of error codes from the miner.  Replaced `summary` as the location of error codes with API version 2.0.4.

        Returns:
            A list of error codes on the miner.
        </details>
        """
        return await self.send_command("get_error_code", allow_warning=False)

adjust_power_limit(power_limit) async

Set the upper limit of the miner's power. Cannot be higher than the ordinary power of the machine.

Expand

Set the upper limit of the miner's power, only works after changing the password of the miner using the Whatsminer tool. The miner will reboot after this is set.

Parameters:

Name	Type	Description	Default
power_limit	int	New power limit.	required

Returns: A reply informing of the status of setting power limit.

Source code in pyasic/rpc/btminer.py

@api_min_version("2.0.5")
async def adjust_power_limit(self, power_limit: int) -> dict:
    """Set the upper limit of the miner's power. Cannot be higher than the ordinary power of the machine.

    <details>
        <summary>Expand</summary>

    Set the upper limit of the miner's power, only works after
    changing the password of the miner using the Whatsminer tool.
    The miner will reboot after this is set.

    Parameters:
        power_limit: New power limit.
    Returns:
        A reply informing of the status of setting power limit.
    </details>

    """
    return await self.send_privileged_command(
        "adjust_power_limit", power_limit=str(power_limit)
    )

adjust_upfreq_speed(upfreq_speed) async

Set the upfreq speed, 0 is the normal speed, 9 is the fastest speed.

Expand

Set the upfreq speed, 0 is the normal speed, 9 is the fastest speed, only works after changing the password of the miner using the Whatsminer tool. The faster the speed, the greater the final hash rate and power deviation, and the stability may be impacted. Fast boot mode cannot be used at the same time.

Parameters:

Name	Type	Description	Default
upfreq_speed	int	New upfreq speed.	required

Returns: A reply informing of the status of setting upfreq speed.

Source code in pyasic/rpc/btminer.py

@api_min_version("2.0.5")
async def adjust_upfreq_speed(self, upfreq_speed: int) -> dict:
    """Set the upfreq speed, 0 is the normal speed, 9 is the fastest speed.

    <details>
        <summary>Expand</summary>

    Set the upfreq speed, 0 is the normal speed, 9 is the fastest speed, only works after
    changing the password of the miner using the Whatsminer tool.
    The faster the speed, the greater the final hash rate and power deviation, and the stability
    may be impacted. Fast boot mode cannot be used at the same time.

    Parameters:
        upfreq_speed: New upfreq speed.
    Returns:
        A reply informing of the status of setting upfreq speed.
    </details>
    """
    if not 0 < upfreq_speed < 9:
        raise APIError(
            f"Upfreq speed is outside of the allowed "
            f"range.  Please set a number between 0 (Normal) and "
            f"9 (Fastest)."
        )
    return await self.send_privileged_command(
        "adjust_upfreq_speed", upfreq_speed=upfreq_speed
    )

devdetails() async

Get data on all devices with their static details.

Expand

Returns:

Type	Description
dict	Data on all devices with their static details.

Source code in pyasic/rpc/btminer.py

async def devdetails(self) -> dict:
    """Get data on all devices with their static details.
    <details>
        <summary>Expand</summary>

    Returns:
        Data on all devices with their static details.
    </details>
    """
    return await self.send_command("devdetails")

devs() async

Get data on each PGA/ASC with their details.

Expand

Returns:

Type	Description
dict	Data on each PGA/ASC with their details.

Source code in pyasic/rpc/btminer.py

async def devs(self) -> dict:
    """Get data on each PGA/ASC with their details.
    <details>
        <summary>Expand</summary>

    Returns:
        Data on each PGA/ASC with their details.
    </details>
    """
    return await self.send_command("devs")

disable_fast_boot() async

Turn off fast boot.

Expand

Turn off fast boot, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of disabling fast boot.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the status of disabling fast boot.
    </details>
    """
    return await self.send_privileged_command("disable_btminer_fast_boot")

disable_web_pools() async

Turn off web pool updates.

Expand

Turn off web pool updates, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of disabling web pools.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the status of disabling web pools.
    </details>
    """
    return await self.send_privileged_command("disable_web_pools")

edevs() async

Get data on each PGA/ASC with their details, ignoring blacklisted and zombie devices.

Expand

Returns:

Type	Description
dict	Data on each PGA/ASC with their details.

Source code in pyasic/rpc/btminer.py

async def edevs(self) -> dict:
    """Get data on each PGA/ASC with their details, ignoring blacklisted and zombie devices.
    <details>
        <summary>Expand</summary>

    Returns:
        Data on each PGA/ASC with their details.
    </details>
    """
    return await self.send_command("edevs")

enable_fast_boot() async

Turn on fast boot.

Expand

Turn on fast boot, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of enabling fast boot.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the status of enabling fast boot.
    </details>
    """
    return await self.send_privileged_command("enable_btminer_fast_boot")

enable_web_pools() async

Turn on web pool updates.

Expand

Turn on web pool updates, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of enabling web pools.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the status of enabling web pools.
    </details>
    """
    return await self.send_privileged_command("enable_web_pools")

factory_reset() async

Reset the miner to factory defaults.

Expand

Returns:

Type	Description
dict	A reply informing of the status of the reset.

Source code in pyasic/rpc/btminer.py

async def factory_reset(self) -> dict:
    """Reset the miner to factory defaults.
    <details>
        <summary>Expand</summary>

    Returns:
        A reply informing of the status of the reset.
    </details>
    """
    return await self.send_privileged_command("factory_reset")

get_error_code() async

Get a list of error codes from the miner.

Expand

Get a list of error codes from the miner. Replaced `summary` as the location of error codes with API version 2.0.4.

Returns:

Type	Description
dict	A list of error codes on the miner.

Source code in pyasic/rpc/btminer.py

@api_min_version("2.0.1")
async def get_error_code(self) -> dict:
    """Get a list of error codes from the miner.

    <details>
        <summary>Expand</summary>
    Get a list of error codes from the miner.  Replaced `summary` as the location of error codes with API version 2.0.4.

    Returns:
        A list of error codes on the miner.
    </details>
    """
    return await self.send_command("get_error_code", allow_warning=False)

get_miner_info() async

Get general miner info.

Expand

Returns:

Type	Description
dict	General miner info.

Source code in pyasic/rpc/btminer.py

async def get_miner_info(self) -> dict:
    """Get general miner info.
    <details>
        <summary>Expand</summary>

    Returns:
        General miner info.
    </details>
    """
    return await self.send_command("get_miner_info", allow_warning=False)

get_psu() async

Get data on the PSU and power information.

Expand

Returns:

Type	Description
dict	Data on the PSU and power information.

Source code in pyasic/rpc/btminer.py

async def get_psu(self) -> dict:
    """Get data on the PSU and power information.
    <details>
        <summary>Expand</summary>

    Returns:
        Data on the PSU and power information.
    </details>
    """
    return await self.send_command("get_psu")

get_token() async

Gets token information from the API.

Expand

Returns:

Type	Description
dict	An encoded token and md5 password, which are used for the privileged API.

Source code in pyasic/rpc/btminer.py

async def get_token(self) -> dict:
    """Gets token information from the API.
    <details>
        <summary>Expand</summary>

    Returns:
        An encoded token and md5 password, which are used for the privileged API.
    </details>
    """
    logging.debug(f"{self} - (Get Token) - Getting token")
    if self.token:
        if self.token["timestamp"] > datetime.datetime.now() - datetime.timedelta(
            minutes=30
        ):
            return self.token

    # get the token
    data = await self.send_command("get_token")

    # encrypt the admin password with the salt
    pwd = _crypt(self.pwd, "$1$" + data["Msg"]["salt"] + "$")
    pwd = pwd.split("$")

    # take the 4th item from the pwd split
    host_passwd_md5 = pwd[3]

    # encrypt the pwd with the time and new salt
    tmp = _crypt(pwd[3] + data["Msg"]["time"], "$1$" + data["Msg"]["newsalt"] + "$")
    tmp = tmp.split("$")

    # take the 4th item from the encrypted pwd split
    host_sign = tmp[3]

    # set the current token
    self.token = {
        "host_sign": host_sign,
        "host_passwd_md5": host_passwd_md5,
        "timestamp": datetime.datetime.now(),
    }
    logging.debug(f"{self} - (Get Token) - Gathered token data: {self.token}")
    return self.token

get_version() async

Get version data for the miner.

Expand

Returns:

Type	Description
dict	Version data for the miner.

Source code in pyasic/rpc/btminer.py

async def get_version(self) -> dict:
    """Get version data for the miner.
    <details>
        <summary>Expand</summary>

    Returns:
        Version data for the miner.
    </details>
    """
    return await self.send_command("get_version")

multicommand(*commands, allow_warning=True) async

Creates and sends multiple commands as one command to the miner.

Parameters:

Name	Type	Description	Default
*commands	str	The commands to send as a multicommand to the miner.	()
allow_warning	bool	A boolean to supress APIWarnings.	True

Source code in pyasic/rpc/btminer.py

async def multicommand(self, *commands: str, allow_warning: bool = True) -> dict:
    """Creates and sends multiple commands as one command to the miner.

    Parameters:
        *commands: The commands to send as a multicommand to the miner.
        allow_warning: A boolean to supress APIWarnings.
    """
    # make sure we can actually run each command, otherwise they will fail
    commands = self._check_commands(*commands)
    # standard multicommand format is "command1+command2"
    # commands starting with "get_" and the "status" command aren't supported, but we can fake that

    split_commands = []

    for command in list(commands):
        if command.startswith("get_") or command == "status":
            commands.remove(command)
            # send seperately and append later
            split_commands.append(command)

    command = "+".join(commands)

    tasks = []
    if len(split_commands) > 0:
        tasks.append(
            asyncio.create_task(
                self._send_split_multicommand(
                    *split_commands, allow_warning=allow_warning
                )
            )
        )
    tasks.append(
        asyncio.create_task(self.send_command(command, allow_warning=allow_warning))
    )

    try:
        all_data = await asyncio.gather(*tasks)
    except APIError:
        return {}

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

    data["multicommand"] = True
    return data

pools() async

Get the pool status from the miner.

Expand

Returns:

Type	Description
dict	Pool status of the miner.

Source code in pyasic/rpc/btminer.py

async def pools(self) -> dict:
    """Get the pool status from the miner.
    <details>
        <summary>Expand</summary>

    Returns:
        Pool status of the miner.
    </details>
    """
    return await self.send_command("pools")

power_off(respbefore=True) async

Power off the miner using the API.

Expand

Power off the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
respbefore	bool	Whether to respond before powering off.	True

Returns: A reply informing of the status of powering off.

Source code in pyasic/rpc/btminer.py

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.

    Parameters:
        respbefore: Whether to respond before powering off.
    Returns:
        A reply informing of the status of powering off.
    </details>
    """
    if respbefore:
        return await self.send_privileged_command("power_off", respbefore="true")
    return await self.send_privileged_command("power_off", respbefore="false")

power_on() async

Power on the miner using the API.

Expand

Power on the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of powering on.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the status of powering on.
    </details>
    """
    return await self.send_privileged_command("power_on")

pre_power_on(complete, msg) async

Configure or check status of pre power on.

Expand

Configure or check status of pre power on, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
complete	bool	check whether pre power on is complete.	required
msg	PrePowerOnMessage	the message to check.	required

Returns:

Type	Description
dict	A reply informing of the status of pre power on.

Source code in pyasic/rpc/btminer.py

async def pre_power_on(self, complete: bool, msg: PrePowerOnMessage) -> 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.

    Parameters:
        complete: check whether pre power on is complete.
        msg: the message to check.

    Returns:
        A reply informing of the status of pre power on.
    </details>
    """

    if msg not in ("wait for adjust temp", "adjust complete", "adjust continue"):
        raise APIError(
            "Message is incorrect, please choose one of "
            '["wait for adjust temp", '
            '"adjust complete", '
            '"adjust continue"]'
        )
    if complete:
        return await self.send_privileged_command(
            "pre_power_on", complete="true", msg=msg
        )
    return await self.send_privileged_command(
        "pre_power_on", complete="false", msg=msg
    )

reboot(timeout=10) async

Reboot the miner using the API.

Expand

Returns:

Type	Description
dict	A reply informing of the status of the reboot.

Source code in pyasic/rpc/btminer.py

async def reboot(self, timeout: int = 10) -> dict:
    """Reboot the miner using the API.
    <details>
        <summary>Expand</summary>

    Returns:
        A reply informing of the status of the reboot.
    </details>
    """
    try:
        d = await asyncio.wait_for(
            self.send_privileged_command("reboot"), timeout=timeout
        )
    except (asyncio.CancelledError, asyncio.TimeoutError):
        return {}
    else:
        return d

reset_led() async

Reset the LED on the miner using the API.

Expand

Reset the LED on the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of resetting the LED.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the status of resetting the LED.
    </details>
    """
    return await self.set_led(auto=True)

restart() async

Restart BTMiner using the API.

Expand

Restart BTMiner using the API, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the restart.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the restart.
    </details>
    """
    return await self.send_privileged_command("restart_btminer")

set_fan_zero_speed(fan_zero_speed) async

Sets whether the fan speed supports the lowest 0 speed.

Expand

Sets whether the fan speed supports the lowest 0 speed, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
fan_zero_speed	bool	Whether the fan is allowed to support 0 speed.	required

Returns: A reply informing of the status of setting fan minimum speed.

Source code in pyasic/rpc/btminer.py

@api_min_version("2.0.5")
async def set_fan_zero_speed(self, fan_zero_speed: bool) -> dict:
    """Sets whether the fan speed supports the lowest 0 speed.

    <details>
        <summary>Expand</summary>

    Sets whether the fan speed supports the lowest 0 speed, only works after
    changing the password of the miner using the Whatsminer tool.

    Parameters:
        fan_zero_speed: Whether the fan is allowed to support 0 speed.
    Returns:
        A reply informing of the status of setting fan minimum speed.
    </details>

    """
    return await self.send_privileged_command(
        "set_fan_zero_speed", fan_zero_speed=int(fan_zero_speed)
    )

set_high_power() async

Set low power mode on the miner using the API.

Expand

Set low power mode on the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of setting low power mode.

Source code in pyasic/rpc/btminer.py

async def set_high_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.

    Returns:
        A reply informing of the status of setting low power mode.
    </details>
    """
    return await self.send_privileged_command("set_high_power")

set_hostname(hostname) async

Set the hostname of the miner.

Expand

Set the hostname of the miner, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
hostname	str	The new hostname to use.	required

Returns: A reply informing of the status of setting the hostname.

Source code in pyasic/rpc/btminer.py

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:
        A reply informing of the status of setting the hostname.
    </details>
    """
    return await self.send_privileged_command("set_hostname", hostname=hostname)

set_led(auto=True, color='red', period=60, duration=20, start=0) async

Set the LED on the miner using the API.

Expand

Set the LED on the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
auto	bool	Whether or not to reset the LED to auto mode.	True
color	str	The LED color to set, either 'red' or 'green'.	'red'
period	int	The flash cycle in ms.	60
duration	int	LED on time in the cycle in ms.	20
start	int	LED on time offset in the cycle in ms.	0

Returns: A reply informing of the status of setting the LED.

Source code in pyasic/rpc/btminer.py

async def set_led(
    self,
    auto: bool = True,
    color: str = "red",
    period: int = 60,
    duration: int = 20,
    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.

    Parameters:
        auto: Whether or not to reset the LED to auto mode.
        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>
    """
    if auto:
        return await self.send_privileged_command("set_led", param="auto")
    return await self.send_privileged_command(
        "set_led", color=color, period=period, duration=duration, start=start
    )

set_low_power() async

Set low power mode on the miner using the API.

Expand

Set low power mode on the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of setting low power mode.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        A reply informing of the status of setting low power mode.
    </details>
    """
    return await self.send_privileged_command("set_low_power")

set_normal_power() async

Set low power mode on the miner using the API.

Expand

Set low power mode on the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Returns:

Type	Description
dict	A reply informing of the status of setting low power mode.

Source code in pyasic/rpc/btminer.py

async def set_normal_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.

    Returns:
        A reply informing of the status of setting low power mode.
    </details>
    """
    return await self.send_privileged_command("set_normal_power")

set_power_pct(percent) async

Set the power percentage of the miner based on current power. Used for temporary adjustment.

Expand

Set the power percentage of the miner, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
percent	int	The power percentage to set.	required

Returns: A reply informing of the status of setting the power percentage.

Source code in pyasic/rpc/btminer.py

async def set_power_pct(self, percent: int) -> dict:
    """Set the power percentage of the miner based on current power.  Used for temporary adjustment.

    <details>
        <summary>Expand</summary>

    Set the power percentage of the miner, only works after changing
    the password of the miner using the Whatsminer tool.

    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:
        raise APIError(
            f"Power PCT % is outside of the allowed "
            f"range.  Please set a % between 0 and "
            f"100"
        )
    return await self.send_privileged_command("set_power_pct", percent=str(percent))

set_power_pct_v2(percent) async

Set the power percentage of the miner based on current power. Used for temporary adjustment. Added in API v2.0.5.

Expand

Set the power percentage of the miner, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
percent	int	The power percentage to set.	required

Returns: A reply informing of the status of setting the power percentage.

Source code in pyasic/rpc/btminer.py

@api_min_version("2.0.5")
async def set_power_pct_v2(self, percent: int) -> dict:
    """Set the power percentage of the miner based on current power.  Used for temporary adjustment.  Added in API v2.0.5.

    <details>
        <summary>Expand</summary>

    Set the power percentage of the miner, only works after changing
    the password of the miner using the Whatsminer tool.

    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:
        raise APIError(
            f"Power PCT % is outside of the allowed "
            f"range.  Please set a % between 0 and "
            f"100"
        )
    return await self.send_privileged_command(
        "set_power_pct_v2", percent=str(percent)
    )

set_poweroff_cool(poweroff_cool) async

Set whether to cool the machine when mining is stopped.

Expand

Set whether to cool the machine when mining is stopped, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
poweroff_cool	bool	Whether to cool the miner during power off mode.	required

Returns: A reply informing of the status of setting power off cooling mode.

Source code in pyasic/rpc/btminer.py

@api_min_version("2.0.5")
async def set_poweroff_cool(self, poweroff_cool: bool) -> dict:
    """Set whether to cool the machine when mining is stopped.

    <details>
        <summary>Expand</summary>

    Set whether to cool the machine when mining is stopped, only works after
    changing the password of the miner using the Whatsminer tool.

    Parameters:
        poweroff_cool: Whether to cool the miner during power off mode.
    Returns:
        A reply informing of the status of setting power off cooling mode.
    </details>
    """

    return await self.send_privileged_command(
        "set_poweroff_cool", poweroff_cool=int(poweroff_cool)
    )

set_target_freq(percent) async

Update the target frequency.

Expand

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%.

Parameters:

Name	Type	Description	Default
percent	int	The frequency % to set.	required

Returns: A reply informing of the status of setting the frequency.

Source code in pyasic/rpc/btminer.py

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%.

    Parameters:
        percent: The frequency % to set.
    Returns:
        A reply informing of the status of setting the frequency.
    </details>
    """
    if not -100 < percent < 100:
        raise APIError(
            f"Frequency % is outside of the allowed "
            f"range.  Please set a % between -100 and "
            f"100"
        )
    return await self.send_privileged_command(
        "set_target_freq", percent=str(percent)
    )

set_temp_offset(temp_offset) async

Set the offset of miner hash board target temperature.

Expand

Set the offset of miner hash board target temperature, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
temp_offset	int	Target temperature offset.	required

Returns: A reply informing of the status of setting temp offset.

Source code in pyasic/rpc/btminer.py

@api_min_version("2.0.5")
async def set_temp_offset(self, temp_offset: int) -> dict:
    """Set the offset of miner hash board target temperature.

    <details>
        <summary>Expand</summary>

    Set the offset of miner hash board target temperature, only works after
    changing the password of the miner using the Whatsminer tool.

    Parameters:
        temp_offset: Target temperature offset.
    Returns:
        A reply informing of the status of setting temp offset.
    </details>

    """
    if not -30 < temp_offset < 0:
        raise APIError(
            f"Temp offset is outside of the allowed "
            f"range.  Please set a number between -30 and "
            f"0."
        )

    return await self.send_privileged_command(
        "set_temp_offset", temp_offset=temp_offset
    )

status() async

Get BTMiner status and firmware version.

Expand

Returns:

Type	Description
dict	BTMiner status and firmware version.

Source code in pyasic/rpc/btminer.py

async def status(self) -> dict:
    """Get BTMiner status and firmware version.
    <details>
        <summary>Expand</summary>

    Returns:
        BTMiner status and firmware version.
    </details>
    """
    return await self.send_command("status")

summary() async

Get the summary status from the miner.

Expand

Returns:

Type	Description
dict	Summary status of the miner.

Source code in pyasic/rpc/btminer.py

async def summary(self) -> dict:
    """Get the summary status from the miner.
    <details>
        <summary>Expand</summary>

    Returns:
        Summary status of the miner.
    </details>
    """
    return await self.send_command("summary")

update_firmware(firmware) async

Upgrade the firmware running on the miner and using the firmware passed in bytes.

Expand

Set the LED on the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
firmware	bytes	The firmware binary data to be uploaded.	required

Returns: A boolean indicating the success of the firmware upgrade. Raises: APIError: If the miner is not ready for firmware update.

Source code in pyasic/rpc/btminer.py

async def update_firmware(self, firmware: bytes) -> bool:
    """Upgrade the firmware running on the miner and using the firmware passed in bytes.
    <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.

    Parameters:
        firmware (bytes): The firmware binary data to be uploaded.
    Returns:
        A boolean indicating the success of the firmware upgrade.
    Raises:
        APIError: If the miner is not ready for firmware update.
    </details>
    """
    ready = await self.send_privileged_command("upgrade_firmware")
    if not ready.get("Msg") == "ready":
        raise APIError(f"Not ready for firmware update: {self}")
    file_size = struct.pack("<I", len(firmware))
    await self._send_bytes(file_size + firmware)
    return True

update_pools(pool_1, worker_1, passwd_1, pool_2=None, worker_2=None, passwd_2=None, pool_3=None, worker_3=None, passwd_3=None) async

Update the pools of the miner using the API.

Expand

Update the pools of the miner using the API, only works after changing the password of the miner using the Whatsminer tool.

Parameters:

Name	Type	Description	Default
pool_1	str	The URL to update pool 1 to.	required
worker_1	str	The worker name for pool 1 to update to.	required
passwd_1	str	The password for pool 1 to update to.	required
pool_2	str	The URL to update pool 2 to.	None
worker_2	str	The worker name for pool 2 to update to.	None
passwd_2	str	The password for pool 2 to update to.	None
pool_3	str	The URL to update pool 3 to.	None
worker_3	str	The worker name for pool 3 to update to.	None
passwd_3	str	The password for pool 3 to update to.	None

Returns:

Type	Description
dict	A dict from the API to confirm the pools were updated.

Source code in pyasic/rpc/btminer.py

async def update_pools(
    self,
    pool_1: str,
    worker_1: str,
    passwd_1: str,
    pool_2: str = None,
    worker_2: str = None,
    passwd_2: str = None,
    pool_3: str = None,
    worker_3: str = None,
    passwd_3: str = None,
) -> 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.

    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>
    """
    return await self.send_privileged_command(
        "update_pools",
        pool1=pool_1,
        worker1=worker_1,
        passwd1=passwd_1,
        pool2=pool_2,
        worker2=worker_2,
        passwd2=passwd_2,
        pool3=pool_3,
        worker3=worker_3,
        passwd3=passwd_3,
    )

update_pwd(old_pwd, new_pwd) async

Update the admin user's password.

Expand

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.

Parameters:

Name	Type	Description	Default
old_pwd	str	The old admin password.	required
new_pwd	str	The new password to set.	required

Returns: A reply informing of the status of setting the password.

Source code in pyasic/rpc/btminer.py

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.

    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.
    """
    self.pwd = old_pwd
    # check if password length is greater than 8 bytes
    if len(new_pwd.encode("utf-8")) > 8:
        raise APIError(
            f"New password too long, the max length is 8.  "
            f"Password size: {len(new_pwd.encode('utf-8'))}"
        )
    try:
        data = await self.send_privileged_command(
            "update_pwd", old=old_pwd, new=new_pwd
        )
    except APIError as e:
        raise e
    self.pwd = new_pwd
    return data

version() async

Get version data for the miner. Wraps self.get_version().

Expand

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.

Returns:

Type	Description
dict	Version data for the miner.

Source code in pyasic/rpc/btminer.py

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.

    Returns:
        Version data for the miner.
    </details>
    """
    return await self.get_version()