From ba0bb73aa32efd33d086db42997e06bb154c838e Mon Sep 17 00:00:00 2001 From: UpstreamData Date: Mon, 27 Feb 2023 15:44:58 -0700 Subject: [PATCH] format: swap X17 over to the new web api format. --- pyasic/miners/antminer/bmminer/X17/X17.py | 52 ++++-------------- pyasic/web/X17.py | 66 +++++++++++++++++++++++ pyasic/web/X19.py | 18 +++---- 3 files changed, 86 insertions(+), 50 deletions(-) create mode 100644 pyasic/web/X17.py diff --git a/pyasic/miners/antminer/bmminer/X17/X17.py b/pyasic/miners/antminer/bmminer/X17/X17.py index 8fc10c45..8f45e346 100644 --- a/pyasic/miners/antminer/bmminer/X17/X17.py +++ b/pyasic/miners/antminer/bmminer/X17/X17.py @@ -14,45 +14,21 @@ # limitations under the License. - # ------------------------------------------------------------------------------ -import json -from typing import Optional, Union - -import httpx +from typing import Union 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.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 + self.web = X17WebAPI(ip) 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( - "blink", params={"action": "onPageLoaded"} - ) + data = await self.web.get_blink_status() 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( - "blink", params={"action": "onPageLoaded"} - ) + data = await self.web.get_blink_status() 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( - "blink", params={"action": "onPageLoaded"} - ) + data = await self.web.get_blink_status() 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: diff --git a/pyasic/web/X17.py b/pyasic/web/X17.py new file mode 100644 index 00000000..88a2b7a1 --- /dev/null +++ b/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") diff --git a/pyasic/web/X19.py b/pyasic/web/X19.py index 3c6d8912..cb92a6df 100644 --- a/pyasic/web/X19.py +++ b/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,