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

minor: reformat a bunch of files to try to make backends more cohesive, add static dict for get_data instead of using inspect, refactored graphql into a bosminer web API, and added supports_autotuning and supports_shutdown attributes to miners.

Browse Source
This commit is contained in:
UpstreamData
2023-03-28 11:39:03 -06:00
parent 63c8fe6868
commit aa6dc74471
239 changed files with 2106 additions and 4833 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
pyasic/web/S9.py
View File

@@ -25,7 +25,7 @@ from pyasic.web import BaseWebAPI
class S9WebAPI(BaseWebAPI):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.pwd = PyasicSettings().global_x17_password
self.pwd = PyasicSettings().global_antminer_password
async def send_command(
self,
@@ -51,6 +51,28 @@ class S9WebAPI(BaseWebAPI):
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")

72
pyasic/web/X17.py
View File

@@ -1,72 +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 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) -> dict:
if blink:
return await self.send_command("blink", action="startBlink")
return await self.send_command("blink", action="stopBlink")
async def reboot(self) -> dict:
return await self.send_command("reboot")
async def get_blink_status(self) -> dict:
return await self.send_command("blink", action="onPageLoaded")
async def get_miner_conf(self) -> dict:
return await self.send_command("get_miner_conf")
async def set_miner_conf(self, conf: dict) -> dict:
return await self.send_command("set_miner_conf", **conf)

6
pyasic/web/__init__.py
View File

@@ -46,6 +46,12 @@ class BaseWebAPI(ABC):
) -> dict:
pass
@abstractmethod
async def multicommand(
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
) -> dict:
pass
def _check_commands(self, *commands):
allowed_commands = self.get_commands()
return_commands = []

192
pyasic/web/antminer.py Normal file
View File

@@ -0,0 +1,192 @@
# ------------------------------------------------------------------------------
# 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 AntminerModernWebAPI(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=json.dumps(parameters), auth=auth # noqa
)
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)
async with httpx.AsyncClient() as client:
for command in commands:
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_miner_conf(self) -> dict:
return await self.send_command("get_miner_conf")
async def set_miner_conf(self, conf: dict) -> dict:
return await self.send_command("set_miner_conf", **conf)
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) -> dict:
return await self.send_command("reboot")
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")
async def summary(self) -> dict:
return await self.send_command("summary")
async def get_blink_status(self) -> dict:
return await self.send_command("get_blink_status")
async def set_network_conf(
self,
ip: str,
dns: str,
gateway: str,
subnet_mask: str,
hostname: str,
protocol: int,
) -> dict:
return await self.send_command(
"set_network_conf",
ipAddress=ip,
ipDns=dns,
ipGateway=gateway,
ipHost=hostname,
ipPro=protocol,
ipSub=subnet_mask,
)
class AntminerOldWebAPI(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}
auth = httpx.DigestAuth(self.username, self.pwd)
async with httpx.AsyncClient() as client:
for command in commands:
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 blink(self, blink: bool) -> dict:
if blink:
return await self.send_command("blink", action="startBlink")
return await self.send_command("blink", action="stopBlink")
async def reboot(self) -> dict:
return await self.send_command("reboot")
async def get_blink_status(self) -> dict:
return await self.send_command("blink", action="onPageLoaded")
async def get_miner_conf(self) -> dict:
return await self.send_command("get_miner_conf")
async def set_miner_conf(self, conf: dict) -> dict:
return await self.send_command("set_miner_conf", **conf)

106
pyasic/web/X19.py → pyasic/web/bosminer.py
View File

@@ -22,28 +22,39 @@ from pyasic.settings import PyasicSettings
from pyasic.web import BaseWebAPI
class X19WebAPI(BaseWebAPI):
class BOSMinerWebAPI(BaseWebAPI):
def __init__(self, ip: str) -> None:
super().__init__(ip)
self.pwd = PyasicSettings().global_x19_password
self.pwd = PyasicSettings().global_bosminer_password
def parse_command(self, graphql_command: Union[dict, set]) -> str:
if isinstance(graphql_command, dict):
data = []
for key in graphql_command:
if graphql_command[key]:
parsed = self.parse_command(graphql_command[key])
if key.startswith("... on"):
parsed = parsed.replace(",", "")
data.append(key + parsed)
else:
data.append(key)
else:
data = graphql_command
return "{" + ",".join(data) + "}"
async def send_command(
self,
command: Union[str, bytes],
command: dict,
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)
url = f"http://{self.ip}/graphql"
query = self.parse_command(command)
try:
async with httpx.AsyncClient() as client:
if parameters:
data = await client.post(
url, data=json.dumps(parameters), auth=auth # noqa
)
else:
data = await client.get(url, auth=auth)
await self.auth(client)
data = await client.post(url, json={"query": query})
except httpx.HTTPError:
pass
else:
@@ -53,47 +64,36 @@ class X19WebAPI(BaseWebAPI):
except json.decoder.JSONDecodeError:
pass
async def get_miner_conf(self) -> dict:
return await self.send_command("get_miner_conf")
async def set_miner_conf(self, conf: dict) -> dict:
return await self.send_command("set_miner_conf", **conf)
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) -> dict:
return await self.send_command("reboot")
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")
async def summary(self) -> dict:
return await self.send_command("summary")
async def get_blink_status(self) -> dict:
return await self.send_command("get_blink_status")
async def set_network_conf(
self,
ip: str,
dns: str,
gateway: str,
subnet_mask: str,
hostname: str,
protocol: int,
async def multicommand(
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
) -> dict:
return await self.send_command(
"set_network_conf",
ipAddress=ip,
ipDns=dns,
ipGateway=gateway,
ipHost=hostname,
ipPro=protocol,
ipSub=subnet_mask,
def merge(*d: dict):
ret = {}
for i in d:
if i:
for k in i:
if not k in ret:
ret[k] = i[k]
else:
ret[k] = merge(ret[k], i[k])
return None if ret == {} else ret
command = merge(*commands)
data = await self.send_command(command)
if not data:
data = {}
data["multicommand"] = False
return data
async def auth(self, client: httpx.AsyncClient) -> None:
url = f"http://{self.ip}/graphql"
await client.post(
url,
json={
"query": 'mutation{auth{login(username:"'
+ "root"
+ '", password:"'
+ self.pwd
+ '"){__typename}}}'
},
)

25
pyasic/web/goldshell.py
View File

@@ -19,7 +19,6 @@ from typing import Union
import httpx
from pyasic.errors import APIError
from pyasic.settings import PyasicSettings
from pyasic.web import BaseWebAPI
@@ -97,6 +96,30 @@ class GoldshellWebAPI(BaseWebAPI):
except TypeError:
await self.auth()
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
await self.auth()
async with httpx.AsyncClient() as client:
for command in commands:
try:
response = await client.get(
f"http://{self.ip}/mcb/{command}",
headers={"Authorization": "Bearer " + self.jwt},
timeout=5,
)
json_data = response.json()
data[command] = json_data
except httpx.HTTPError:
pass
except json.JSONDecodeError:
pass
except TypeError:
await self.auth()
return data
async def pools(self):
return await self.send_command("pools")

25
pyasic/web/inno.py
View File

@@ -14,7 +14,6 @@
# limitations under the License. -
# ------------------------------------------------------------------------------
import json
import urllib.parse
import warnings
from typing import Union
@@ -85,6 +84,30 @@ class InnosiliconWebAPI(BaseWebAPI):
except json.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
await self.auth()
async with httpx.AsyncClient() as client:
for command in commands:
try:
response = await client.post(
f"http://{self.ip}/api/{command}",
headers={"Authorization": "Bearer " + self.jwt},
timeout=5,
)
json_data = response.json()
data[command] = json_data
except httpx.HTTPError:
pass
except json.JSONDecodeError:
pass
except TypeError:
await self.auth()
return data
async def reboot(self) -> dict:
return await self.send_command("reboot")

9
pyasic/web/vnish.py
View File

@@ -99,6 +99,15 @@ class VNishWebAPI(BaseWebAPI):
except json.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
for command in commands:
data[command] = await self.send_command(command)
return data
async def restart_vnish(self) -> dict:
return await self.send_command("mining/restart", post=True)