bump version number.

README.md

@@ -5,6 +5,8 @@
 [![pypi](https://img.shields.io/pypi/v/pyasic.svg)](https://pypi.org/project/pyasic/)
 [![python](https://img.shields.io/pypi/pyversions/pyasic.svg)](https://pypi.org/project/pyasic/)
 [![Read the Docs](https://img.shields.io/readthedocs/pyasic)](https://pyasic.readthedocs.io/en/latest/)
+[![GitHub](https://img.shields.io/github/license/UpstreamData/pyasic)](https://github.com/UpstreamData/pyasic/blob/master/LICENSE.txt)
+[![CodeFactor Grade](https://img.shields.io/codefactor/grade/github/UpstreamData/pyasic)](https://www.codefactor.io/repository/github/upstreamdata/pyasic)
 ## Documentation
 Documentation is located on Read the Docs as [pyasic](https://pyasic.readthedocs.io/en/latest/)
 

docs/index.md

@@ -5,6 +5,8 @@
 [![pypi](https://img.shields.io/pypi/v/pyasic.svg)](https://pypi.org/project/pyasic/)
 [![python](https://img.shields.io/pypi/pyversions/pyasic.svg)](https://pypi.org/project/pyasic/)
 [![Read the Docs](https://img.shields.io/readthedocs/pyasic)](https://pyasic.readthedocs.io/en/latest/)
+[![GitHub](https://img.shields.io/github/license/UpstreamData/pyasic)](https://github.com/UpstreamData/pyasic/blob/master/LICENSE.txt)
+[![CodeFactor Grade](https://img.shields.io/codefactor/grade/github/UpstreamData/pyasic)](https://www.codefactor.io/repository/github/upstreamdata/pyasic)
 
 ## Intro
 Welcome to pyasic!  Pyasic uses an asynchronous method of communicating with asic miners on your network, which makes it super fast.

docs/miners/supported_types.md

@@ -27,6 +27,8 @@ Supported miner types are here on this list.  If your miner (or miner version) i
             * [VE20][pyasic.miners.whatsminer.btminer.M3X.M31S_Plus.BTMinerM31SPlusVE20]
         * [M32S][pyasic.miners.whatsminer.btminer.M3X.M32S.BTMinerM32S]
     * M2X Series:
+        * [M20][pyasic.miners.whatsminer.btminer.M2X.M20.BTMinerM20]:
+            * [V10][pyasic.miners.whatsminer.btminer.M2X.M20.BTMinerM20V10]
         * [M20S][pyasic.miners.whatsminer.btminer.M2X.M20S.BTMinerM20S]:
             * [V10][pyasic.miners.whatsminer.btminer.M2X.M20S.BTMinerM20SV10]
             * [V20][pyasic.miners.whatsminer.btminer.M2X.M20S.BTMinerM20SV20]

docs/miners/whatsminer/M2X.md

@@ -1,6 +1,22 @@
 # pyasic
 ## M2X Models
 
+## M20
+
+::: pyasic.miners.whatsminer.btminer.M2X.M20.BTMinerM20
+    handler: python
+    options:
+        show_root_heading: false
+        heading_level: 4
+
+## M20V10
+
+::: pyasic.miners.whatsminer.btminer.M2X.M20.BTMinerM20V10
+    handler: python
+    options:
+        show_root_heading: false
+        heading_level: 4
+
 ## M20S
 
 ::: pyasic.miners.whatsminer.btminer.M2X.M20S.BTMinerM20S

pyasic/API/__init__.py

@@ -111,7 +111,7 @@ If you are sure you want to use this command please use API.send_command("{comma
             data = await self.send_command(command, x19_command=ignore_x19_error)
         except APIError:
             logging.debug(f"{self.ip}: Handling X19 multicommand.")
-            data = await self._x19_multicommand(command.split("+"))
+            data = await self._x19_multicommand(*command.split("+"))
         logging.debug(f"{self.ip}: Received multicommand data.")
         return data
 

pyasic/data/__init__.py

@@ -12,6 +12,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+from typing import Union
 from dataclasses import dataclass, field, asdict
 from datetime import datetime
 
@@ -48,7 +49,7 @@ class MinerData:
         right_chips: The number of chips online in the left board as an int.
         total_chips: The total number of chips on all boards.  Calculated automatically.
         ideal_chips: The ideal number of chips in the miner as an int.
-        perecent_ideal: The percent of total chips out of the ideal count.  Calculated automatically.
+        percent_ideal: The percent of total chips out of the ideal count.  Calculated automatically.
         nominal: The nominal amount of chips in the miner.  Calculated automatically.
         pool_split: The pool split as a str.
         pool_1_url: The first pool url on the miner as a str.
@@ -56,6 +57,7 @@ class MinerData:
         pool_2_url: The second pool url on the miner as a str.
         pool_2_user: The second pool user on the miner as a str.
         errors: A list of errors on the miner.
+        fault_light: Whether or not the fault light is on as a boolean.
     """
 
     ip: str
@@ -94,6 +96,7 @@ class MinerData:
     pool_2_url: str = ""
     pool_2_user: str = ""
     errors: list = field(default_factory=list)
+    fault_light: Union[bool, None] = None
 
     def __post_init__(self):
         self.datetime = datetime.now()

pyasic/miners/_backends/bmminer.py

@@ -178,6 +178,8 @@ class BMMiner(BaseMiner):
         if mac:
             data.mac = mac
 
+        data.fault_light = await self.check_light()
+
         miner_data = None
         for i in range(PyasicSettings().miner_get_data_retries):
             miner_data = await self.api.multicommand(

pyasic/miners/_backends/bosminer.py

@@ -264,6 +264,8 @@ class BOSMiner(BaseMiner):
         if mac:
             data.mac = mac
 
+        data.fault_light = await self.check_light()
+
         miner_data = None
         for i in range(PyasicSettings().miner_get_data_retries):
             try:

pyasic/miners/_backends/btminer.py

@@ -129,6 +129,8 @@ class BTMiner(BaseMiner):
         if hostname:
             data.hostname = hostname
 
+        data.fault_light = await self.check_light()
+
         miner_data = None
         for i in range(PyasicSettings().miner_get_data_retries):
             try:

pyasic/miners/_backends/cgminer.py

@@ -175,6 +175,8 @@ class CGMiner(BaseMiner):
         if mac:
             data.mac = mac
 
+        data.fault_light = await self.check_light()
+
         miner_data = None
         for i in range(PyasicSettings().miner_get_data_retries):
             miner_data = await self.api.multicommand(

pyasic/miners/_types/whatsminer/M2X/M20.py

@@ -0,0 +1,33 @@
+#  Copyright 2022 Upstream Data Inc
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from pyasic.miners import BaseMiner
+
+
+class M20(BaseMiner):
+    def __init__(self, ip: str):
+        super().__init__()
+        self.ip = ip
+        self.model = "M20"
+        self.nominal_chips = 70
+        self.fan_count = 2
+
+
+class M20V10(BaseMiner):
+    def __init__(self, ip: str):
+        super().__init__()
+        self.ip = ip
+        self.model = "M20 V10"
+        self.nominal_chips = 70
+        self.fan_count = 2

pyasic/miners/_types/whatsminer/M2X/M20S.py

@@ -28,7 +28,7 @@ class M20SV10(BaseMiner):
     def __init__(self, ip: str):
         super().__init__()
         self.ip = ip
-        self.model = "M20S"
+        self.model = "M20S V10"
         self.nominal_chips = 105
         self.fan_count = 2
 
@@ -37,6 +37,6 @@ class M20SV20(BaseMiner):
     def __init__(self, ip: str):
         super().__init__()
         self.ip = ip
-        self.model = "M20S"
+        self.model = "M20S V20"
         self.nominal_chips = 111
         self.fan_count = 2

pyasic/miners/_types/whatsminer/M2X/__init__.py

@@ -12,6 +12,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+from .M20 import M20, M20V10
 from .M20S import M20S, M20SV10, M20SV20
 from .M20S_Plus import M20SPlus
 

pyasic/miners/antminer/bmminer/X17/X17.py

@@ -15,6 +15,7 @@
 from pyasic.miners._backends import BMMiner  # noqa - Ignore access to _module
 
 import httpx
+from typing import Union
 
 
 class BMMinerX17(BMMiner):
@@ -22,7 +23,7 @@ class BMMinerX17(BMMiner):
         super().__init__(ip)
         self.ip = ip
 
-    async def get_hostname(self) -> str or None:
+    async def get_hostname(self) -> Union[str, None]:
         hostname = None
         url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
         auth = httpx.DigestAuth("root", "root")
@@ -35,7 +36,7 @@ class BMMinerX17(BMMiner):
                     hostname = data["hostname"]
         return hostname
 
-    async def get_mac(self):
+    async def get_mac(self) -> Union[str, None]:
         mac = None
         url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
         auth = httpx.DigestAuth("root", "root")
@@ -61,6 +62,7 @@ class BMMinerX17(BMMiner):
         if data.status_code == 200:
             data = data.json()
             if data["isBlinking"]:
+                self.light = True
                 return True
         return False
 
@@ -73,10 +75,13 @@ class BMMinerX17(BMMiner):
         if data.status_code == 200:
             data = data.json()
             if not data["isBlinking"]:
+                self.light = False
                 return True
         return False
 
-    async def check_light(self):
+    async def check_light(self) -> Union[bool, None]:
+        if self.light:
+            return self.light
         url = f"http://{self.ip}/cgi-bin/blink.cgi"
         auth = httpx.DigestAuth("root", "root")
         async with httpx.AsyncClient() as client:
@@ -84,8 +89,12 @@ class BMMinerX17(BMMiner):
         if data.status_code == 200:
             data = data.json()
             if data["isBlinking"]:
+                self.light = True
                 return True
+            else:
+                self.light = False
                 return False
+        return None
 
     async def reboot(self) -> bool:
         url = f"http://{self.ip}/cgi-bin/reboot.cgi"

pyasic/miners/antminer/bmminer/X19/X19.py

@@ -19,6 +19,7 @@ from pyasic.config import MinerConfig
 import httpx
 import json
 import asyncio
+from typing import Union
 
 
 class BMMinerX19(BMMiner):
@@ -26,6 +27,20 @@ class BMMinerX19(BMMiner):
         super().__init__(ip)
         self.ip = ip
 
+    async def check_light(self) -> Union[bool, None]:
+        if self.light:
+            return self.light
+        url = f"http://{self.ip}/cgi-bin/get_blink_status.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        async with httpx.AsyncClient() as client:
+            data = await client.get(url, auth=auth)
+        if data.status_code == 200:
+            data = data.json()
+            light = data["blink"]
+            self.light = light
+            return light
+        return None
+
     async def get_config(self) -> MinerConfig:
         url = f"http://{self.ip}/cgi-bin/get_miner_conf.cgi"
         auth = httpx.DigestAuth("root", "root")
@@ -56,7 +71,7 @@ class BMMinerX19(BMMiner):
                 break
             await asyncio.sleep(1)
 
-    async def get_hostname(self) -> str or None:
+    async def get_hostname(self) -> Union[str, None]:
         hostname = None
         url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
         auth = httpx.DigestAuth("root", "root")
@@ -69,7 +84,7 @@ class BMMinerX19(BMMiner):
                     hostname = data["hostname"]
         return hostname
 
-    async def get_mac(self):
+    async def get_mac(self) -> Union[str, None]:
         mac = None
         url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
         auth = httpx.DigestAuth("root", "root")
@@ -91,6 +106,7 @@ class BMMinerX19(BMMiner):
         if data.status_code == 200:
             data = data.json()
             if data.get("code") == "B000":
+                self.light = True
                 return True
         return False
 
@@ -103,6 +119,7 @@ class BMMinerX19(BMMiner):
         if data.status_code == 200:
             data = data.json()
             if data.get("code") == "B100":
+                self.light = False
                 return True
         return False
 

pyasic/miners/avalonminer/cgminer/A10X/A1026.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A10X import CGMinerA10X
 from pyasic.miners._types import Avalon1026  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon1026(CGMinerA10X, Avalon1026):
-class CGMinerAvalon1026(CGMiner, Avalon1026):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A10X/A1047.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A10X import CGMinerA10X
 from pyasic.miners._types import Avalon1047  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon1047(CGMinerA10X, Avalon1047):
-class CGMinerAvalon1047(CGMiner, Avalon1047):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A10X/A1066.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A10X import CGMinerA10X
 from pyasic.miners._types import Avalon1066  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon1066(CGMinerA10X, Avalon1066):
-class CGMinerAvalon1066(CGMiner, Avalon1066):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A10X/A10X.py

@@ -0,0 +1,261 @@
+#  Copyright 2022 Upstream Data Inc
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+
+from pyasic.data import MinerData
+from pyasic.settings import PyasicSettings
+import re
+from pyasic.config import MinerConfig
+import logging
+
+
+class CGMinerA10X(CGMiner):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.ip = ip
+
+    async def check_light(self) -> bool:
+        if self.light:
+            return self.light
+        data = await self.api.ascset(0, "led", "1-255")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set info: LED[1]":
+            return True
+        return False
+
+    async def fault_light_on(self) -> bool:
+        data = await self.api.ascset(0, "led", "1-1")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
+            return True
+        return False
+
+    async def fault_light_off(self) -> bool:
+        data = await self.api.ascset(0, "led", "1-0")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
+            return True
+        return False
+
+    async def reboot(self) -> bool:
+        if (await self.api.restart())["STATUS"] == "RESTART":
+            return True
+        return False
+
+    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
+        """Configures miner with yaml config."""
+        raise NotImplementedError
+        logging.debug(f"{self}: Sending config.")
+        if ip_user:
+            suffix = str(self.ip).split(".")[-1]
+            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
+        else:
+            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
+        data = await self.api.ascset(
+            0, "setpool", f"root,root,{conf}"
+        )  # this should work but doesn't
+        return data
+
+    async def get_mac(self) -> str:
+        mac = None
+        version = await self.api.version()
+        if version:
+            if "VERSION" in version.keys():
+                if "MAC" in version["VERSION"][0].keys():
+                    base_mac = version["VERSION"][0]["MAC"].upper()
+                    # parse the MAC into a recognizable form
+                    mac = ":".join(
+                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
+                    )
+        return mac
+
+    async def get_data(self):
+        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
+
+        model = await self.get_model()
+        mac = None
+
+        if model:
+            data.model = model
+
+        miner_data = None
+        for i in range(PyasicSettings().miner_get_data_retries):
+            miner_data = await self.api.multicommand(
+                "version", "summary", "pools", "stats"
+            )
+            if miner_data:
+                break
+        if not miner_data:
+            hostname = await self.get_hostname()
+            mac = await self.get_mac()
+
+            if hostname and not hostname == "?":
+                data.hostname = hostname
+            elif mac:
+                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
+            if mac:
+                data.mac = mac
+            return data
+
+        data.fault_light = await self.check_light()
+
+        summary = miner_data.get("summary")
+        version = miner_data.get("version")
+        pools = miner_data.get("pools")
+        stats = miner_data.get("stats")
+
+        if summary:
+            hr = summary[0].get("SUMMARY")
+            if hr:
+                if len(hr) > 0:
+                    hr = hr[0].get("MHS 1m")
+                    if hr:
+                        data.hashrate = round(hr / 1000000, 2)
+
+        if version:
+            if "VERSION" in version[0].keys():
+                if "MAC" in version[0]["VERSION"][0].keys():
+                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
+                    # parse the MAC into a recognizable form
+                    mac = ":".join(
+                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
+                    )
+
+        if stats:
+            stats_data = stats[0].get("STATS")
+            if stats_data:
+                for key in stats_data[0].keys():
+                    if key.startswith("MM ID"):
+                        raw_data = self.parse_stats(stats_data[0][key])
+                        for fan in range(self.fan_count):
+                            if f"Fan{fan+1}" in raw_data:
+                                setattr(
+                                    data,
+                                    f"fan_{fan+1}",
+                                    int(raw_data[f"Fan{fan+1}"]),
+                                )
+                            if "MTmax" in raw_data.keys():
+                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
+                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
+                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
+                            if "MTavg" in raw_data.keys():
+                                data.left_board_temp = int(raw_data["MTavg"][0])
+                                data.center_board_temp = int(raw_data["MTavg"][1])
+                                data.right_board_temp = int(raw_data["MTavg"][2])
+
+                        if "PVT_T0" in raw_data:
+                            data.left_chips = len(
+                                [item for item in raw_data["PVT_T0"] if not item == "0"]
+                            )
+                        if "PVT_T1" in raw_data:
+                            data.center_chips = len(
+                                [item for item in raw_data["PVT_T1"] if not item == "0"]
+                            )
+                        if "PVT_T2" in raw_data:
+                            data.right_chips = len(
+                                [item for item in raw_data["PVT_T2"] if not item == "0"]
+                            )
+
+        if pools:
+            pool_1 = None
+            pool_2 = None
+            pool_1_user = None
+            pool_2_user = None
+            pool_1_quota = 1
+            pool_2_quota = 1
+            quota = 0
+            for pool in pools[0].get("POOLS"):
+                if not pool_1_user:
+                    pool_1_user = pool.get("User")
+                    pool_1 = pool["URL"]
+                    pool_1_quota = pool["Quota"]
+                elif not pool_2_user:
+                    pool_2_user = pool.get("User")
+                    pool_2 = pool["URL"]
+                    pool_2_quota = pool["Quota"]
+                if not pool.get("User") == pool_1_user:
+                    if not pool_2_user == pool.get("User"):
+                        pool_2_user = pool.get("User")
+                        pool_2 = pool["URL"]
+                        pool_2_quota = pool["Quota"]
+            if pool_2_user and not pool_2_user == pool_1_user:
+                quota = f"{pool_1_quota}/{pool_2_quota}"
+
+            if pool_1:
+                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
+                    "stratum2+tcp://", ""
+                )
+                data.pool_1_url = pool_1
+
+            if pool_1_user:
+                data.pool_1_user = pool_1_user
+
+            if pool_2:
+                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
+                    "stratum2+tcp://", ""
+                )
+                data.pool_2_url = pool_2
+
+            if pool_2_user:
+                data.pool_2_user = pool_2_user
+
+            if quota:
+                data.pool_split = str(quota)
+
+        hostname = await self.get_hostname()
+
+        if mac:
+            data.mac = mac
+        else:
+            mac = await self.get_mac()
+            if mac:
+                data.mac = mac
+        if hostname and not hostname == "?":
+            data.hostname = hostname
+        elif mac:
+            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
+
+        return data
+
+    @staticmethod
+    def parse_stats(stats):
+        _stats_items = re.findall(".+?\[*?]", stats)
+        stats_items = []
+        stats_dict = {}
+        for item in _stats_items:
+            if ":" in item:
+                data = item.replace("]", "").split("[")
+                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
+                data_dict = {}
+                for key, val in [tuple(item) for item in data_list]:
+                    data_dict[key] = val
+                raw_data = [data[0].strip(), data_dict]
+            else:
+                raw_data = [
+                    value
+                    for value in item.replace("[", " ")
+                    .replace("]", " ")
+                    .split(" ")[:-1]
+                    if value != ""
+                ]
+                if len(raw_data) == 1:
+                    raw_data.append("")
+            if raw_data[0] == "":
+                raw_data = raw_data[1:]
+
+            if len(raw_data) == 2:
+                stats_dict[raw_data[0]] = raw_data[1]
+            else:
+                stats_dict[raw_data[0]] = raw_data[1:]
+            stats_items.append(raw_data)
+
+        return stats_dict

pyasic/miners/avalonminer/cgminer/A7X/A721.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A7X import CGMinerA7X  # noqa - Ignore access to _module
 from pyasic.miners._types import Avalon721  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon721(CGMinerA7X, Avalon721):
-class CGMinerAvalon721(CGMiner, Avalon721):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A7X/A741.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A7X import CGMinerA7X  # noqa - Ignore access to _module
 from pyasic.miners._types import Avalon741  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon741(CGMinerA7X, Avalon741):
-class CGMinerAvalon741(CGMiner, Avalon741):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A7X/A761.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A7X import CGMinerA7X  # noqa - Ignore access to _module
 from pyasic.miners._types import Avalon761  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon761(CGMinerA7X, Avalon761):
-class CGMinerAvalon761(CGMiner, Avalon761):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A7X/A7X.py

@@ -0,0 +1,261 @@
+#  Copyright 2022 Upstream Data Inc
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+
+from pyasic.data import MinerData
+from pyasic.settings import PyasicSettings
+import re
+from pyasic.config import MinerConfig
+import logging
+
+
+class CGMinerA7X(CGMiner):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.ip = ip
+
+    async def check_light(self) -> bool:
+        if self.light:
+            return self.light
+        data = await self.api.ascset(0, "led", "1-255")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set info: LED[1]":
+            return True
+        return False
+
+    async def fault_light_on(self) -> bool:
+        data = await self.api.ascset(0, "led", "1-1")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
+            return True
+        return False
+
+    async def fault_light_off(self) -> bool:
+        data = await self.api.ascset(0, "led", "1-0")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
+            return True
+        return False
+
+    async def reboot(self) -> bool:
+        if (await self.api.restart())["STATUS"] == "RESTART":
+            return True
+        return False
+
+    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
+        """Configures miner with yaml config."""
+        raise NotImplementedError
+        logging.debug(f"{self}: Sending config.")
+        if ip_user:
+            suffix = str(self.ip).split(".")[-1]
+            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
+        else:
+            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
+        data = await self.api.ascset(
+            0, "setpool", f"root,root,{conf}"
+        )  # this should work but doesn't
+        return data
+
+    async def get_mac(self) -> str:
+        mac = None
+        version = await self.api.version()
+        if version:
+            if "VERSION" in version.keys():
+                if "MAC" in version["VERSION"][0].keys():
+                    base_mac = version["VERSION"][0]["MAC"].upper()
+                    # parse the MAC into a recognizable form
+                    mac = ":".join(
+                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
+                    )
+        return mac
+
+    async def get_data(self):
+        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
+
+        model = await self.get_model()
+        mac = None
+
+        if model:
+            data.model = model
+
+        miner_data = None
+        for i in range(PyasicSettings().miner_get_data_retries):
+            miner_data = await self.api.multicommand(
+                "version", "summary", "pools", "stats"
+            )
+            if miner_data:
+                break
+        if not miner_data:
+            hostname = await self.get_hostname()
+            mac = await self.get_mac()
+
+            if hostname and not hostname == "?":
+                data.hostname = hostname
+            elif mac:
+                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
+            if mac:
+                data.mac = mac
+            return data
+
+        data.fault_light = await self.check_light()
+
+        summary = miner_data.get("summary")
+        version = miner_data.get("version")
+        pools = miner_data.get("pools")
+        stats = miner_data.get("stats")
+
+        if summary:
+            hr = summary[0].get("SUMMARY")
+            if hr:
+                if len(hr) > 0:
+                    hr = hr[0].get("MHS 1m")
+                    if hr:
+                        data.hashrate = round(hr / 1000000, 2)
+
+        if version:
+            if "VERSION" in version[0].keys():
+                if "MAC" in version[0]["VERSION"][0].keys():
+                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
+                    # parse the MAC into a recognizable form
+                    mac = ":".join(
+                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
+                    )
+
+        if stats:
+            stats_data = stats[0].get("STATS")
+            if stats_data:
+                for key in stats_data[0].keys():
+                    if key.startswith("MM ID"):
+                        raw_data = self.parse_stats(stats_data[0][key])
+                        for fan in range(self.fan_count):
+                            if f"Fan{fan+1}" in raw_data:
+                                setattr(
+                                    data,
+                                    f"fan_{fan+1}",
+                                    int(raw_data[f"Fan{fan+1}"]),
+                                )
+                            if "MTmax" in raw_data.keys():
+                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
+                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
+                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
+                            if "MTavg" in raw_data.keys():
+                                data.left_board_temp = int(raw_data["MTavg"][0])
+                                data.center_board_temp = int(raw_data["MTavg"][1])
+                                data.right_board_temp = int(raw_data["MTavg"][2])
+
+                        if "PVT_T0" in raw_data:
+                            data.left_chips = len(
+                                [item for item in raw_data["PVT_T0"] if not item == "0"]
+                            )
+                        if "PVT_T1" in raw_data:
+                            data.center_chips = len(
+                                [item for item in raw_data["PVT_T1"] if not item == "0"]
+                            )
+                        if "PVT_T2" in raw_data:
+                            data.right_chips = len(
+                                [item for item in raw_data["PVT_T2"] if not item == "0"]
+                            )
+
+        if pools:
+            pool_1 = None
+            pool_2 = None
+            pool_1_user = None
+            pool_2_user = None
+            pool_1_quota = 1
+            pool_2_quota = 1
+            quota = 0
+            for pool in pools[0].get("POOLS"):
+                if not pool_1_user:
+                    pool_1_user = pool.get("User")
+                    pool_1 = pool["URL"]
+                    pool_1_quota = pool["Quota"]
+                elif not pool_2_user:
+                    pool_2_user = pool.get("User")
+                    pool_2 = pool["URL"]
+                    pool_2_quota = pool["Quota"]
+                if not pool.get("User") == pool_1_user:
+                    if not pool_2_user == pool.get("User"):
+                        pool_2_user = pool.get("User")
+                        pool_2 = pool["URL"]
+                        pool_2_quota = pool["Quota"]
+            if pool_2_user and not pool_2_user == pool_1_user:
+                quota = f"{pool_1_quota}/{pool_2_quota}"
+
+            if pool_1:
+                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
+                    "stratum2+tcp://", ""
+                )
+                data.pool_1_url = pool_1
+
+            if pool_1_user:
+                data.pool_1_user = pool_1_user
+
+            if pool_2:
+                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
+                    "stratum2+tcp://", ""
+                )
+                data.pool_2_url = pool_2
+
+            if pool_2_user:
+                data.pool_2_user = pool_2_user
+
+            if quota:
+                data.pool_split = str(quota)
+
+        hostname = await self.get_hostname()
+
+        if mac:
+            data.mac = mac
+        else:
+            mac = await self.get_mac()
+            if mac:
+                data.mac = mac
+        if hostname and not hostname == "?":
+            data.hostname = hostname
+        elif mac:
+            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
+
+        return data
+
+    @staticmethod
+    def parse_stats(stats):
+        _stats_items = re.findall(".+?\[*?]", stats)
+        stats_items = []
+        stats_dict = {}
+        for item in _stats_items:
+            if ":" in item:
+                data = item.replace("]", "").split("[")
+                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
+                data_dict = {}
+                for key, val in [tuple(item) for item in data_list]:
+                    data_dict[key] = val
+                raw_data = [data[0].strip(), data_dict]
+            else:
+                raw_data = [
+                    value
+                    for value in item.replace("[", " ")
+                    .replace("]", " ")
+                    .split(" ")[:-1]
+                    if value != ""
+                ]
+                if len(raw_data) == 1:
+                    raw_data.append("")
+            if raw_data[0] == "":
+                raw_data = raw_data[1:]
+
+            if len(raw_data) == 2:
+                stats_dict[raw_data[0]] = raw_data[1]
+            else:
+                stats_dict[raw_data[0]] = raw_data[1:]
+            stats_items.append(raw_data)
+
+        return stats_dict

pyasic/miners/avalonminer/cgminer/A8X/A821.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A8X import CGMinerA8X  # noqa - Ignore access to _module
 from pyasic.miners._types import Avalon821  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon821(CGMinerA8X, Avalon821):
-class CGMinerAvalon821(CGMiner, Avalon821):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A8X/A841.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A8X import CGMinerA8X  # noqa - Ignore access to _module
 from pyasic.miners._types import Avalon841  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon841(CGMinerA8X, Avalon841):
-class CGMinerAvalon841(CGMiner, Avalon841):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname:
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A8X/A851.py

@@ -12,241 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+from .A8X import CGMinerA8X  # noqa - Ignore access to _module
 from pyasic.miners._types import Avalon851  # noqa - Ignore access to _module
 
-from pyasic.data import MinerData
-from pyasic.settings import PyasicSettings
-import re
-from pyasic.config import MinerConfig
-import logging
 
-
+class CGMinerAvalon851(CGMinerA8X, Avalon851):
-class CGMinerAvalon851(CGMiner, Avalon851):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def fault_light_on(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-1")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        data = await self.api.ascset(0, "led", "1-0")
-        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
-            return True
-        return False
-
-    async def reboot(self) -> bool:
-        if (await self.api.restart())["STATUS"] == "RESTART":
-            return True
-        return False
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        """Configures miner with yaml config."""
-        raise NotImplementedError
-        logging.debug(f"{self}: Sending config.")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
-        data = await self.api.ascset(
-            0, "setpool", f"root,root,{conf}"
-        )  # this should work but doesn't
-        return data
-
-    async def get_mac(self) -> str:
-        mac = None
-        version = await self.api.version()
-        if version:
-            if "VERSION" in version.keys():
-                if "MAC" in version["VERSION"][0].keys():
-                    base_mac = version["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-        return mac
-
-    async def get_data(self):
-        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
-
-        model = await self.get_model()
-        mac = None
-
-        if model:
-            data.model = model
-
-        miner_data = None
-        for i in range(PyasicSettings().miner_get_data_retries):
-            miner_data = await self.api.multicommand(
-                "version", "summary", "pools", "stats"
-            )
-            if miner_data:
-                break
-        if not miner_data:
-            hostname = await self.get_hostname()
-            mac = await self.get_mac()
-
-            if hostname and not hostname == "?":
-                data.hostname = hostname
-            elif mac:
-                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-            if mac:
-                data.mac = mac
-            return data
-
-        summary = miner_data.get("summary")
-        version = miner_data.get("version")
-        pools = miner_data.get("pools")
-        stats = miner_data.get("stats")
-
-        if summary:
-            hr = summary[0].get("SUMMARY")
-            if hr:
-                if len(hr) > 0:
-                    hr = hr[0].get("MHS 1m")
-                    if hr:
-                        data.hashrate = round(hr / 1000000, 2)
-
-        if version:
-            if "VERSION" in version[0].keys():
-                if "MAC" in version[0]["VERSION"][0].keys():
-                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
-                    # parse the MAC into a recognizable form
-                    mac = ":".join(
-                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
-                    )
-
-        if stats:
-            stats_data = stats[0].get("STATS")
-            if stats_data:
-                for key in stats_data[0].keys():
-                    if key.startswith("MM ID"):
-                        raw_data = self.parse_stats(stats_data[0][key])
-                        for fan in range(self.fan_count):
-                            if f"Fan{fan+1}" in raw_data:
-                                setattr(
-                                    data,
-                                    f"fan_{fan+1}",
-                                    int(raw_data[f"Fan{fan+1}"]),
-                                )
-                            if "MTmax" in raw_data.keys():
-                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
-                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
-                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
-                            if "MTavg" in raw_data.keys():
-                                data.left_board_temp = int(raw_data["MTavg"][0])
-                                data.center_board_temp = int(raw_data["MTavg"][1])
-                                data.right_board_temp = int(raw_data["MTavg"][2])
-
-                        if "PVT_T0" in raw_data:
-                            data.left_chips = len(
-                                [item for item in raw_data["PVT_T0"] if not item == "0"]
-                            )
-                        if "PVT_T1" in raw_data:
-                            data.center_chips = len(
-                                [item for item in raw_data["PVT_T1"] if not item == "0"]
-                            )
-                        if "PVT_T2" in raw_data:
-                            data.right_chips = len(
-                                [item for item in raw_data["PVT_T2"] if not item == "0"]
-                            )
-
-        if pools:
-            pool_1 = None
-            pool_2 = None
-            pool_1_user = None
-            pool_2_user = None
-            pool_1_quota = 1
-            pool_2_quota = 1
-            quota = 0
-            for pool in pools[0].get("POOLS"):
-                if not pool_1_user:
-                    pool_1_user = pool.get("User")
-                    pool_1 = pool["URL"]
-                    pool_1_quota = pool["Quota"]
-                elif not pool_2_user:
-                    pool_2_user = pool.get("User")
-                    pool_2 = pool["URL"]
-                    pool_2_quota = pool["Quota"]
-                if not pool.get("User") == pool_1_user:
-                    if not pool_2_user == pool.get("User"):
-                        pool_2_user = pool.get("User")
-                        pool_2 = pool["URL"]
-                        pool_2_quota = pool["Quota"]
-            if pool_2_user and not pool_2_user == pool_1_user:
-                quota = f"{pool_1_quota}/{pool_2_quota}"
-
-            if pool_1:
-                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_1_url = pool_1
-
-            if pool_1_user:
-                data.pool_1_user = pool_1_user
-
-            if pool_2:
-                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
-                    "stratum2+tcp://", ""
-                )
-                data.pool_2_url = pool_2
-
-            if pool_2_user:
-                data.pool_2_user = pool_2_user
-
-            if quota:
-                data.pool_split = str(quota)
-
-        hostname = await self.get_hostname()
-
-        if mac:
-            data.mac = mac
-        else:
-            mac = await self.get_mac()
-            if mac:
-                data.mac = mac
-        if hostname and not hostname == "?":
-            data.hostname = hostname
-        elif mac:
-            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
-
-        return data
-
-    @staticmethod
-    def parse_stats(stats):
-        _stats_items = re.findall(".+?\[*?]", stats)
-        stats_items = []
-        stats_dict = {}
-        for item in _stats_items:
-            if ":" in item:
-                data = item.replace("]", "").split("[")
-                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
-                data_dict = {}
-                for key, val in [tuple(item) for item in data_list]:
-                    data_dict[key] = val
-                raw_data = [data[0].strip(), data_dict]
-            else:
-                raw_data = [
-                    value
-                    for value in item.replace("[", " ")
-                    .replace("]", " ")
-                    .split(" ")[:-1]
-                    if value != ""
-                ]
-                if len(raw_data) == 1:
-                    raw_data.append("")
-            if raw_data[0] == "":
-                raw_data = raw_data[1:]
-
-            if len(raw_data) == 2:
-                stats_dict[raw_data[0]] = raw_data[1]
-            else:
-                stats_dict[raw_data[0]] = raw_data[1:]
-            stats_items.append(raw_data)
-
-        return stats_dict

pyasic/miners/avalonminer/cgminer/A8X/A8X.py

@@ -0,0 +1,261 @@
+#  Copyright 2022 Upstream Data Inc
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
+
+from pyasic.data import MinerData
+from pyasic.settings import PyasicSettings
+import re
+from pyasic.config import MinerConfig
+import logging
+
+
+class CGMinerA8X(CGMiner):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.ip = ip
+
+    async def check_light(self) -> bool:
+        if self.light:
+            return self.light
+        data = await self.api.ascset(0, "led", "1-255")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set info: LED[1]":
+            return True
+        return False
+
+    async def fault_light_on(self) -> bool:
+        data = await self.api.ascset(0, "led", "1-1")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
+            return True
+        return False
+
+    async def fault_light_off(self) -> bool:
+        data = await self.api.ascset(0, "led", "1-0")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
+            return True
+        return False
+
+    async def reboot(self) -> bool:
+        if (await self.api.restart())["STATUS"] == "RESTART":
+            return True
+        return False
+
+    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
+        """Configures miner with yaml config."""
+        raise NotImplementedError
+        logging.debug(f"{self}: Sending config.")
+        if ip_user:
+            suffix = str(self.ip).split(".")[-1]
+            conf = MinerConfig().from_yaml(yaml_config).as_avalon(user_suffix=suffix)
+        else:
+            conf = MinerConfig().from_yaml(yaml_config).as_avalon()
+        data = await self.api.ascset(
+            0, "setpool", f"root,root,{conf}"
+        )  # this should work but doesn't
+        return data
+
+    async def get_mac(self) -> str:
+        mac = None
+        version = await self.api.version()
+        if version:
+            if "VERSION" in version.keys():
+                if "MAC" in version["VERSION"][0].keys():
+                    base_mac = version["VERSION"][0]["MAC"].upper()
+                    # parse the MAC into a recognizable form
+                    mac = ":".join(
+                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
+                    )
+        return mac
+
+    async def get_data(self):
+        data = MinerData(ip=str(self.ip), ideal_chips=self.nominal_chips * 3)
+
+        model = await self.get_model()
+        mac = None
+
+        if model:
+            data.model = model
+
+        miner_data = None
+        for i in range(PyasicSettings().miner_get_data_retries):
+            miner_data = await self.api.multicommand(
+                "version", "summary", "pools", "stats"
+            )
+            if miner_data:
+                break
+        if not miner_data:
+            hostname = await self.get_hostname()
+            mac = await self.get_mac()
+
+            if hostname and not hostname == "?":
+                data.hostname = hostname
+            elif mac:
+                data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
+            if mac:
+                data.mac = mac
+            return data
+
+        data.fault_light = await self.check_light()
+
+        summary = miner_data.get("summary")
+        version = miner_data.get("version")
+        pools = miner_data.get("pools")
+        stats = miner_data.get("stats")
+
+        if summary:
+            hr = summary[0].get("SUMMARY")
+            if hr:
+                if len(hr) > 0:
+                    hr = hr[0].get("MHS 1m")
+                    if hr:
+                        data.hashrate = round(hr / 1000000, 2)
+
+        if version:
+            if "VERSION" in version[0].keys():
+                if "MAC" in version[0]["VERSION"][0].keys():
+                    base_mac = version[0]["VERSION"][0]["MAC"].upper()
+                    # parse the MAC into a recognizable form
+                    mac = ":".join(
+                        [base_mac[i : (i + 2)] for i in range(0, len(base_mac), 2)]
+                    )
+
+        if stats:
+            stats_data = stats[0].get("STATS")
+            if stats_data:
+                for key in stats_data[0].keys():
+                    if key.startswith("MM ID"):
+                        raw_data = self.parse_stats(stats_data[0][key])
+                        for fan in range(self.fan_count):
+                            if f"Fan{fan+1}" in raw_data:
+                                setattr(
+                                    data,
+                                    f"fan_{fan+1}",
+                                    int(raw_data[f"Fan{fan+1}"]),
+                                )
+                            if "MTmax" in raw_data.keys():
+                                data.left_board_chip_temp = int(raw_data["MTmax"][0])
+                                data.center_board_chip_temp = int(raw_data["MTmax"][1])
+                                data.right_board_chip_temp = int(raw_data["MTmax"][2])
+                            if "MTavg" in raw_data.keys():
+                                data.left_board_temp = int(raw_data["MTavg"][0])
+                                data.center_board_temp = int(raw_data["MTavg"][1])
+                                data.right_board_temp = int(raw_data["MTavg"][2])
+
+                        if "PVT_T0" in raw_data:
+                            data.left_chips = len(
+                                [item for item in raw_data["PVT_T0"] if not item == "0"]
+                            )
+                        if "PVT_T1" in raw_data:
+                            data.center_chips = len(
+                                [item for item in raw_data["PVT_T1"] if not item == "0"]
+                            )
+                        if "PVT_T2" in raw_data:
+                            data.right_chips = len(
+                                [item for item in raw_data["PVT_T2"] if not item == "0"]
+                            )
+
+        if pools:
+            pool_1 = None
+            pool_2 = None
+            pool_1_user = None
+            pool_2_user = None
+            pool_1_quota = 1
+            pool_2_quota = 1
+            quota = 0
+            for pool in pools[0].get("POOLS"):
+                if not pool_1_user:
+                    pool_1_user = pool.get("User")
+                    pool_1 = pool["URL"]
+                    pool_1_quota = pool["Quota"]
+                elif not pool_2_user:
+                    pool_2_user = pool.get("User")
+                    pool_2 = pool["URL"]
+                    pool_2_quota = pool["Quota"]
+                if not pool.get("User") == pool_1_user:
+                    if not pool_2_user == pool.get("User"):
+                        pool_2_user = pool.get("User")
+                        pool_2 = pool["URL"]
+                        pool_2_quota = pool["Quota"]
+            if pool_2_user and not pool_2_user == pool_1_user:
+                quota = f"{pool_1_quota}/{pool_2_quota}"
+
+            if pool_1:
+                pool_1 = pool_1.replace("stratum+tcp://", "").replace(
+                    "stratum2+tcp://", ""
+                )
+                data.pool_1_url = pool_1
+
+            if pool_1_user:
+                data.pool_1_user = pool_1_user
+
+            if pool_2:
+                pool_2 = pool_2.replace("stratum+tcp://", "").replace(
+                    "stratum2+tcp://", ""
+                )
+                data.pool_2_url = pool_2
+
+            if pool_2_user:
+                data.pool_2_user = pool_2_user
+
+            if quota:
+                data.pool_split = str(quota)
+
+        hostname = await self.get_hostname()
+
+        if mac:
+            data.mac = mac
+        else:
+            mac = await self.get_mac()
+            if mac:
+                data.mac = mac
+        if hostname and not hostname == "?":
+            data.hostname = hostname
+        elif mac:
+            data.hostname = f"Avalon{mac.replace(':', '')[-6:]}"
+
+        return data
+
+    @staticmethod
+    def parse_stats(stats):
+        _stats_items = re.findall(".+?\[*?]", stats)
+        stats_items = []
+        stats_dict = {}
+        for item in _stats_items:
+            if ":" in item:
+                data = item.replace("]", "").split("[")
+                data_list = [i.split(": ") for i in data[1].strip().split(", ")]
+                data_dict = {}
+                for key, val in [tuple(item) for item in data_list]:
+                    data_dict[key] = val
+                raw_data = [data[0].strip(), data_dict]
+            else:
+                raw_data = [
+                    value
+                    for value in item.replace("[", " ")
+                    .replace("]", " ")
+                    .split(" ")[:-1]
+                    if value != ""
+                ]
+                if len(raw_data) == 1:
+                    raw_data.append("")
+            if raw_data[0] == "":
+                raw_data = raw_data[1:]
+
+            if len(raw_data) == 2:
+                stats_dict[raw_data[0]] = raw_data[1]
+            else:
+                stats_dict[raw_data[0]] = raw_data[1:]
+            stats_items.append(raw_data)
+
+        return stats_dict

pyasic/miners/avalonminer/cgminer/A9X/A921.py

@@ -27,6 +27,14 @@ class CGMinerAvalon921(CGMiner, Avalon921):
         super().__init__(ip)
         self.ip = ip
 
+    async def check_light(self) -> bool:
+        if self.light:
+            return self.light
+        data = await self.api.ascset(0, "led", "1-255")
+        if data["STATUS"][0]["Msg"] == "ASC 0 set info: LED[1]":
+            return True
+        return False
+
     async def fault_light_on(self) -> bool:
         data = await self.api.ascset(0, "led", "1-1")
         if data["STATUS"][0]["Msg"] == "ASC 0 set OK":
@@ -80,6 +88,8 @@ class CGMinerAvalon921(CGMiner, Avalon921):
         if model:
             data.model = model
 
+        data.fault_light = await self.check_light()
+
         miner_data = None
         for i in range(PyasicSettings().miner_get_data_retries):
             miner_data = await self.api.multicommand(

pyasic/miners/miner_factory.py

@@ -45,93 +45,98 @@ import asyncssh
 AnyMiner = TypeVar("AnyMiner", bound=BaseMiner)
 
 MINER_CLASSES = {
-    "Antminer S9": {
+    "ANTMINER S9": {
         "Default": BOSMinerS9,
         "BOSMiner": BOSMinerOld,
         "BOSMiner+": BOSMinerS9,
         "BMMiner": BMMinerS9,
         "CGMiner": CGMinerS9,
     },
-    "Antminer S9i": {
+    "ANTMINER S9I": {
         "Default": BMMinerS9i,
         "BMMiner": BMMinerS9i,
     },
-    "Antminer S17": {
+    "ANTMINER S17": {
         "Default": BMMinerS17,
         "BOSMiner+": BOSMinerS17,
         "BMMiner": BMMinerS17,
         "CGMiner": CGMinerS17,
     },
-    "Antminer S17+": {
+    "ANTMINER S17+": {
         "Default": BMMinerS17Plus,
         "BOSMiner+": BOSMinerS17Plus,
         "BMMiner": BMMinerS17Plus,
         "CGMiner": CGMinerS17Plus,
     },
-    "Antminer S17 Pro": {
+    "ANTMINER S17 Pro": {
         "Default": BMMinerS17Pro,
         "BOSMiner+": BOSMinerS17Pro,
         "BMMiner": BMMinerS17Pro,
         "CGMiner": CGMinerS17Pro,
     },
-    "Antminer S17e": {
+    "ANTMINER S17E": {
         "Default": BMMinerS17e,
         "BOSMiner+": BOSMinerS17e,
         "BMMiner": BMMinerS17e,
         "CGMiner": CGMinerS17e,
     },
-    "Antminer T17": {
+    "ANTMINER T17": {
         "Default": BMMinerT17,
         "BOSMiner+": BOSMinerT17,
         "BMMiner": BMMinerT17,
         "CGMiner": CGMinerT17,
     },
-    "Antminer T17+": {
+    "ANTMINER T17+": {
         "Default": BMMinerT17Plus,
         "BOSMiner+": BOSMinerT17Plus,
         "BMMiner": BMMinerT17Plus,
         "CGMiner": CGMinerT17Plus,
     },
-    "Antminer T17e": {
+    "ANTMINER T17E": {
         "Default": BMMinerT17e,
         "BOSMiner+": BOSMinerT17e,
         "BMMiner": BMMinerT17e,
         "CGMiner": CGMinerT17e,
     },
-    "Antminer S19": {
+    "ANTMINER S19": {
         "Default": BMMinerS19,
         "BOSMiner+": BOSMinerS19,
         "BMMiner": BMMinerS19,
         "CGMiner": CGMinerS19,
     },
-    "Antminer S19 Pro": {
+    "ANTMINER S19 Pro": {
         "Default": BMMinerS19Pro,
         "BOSMiner+": BOSMinerS19Pro,
         "BMMiner": BMMinerS19Pro,
         "CGMiner": CGMinerS19Pro,
     },
-    "Antminer S19j": {
+    "ANTMINER S19J": {
         "Default": BMMinerS19j,
         "BOSMiner+": BOSMinerS19j,
         "BMMiner": BMMinerS19j,
         "CGMiner": CGMinerS19j,
     },
-    "Antminer S19j Pro": {
+    "ANTMINER S19J Pro": {
         "Default": BMMinerS19jPro,
         "BOSMiner+": BOSMinerS19jPro,
         "BMMiner": BMMinerS19jPro,
         "CGMiner": CGMinerS19jPro,
     },
-    "Antminer S19a": {
+    "ANTMINER S19A": {
         "Default": BMMinerS19a,
         "BMMiner": BMMinerS19a,
     },
-    "Antminer T19": {
+    "ANTMINER T19": {
         "Default": BMMinerT19,
         "BOSMiner+": BOSMinerT19,
         "BMMiner": BMMinerT19,
         "CGMiner": CGMinerT19,
     },
+    "M20": {
+        "Default": BTMinerM20,
+        "BTMiner": BTMinerM20,
+        "10": BTMinerM20V10,
+    },
     "M20S": {
         "Default": BTMinerM20S,
         "BTMiner": BTMinerM20S,
@@ -362,8 +367,8 @@ class MinerFactory(metaclass=Singleton):
         self.miners = {}
 
     async def _get_miner_type(
-        self, ip: ipaddress.ip_address or str
+        self, ip: Union[ipaddress.ip_address, str]
-    ) -> Tuple[str or None, str or None, str or None]:
+    ) -> Tuple[Union[str, None], Union[str, None], Union[str, None]]:
         data = None
 
         model = None
@@ -426,9 +431,9 @@ class MinerFactory(metaclass=Singleton):
 
                 if board_name:
                     if board_name == "am1-s9":
-                        model = "Antminer S9"
+                        model = "ANTMINER S9"
                     if board_name == "am2-s17":
-                        model = "Antminer S17"
+                        model = "ANTMINER S17"
                     api = "BOSMiner+"
                     return model, api, None
 
@@ -441,10 +446,11 @@ class MinerFactory(metaclass=Singleton):
                     if data.status_code == 200:
                         data = data.json()
                     if "minertype" in data.keys():
-                        model = data["minertype"]
+                        model = data["minertype"].upper()
                     if "bmminer" in "\t".join(data.keys()):
                         api = "BMMiner"
-                except:
+                except Exception as e:
+                    logging.debug(f"Unable to get miner - {e}")
                     return None, None, None
 
         # if we have devdetails, we can get model data from there
@@ -453,15 +459,15 @@ class MinerFactory(metaclass=Singleton):
                 # check for model, for most miners
                 if not devdetails["DEVDETAILS"][0]["Model"] == "":
                     # model of most miners
-                    model = devdetails["DEVDETAILS"][0]["Model"]
+                    model = devdetails["DEVDETAILS"][0]["Model"].upper()
 
                 # if model fails, try driver
                 else:
                     # some avalonminers have model in driver
-                    model = devdetails["DEVDETAILS"][0]["Driver"]
+                    model = devdetails["DEVDETAILS"][0]["Driver"].upper()
             else:
                 if "s9" in devdetails["STATUS"][0]["Description"]:
-                    model = "Antminer S9"
+                    model = "ANTMINER S9"
 
         # if we have version we can get API type from here
         if version:
@@ -496,7 +502,7 @@ class MinerFactory(metaclass=Singleton):
                 # check for avalonminers
                 if version["VERSION"][0].get("PROD"):
                     _data = version["VERSION"][0]["PROD"].split("-")
-                    model = _data[0]
+                    model = _data[0].upper()
                     if len(data) > 1:
                         ver = _data[1]
                 elif version["VERSION"][0].get("MODEL"):
@@ -521,11 +527,11 @@ class MinerFactory(metaclass=Singleton):
             ):
                 # try to get "Type" which is model
                 if version["VERSION"][0].get("Type"):
-                    model = version["VERSION"][0]["Type"]
+                    model = version["VERSION"][0]["Type"].upper()
 
                 # braiins OS bug check just in case
                 elif "am2-s17" in version["STATUS"][0]["Description"]:
-                    model = "Antminer S17"
+                    model = "ANTMINER S17"
 
         if model:
             # whatsminer have a V in their version string (M20SV41), remove everything after it
@@ -534,13 +540,13 @@ class MinerFactory(metaclass=Singleton):
                 if len(_ver) > 1:
                     ver = model.split("V")[1]
                     model = model.split("V")[0]
-            # don't need "Bitmain", just "Antminer XX" as model
+            # don't need "Bitmain", just "ANTMINER XX" as model
             if "Bitmain " in model:
                 model = model.replace("Bitmain ", "")
         return model, api, ver
 
     @staticmethod
-    async def _validate_command(data: dict) -> Tuple[bool, str or None]:
+    async def _validate_command(data: dict) -> Tuple[bool, Union[str, None]]:
         """Check if the returned command output is correctly formatted."""
         # check if the data returned is correct or an error
         if not data or data == {}:
@@ -567,7 +573,9 @@ class MinerFactory(metaclass=Singleton):
         return True, None
 
     @staticmethod
-    async def _send_api_command(ip: ipaddress.ip_address or str, command: str) -> dict:
+    async def _send_api_command(
+        ip: Union[ipaddress.ip_address, str], command: str
+    ) -> dict:
         try:
             # get reader and writer streams
             reader, writer = await asyncio.open_connection(str(ip), 4028)

pyasic/miners/whatsminer/btminer/M2X/M20.py

@@ -0,0 +1,31 @@
+#  Copyright 2022 Upstream Data Inc
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from pyasic.miners._backends import BTMiner  # noqa - Ignore access to _module
+from pyasic.miners._types import (  # noqa - Ignore access to _module
+    M20,
+    M20V10,
+)
+
+
+class BTMinerM20(BTMiner, M20):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.ip = ip
+
+
+class BTMinerM20V10(BTMiner, M20V10):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.ip = ip

pyasic/miners/whatsminer/btminer/M2X/__init__.py

@@ -12,6 +12,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+from .M20 import BTMinerM20, BTMinerM20V10
 from .M20S import BTMinerM20S, BTMinerM20SV10, BTMinerM20SV20
 from .M20S_Plus import BTMinerM20SPlus
 

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyasic"
-version = "0.12.6"
+version = "0.14.1"
 description = "A set of modules for interfacing with many common types of ASIC bitcoin miners, using both their API and SSH."
 authors = ["UpstreamData <brett@upstreamdata.ca>"]
 repository = "https://github.com/UpstreamData/pyasic"