From c2b6cc746875e2d830a4d23fb0b72dafbdc6031d Mon Sep 17 00:00:00 2001 From: UpstreamData Date: Fri, 26 Jan 2024 09:51:09 -0700 Subject: [PATCH] refactor: improve validate_command_output, and move it out of the miner rpc api. --- pyasic/misc/__init__.py | 27 ++++++++++++++++++++++ pyasic/rpc/base.py | 45 ++----------------------------------- pyasic/web/auradine.py | 45 ++----------------------------------- tests/api_tests/__init__.py | 17 ++++++++++++++ 4 files changed, 48 insertions(+), 86 deletions(-) diff --git a/pyasic/misc/__init__.py b/pyasic/misc/__init__.py index 9d9015a0..0ebc679d 100644 --- a/pyasic/misc/__init__.py +++ b/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"] diff --git a/pyasic/rpc/base.py b/pyasic/rpc/base.py index e68a32a6..af9b5e52 100644 --- a/pyasic/rpc/base.py +++ b/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 diff --git a/pyasic/web/auradine.py b/pyasic/web/auradine.py index 7c52e48d..bdfbd9f0 100644 --- a/pyasic/web/auradine.py +++ b/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) diff --git a/tests/api_tests/__init__.py b/tests/api_tests/__init__.py index 0e6969d5..ff6b0114 100644 --- a/tests/api_tests/__init__.py +++ b/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": [