From 472a15f4ca439a93064edec265a9a4b8da21116d Mon Sep 17 00:00:00 2001 From: UpstreamData Date: Wed, 1 Jun 2022 10:54:45 -0600 Subject: [PATCH] added fault light function for X17 BMMiner models --- miners/antminer/bmminer/X17/S17.py | 54 +++++++++++++++++++++++++ miners/antminer/bmminer/X17/S17_Plus.py | 54 +++++++++++++++++++++++++ miners/antminer/bmminer/X17/S17_Pro.py | 54 +++++++++++++++++++++++++ miners/antminer/bmminer/X17/S17e.py | 54 +++++++++++++++++++++++++ miners/antminer/bmminer/X17/T17.py | 54 +++++++++++++++++++++++++ miners/antminer/bmminer/X17/T17_Plus.py | 54 +++++++++++++++++++++++++ miners/antminer/bmminer/X17/T17e.py | 54 +++++++++++++++++++++++++ 7 files changed, 378 insertions(+) diff --git a/miners/antminer/bmminer/X17/S17.py b/miners/antminer/bmminer/X17/S17.py index d0b36faa..5f2ed7c6 100644 --- a/miners/antminer/bmminer/X17/S17.py +++ b/miners/antminer/bmminer/X17/S17.py @@ -1,8 +1,62 @@ from miners._backends import BMMiner # noqa - Ignore access to _module from miners._types import S17 # noqa - Ignore access to _module +import httpx + class BMMinerS17(BMMiner, S17): def __init__(self, ip: str) -> None: super().__init__(ip) self.ip = ip + + async def get_hostname(self) -> str or None: + hostname = None + url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.get(url, auth=auth) + if data.status_code == 200: + data = data.json() + if len(data.keys()) > 0: + if "hostname" in data.keys(): + hostname = data["hostname"] + return hostname + + async def fault_light_on(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + try: + await client.post(url, data={"action": "startBlink"}, auth=auth) + except httpx.ReadTimeout: + # Expected behaviour + pass + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False + + async def fault_light_off(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + await client.post(url, data={"action": "stopBlink"}, auth=auth) + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if not data["isBlinking"]: + return True + return False + + async def check_light(self): + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False diff --git a/miners/antminer/bmminer/X17/S17_Plus.py b/miners/antminer/bmminer/X17/S17_Plus.py index d976db5b..bdf3e01c 100644 --- a/miners/antminer/bmminer/X17/S17_Plus.py +++ b/miners/antminer/bmminer/X17/S17_Plus.py @@ -1,8 +1,62 @@ from miners._backends import BMMiner # noqa - Ignore access to _module from miners._types import S17Plus # noqa - Ignore access to _module +import httpx + class BMMinerS17Plus(BMMiner, S17Plus): def __init__(self, ip: str) -> None: super().__init__(ip) self.ip = ip + + async def get_hostname(self) -> str or None: + hostname = None + url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.get(url, auth=auth) + if data.status_code == 200: + data = data.json() + if len(data.keys()) > 0: + if "hostname" in data.keys(): + hostname = data["hostname"] + return hostname + + async def fault_light_on(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + try: + await client.post(url, data={"action": "startBlink"}, auth=auth) + except httpx.ReadTimeout: + # Expected behaviour + pass + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False + + async def fault_light_off(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + await client.post(url, data={"action": "stopBlink"}, auth=auth) + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if not data["isBlinking"]: + return True + return False + + async def check_light(self): + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False diff --git a/miners/antminer/bmminer/X17/S17_Pro.py b/miners/antminer/bmminer/X17/S17_Pro.py index 9916303a..6426ca80 100644 --- a/miners/antminer/bmminer/X17/S17_Pro.py +++ b/miners/antminer/bmminer/X17/S17_Pro.py @@ -1,8 +1,62 @@ from miners._backends import BMMiner # noqa - Ignore access to _module from miners._types import S17Pro # noqa - Ignore access to _module +import httpx + class BMMinerS17Pro(BMMiner, S17Pro): def __init__(self, ip: str) -> None: super().__init__(ip) self.ip = ip + + async def get_hostname(self) -> str or None: + hostname = None + url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.get(url, auth=auth) + if data.status_code == 200: + data = data.json() + if len(data.keys()) > 0: + if "hostname" in data.keys(): + hostname = data["hostname"] + return hostname + + async def fault_light_on(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + try: + await client.post(url, data={"action": "startBlink"}, auth=auth) + except httpx.ReadTimeout: + # Expected behaviour + pass + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False + + async def fault_light_off(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + await client.post(url, data={"action": "stopBlink"}, auth=auth) + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if not data["isBlinking"]: + return True + return False + + async def check_light(self): + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False diff --git a/miners/antminer/bmminer/X17/S17e.py b/miners/antminer/bmminer/X17/S17e.py index 13ddea36..32178f13 100644 --- a/miners/antminer/bmminer/X17/S17e.py +++ b/miners/antminer/bmminer/X17/S17e.py @@ -1,8 +1,62 @@ from miners._backends import BMMiner # noqa - Ignore access to _module from miners._types import S17e # noqa - Ignore access to _module +import httpx + class BMMinerS17e(BMMiner, S17e): def __init__(self, ip: str) -> None: super().__init__(ip) self.ip = ip + + async def get_hostname(self) -> str or None: + hostname = None + url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.get(url, auth=auth) + if data.status_code == 200: + data = data.json() + if len(data.keys()) > 0: + if "hostname" in data.keys(): + hostname = data["hostname"] + return hostname + + async def fault_light_on(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + try: + await client.post(url, data={"action": "startBlink"}, auth=auth) + except httpx.ReadTimeout: + # Expected behaviour + pass + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False + + async def fault_light_off(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + await client.post(url, data={"action": "stopBlink"}, auth=auth) + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if not data["isBlinking"]: + return True + return False + + async def check_light(self): + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False diff --git a/miners/antminer/bmminer/X17/T17.py b/miners/antminer/bmminer/X17/T17.py index eb6db4a6..22db3774 100644 --- a/miners/antminer/bmminer/X17/T17.py +++ b/miners/antminer/bmminer/X17/T17.py @@ -1,8 +1,62 @@ from miners._backends import BMMiner # noqa - Ignore access to _module from miners._types import T17 # noqa - Ignore access to _module +import httpx + class BMMinerT17(BMMiner, T17): def __init__(self, ip: str) -> None: super().__init__(ip) self.ip = ip + + async def get_hostname(self) -> str or None: + hostname = None + url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.get(url, auth=auth) + if data.status_code == 200: + data = data.json() + if len(data.keys()) > 0: + if "hostname" in data.keys(): + hostname = data["hostname"] + return hostname + + async def fault_light_on(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + try: + await client.post(url, data={"action": "startBlink"}, auth=auth) + except httpx.ReadTimeout: + # Expected behaviour + pass + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False + + async def fault_light_off(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + await client.post(url, data={"action": "stopBlink"}, auth=auth) + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if not data["isBlinking"]: + return True + return False + + async def check_light(self): + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False diff --git a/miners/antminer/bmminer/X17/T17_Plus.py b/miners/antminer/bmminer/X17/T17_Plus.py index 97f5e99b..c9edc5b6 100644 --- a/miners/antminer/bmminer/X17/T17_Plus.py +++ b/miners/antminer/bmminer/X17/T17_Plus.py @@ -1,8 +1,62 @@ from miners._backends import BMMiner # noqa - Ignore access to _module from miners._types import T17Plus # noqa - Ignore access to _module +import httpx + class BMMinerT17Plus(BMMiner, T17Plus): def __init__(self, ip: str) -> None: super().__init__(ip) self.ip = ip + + async def get_hostname(self) -> str or None: + hostname = None + url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.get(url, auth=auth) + if data.status_code == 200: + data = data.json() + if len(data.keys()) > 0: + if "hostname" in data.keys(): + hostname = data["hostname"] + return hostname + + async def fault_light_on(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + try: + await client.post(url, data={"action": "startBlink"}, auth=auth) + except httpx.ReadTimeout: + # Expected behaviour + pass + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False + + async def fault_light_off(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + await client.post(url, data={"action": "stopBlink"}, auth=auth) + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if not data["isBlinking"]: + return True + return False + + async def check_light(self): + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False diff --git a/miners/antminer/bmminer/X17/T17e.py b/miners/antminer/bmminer/X17/T17e.py index cfc20b12..90948104 100644 --- a/miners/antminer/bmminer/X17/T17e.py +++ b/miners/antminer/bmminer/X17/T17e.py @@ -1,8 +1,62 @@ from miners._backends import BMMiner # noqa - Ignore access to _module from miners._types import T17e # noqa - Ignore access to _module +import httpx + class BMMinerT17e(BMMiner, T17e): def __init__(self, ip: str) -> None: super().__init__(ip) self.ip = ip + + async def get_hostname(self) -> str or None: + hostname = None + url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.get(url, auth=auth) + if data.status_code == 200: + data = data.json() + if len(data.keys()) > 0: + if "hostname" in data.keys(): + hostname = data["hostname"] + return hostname + + async def fault_light_on(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + try: + await client.post(url, data={"action": "startBlink"}, auth=auth) + except httpx.ReadTimeout: + # Expected behaviour + pass + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False + + async def fault_light_off(self) -> bool: + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + await client.post(url, data={"action": "stopBlink"}, auth=auth) + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if not data["isBlinking"]: + return True + return False + + async def check_light(self): + url = f"http://{self.ip}/cgi-bin/blink.cgi" + auth = httpx.DigestAuth("root", "root") + async with httpx.AsyncClient() as client: + data = await client.post(url, data={"action": "onPageLoaded"}, auth=auth) + if data.status_code == 200: + data = data.json() + if data["isBlinking"]: + return True + return False