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

improve some type hinting compatibility

Browse Source
This commit is contained in:
UpstreamData
2022-07-20 11:19:13 -06:00
parent 2e3991355b
commit 3d3064d78e
2 changed files with 10 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pyasic/miners/antminer/bmminer/X17/X17.py
View File

@@ -15,6 +15,7 @@
from pyasic.miners._backends import BMMiner # noqa - Ignore access to _module from pyasic.miners._backends import BMMiner # noqa - Ignore access to _module
import httpx import httpx
from typing import Union
class BMMinerX17(BMMiner): class BMMinerX17(BMMiner):
@@ -22,7 +23,7 @@ class BMMinerX17(BMMiner):
super().__init__(ip) super().__init__(ip)
self.ip = ip self.ip = ip
async def get_hostname(self) -> str or None: async def get_hostname(self) -> Union[str, None]:
hostname = None hostname = None
url = f"http://{self.ip}/cgi-bin/get_system_info.cgi" url = f"http://{self.ip}/cgi-bin/get_system_info.cgi"
auth = httpx.DigestAuth("root", "root") auth = httpx.DigestAuth("root", "root")

13
pyasic/miners/miner_factory.py
View File

@@ -362,8 +362,8 @@ class MinerFactory(metaclass=Singleton):
self.miners = {} self.miners = {}
async def _get_miner_type( async def _get_miner_type(
self, ip: ipaddress.ip_address or str self, ip: Union[ipaddress.ip_address, str]
) -> Tuple[str or None, str or None, str or None]: ) -> Tuple[Union[str, None], Union[str, None], Union[str, None]]:
data = None data = None
model = None model = None
@@ -444,7 +444,8 @@ class MinerFactory(metaclass=Singleton):
model = data["minertype"] model = data["minertype"]
if "bmminer" in "\t".join(data.keys()): if "bmminer" in "\t".join(data.keys()):
api = "BMMiner" api = "BMMiner"
except: except Exception as e:
logging.debug(f"Unable to get miner - {e}")
return None, None, None return None, None, None
# if we have devdetails, we can get model data from there # if we have devdetails, we can get model data from there
@@ -540,7 +541,7 @@ class MinerFactory(metaclass=Singleton):
return model, api, ver return model, api, ver
@staticmethod @staticmethod
async def _validate_command(data: dict) -> Tuple[bool, str or None]: async def _validate_command(data: dict) -> Tuple[bool, Union[str, None]]:
"""Check if the returned command output is correctly formatted.""" """Check if the returned command output is correctly formatted."""
# check if the data returned is correct or an error # check if the data returned is correct or an error
if not data or data == {}: if not data or data == {}:
@@ -567,7 +568,9 @@ class MinerFactory(metaclass=Singleton):
return True, None return True, None
@staticmethod @staticmethod
async def _send_api_command(ip: ipaddress.ip_address or str, command: str) -> dict: async def _send_api_command(
ip: Union[ipaddress.ip_address, str], command: str
) -> dict:
try: try:
# get reader and writer streams # get reader and writer streams
reader, writer = await asyncio.open_connection(str(ip), 4028) reader, writer = await asyncio.open_connection(str(ip), 4028)