Merge pull request #166 from UpstreamData/dev_bitaxe

feature: Add basic bitaxe support.

pyasic/config/mining/algo.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from dataclasses import dataclass, field
 
 from pyasic.config.base import MinerConfigOption, MinerConfigValue

pyasic/device/makes.py

@@ -25,6 +25,7 @@ class MinerMake(str, Enum):
     GOLDSHELL = "Goldshell"
     AURADINE = "Auradine"
     EPIC = "ePIC"
+    BITAXE = "BitAxe"
 
     def __str__(self):
         return self.value

pyasic/device/models.py

@@ -329,6 +329,15 @@ class AuradineModels(str, Enum):
         return self.value
 
 
+class BitAxeModels(str, Enum):
+    BM1366 = "Ultra"
+    BM1368 = "Supra"
+    BM1397 = "Max"
+
+    def __str__(self):
+        return self.value
+
+
 class MinerModel:
     ANTMINER = AntminerModels
     WHATSMINER = WhatsminerModels
@@ -337,3 +346,4 @@ class MinerModel:
     GOLDSHELL = GoldshellModels
     AURADINE = AuradineModels
     EPIC = ePICModels
+    BITAXE = BitAxeModels

pyasic/miners/backends/bitaxe.py

@@ -0,0 +1,175 @@
+from typing import List, Optional
+
+from pyasic import APIError
+from pyasic.data import AlgoHashRate, Fan, HashBoard, HashUnit
+from pyasic.miners.base import BaseMiner
+from pyasic.miners.data import DataFunction, DataLocations, DataOptions, WebAPICommand
+from pyasic.web.bitaxe import BitAxeWebAPI
+
+BITAXE_DATA_LOC = DataLocations(
+    **{
+        str(DataOptions.HASHRATE): DataFunction(
+            "_get_hashrate",
+            [WebAPICommand("web_system_info", "system/info")],
+        ),
+        str(DataOptions.WATTAGE): DataFunction(
+            "_get_wattage",
+            [WebAPICommand("web_system_info", "system/info")],
+        ),
+        str(DataOptions.UPTIME): DataFunction(
+            "_get_uptime",
+            [WebAPICommand("web_system_info", "system/info")],
+        ),
+        str(DataOptions.HASHBOARDS): DataFunction(
+            "_get_hashboards",
+            [WebAPICommand("web_system_info", "system/info")],
+        ),
+        str(DataOptions.FANS): DataFunction(
+            "_get_fans",
+            [WebAPICommand("web_system_info", "system/info")],
+        ),
+        str(DataOptions.FW_VERSION): DataFunction(
+            "_get_fw_ver",
+            [WebAPICommand("web_system_info", "system/info")],
+        ),
+        str(DataOptions.API_VERSION): DataFunction(
+            "_get_api_ver",
+            [WebAPICommand("web_system_info", "system/info")],
+        ),
+    }
+)
+
+
+class BitAxe(BaseMiner):
+    """Handler for BitAxe"""
+
+    web: BitAxeWebAPI
+    _web_cls = BitAxeWebAPI
+
+    data_locations = BITAXE_DATA_LOC
+
+    async def reboot(self) -> bool:
+        await self.web.restart()
+        return True
+
+    async def _get_wattage(self, web_system_info: dict = None) -> Optional[int]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+        if web_system_info is not None:
+            try:
+                return round(web_system_info["power"])
+            except KeyError:
+                pass
+
+    async def _get_hashrate(
+        self, web_system_info: dict = None
+    ) -> Optional[AlgoHashRate]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+
+        if web_system_info is not None:
+            try:
+                return AlgoHashRate.SHA256(
+                    web_system_info["hashRate"], HashUnit.SHA256.GH
+                ).into(self.algo.unit.default)
+            except KeyError:
+                pass
+
+    async def _get_uptime(self, web_system_info: dict = None) -> Optional[int]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+
+        if web_system_info is not None:
+            try:
+                return web_system_info["uptimeSeconds"]
+            except KeyError:
+                pass
+
+    async def _get_hashboards(self, web_system_info: dict = None) -> List[HashBoard]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+
+        if web_system_info is not None:
+            try:
+                return [
+                    HashBoard(
+                        hashrate=AlgoHashRate.SHA256(
+                            web_system_info["hashRate"], HashUnit.SHA256.GH
+                        ).into(self.algo.unit.default),
+                        chip_temp=web_system_info.get("temp"),
+                        temp=web_system_info.get("vrTemp"),
+                        chips=web_system_info.get("asicCount"),
+                        expected_chips=self.expected_chips,
+                        missing=False,
+                        active=True,
+                        voltage=web_system_info.get("voltage"),
+                    )
+                ]
+            except KeyError:
+                pass
+        return []
+
+    async def _get_fans(self, web_system_info: dict = None) -> List[Fan]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+
+        if web_system_info is not None:
+            try:
+                return [Fan(speed=web_system_info["fanrpm"])]
+            except KeyError:
+                pass
+        return []
+
+    async def _get_hostname(self, web_system_info: dict = None) -> Optional[str]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+
+        if web_system_info is not None:
+            try:
+                return web_system_info["hostname"]
+            except KeyError:
+                pass
+
+    async def _get_api_ver(self, web_system_info: dict = None) -> Optional[str]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+
+        if web_system_info is not None:
+            try:
+                return web_system_info["version"]
+            except KeyError:
+                pass
+
+    async def _get_fw_ver(self, web_system_info: dict = None) -> Optional[str]:
+        if web_system_info is None:
+            try:
+                web_system_info = await self.web.system_info()
+            except APIError:
+                pass
+
+        if web_system_info is not None:
+            try:
+                return web_system_info["version"]
+            except KeyError:
+                pass

pyasic/miners/backends/whatsminer.py

@@ -29,4 +29,4 @@ class M3X(BTMiner):
 
 
 class M2X(BTMiner):
-    pass
+    pass

pyasic/miners/bitaxe/__init__.py

@@ -0,0 +1 @@
+from .espminer import *

pyasic/miners/bitaxe/espminer/BM/BM1366.py

@@ -0,0 +1,6 @@
+from pyasic.miners.backends.bitaxe import BitAxe
+from pyasic.miners.device.models.bitaxe import Ultra
+
+
+class BitAxeUltra(BitAxe, Ultra):
+    pass

pyasic/miners/bitaxe/espminer/BM/BM1368.py

@@ -0,0 +1,6 @@
+from pyasic.miners.backends.bitaxe import BitAxe
+from pyasic.miners.device.models.bitaxe import Supra
+
+
+class BitAxeSupra(BitAxe, Supra):
+    pass

pyasic/miners/bitaxe/espminer/BM/BM1397.py

@@ -0,0 +1,6 @@
+from pyasic.miners.backends.bitaxe import BitAxe
+from pyasic.miners.device.models.bitaxe import Max
+
+
+class BitAxeMax(BitAxe, Max):
+    pass

pyasic/miners/bitaxe/espminer/BM/__init__.py

@@ -0,0 +1,3 @@
+from .BM1366 import BitAxeUltra
+from .BM1368 import BitAxeSupra
+from .BM1397 import BitAxeMax

pyasic/miners/bitaxe/espminer/__init__.py

@@ -0,0 +1 @@
+from .BM import *

pyasic/miners/device/makes.py

@@ -44,3 +44,7 @@ class AuradineMake(BaseMiner):
 
 class ePICMake(BaseMiner):
     make = MinerMake.EPIC
+
+
+class BitAxeMake(BaseMiner):
+    make = MinerMake.BITAXE

pyasic/miners/device/models/bitaxe/BM/BM1366.py

@@ -0,0 +1,10 @@
+from pyasic.device.models import MinerModel
+from pyasic.miners.device.makes import BitAxeMake
+
+
+class Ultra(BitAxeMake):
+    raw_model = MinerModel.BITAXE.BM1366
+
+    expected_hashboards = 1
+    expected_chips = 1
+    expected_fans = 1

pyasic/miners/device/models/bitaxe/BM/BM1368.py

@@ -0,0 +1,10 @@
+from pyasic.device.models import MinerModel
+from pyasic.miners.device.makes import BitAxeMake
+
+
+class Supra(BitAxeMake):
+    raw_model = MinerModel.BITAXE.BM1368
+
+    expected_hashboards = 1
+    expected_chips = 1
+    expected_fans = 1

pyasic/miners/device/models/bitaxe/BM/BM1397.py

@@ -0,0 +1,10 @@
+from pyasic.device.models import MinerModel
+from pyasic.miners.device.makes import BitAxeMake
+
+
+class Max(BitAxeMake):
+    raw_model = MinerModel.BITAXE.BM1397
+
+    expected_hashboards = 1
+    expected_chips = 1
+    expected_fans = 1

pyasic/miners/device/models/bitaxe/BM/__init__.py

@@ -0,0 +1,3 @@
+from .BM1366 import Ultra
+from .BM1368 import Supra
+from .BM1397 import Max

pyasic/miners/device/models/bitaxe/__init__.py

@@ -0,0 +1 @@
+from .BM import *

pyasic/miners/factory.py

@@ -31,8 +31,10 @@ from pyasic.miners.antminer import *
 from pyasic.miners.auradine import *
 from pyasic.miners.avalonminer import *
 from pyasic.miners.backends import *
