CAT_TOM/pyasic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: refactor miner types.

Browse Source
This commit is contained in:
UpstreamData
2023-06-12 11:02:51 -06:00
parent 134c44aedc
commit e446176922
342 changed files with 628 additions and 2906 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
pyasic/web/S9.py
View File

@@ -1,80 +0,0 @@
# ------------------------------------------------------------------------------
# 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 S9WebAPI(BaseWebAPI):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.pwd = PyasicSettings().global_antminer_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 multicommand(
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
) -> dict:
data = {k: None for k in commands}
data["multicommand"] = True
auth = httpx.DigestAuth(self.username, self.pwd)
for command in commands:
async with httpx.AsyncClient() as client:
try:
url = f"http://{self.ip}/cgi-bin/{command}.cgi"
ret = await client.get(url, auth=auth)
except httpx.HTTPError:
pass
else:
if ret.status_code == 200:
try:
json_data = ret.json()
data[command] = json_data
except json.decoder.JSONDecodeError:
pass
return data
async def get_system_info(self) -> dict:
return await self.send_command("get_system_info")
async def get_network_info(self) -> dict:
return await self.send_command("get_network_info")

4
pyasic/web/antminer.py
View File

@@ -138,7 +138,9 @@ class AntminerOldWebAPI(BaseWebAPI):
try:
async with httpx.AsyncClient() as client:
if parameters:
data = await client.post(url, data=parameters, auth=auth, timeout=15)
data = await client.post(
url, data=parameters, auth=auth, timeout=15
)
else:
data = await client.get(url, auth=auth)
except httpx.HTTPError: