refactor: improve validate_command_output, and move it out of the miner rpc api.

pyasic/misc/__init__.py

@@ -13,6 +13,8 @@
 #  See the License for the specific language governing permissions and         -
 #  limitations under the License.                                              -
 # ------------------------------------------------------------------------------
+from __future__ import annotations
+
 from copy import deepcopy
 
 from pyasic.errors import APIError
@@ -80,3 +82,28 @@ def merge_dicts(a: dict, b: dict) -> dict:
         else:
             result[b_key] = deepcopy(b_val)
     return result
+
+
+def validate_command_output(data: dict) -> tuple[bool, str | None]:
+    if "STATUS" in data.keys():
+        status = data["STATUS"]
+        if isinstance(status, str):
+            if status in ["RESTART"]:
+                return True, None
+            status = data
+        if isinstance(status, list):
+            status = status[0]
+
+        if status.get("STATUS") in ["S", "I"]:
+            return True, None
+        else:
+            return False, status.get("Msg", "Unknown error")
+    else:
+        for key in data.keys():
+            # make sure not to try to turn id into a dict
+            if key == "id":
+                continue
+            if "STATUS" in data[key][0].keys():
+                if data[key][0]["STATUS"][0]["STATUS"] not in ["S", "I"]:
+                    # this is an error
+                    return False, f"{key}: " + data[key][0]["STATUS"][0]["Msg"]

pyasic/rpc/base.py

@@ -23,6 +23,7 @@ import warnings
 from typing import Union
 
 from pyasic.errors import APIError, APIWarning
+from pyasic.misc import validate_command_output
 
 
 class BaseMinerRPCAPI:
@@ -85,7 +86,7 @@ class BaseMinerRPCAPI:
         data = self._load_api_data(data)
 
         # check for if the user wants to allow errors to return
-        validation = self._validate_command_output(data)
+        validation = validate_command_output(data)
         if not validation[0]:
             if not ignore_errors:
                 # validate the command succeeded
@@ -248,48 +249,6 @@ If you are sure you want to use this command please use API.send_command("{comma
 
         return ret_data
 
-    @staticmethod
-    def _validate_command_output(data: dict) -> tuple:
-        # check if the data returned is correct or an error
-        # if status isn't a key, it is a multicommand
-        if "STATUS" not in data.keys():
-            for key in data.keys():
-                # make sure not to try to turn id into a dict
-                if not key == "id":
-                    # make sure they succeeded
-                    if "STATUS" in data[key][0].keys():
-                        if data[key][0]["STATUS"][0]["STATUS"] not in ["S", "I"]:
-                            # this is an error
-                            return False, f"{key}: " + data[key][0]["STATUS"][0]["Msg"]
-        elif "id" not in data.keys():
-            if isinstance(data["STATUS"], list):
-                if data["STATUS"][0].get("STATUS", None) in ["S", "I"]:
-                    return True, None
-                else:
-                    return False, data["STATUS"][0]["Msg"]
-
-            elif isinstance(data["STATUS"], dict):
-                # new style X19 command
-                if data["STATUS"]["STATUS"] not in ["S", "I"]:
-                    return False, data["STATUS"]["Msg"]
-                return True, None
-
-            if data["STATUS"] not in ["S", "I"]:
-                return False, data["Msg"]
-        else:
-            # make sure the command succeeded
-            if isinstance(data["STATUS"], str):
-                if data["STATUS"] in ["RESTART"]:
-                    return True, None
-            elif isinstance(data["STATUS"], dict):
-                if data["STATUS"].get("STATUS") in ["S", "I"]:
-                    return True, None
-            elif data["STATUS"][0]["STATUS"] not in ("S", "I"):
-                # this is an error
-                if data["STATUS"][0]["STATUS"] not in ("S", "I"):
-                    return False, data["STATUS"][0]["Msg"]
-        return True, None
-
     @staticmethod
     def _load_api_data(data: bytes) -> dict:
         # some json from the API returns with a null byte (\x00) on the end

pyasic/web/auradine.py

@@ -24,6 +24,7 @@ import httpx
 
 from pyasic import settings
 from pyasic.errors import APIError
+from pyasic.misc import validate_command_output
 from pyasic.web.base import BaseWebAPI
 
 
@@ -87,7 +88,7 @@ class AuradineWebAPI(BaseWebAPI):
                             timeout=settings.get("api_function_timeout", 5),
                         )
                     json_data = response.json()
-                    validation = self._validate_command_output(json_data)
+                    validation = validate_command_output(json_data)
                     if not validation[0]:
                         if i == settings.get("get_data_retries", 1):
                             raise APIError(validation[1])
@@ -122,48 +123,6 @@ class AuradineWebAPI(BaseWebAPI):
 
         return data
 
-    @staticmethod
-    def _validate_command_output(data: dict) -> tuple:
-        # check if the data returned is correct or an error
-        # if status isn't a key, it is a multicommand
-        if "STATUS" not in data.keys():
-            for key in data.keys():
-                # make sure not to try to turn id into a dict
-                if not key == "id":
-                    # make sure they succeeded
-                    if "STATUS" in data[key][0].keys():
-                        if data[key][0]["STATUS"][0]["STATUS"] not in ["S", "I"]:
-                            # this is an error
-                            return False, f"{key}: " + data[key][0]["STATUS"][0]["Msg"]
-        elif "id" not in data.keys():
-            if isinstance(data["STATUS"], list):
-                if data["STATUS"][0].get("STATUS", None) in ["S", "I"]:
-                    return True, None
-                else:
-                    return False, data["STATUS"][0]["Msg"]
-
-            elif isinstance(data["STATUS"], dict):
-                # new style X19 command
-                if data["STATUS"]["STATUS"] not in ["S", "I"]:
-                    return False, data["STATUS"]["Msg"]
-                return True, None
-
-            if data["STATUS"] not in ["S", "I"]:
-                return False, data["Msg"]
-        else:
-            # make sure the command succeeded
-            if isinstance(data["STATUS"], str):
-                if data["STATUS"] in ["RESTART"]:
-                    return True, None
-            elif isinstance(data["STATUS"], dict):
-                if data["STATUS"].get("STATUS") in ["S", "I"]:
-                    return True, None
-            elif data["STATUS"][0]["STATUS"] not in ("S", "I"):
-                # this is an error
-                if data["STATUS"][0]["STATUS"] not in ("S", "I"):
-                    return False, data["STATUS"][0]["Msg"]
-        return True, None
-
     async def factory_reset(self) -> dict:
         return await self.send_command("factory-reset", privileged=True)
 

tests/api_tests/__init__.py

@@ -49,6 +49,23 @@ class TestAPIBase(unittest.IsolatedAsyncioTestCase):
         ).encode("utf-8")
 
     def get_success_value(self, command: str):
+        if self.api_str == "BTMiner" and command == "status":
+            return json.dumps(
+                {
+                    "STATUS": "S",
+                    "When": 1706287567,
+                    "Code": 131,
+                    "Msg": {
+                        "mineroff": "false",
+                        "mineroff_reason": "",
+                        "mineroff_time": "",
+                        "FirmwareVersion": "20230911.12.Rel",
+                        "power_mode": "",
+                        "hash_percent": "",
+                    },
+                    "Description": "",
+                }
+            ).encode("utf-8")
         return json.dumps(
             {
                 "STATUS": [