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

pyasic/config/__init__.py

@@ -455,7 +455,7 @@ class MinerConfig:
         logging.debug(f"MinerConfig - (As Inno) - Generating Innosilicon config")
         return self.pool_groups[0].as_inno(user_suffix=user_suffix)
 
-    def as_x19(self, user_suffix: str = None) -> str:
+    def as_x19(self, user_suffix: str = None) -> dict:
         """Convert the data in this class to a config usable by an X19 device.
 
         Parameters:
@@ -475,7 +475,7 @@ class MinerConfig:
         if self.fan_speed:
             cfg["bitmain-fan-ctrl"] = str(self.fan_speed)
 
-        return json.dumps(cfg)
+        return cfg
 
     def as_avalon(self, user_suffix: str = None) -> str:
         """Convert the data in this class to a config usable by an Avalonminer device.

pyasic/miners/_backends/X19.py

@@ -15,47 +15,23 @@
 # ------------------------------------------------------------------------------
 
 import asyncio
-import json
 from typing import List, Optional, Union
 
-import httpx
-
 from pyasic.API import APIError
 from pyasic.config import MinerConfig, X19PowerMode
 from pyasic.data.error_codes import MinerErrorData, X19Error
 from pyasic.miners._backends import BMMiner  # noqa - Ignore access to _module
-from pyasic.settings import PyasicSettings
+from pyasic.web.X19 import X19WebAPI
 
 
 class X19(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 = X19WebAPI(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_config(self) -> MinerConfig:
-        data = await self.send_web_command("get_miner_conf")
+        data = await self.web.get_miner_conf()
         if data:
             self.config = MinerConfig().from_raw(data)
         return self.config
@@ -63,9 +39,7 @@ class X19(BMMiner):
     async def send_config(self, config: MinerConfig, user_suffix: str = None) -> None:
         self.config = config
         conf = config.as_x19(user_suffix=user_suffix)
-        await self.send_web_command(
+        await self.web.set_miner_conf(conf)
-            "set_miner_conf", params=conf  # noqa: ignore conf being a str
-        )
 
         for i in range(7):
             data = await self.get_config()
@@ -74,27 +48,21 @@ class X19(BMMiner):
             await asyncio.sleep(1)
 
     async def fault_light_on(self) -> bool:
-        data = await self.send_web_command(
+        data = await self.web.blink(blink=True)
-            "blink",
-            params=json.dumps({"blink": "true"}),  # noqa - ignore params being a str
-        )
         if data:
             if data.get("code") == "B000":
                 self.light = True
         return self.light
 
     async def fault_light_off(self) -> bool:
-        data = await self.send_web_command(
+        data = await self.web.blink(blink=False)
-            "blink",
-            params=json.dumps({"blink": "false"}),  # noqa - ignore params being a str
-        )
         if data:
             if data.get("code") == "B100":
                 self.light = True
         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
@@ -113,7 +81,7 @@ class X19(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:
@@ -121,14 +89,14 @@ class X19(BMMiner):
 
     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:
@@ -136,7 +104,7 @@ class X19(BMMiner):
 
     async def get_errors(self) -> List[MinerErrorData]:
         errors = []
-        data = await self.send_web_command("summary")
+        data = await self.web.summary()
         if data:
             try:
                 for item in data["SUMMARY"][0]["status"]:
@@ -153,7 +121,7 @@ class X19(BMMiner):
         if self.light:
             return self.light
         try:
-            data = await self.send_web_command("get_blink_status")
+            data = await self.web.get_blink_status()
             if data:
                 self.light = data["blink"]
         except KeyError:
@@ -193,42 +161,34 @@ class X19(BMMiner):
     ):
         if not hostname:
             hostname = await self.get_hostname()
-        payload = {
+        await self.web.set_network_conf(
-            "ipAddress": ip,
+            ip=ip,
-            "ipDns": dns,
+            dns=dns,
-            "ipGateway": gateway,
+            gateway=gateway,
-            "ipHost": hostname,
+            subnet_mask=subnet_mask,
-            "ipPro": 2,  # static
+            hostname=hostname,
-            "ipSub": subnet_mask,
+            protocol=2,
-        }
+        )
-        await self.send_web_command("set_network_conf", params=payload)
 
     async def set_dhcp(self, hostname: str = None):
         if not hostname:
             hostname = await self.get_hostname()
-        payload = {
+        await self.web.set_network_conf(
-            "ipAddress": "",
+            ip=ip, dns="", gateway="", subnet_mask="", hostname=hostname, protocol=1
-            "ipDns": "",
+        )
-            "ipGateway": "",
-            "ipHost": hostname,
-            "ipPro": 1,  # DHCP
-            "ipSub": "",
-        }
-        await self.send_web_command("set_network_conf", params=payload)
 
     async def set_hostname(self, hostname: str):
-        cfg = await self.send_web_command("get_network_info")
+        cfg = await self.web.get_network_info()
         dns = cfg["conf_dnsservers"]
         gateway = cfg["conf_gateway"]
         ip = cfg["conf_ipaddress"]
         subnet_mask = cfg["conf_netmask"]
         protocol = 1 if cfg["conf_nettype"] == "DHCP" else 2
-        payload = {
+        await self.web.set_network_conf(
-            "ipAddress": ip,
+            ip=ip,
-            "ipDns": dns,
+            dns=dns,
-            "ipGateway": gateway,
+            gateway=gateway,
-            "ipHost": hostname,
+            subnet_mask=subnet_mask,
-            "ipPro": protocol,
+            hostname=hostname,
-            "ipSub": subnet_mask,
+            protocol=protocol,
-        }
+        )
-        await self.send_web_command("set_network_conf", params=payload)

pyasic/miners/base.py

@@ -31,9 +31,8 @@ from pyasic.errors import APIError
 class BaseMiner(ABC):
     def __init__(self, *args, **kwargs) -> None:
         self.ip = None
-        self.uname = "root"
-        self.pwd = "admin"
         self.api = None
+        self.web = None
         self.api_type = None
         self.api_ver = None
         self.fw_ver = None

pyasic/web/X19.py

@@ -0,0 +1,99 @@
+# ------------------------------------------------------------------------------
+#  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 X19WebAPI(BaseWebAPI):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.pwd = PyasicSettings().global_x19_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=json.dumps(parameters), auth=auth  # noqa
+                    )
+                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_miner_conf(self):
+        return await self.send_command("get_miner_conf")
+
+    async def set_miner_conf(self, conf: dict):
+        return await self.send_command("set_miner_conf", **conf)
+
+    async def blink(self, blink: bool):
+        if blink:
+            return await self.send_command("blink", blink="true")
+        return await self.send_command("blink", blink="false")
+
+    async def reboot(self):
+        return await self.send_command("reboot")
+
+    async def get_system_info(self):
+        return await self.send_command("get_system_info")
+
+    async def get_network_info(self):
+        return await self.send_command("get_network_info")
+
+    async def summary(self):
+        return await self.send_command("summary")
+
+    async def get_blink_status(self):
+        return await self.send_command("get_blink_status")
+
+    async def set_network_conf(
+        self,
+        ip: str,
+        dns: str,
+        gateway: str,
+        subnet_mask: str,
+        hostname: str,
+        protocol: int,
+    ):
+        return await self.send_command(
+            "set_network_conf",
+            ipAddress=ip,
+            ipDns=dns,
+            ipGateway=gateway,
+            ipHost=hostname,
+            ipPro=protocol,
+            ipSub=subnet_mask,
+        )

pyasic/web/__init__.py

@@ -0,0 +1,88 @@
+# ------------------------------------------------------------------------------
+#  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 ipaddress
+import warnings
+from abc import ABC, abstractmethod
+from typing import Union
+
+from pyasic.errors import APIWarning
+from pyasic.settings import PyasicSettings
+
+
+class BaseWebAPI(ABC):
+    def __init__(self, ip: str) -> None:
+        # ip address of the miner
+        self.ip = ipaddress.ip_address(ip)
+        self.username = "root"
+        self.pwd = "root"
+
+    def __new__(cls, *args, **kwargs):
+        if cls is BaseWebAPI:
+            raise TypeError(f"Only children of '{cls.__name__}' may be instantiated")
+        return object.__new__(cls)
+
+    def __repr__(self):
+        return f"{self.__class__.__name__}: {str(self.ip)}"
+
+    @abstractmethod
+    async def send_command(
+        self,
+        command: Union[str, bytes],
+        ignore_errors: bool = False,
+        allow_warning: bool = True,
+        **parameters: Union[str, int, bool],
+    ) -> dict:
+        pass
+
+    def _check_commands(self, *commands):
+        allowed_commands = self.get_commands()
+        return_commands = []
+        for command in [*commands]:
+            if command in allowed_commands:
+                return_commands.append(command)
+            else:
+                warnings.warn(
+                    f"""Removing incorrect command: {command}
+If you are sure you want to use this command please use WebAPI.send_command("{command}", ignore_errors=True) instead.""",
+                    APIWarning,
+                )
+        return return_commands
+
+    @property
+    def commands(self) -> list:
+        return self.get_commands()
+
+    def get_commands(self) -> list:
+        """Get a list of command accessible to a specific type of web API on the miner.
+
+        Returns:
+            A list of all web commands that the miner supports.
+        """
+        return [
+            func
+            for func in
+            # each function in self
+            dir(self)
+            if not func == "commands"
+            if callable(getattr(self, func)) and
+            # no __ or _ methods
+            not func.startswith("__") and not func.startswith("_") and
+            # remove all functions that are in this base class
+            func
+            not in [
+                func for func in dir(BaseWebAPI) if callable(getattr(BaseWebAPI, func))
+            ]
+        ]