+from pyasic.miners.backends.bitaxe import BitAxe
 from pyasic.miners.backends.unknown import UnknownMiner
 from pyasic.miners.base import AnyMiner
+from pyasic.miners.bitaxe import *
 from pyasic.miners.blockminer import *
 from pyasic.miners.device.makes import *
 from pyasic.miners.goldshell import *
@@ -53,6 +55,7 @@ class MinerTypes(enum.Enum):
     EPIC = 9
     AURADINE = 10
     MARATHON = 11
+    BITAXE = 12
 
 
 MINER_CLASSES = {
@@ -438,6 +441,12 @@ MINER_CLASSES = {
         "ANTMINER S21": MaraS21,
         "ANTMINER T21": MaraT21,
     },
+    MinerTypes.BITAXE: {
+        None: BitAxe,
+        "SUPRA": BitAxeSupra,
+        "ULTRA": BitAxeUltra,
+        "MAX": BitAxeMax,
+    },
 }
 
 
@@ -514,6 +523,7 @@ class MinerFactory:
                 MinerTypes.LUX_OS: self.get_miner_model_luxos,
                 MinerTypes.AURADINE: self.get_miner_model_auradine,
                 MinerTypes.MARATHON: self.get_miner_model_marathon,
+                MinerTypes.BITAXE: self.get_miner_model_bitaxe,
             }
             fn = miner_model_fns.get(miner_type)
 
@@ -595,6 +605,8 @@ class MinerFactory:
                 return MinerTypes.WHATSMINER
         if "Braiins OS" in web_text:
             return MinerTypes.BRAIINS_OS
+        if "AxeOS" in web_text:
+            return MinerTypes.BITAXE
         if "cloud-box" in web_text:
             return MinerTypes.GOLDSHELL
         if "AnthillOS" in web_text:
@@ -1008,6 +1020,18 @@ class MinerFactory:
         except (TypeError, LookupError):
             pass
 
+    async def get_miner_model_bitaxe(self, ip: str) -> str | None:
+        web_json_data = await self.send_web_command(ip, "/api/system/info")
+
+        try:
+            miner_model = web_json_data["devicemodel"]
+            if miner_model == "":
+                return None
+
+            return miner_model
+        except (TypeError, LookupError):
+            pass
+
 
 miner_factory = MinerFactory()
 

pyasic/ssh/braiins_os.py

@@ -92,4 +92,4 @@ class BOSMinerSSH(BaseSSH):
         Returns:
             str: Status of the LED.
         """
-        return await self.send_command("cat /sys/class/leds/'Red LED'/delay_off")
+        return await self.send_command("cat /sys/class/leds/'Red LED'/delay_off")

pyasic/web/bitaxe.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from typing import Any
+
+import httpx
+
+from pyasic import APIError, settings
+from pyasic.web.base import BaseWebAPI
+
+
+class BitAxeWebAPI(BaseWebAPI):
+    async def send_command(
+        self,
+        command: str | bytes,
+        ignore_errors: bool = False,
+        allow_warning: bool = True,
+        privileged: bool = False,
+        **parameters: Any,
+    ) -> dict:
+        url = f"http://{self.ip}:{self.port}/api/{command}"
+        try:
+            async with httpx.AsyncClient(
+                transport=settings.transport(),
+            ) as client:
+                if parameters.get("post", False):
+                    data = await client.post(
+                        url,
+                        timeout=settings.get("api_function_timeout", 3),
+                        json=parameters,
+                    )
+                else:
+                    data = await client.get(url)
+        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 on the BitAxe miner.
+
+        Args:
+            *commands (str): Commands to execute.
+            ignore_errors (bool): Whether to ignore errors during command execution.
+            allow_warning (bool): Whether to proceed despite warnings.
+
+        Returns:
+            dict: A dictionary containing responses for all commands executed.
+        """
+        tasks = {}
+        # send all commands individually
+        for cmd in commands:
+            tasks[cmd] = asyncio.create_task(
+                self.send_command(cmd, allow_warning=allow_warning)
+            )
+
+        await asyncio.gather(*[tasks[cmd] for cmd in tasks], return_exceptions=True)
+
+        data = {"multicommand": True}
+        for cmd in tasks:
+            try:
+                result = tasks[cmd].result()
+                if result is None or result == {}:
+                    result = {}
+                data[cmd] = result
+            except APIError:
+                pass
+
+        return data
+
+    async def system_info(self):
+        return await self.send_command("system/info")
+
+    async def swarm_info(self):
+        return await self.send_command("swarm/info")
+
+    async def restart(self):
+        return await self.send_command("system/restart", post=True)