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

improve more typing

Browse Source
This commit is contained in:
UpstreamData
2022-07-18 14:46:17 -06:00
parent 43b4992cee
commit f8590b0c5f
3 changed files with 25 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
pyasic/misc/__init__.py
View File

70
pyasic/misc/bos.py
View File

@@ -1,70 +0,0 @@
import asyncio
from pyasic.network import MinerNetwork
from pyasic.miners._backends.bosminer import BOSMiner # noqa - Ignore access to _module
async def get_bos_bad_tuners(ip: str = "192.168.1.0", mask: int = 24):
# create a miner network
miner_network = MinerNetwork(ip, mask=mask)
# scan for miners
miners = await miner_network.scan_network_for_miners()
# create an empty list of tasks
tuner_tasks = []
# loop checks if the miner is a BOSMiner
for miner in miners:
# can only do this if its a subclass of BOSMiner
if BOSMiner in type(miner).__bases__:
tuner_tasks.append(_get_tuner_status(miner))
# run all the tuner status commands
tuner_status = await asyncio.gather(*tuner_tasks)
# create a list of all miners with bad board tuner status
bad_tuner_miners = []
for item in tuner_status:
# loop through and get each miners' bad board count
bad_boards = []
for board in item["tuner_status"]:
# if its not stable or still testing, its bad
if board["status"] not in [
"Stable",
"Testing performance profile",
"Tuning individual chips",
]:
# remove the part about the board refusing to start
bad_boards.append(
{
"board": board["board"],
"error": board["status"].replace(
"Hashchain refused to start: ", ""
),
}
)
# if this miner has bad boards, add it to the list of bad board miners
if len(bad_boards) > 0:
bad_tuner_miners.append({"ip": item["ip"], "boards": bad_boards})
# return the list of bad board miners
return bad_tuner_miners
async def _get_tuner_status(miner):
# run the tunerstatus command, since the miner will always be BOSMiner
tuner_status = await miner.api.tunerstatus()
# create a list to add the tuner data to
tuner_data = []
# if we have data, loop through to get the hashchain status
if tuner_status:
for board in tuner_status["TUNERSTATUS"][0]["TunerChainStatus"]:
tuner_data.append(
{"board": board["HashchainIndex"], "status": board["Status"]}
)
# return the data along with the IP or later tracking
return {"ip": str(miner.ip), "tuner_status": tuner_data}

37
pyasic/network/__init__.py
View File

@@ -1,7 +1,7 @@
import ipaddress
import asyncio
import logging
from typing import Union
from typing import Union, List, AsyncIterator
from pyasic.network.net_range import MinerNetworkRange
from pyasic.miners.miner_factory import MinerFactory, AnyMiner
@@ -40,7 +40,11 @@ class MinerNetwork:
return str(self.network)
def get_network(self) -> ipaddress.ip_network:
"""Get the network using the information passed to the MinerNetwork or from cache."""
"""Get the network using the information passed to the MinerNetwork or from cache.
Returns:
The proper network to be able to scan.
"""
# if we have a network cached already, use that
if self.network:
return self.network
@@ -68,13 +72,19 @@ class MinerNetwork:
self.network = ipaddress.ip_network(
f"{default_gateway}/{subnet_mask}", strict=False
)
logging.debug(f"Setting MinerNetwork: {self.network}")
return self.network
async def scan_network_for_miners(self) -> None or list:
"""Scan the network for miners, and return found miners as a list."""
async def scan_network_for_miners(self) -> List[AnyMiner]:
"""Scan the network for miners, and return found miners as a list.
Returns:
A list of found miners.
"""
# get the network
local_network = self.get_network()
print(f"Scanning {local_network} for miners...")
logging.debug(f"Scanning {local_network} for miners")
# clear cached miners
MinerFactory().clear_cached_miners()
@@ -103,16 +113,17 @@ class MinerNetwork:
# remove all None from the miner list
miners = list(filter(None, miners))
print(f"Found {len(miners)} connected miners...")
logging.debug(f"Found {len(miners)} connected miners")
# return the miner objects
return miners
async def scan_network_generator(self):
async def scan_network_generator(self) -> AsyncIterator[AnyMiner]:
"""
Scan the network for miners using an async generator.
Returns an asynchronous generator containing found miners.
Returns:
An asynchronous generator containing found miners.
"""
# get the current event loop
loop = asyncio.get_event_loop()
@@ -145,7 +156,7 @@ class MinerNetwork:
yield await miner
@staticmethod
async def ping_miner(ip: ipaddress.ip_address) -> None or ipaddress.ip_address:
async def ping_miner(ip: ipaddress.ip_address) -> Union[None, ipaddress.ip_address]:
tasks = [ping_miner(ip, port=port) for port in [4028, 4029, 8889]]
for miner in asyncio.as_completed(tasks):
miner = await miner
@@ -155,7 +166,7 @@ class MinerNetwork:
@staticmethod
async def ping_and_get_miner(
ip: ipaddress.ip_address,
) -> None or AnyMiner:
) -> Union[None, AnyMiner]:
tasks = [ping_and_get_miner(ip, port=port) for port in [4028, 4029, 8889]]
for miner in asyncio.as_completed(tasks):
miner = await miner
@@ -165,7 +176,7 @@ class MinerNetwork:
async def ping_miner(
ip: ipaddress.ip_address, port=4028
) -> None or ipaddress.ip_address:
) -> Union[None, ipaddress.ip_address]:
for i in range(PyasicSettings().network_ping_retries):
connection_fut = asyncio.open_connection(str(ip), port)
try:
@@ -192,7 +203,9 @@ async def ping_miner(
return
async def ping_and_get_miner(ip: ipaddress.ip_address, port=4028) -> None or AnyMiner:
async def ping_and_get_miner(
ip: ipaddress.ip_address, port=4028
) -> Union[None, AnyMiner]:
for i in range(PyasicSettings().network_ping_retries):
connection_fut = asyncio.open_connection(str(ip), port)
try: