format: swap X9 and Innosilicon over to the new web api format.

pyasic/miners/antminer/bmminer/X9/S9.py

@@ -21,46 +21,25 @@ import httpx
 
 from pyasic.miners._backends import BMMiner  # noqa - Ignore access to _module
 from pyasic.miners._types import S9  # noqa - Ignore access to _module
-from pyasic.settings import PyasicSettings
+from pyasic.web.S9 import S9WebAPI
 
 
 class BMMinerS9(BMMiner, S9):
     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 = S9WebAPI(ip)
-        self.pwd = PyasicSettings().global_x19_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:
             pass
 
         try:
-            data = await self.send_web_command("get_network_info")
+            data = await self.web.get_network_info()
             if data:
                 return data["macaddr"]
         except KeyError:

pyasic/miners/base.py

@@ -395,7 +395,12 @@ class BaseMiner(ABC):
 
         web_data = {}
         for command in web_params:
-            data = await self.send_web_command(command)  # noqa: web only anyway
+            try:
+                cmd_func = getattr(self.web, command)
+                data = await cmd_func()  # noqa: web only anyway
+            except (LookupError, APIError):
+                pass
+            else:
                 web_data[command] = data
         for data_name in data_to_get:
             function = getattr(self, "get_" + data_name)

pyasic/miners/innosilicon/cgminer/T3X/T3H_Plus.py

@@ -28,64 +28,14 @@ from pyasic.errors import APIError
 from pyasic.miners._backends import CGMiner  # noqa - Ignore access to _module
 from pyasic.miners._types import InnosiliconT3HPlus  # noqa - Ignore access to _module
 from pyasic.settings import PyasicSettings
+from pyasic.web.Inno import InnosiliconWebAPI
 
 
 class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
     def __init__(self, ip: str, api_ver: str = "0.0.0") -> None:
         super().__init__(ip, api_ver=api_ver)
         self.ip = ip
-        self.uname = "admin"
+        self.web = InnosiliconWebAPI(ip)
-        self.pwd = PyasicSettings().global_innosilicon_password
-        self.jwt = None
-
-    async def auth(self):
-        async with httpx.AsyncClient() as client:
-            try:
-                auth = await client.post(
-                    f"http://{self.ip}/api/auth",
-                    data={"username": self.uname, "password": self.pwd},
-                )
-            except httpx.HTTPError:
-                warnings.warn(f"Could not authenticate web token with miner: {self}")
-            else:
-                json_auth = auth.json()
-                self.jwt = json_auth.get("jwt")
-            return self.jwt
-
-    async def send_web_command(self, command: str, data: Union[dict, None] = None):
-        if not self.jwt:
-            await self.auth()
-        if not data:
-            data = {}
-        async with httpx.AsyncClient() as client:
-            for i in range(PyasicSettings().miner_get_data_retries):
-                try:
-                    response = await client.post(
-                        f"http://{self.ip}/api/{command}",
-                        headers={"Authorization": "Bearer " + self.jwt},
-                        timeout=5,
-                        data=data,
-                    )
-                    json_data = response.json()
-                    if (
-                        not json_data.get("success")
-                        and "token" in json_data
-                        and json_data.get("token") == "expired"
-                    ):
-                        # refresh the token, retry
-                        await self.auth()
-                        continue
-                    if not json_data.get("success"):
-                        if json_data.get("msg"):
-                            raise APIError(json_data["msg"])
-                        elif json_data.get("message"):
-                            raise APIError(json_data["message"])
-                        raise APIError("Innosilicon web api command failed.")
-                    return json_data
-                except httpx.HTTPError:
-                    pass
-                except json.JSONDecodeError:
-                    pass
 
     async def fault_light_on(self) -> bool:
         return False
@@ -108,7 +58,7 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
 
     async def reboot(self) -> bool:
         try:
-            data = await self.send_web_command("reboot")
+            data = await self.web.reboot()
         except APIError:
             pass
         else:
@@ -116,7 +66,7 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
 
     async def restart_cgminer(self) -> bool:
         try:
-            data = await self.send_web_command("restartCgMiner")
+            data = await self.web.restart_cgminer()
         except APIError:
             pass
         else:
@@ -127,9 +77,7 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
 
     async def send_config(self, config: MinerConfig, user_suffix: str = None) -> None:
         self.config = config
-        await self.send_web_command(
+        await self.web.update_pools(config.as_inno(user_suffix=user_suffix))
-            "updatePools", data=config.as_inno(user_suffix=user_suffix)
-        )
 
     ##################################################
     ### DATA GATHERING FUNCTIONS (get_{some_data}) ###
@@ -137,19 +85,18 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
 
     async def get_mac(
         self,
-        web_getAll: dict = None,  # noqa
+        web_get_all: dict = None,  # noqa
         web_overview: dict = None,  # noqa: named this way for automatic functionality
     ) -> Optional[str]:
-        web_all_data = web_getAll.get("all")
+        if not web_get_all and not web_overview:
-        if not web_all_data and not web_overview:
             try:
-                web_overview = await self.send_web_command("overview")
+                web_overview = await self.web.overview()
             except APIError:
                 pass
 
-        if web_all_data:
+        if web_get_all:
             try:
-                mac = web_all_data["mac"]
+                mac = web_get_all["mac"]
                 return mac.upper()
             except KeyError:
                 pass
@@ -168,7 +115,7 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
 
         if not web_type:
             try:
-                web_type = await self.send_web_command("type")
+                web_type = await self.web.type()
             except APIError:
                 pass
 
@@ -182,19 +129,18 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
     async def get_hashrate(
         self,
         api_summary: dict = None,
-        web_getAll: dict = None,  # noqa: named this way for automatic functionality
+        web_get_all: dict = None,  # noqa: named this way for automatic functionality
     ) -> Optional[float]:
-        web_all_data = web_getAll.get("all")
+        if not api_summary and not web_get_all:
-        if not api_summary and not web_all_data:
             try:
                 api_summary = await self.api.summary()
             except APIError:
                 pass
 
-        if web_all_data:
+        if web_get_all:
             try:
                 return round(
-                    float(web_all_data["total_hash"]["Hash Rate H"] / 1000000000000), 2
+                    float(web_get_all["total_hash"]["Hash Rate H"] / 1000000000000), 2
                 )
             except KeyError:
                 pass
@@ -208,9 +154,8 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
     async def get_hashboards(
         self,
         api_stats: dict = None,
-        web_getAll: dict = None,  # noqa: named this way for automatic functionality
+        web_get_all: dict = None,  # noqa: named this way for automatic functionality
     ) -> List[HashBoard]:
-        web_all_data = web_getAll.get("all")
         hashboards = [
             HashBoard(slot=i, expected_chips=self.nominal_chips)
             for i in range(self.ideal_hashboards)
@@ -222,13 +167,13 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
             except APIError:
                 pass
 
-        if not web_all_data:
+        if not web_get_all:
             try:
-                web_all_data = await self.send_web_command("getAll")
+                web_get_all = await self.web.get_all()
             except APIError:
                 pass
             else:
-                web_all_data = web_all_data["all"]
+                web_get_all = web_get_all["all"]
 
         if api_stats:
             if api_stats.get("STATS"):
@@ -242,9 +187,9 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
                         hashboards[idx].chips = chips
                         hashboards[idx].missing = False
 
-        if web_all_data:
+        if web_get_all:
-            if web_all_data.get("chain"):
+            if web_get_all.get("chain"):
-                for board in web_all_data["chain"]:
+                for board in web_get_all["chain"]:
                     idx = board.get("ASC")
                     if idx is not None:
                         temp = board.get("Temp min")
@@ -265,21 +210,20 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
 
     async def get_wattage(
         self,
-        web_getAll: dict = None,
+        web_get_all: dict = None,
         api_stats: dict = None,  # noqa: named this way for automatic functionality
     ) -> Optional[int]:
-        web_all_data = web_getAll.get("all")
+        if not web_get_all:
-        if not web_all_data:
             try:
-                web_all_data = await self.send_web_command("getAll")
+                web_get_all = await self.web.get_all()
             except APIError:
                 pass
             else:
-                web_all_data = web_all_data["all"]
+                web_get_all = web_get_all["all"]
 
-        if web_all_data:
+        if web_get_all:
             try:
-                return web_all_data["power"]
+                return web_get_all["power"]
             except KeyError:
                 pass
 
@@ -302,21 +246,20 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
 
     async def get_fans(
         self,
-        web_getAll: dict = None,  # noqa: named this way for automatic functionality
+        web_get_all: dict = None,  # noqa: named this way for automatic functionality
     ) -> List[Fan]:
-        web_all_data = web_getAll.get("all")
+        if not web_get_all:
-        if not web_all_data:
             try:
-                web_all_data = await self.send_web_command("getAll")
+                web_get_all = await self.web.get_all()
             except APIError:
                 pass
             else:
-                web_all_data = web_all_data["all"]
+                web_get_all = web_get_all["all"]
 
         fan_data = [Fan(), Fan(), Fan(), Fan()]
-        if web_all_data:
+        if web_get_all:
             try:
-                spd = web_all_data["fansSpeed"]
+                spd = web_get_all["fansSpeed"]
             except KeyError:
                 pass
             else:
@@ -353,21 +296,20 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
         return groups
 
     async def get_errors(
-        self, web_getErrorDetail: dict = None
+        self, web_get_error_detail: dict = None
     ) -> List[MinerErrorData]:  # noqa: named this way for automatic functionality
-        web_error_details = web_getErrorDetail
         errors = []
-        if not web_error_details:
+        if not web_get_error_detail:
             try:
-                web_error_details = await self.send_web_command("getErrorDetail")
+                web_get_error_detail = await self.web.get_error_detail()
             except APIError:
                 pass
 
-        if web_error_details:
+        if web_get_error_detail:
             try:
                 # only 1 error?
                 # TODO: check if this should be a loop, can't remember.
-                err = web_error_details["code"]
+                err = web_get_error_detail["code"]
             except KeyError:
                 pass
             else:
@@ -376,19 +318,18 @@ class CGMinerInnosiliconT3HPlus(CGMiner, InnosiliconT3HPlus):
                     errors.append(InnosiliconError(error_code=err))
         return errors
 
-    async def get_wattage_limit(self, web_getAll: dict = None) -> Optional[int]:
+    async def get_wattage_limit(self, web_get_all: dict = None) -> Optional[int]:
-        web_all_data = web_getAll.get("all")
+        if not web_get_all:
-        if not web_all_data:
             try:
-                web_all_data = await self.send_web_command("getAll")
+                web_get_all = await self.web.get_all()
             except APIError:
                 pass
             else:
-                web_all_data = web_all_data["all"]
+                web_get_all = web_get_all["all"]
 
-        if web_all_data:
+        if web_get_all:
             try:
-                level = web_all_data["running_mode"]["level"]
+                level = web_get_all["running_mode"]["level"]
             except KeyError:
                 pass
             else:

pyasic/web/Inno.py

@@ -0,0 +1,106 @@
+# ------------------------------------------------------------------------------
+#  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
+import warnings
+from typing import Union
+
+import httpx
+
+from pyasic.errors import APIError
+from pyasic.settings import PyasicSettings
+from pyasic.web import BaseWebAPI
+
+
+class InnosiliconWebAPI(BaseWebAPI):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.username = "admin"
+        self.pwd = PyasicSettings().global_innosilicon_password
+        self.jwt = None
+
+    async def auth(self):
+        async with httpx.AsyncClient() as client:
+            try:
+                auth = await client.post(
+                    f"http://{self.ip}/api/auth",
+                    data={"username": self.username, "password": self.pwd},
+                )
+            except httpx.HTTPError:
+                warnings.warn(f"Could not authenticate web token with miner: {self}")
+            else:
+                json_auth = auth.json()
+                self.jwt = json_auth.get("jwt")
+            return self.jwt
+
+    async def send_command(
+        self,
+        command: Union[str, bytes],
+        ignore_errors: bool = False,
+        allow_warning: bool = True,
+        **parameters: Union[str, int, bool],
+    ) -> dict:
+        if not self.jwt:
+            await self.auth()
+        async with httpx.AsyncClient() as client:
+            for i in range(PyasicSettings().miner_get_data_retries):
+                try:
+                    response = await client.post(
+                        f"http://{self.ip}/api/{command}",
+                        headers={"Authorization": "Bearer " + self.jwt},
+                        timeout=5,
+                        data=parameters,
+                    )
+                    json_data = response.json()
+                    if (
+                        not json_data.get("success")
+                        and "token" in json_data
+                        and json_data.get("token") == "expired"
+                    ):
+                        # refresh the token, retry
+                        await self.auth()
+                        continue
+                    if not json_data.get("success"):
+                        if json_data.get("msg"):
+                            raise APIError(json_data["msg"])
+                        elif json_data.get("message"):
+                            raise APIError(json_data["message"])
+                        raise APIError("Innosilicon web api command failed.")
+                    return json_data
+                except httpx.HTTPError:
+                    pass
+                except json.JSONDecodeError:
+                    pass
+
+    async def reboot(self) -> dict:
+        return await self.send_command("reboot")
+
+    async def restart_cgminer(self) -> dict:
+        return await self.send_command("restartCgMiner")
+
+    async def update_pools(self, conf: dict) -> dict:
+        return await self.send_command("updatePools", **conf)
+
+    async def overview(self) -> dict:
+        return await self.send_command("overview")
+
+    async def type(self) -> dict:
+        return await self.send_command("type")
+
+    async def get_all(self):
+        return await self.send_command("getAll")
+
+    async def get_error_detail(self):
+        return await self.send_command("getErrorDetail")

pyasic/web/S9.py

@@ -0,0 +1,58 @@
+# ------------------------------------------------------------------------------
+#  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 S9WebAPI(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 get_network_info(self) -> dict:
+        return await self.send_command("get_network_info")

pyasic/web/X17.py

@@ -54,12 +54,12 @@ class X17WebAPI(BaseWebAPI):
     async def get_system_info(self) -> dict:
         return await self.send_command("get_system_info")
 
-    async def blink(self, blink: bool):
+    async def blink(self, blink: bool) -> dict:
         if blink:
             return await self.send_command("blink", action="startBlink")
         return await self.send_command("blink", action="stopBlink")
 
-    async def reboot(self):
+    async def reboot(self) -> dict:
         return await self.send_command("reboot")
 
     async def get_blink_status(self) -> dict: