added estimate env temp for X19 and change format of X19 and X17 files

data/__init__.py

@@ -40,7 +40,7 @@ class MinerData:
     hostname: str = "Unknown"
     hashrate: float = 0
     temperature_avg: int = field(init=False)
-    env_temp: int = 0
+    env_temp: float = 0
     left_board_temp: int = 0
     left_board_chip_temp: int = 0
     center_board_temp: int = 0

miners/_backends/bmminer.py

@@ -203,11 +203,17 @@ class BMMiner(BaseMiner):
                         )
 
                     board_map = {0: "left_board", 1: "center_board", 2: "right_board"}
+                    env_temp_list = []
                     for item in range(3):
                         board_temp = temp[1].get(f"temp{item + board_offset}")
                         chip_temp = temp[1].get(f"temp2_{item + board_offset}")
                         setattr(data, f"{board_map[item]}_chip_temp", chip_temp)
                         setattr(data, f"{board_map[item]}_temp", board_temp)
+                        if f"temp_pcb{item}" in temp[1].keys():
+                            env_temp = temp[1][f"temp_pcb{item}"].split("-")[0]
+                            if not env_temp == 0:
+                                env_temp_list.append(int(env_temp))
+                    data.env_temp = sum(env_temp_list) / len(env_temp_list)
 
         if pools:
             pool_1 = None

miners/antminer/bmminer/X17/S17.py

@@ -1,87 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X17 import BMMinerX17
 from miners._types import S17  # noqa - Ignore access to _module
 
-import httpx
 
-
-# TODO add config
-
-
-class BMMinerS17(BMMiner, S17):
+class BMMinerS17(BMMinerX17, S17):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            try:
-                await client.post(url, data={"action": "startBlink"}, auth=auth)
-            except httpx.ReadTimeout:
-                # Expected behaviour
-                pass
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            await client.post(url, data={"action": "stopBlink"}, auth=auth)
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if not data["isBlinking"]:
-                return True
-        return False
-
-    async def check_light(self):
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X17/S17_Plus.py

@@ -1,84 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X17 import BMMinerX17
 from miners._types import S17Plus  # noqa - Ignore access to _module
 
-import httpx
 
-
-class BMMinerS17Plus(BMMiner, S17Plus):
+class BMMinerS17Plus(BMMinerX17, S17Plus):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            try:
-                await client.post(url, data={"action": "startBlink"}, auth=auth)
-            except httpx.ReadTimeout:
-                # Expected behaviour
-                pass
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            await client.post(url, data={"action": "stopBlink"}, auth=auth)
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if not data["isBlinking"]:
-                return True
-        return False
-
-    async def check_light(self):
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X17/S17_Pro.py

@@ -1,84 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X17 import BMMinerX17
 from miners._types import S17Pro  # noqa - Ignore access to _module
 
-import httpx
 
-
-class BMMinerS17Pro(BMMiner, S17Pro):
+class BMMinerS17Pro(BMMinerX17, S17Pro):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            try:
-                await client.post(url, data={"action": "startBlink"}, auth=auth)
-            except httpx.ReadTimeout:
-                # Expected behaviour
-                pass
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            await client.post(url, data={"action": "stopBlink"}, auth=auth)
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if not data["isBlinking"]:
-                return True
-        return False
-
-    async def check_light(self):
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X17/S17e.py

@@ -1,84 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X17 import BMMinerX17
 from miners._types import S17e  # noqa - Ignore access to _module
 
-import httpx
 
-
-class BMMinerS17e(BMMiner, S17e):
+class BMMinerS17e(BMMinerX17, S17e):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            try:
-                await client.post(url, data={"action": "startBlink"}, auth=auth)
-            except httpx.ReadTimeout:
-                # Expected behaviour
-                pass
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            await client.post(url, data={"action": "stopBlink"}, auth=auth)
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if not data["isBlinking"]:
-                return True
-        return False
-
-    async def check_light(self):
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X17/T17.py

@@ -1,84 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X17 import BMMinerX17
 from miners._types import T17  # noqa - Ignore access to _module
 
-import httpx
 
-
-class BMMinerT17(BMMiner, T17):
+class BMMinerT17(BMMinerX17, T17):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            try:
-                await client.post(url, data={"action": "startBlink"}, auth=auth)
-            except httpx.ReadTimeout:
-                # Expected behaviour
-                pass
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            await client.post(url, data={"action": "stopBlink"}, auth=auth)
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if not data["isBlinking"]:
-                return True
-        return False
-
-    async def check_light(self):
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X17/T17_Plus.py

@@ -1,84 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X17 import BMMinerX17
 from miners._types import T17Plus  # noqa - Ignore access to _module
 
-import httpx
 
-
-class BMMinerT17Plus(BMMiner, T17Plus):
+class BMMinerT17Plus(BMMinerX17, T17Plus):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            try:
-                await client.post(url, data={"action": "startBlink"}, auth=auth)
-            except httpx.ReadTimeout:
-                # Expected behaviour
-                pass
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            await client.post(url, data={"action": "stopBlink"}, auth=auth)
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if not data["isBlinking"]:
-                return True
-        return False
-
-    async def check_light(self):
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X17/T17e.py

@@ -1,84 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X17 import BMMinerX17
 from miners._types import T17e  # noqa - Ignore access to _module
 
-import httpx
 
-
-class BMMinerT17e(BMMiner, T17e):
+class BMMinerT17e(BMMinerX17, T17e):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            try:
-                await client.post(url, data={"action": "startBlink"}, auth=auth)
-            except httpx.ReadTimeout:
-                # Expected behaviour
-                pass
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            await client.post(url, data={"action": "stopBlink"}, auth=auth)
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if not data["isBlinking"]:
-                return True
-        return False
-
-    async def check_light(self):
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data["isBlinking"]:
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X17/X17.py

@@ -0,0 +1,83 @@
+from miners._backends import BMMiner  # noqa - Ignore access to _module
+
+import httpx
+
+
+class BMMinerX17(BMMiner):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.ip = ip
+
+    async def get_hostname(self) -> str or None:
+        hostname = None
+        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
+            if len(data.keys()) > 0:
+                if "hostname" in data.keys():
+                    hostname = data["hostname"]
+        return hostname
+
+    async def get_mac(self):
+        mac = None
+        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
+            if len(data.keys()) > 0:
+                if "macaddr" in data.keys():
+                    mac = data["macaddr"]
+        return mac
+
+    async def fault_light_on(self) -> bool:
+        url = f"http://{self.ip}/cgi-bin/blink.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        async with httpx.AsyncClient() as client:
+            try:
+                await client.post(url, data={"action": "startBlink"}, auth=auth)
+            except httpx.ReadTimeout:
+                # Expected behaviour
+                pass
+            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
+        if data.status_code == 200:
+            data = data.json()
+            if data["isBlinking"]:
+                return True
+        return False
+
+    async def fault_light_off(self) -> bool:
+        url = f"http://{self.ip}/cgi-bin/blink.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        async with httpx.AsyncClient() as client:
+            await client.post(url, data={"action": "stopBlink"}, auth=auth)
+            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
+        if data.status_code == 200:
+            data = data.json()
+            if not data["isBlinking"]:
+                return True
+        return False
+
+    async def check_light(self):
+        url = f"http://{self.ip}/cgi-bin/blink.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        async with httpx.AsyncClient() as client:
+            data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth)
+        if data.status_code == 200:
+            data = data.json()
+            if data["isBlinking"]:
+                return True
+        return False
+
+    async def reboot(self) -> bool:
+        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        async with httpx.AsyncClient() as client:
+            data = await client.get(url, auth=auth)
+        if data.status_code == 200:
+            return True
+        return False

miners/antminer/bmminer/X19/S19.py

@@ -1,103 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X19 import BMMinerX19
 from miners._types import S19  # noqa - Ignore access to _module
 
-from config import MinerConfig
 
-import httpx
-import json
-import asyncio
-
-
-class BMMinerS19(BMMiner, S19):
+class BMMinerS19(BMMinerX19, S19):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_config(self) -> MinerConfig:
-        url = f"http://{self.ip}/cgi-bin/get_miner_conf.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()
-            self.config = MinerConfig().from_raw(data)
-        return self.config
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_x19()
-
-        try:
-            async with httpx.AsyncClient() as client:
-                await client.post(url, data=conf, auth=auth)
-        except httpx.ReadTimeout:
-            pass
-        for i in range(7):
-            data = await self.get_config()
-            if data.as_x19() == conf:
-                break
-            await asyncio.sleep(1)
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "true"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B000":
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "false"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B100":
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X19/S19_Pro.py

@@ -1,103 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X19 import BMMinerX19
 from miners._types import S19Pro  # noqa - Ignore access to _module
 
-from config import MinerConfig
 
-import httpx
-import json
-import asyncio
-
-
-class BMMinerS19Pro(BMMiner, S19Pro):
+class BMMinerS19Pro(BMMinerX19, S19Pro):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_config(self) -> MinerConfig:
-        url = f"http://{self.ip}/cgi-bin/get_miner_conf.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()
-            self.config = MinerConfig().from_raw(data)
-        return self.config
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_x19()
-
-        try:
-            async with httpx.AsyncClient() as client:
-                await client.post(url, data=conf, auth=auth)
-        except httpx.ReadTimeout:
-            pass
-        for i in range(7):
-            data = await self.get_config()
-            if data.as_x19() == conf:
-                break
-            await asyncio.sleep(1)
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "true"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B000":
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "false"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B100":
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X19/S19a.py

@@ -1,103 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X19 import BMMinerX19
 from miners._types import S19a  # noqa - Ignore access to _module
 
-from config import MinerConfig
 
-import httpx
-import json
-import asyncio
-
-
-class BMMinerS19a(BMMiner, S19a):
+class BMMinerS19a(BMMinerX19, S19a):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_config(self) -> MinerConfig:
-        url = f"http://{self.ip}/cgi-bin/get_miner_conf.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()
-            self.config = MinerConfig().from_raw(data)
-        return self.config
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_x19()
-
-        try:
-            async with httpx.AsyncClient() as client:
-                await client.post(url, data=conf, auth=auth)
-        except httpx.ReadTimeout:
-            pass
-        for i in range(7):
-            data = await self.get_config()
-            if data.as_x19() == conf:
-                break
-            await asyncio.sleep(1)
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "true"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B000":
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "false"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B100":
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X19/S19j.py

@@ -1,103 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X19 import BMMinerX19
 from miners._types import S19j  # noqa - Ignore access to _module
 
-from config import MinerConfig
 
-import httpx
-import json
-import asyncio
-
-
-class BMMinerS19j(BMMiner, S19j):
+class BMMinerS19j(BMMinerX19, S19j):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_config(self) -> MinerConfig:
-        url = f"http://{self.ip}/cgi-bin/get_miner_conf.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()
-            self.config = MinerConfig().from_raw(data)
-        return self.config
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_x19()
-
-        try:
-            async with httpx.AsyncClient() as client:
-                await client.post(url, data=conf, auth=auth)
-        except httpx.ReadTimeout:
-            pass
-        for i in range(7):
-            data = await self.get_config()
-            if data.as_x19() == conf:
-                break
-            await asyncio.sleep(1)
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "true"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B000":
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "false"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B100":
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X19/S19j_Pro.py

@@ -1,103 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X19 import BMMinerX19
 from miners._types import S19jPro  # noqa - Ignore access to _module
 
-from config import MinerConfig
 
-import httpx
-import json
-import asyncio
-
-
-class BMMinerS19jPro(BMMiner, S19jPro):
+class BMMinerS19jPro(BMMinerX19, S19jPro):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_config(self) -> MinerConfig:
-        url = f"http://{self.ip}/cgi-bin/get_miner_conf.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()
-            self.config = MinerConfig().from_raw(data)
-        return self.config
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_x19()
-
-        try:
-            async with httpx.AsyncClient() as client:
-                await client.post(url, data=conf, auth=auth)
-        except httpx.ReadTimeout:
-            pass
-        for i in range(7):
-            data = await self.get_config()
-            if data.as_x19() == conf:
-                break
-            await asyncio.sleep(1)
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "true"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B000":
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "false"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B100":
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X19/T19.py

@@ -1,103 +1,8 @@
-from miners._backends import BMMiner  # noqa - Ignore access to _module
+from .X19 import BMMinerX19
 from miners._types import T19  # noqa - Ignore access to _module
 
-from config import MinerConfig
 
-import httpx
-import json
-import asyncio
-
-
-class BMMinerT19(BMMiner, T19):
+class BMMinerT19(BMMinerX19, T19):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.ip = ip
-
-    async def get_config(self) -> MinerConfig:
-        url = f"http://{self.ip}/cgi-bin/get_miner_conf.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()
-            self.config = MinerConfig().from_raw(data)
-        return self.config
-
-    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
-        url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        if ip_user:
-            suffix = str(self.ip).split(".")[-1]
-            conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
-        else:
-            conf = MinerConfig().from_yaml(yaml_config).as_x19()
-
-        try:
-            async with httpx.AsyncClient() as client:
-                await client.post(url, data=conf, auth=auth)
-        except httpx.ReadTimeout:
-            pass
-        for i in range(7):
-            data = await self.get_config()
-            if data.as_x19() == conf:
-                break
-            await asyncio.sleep(1)
-
-    async def get_hostname(self) -> str or None:
-        hostname = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "hostname" in data.keys():
-                    hostname = data["hostname"]
-        return hostname
-
-    async def get_mac(self):
-        mac = None
-        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
-            if len(data.keys()) > 0:
-                if "macaddr" in data.keys():
-                    mac = data["macaddr"]
-        return mac
-
-    async def fault_light_on(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "true"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B000":
-                return True
-        return False
-
-    async def fault_light_off(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/blink.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        data = json.dumps({"blink": "false"})
-        async with httpx.AsyncClient() as client:
-            data = await client.post(url, data=data, auth=auth)
-        if data.status_code == 200:
-            data = data.json()
-            if data.get("code") == "B100":
-                return True
-        return False
-
-    async def reboot(self) -> bool:
-        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
-        auth = httpx.DigestAuth("root", "root")
-        async with httpx.AsyncClient() as client:
-            data = await client.get(url, auth=auth)
-        if data.status_code == 200:
-            return True
-        return False

miners/antminer/bmminer/X19/X19.py

@@ -0,0 +1,102 @@
+from miners._backends import BMMiner  # noqa - Ignore access to _module
+
+from config import MinerConfig
+
+import httpx
+import json
+import asyncio
+
+
+class BMMinerX19(BMMiner):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.ip = ip
+
+    async def get_config(self) -> MinerConfig:
+        url = f"http://{self.ip}/cgi-bin/get_miner_conf.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()
+            self.config = MinerConfig().from_raw(data)
+        return self.config
+
+    async def send_config(self, yaml_config, ip_user: bool = False) -> None:
+        url = f"http://{self.ip}/cgi-bin/set_miner_conf.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        if ip_user:
+            suffix = str(self.ip).split(".")[-1]
+            conf = MinerConfig().from_yaml(yaml_config).as_x19(user_suffix=suffix)
+        else:
+            conf = MinerConfig().from_yaml(yaml_config).as_x19()
+
+        try:
+            async with httpx.AsyncClient() as client:
+                await client.post(url, data=conf, auth=auth)
+        except httpx.ReadTimeout:
+            pass
+        for i in range(7):
+            data = await self.get_config()
+            if data.as_x19() == conf:
+                break
+            await asyncio.sleep(1)
+
+    async def get_hostname(self) -> str or None:
+        hostname = None
+        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
+            if len(data.keys()) > 0:
+                if "hostname" in data.keys():
+                    hostname = data["hostname"]
+        return hostname
+
+    async def get_mac(self):
+        mac = None
+        url = f"http://{self.ip}/cgi-bin/get_system_info.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()
+            if len(data.keys()) > 0:
+                if "macaddr" in data.keys():
+                    mac = data["macaddr"]
+        return mac
+
+    async def fault_light_on(self) -> bool:
+        url = f"http://{self.ip}/cgi-bin/blink.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        data = json.dumps({"blink": "true"})
+        async with httpx.AsyncClient() as client:
+            data = await client.post(url, data=data, auth=auth)
+        if data.status_code == 200:
+            data = data.json()
+            if data.get("code") == "B000":
+                return True
+        return False
+
+    async def fault_light_off(self) -> bool:
+        url = f"http://{self.ip}/cgi-bin/blink.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        data = json.dumps({"blink": "false"})
+        async with httpx.AsyncClient() as client:
+            data = await client.post(url, data=data, auth=auth)
+        if data.status_code == 200:
+            data = data.json()
+            if data.get("code") == "B100":
+                return True
+        return False
+
+    async def reboot(self) -> bool:
+        url = f"http://{self.ip}/cgi-bin/reboot.cgi"
+        auth = httpx.DigestAuth("root", "root")
+        async with httpx.AsyncClient() as client:
+            data = await client.get(url, auth=auth)
+        if data.status_code == 200:
+            return True
+        return False

tools/cfg_util/configure/__init__.py

@@ -1,13 +1,13 @@
-import PySimpleGUI as sg
-from config import MinerConfig
-import time
-from tools.cfg_util.layout import window, update_prog_bar
-from tools.cfg_util.decorators import disable_buttons
-from miners.miner_factory import MinerFactory
 import asyncio
-from settings import CFG_UTIL_CONFIG_THREADS as CONFIG_THREADS
-from tools.cfg_util.general import update_miners_data
 
+import PySimpleGUI as sg
+
+from config import MinerConfig
+from miners.miner_factory import MinerFactory
+from settings import CFG_UTIL_CONFIG_THREADS as CONFIG_THREADS
+from tools.cfg_util.decorators import disable_buttons
+from tools.cfg_util.general import update_miners_data
+from tools.cfg_util.layout import window, update_prog_bar
 
 progress_bar_len = 0
 

tools/web_testbench/feeds.py

@@ -1,12 +1,12 @@
-import aiohttp
-import httpx
-import shutil
-import aiofiles
 import asyncio
-from bs4 import BeautifulSoup
-import re
-import os
 import logging
+import os
+import re
+import shutil
+
+import aiofiles
+import httpx
+from bs4 import BeautifulSoup
 
 
 async def get_latest_version(session):