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

feature: add is_mining to all miner types.

Browse Source
This commit is contained in:
UpstreamData
2023-06-23 14:38:38 -06:00
parent 7d1a702804
commit f2abe9fd9e
8 changed files with 37 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
pyasic/miners/backends/bfgminer.py
View File

@@ -318,3 +318,6 @@ class BFGMiner(BaseMiner):
return round(ideal_rate, 2)
except (KeyError, IndexError):
pass
async def is_mining(self, *args, **kwargs) -> Optional[bool]:
return None

5
pyasic/miners/backends/bfgminer_goldshell.py
View File

@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from typing import List
from typing import List, Optional
from pyasic.config import MinerConfig
from pyasic.data import HashBoard
@@ -157,3 +157,6 @@ class BFGMinerGoldshell(BFGMiner):
print(self, api_devdetails)
return hashboards
async def is_mining(self, *args, **kwargs) -> Optional[bool]:
return None

3
pyasic/miners/backends/bmminer.py
View File

@@ -352,3 +352,6 @@ class BMMiner(BaseMiner):
return round(ideal_rate, 2)
except (KeyError, IndexError):
pass
async def is_mining(self, *args, **kwargs) -> Optional[bool]:
return None

3
pyasic/miners/backends/bosminer_old.py
View File

@@ -150,3 +150,6 @@ class BOSMinerOld(BOSMiner):
async def get_data(self, allow_warning: bool = False, **kwargs) -> MinerData:
return MinerData(ip=str(self.ip))
async def is_mining(self, *args, **kwargs) -> Optional[bool]:
return None

3
pyasic/miners/backends/cgminer.py
View File

@@ -375,3 +375,6 @@ class CGMiner(BaseMiner):
return round(ideal_rate, 2)
except (KeyError, IndexError):
pass
async def is_mining(self, *args, **kwargs) -> Optional[bool]:
return None

3
pyasic/miners/backends/cgminer_avalon.py
View File

@@ -371,3 +371,6 @@ class CGMinerAvalon(CGMiner):
except LookupError:
pass
return False
async def is_mining(self, *args, **kwargs) -> Optional[bool]:
return None

3
pyasic/miners/backends/vnish.py
View File

@@ -169,3 +169,6 @@ class VNish(BMMiner):
return fw_ver
except KeyError:
pass
async def is_mining(self, *args, **kwargs) -> Optional[bool]:
return None

166
tests/miners_tests/__init__.py
View File

@@ -23,157 +23,21 @@ from pyasic.miners.backends import CGMiner # noqa
from pyasic.miners.base import BaseMiner
from pyasic.miners.miner_factory import MINER_CLASSES, MinerFactory
# class MinersTest(unittest.TestCase):
# def test_miner_model_creation(self):
# warnings.filterwarnings("ignore")
# for miner_model in MINER_CLASSES.keys():
# for miner_api in MINER_CLASSES[miner_model].keys():
# with self.subTest(
# msg=f"Creation of miner using model={miner_model}, api={miner_api}",
# miner_model=miner_model,
# miner_api=miner_api,
# ):
# miner = MINER_CLASSES[miner_model][miner_api]("127.0.0.1")
# self.assertTrue(
# isinstance(miner, MINER_CLASSES[miner_model][miner_api])
# )
#
# def test_miner_backend_backup_creation(self):
# warnings.filterwarnings("ignore")
#
# backends = [
# list(
# inspect.getmembers(
# sys.modules[f"pyasic.miners.backends"], inspect.isclass
# )
# )
# ]
# backends = [item for sublist in backends for item in sublist]
# for backend in backends:
# miner_class = backend[1]
# with self.subTest(miner_class=miner_class):
# miner = miner_class("127.0.0.1")
# self.assertTrue(isinstance(miner, miner_class))
#
# def test_miner_type_creation_failure(self):
# warnings.filterwarnings("ignore")
#
# backends = [
# list(
# inspect.getmembers(
# sys.modules[f"pyasic.miners.{algo}._types"], inspect.isclass
# )
# )
# for algo in ["btc", "zec", "ltc"]
# ]
# backends = [item for sublist in backends for item in sublist]
# for backend in backends:
# miner_class = backend[1]
# with self.subTest(miner_class=miner_class):
# with self.assertRaises(TypeError):
# miner_class("127.0.0.1")
# with self.assertRaises(TypeError):
# BaseMiner("127.0.0.1")
#
# def test_miner_comparisons(self):
# miner_1 = CGMiner("1.1.1.1")
# miner_2 = CGMiner("2.2.2.2")
# miner_3 = CGMiner("1.1.1.1")
# self.assertEqual(miner_1, miner_3)
# self.assertGreater(miner_2, miner_1)
# self.assertLess(miner_3, miner_2)
#
#
# class MinerFactoryTest(unittest.TestCase):
# def test_miner_factory_creation(self):
# warnings.filterwarnings("ignore")
#
# self.assertDictEqual(MinerFactory().miners, {})
# miner_factory = MinerFactory()
# self.assertIs(MinerFactory(), miner_factory)
#
# def test_get_miner_generator(self):
# async def _coro():
# gen = MinerFactory().get_miner_generator([])
# miners = []
# async for miner in gen:
# miners.append(miner)
# return miners
#
# _miners = asyncio.run(_coro())
# self.assertListEqual(_miners, [])
#
# def test_miner_selection(self):
# warnings.filterwarnings("ignore")
#
# for miner_model in MINER_CLASSES.keys():
# with self.subTest():
# miner = MinerFactory()._select_miner_from_classes(
# "127.0.0.1", miner_model, None, None
# )
# self.assertIsInstance(miner, BaseMiner)
# for api in ["BOSMiner+", "BOSMiner", "CGMiner", "BTMiner", "BMMiner"]:
# with self.subTest():
# miner = MinerFactory()._select_miner_from_classes(
# "127.0.0.1", None, api, None
# )
# self.assertIsInstance(miner, BaseMiner)
#
# with self.subTest():
# miner = MinerFactory()._select_miner_from_classes(
# "127.0.0.1", "ANTMINER S17+", "Fake API", None
# )
# self.assertIsInstance(miner, BaseMiner)
#
# with self.subTest():
# miner = MinerFactory()._select_miner_from_classes(
# "127.0.0.1", "M30S", "BTMiner", "G20"
# )
# self.assertIsInstance(miner, BaseMiner)
#
# def test_validate_command(self):
# bad_test_data_returns = [
# {},
# {
# "cmd": [
# {
# "STATUS": [
# {"STATUS": "E", "Msg": "Command failed for some reason."}
# ]
# }
# ]
# },
# {"STATUS": "E", "Msg": "Command failed for some reason."},
# {
# "STATUS": [{"STATUS": "E", "Msg": "Command failed for some reason."}],
# "id": 1,
# },
# ]
# for data in bad_test_data_returns:
# with self.subTest():
#
# async def _coro(miner_ret):
# _data = await MinerFactory()._validate_command(miner_ret)
# return _data
#
# ret = asyncio.run(_coro(data))
# self.assertFalse(ret[0])
#
# good_test_data_returns = [
# {
# "STATUS": [{"STATUS": "S", "Msg": "Yay! Command succeeded."}],
# "id": 1,
# },
# ]
# for data in good_test_data_returns:
# with self.subTest():
#
# async def _coro(miner_ret):
# _data = await MinerFactory()._validate_command(miner_ret)
# return _data
#
# ret = asyncio.run(_coro(data))
# self.assertTrue(ret[0])
class MinersTest(unittest.TestCase):
def test_miner_model_creation(self):
warnings.filterwarnings("ignore")
for miner_model in MINER_CLASSES.keys():
for miner_api in MINER_CLASSES[miner_model].keys():
with self.subTest(
msg=f"Creation of miner using model={miner_model}, api={miner_api}",
miner_model=miner_model,
miner_api=miner_api,
):
miner = MINER_CLASSES[miner_model][miner_api]("127.0.0.1")
self.assertTrue(
isinstance(miner, MINER_CLASSES[miner_model][miner_api])
)
if __name__ == "__main__":