format: swap X17 over to the new web api format.

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

@@ -14,45 +14,21 @@
 #  limitations under the License.                                              -
 # ------------------------------------------------------------------------------
 
-import json
+from typing import Union
-from typing import Optional, Union
-
-import httpx
 
 from pyasic.miners._backends import BMMiner  # noqa - Ignore access to _module
-from pyasic.settings import PyasicSettings
+from pyasic.web.X17 import X17WebAPI
 
 
 class BMMinerX17(BMMiner):
     def __init__(self, ip: str, api_ver: str = "0.0.0") -> None:
         super().__init__(ip, api_ver=api_ver)
         self.ip = ip
-        self.uname = "root"
+        self.web = X17WebAPI(ip)
-        self.pwd = PyasicSettings().global_x17_password
-
-    async def send_web_command(
-        self, command: str, params: dict = None
-    ) -> Optional[dict]:
-        url = f"http://{self.ip}/cgi-bin/{command}.cgi"
-        auth = httpx.DigestAuth(self.uname, self.pwd)
-        try:
-            async with httpx.AsyncClient() as client:
-                if params:
-                    data = await client.post(url, data=params, auth=auth)
-                else:
-                    data = await client.get(url, auth=auth)
-        except httpx.HTTPError:
-            pass
-        else:
-            if data.status_code == 200:
-                try:
-                    return data.json()
-                except json.decoder.JSONDecodeError:
-                    pass
 
     async def get_mac(self) -> Union[str, None]:
         try:
-            data = await self.send_web_command("get_system_info")
+            data = await self.web.get_system_info()
             if data:
                 return data["macaddr"]
         except KeyError:
@@ -60,11 +36,9 @@ class BMMinerX17(BMMiner):
 
     async def fault_light_on(self) -> bool:
         # this should time out, after it does do a check
-        await self.send_web_command("blink", params={"action": "startBlink"})
+        await self.web.blink(blink=True)
         try:
-            data = await self.send_web_command(
+            data = await self.web.get_blink_status()
-                "blink", params={"action": "onPageLoaded"}
-            )
             if data:
                 if data["isBlinking"]:
                     self.light = True
@@ -73,11 +47,9 @@ class BMMinerX17(BMMiner):
         return self.light
 
     async def fault_light_off(self) -> bool:
-        await self.send_web_command("blink", params={"action": "stopBlink"})
+        await self.web.blink(blink=False)
         try:
-            data = await self.send_web_command(
+            data = await self.web.get_blink_status()
-                "blink", params={"action": "onPageLoaded"}
-            )
             if data:
                 if not data["isBlinking"]:
                     self.light = False
@@ -86,7 +58,7 @@ class BMMinerX17(BMMiner):
         return self.light
 
     async def reboot(self) -> bool:
-        data = await self.send_web_command("reboot")
+        data = await self.web.reboot()
         if data:
             return True
         return False
@@ -95,9 +67,7 @@ class BMMinerX17(BMMiner):
         if self.light:
             return self.light
         try:
-            data = await self.send_web_command(
+            data = await self.web.get_blink_status()
-                "blink", params={"action": "onPageLoaded"}
-            )
             if data:
                 self.light = data["isBlinking"]
         except KeyError:
@@ -106,7 +76,7 @@ class BMMinerX17(BMMiner):
 
     async def get_hostname(self) -> Union[str, None]:
         try:
-            data = await self.send_web_command("get_system_info")
+            data = await self.web.get_system_info()
             if data:
                 return data["hostname"]
         except KeyError:

pyasic/web/X17.py

@@ -0,0 +1,66 @@
+# ------------------------------------------------------------------------------
+#  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.                                              -
+# ------------------------------------------------------------------------------
+import json
+from typing import Union
+
+import httpx
+
+from pyasic.settings import PyasicSettings
+from pyasic.web import BaseWebAPI
+
+
+class X17WebAPI(BaseWebAPI):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.pwd = PyasicSettings().global_x17_password
+
+    async def send_command(
+        self,
+        command: Union[str, bytes],
+        ignore_errors: bool = False,
+        allow_warning: bool = True,
+        **parameters: Union[str, int, bool],
+    ) -> dict:
+        url = f"http://{self.ip}/cgi-bin/{command}.cgi"
+        auth = httpx.DigestAuth(self.username, self.pwd)
+        try:
+            async with httpx.AsyncClient() as client:
+                if parameters:
+                    data = await client.post(url, data=parameters, auth=auth)
+                else:
+                    data = await client.get(url, auth=auth)
+        except httpx.HTTPError:
+            pass
+        else:
+            if data.status_code == 200:
+                try:
+                    return data.json()
+                except json.decoder.JSONDecodeError:
+                    pass
+
+    async def get_system_info(self) -> dict:
+        return await self.send_command("get_system_info")
+
+    async def blink(self, blink: bool):
+        if blink:
+            return await self.send_command("blink", action="startBlink")
+        return await self.send_command("blink", action="stopBlink")
+
+    async def reboot(self):
+        return await self.send_command("reboot")
+
+    async def get_blink_status(self) -> dict:
+        return await self.send_command("blink", action="onPageLoaded")

pyasic/web/X19.py

@@ -53,30 +53,30 @@ class X19WebAPI(BaseWebAPI):
                 except json.decoder.JSONDecodeError:
                     pass
 
-    async def get_miner_conf(self):
+    async def get_miner_conf(self) -> dict:
         return await self.send_command("get_miner_conf")
 
-    async def set_miner_conf(self, conf: dict):
+    async def set_miner_conf(self, conf: dict) -> dict:
         return await self.send_command("set_miner_conf", **conf)
 
-    async def blink(self, blink: bool):
+    async def blink(self, blink: bool) -> dict:
         if blink:
             return await self.send_command("blink", blink="true")
         return await self.send_command("blink", blink="false")
 
-    async def reboot(self):
+    async def reboot(self) -> dict:
         return await self.send_command("reboot")
 
-    async def get_system_info(self):
+    async def get_system_info(self) -> dict:
         return await self.send_command("get_system_info")
 
-    async def get_network_info(self):
+    async def get_network_info(self) -> dict:
         return await self.send_command("get_network_info")
 
-    async def summary(self):
+    async def summary(self) -> dict:
         return await self.send_command("summary")
 
-    async def get_blink_status(self):
+    async def get_blink_status(self) -> dict:
         return await self.send_command("get_blink_status")
 
     async def set_network_conf(
@@ -87,7 +87,7 @@ class X19WebAPI(BaseWebAPI):
         subnet_mask: str,
         hostname: str,
         protocol: int,
-    ):
+    ) -> dict:
         return await self.send_command(
             "set_network_conf",
             ipAddress=ip,