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

update documentation and make BaseMiner and BaseMinerAPI unable to be instantiated directly. Add more unittests for miners.

Browse Source
This commit is contained in:
UpstreamData
2022-08-08 13:19:59 -06:00
parent 62238192ce
commit 8379359caf
24 changed files with 365 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
pyasic/API/__init__.py
View File

@@ -55,6 +55,11 @@ class BaseMinerAPI:
# ip address of the miner
self.ip = ipaddress.ip_address(ip)
def __new__(cls, *args, **kwargs):
if cls is BaseMinerAPI:
raise TypeError(f"Only children of '{cls.__name__}' may be instantiated")
return object.__new__(cls)
def get_commands(self) -> list:
"""Get a list of command accessible to a specific type of API on the miner.

5
pyasic/API/btminer.py
View File

@@ -19,6 +19,7 @@ import hashlib
import binascii
import base64
import logging
from typing import Union
from passlib.handlers.md5_crypt import md5_crypt
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
@@ -187,8 +188,8 @@ class BTMinerAPI(BaseMinerAPI):
async def send_command(
self,
command: str or bytes,
parameters: str or int or bool = None,
command: Union[str, bytes],
parameters: Union[str, int, bool] = None,
ignore_errors: bool = False,
**kwargs,
) -> dict:

9
pyasic/config/__init__.py
View File

@@ -287,6 +287,15 @@ class MinerConfig:
self.pool_groups = pool_groups
return self
def from_api(self, pools: list):
_pools = []
for pool in pools:
url = pool.get("URL")
user = pool.get("User")
_pools.append({"url": url, "user": user, "pass": "123"})
self.pool_groups = [_PoolGroup().from_dict({"pools": _pools})]
return self
def from_dict(self, data: dict):
"""Convert an output dict of this class back into usable data and save it to this class.

8
pyasic/data/__init__.py
View File

@@ -12,10 +12,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import Union
from typing import Union, List
from dataclasses import dataclass, field, asdict
from datetime import datetime
from .error_codes import X19Error, WhatsminerError, BraiinsOSError
@dataclass
class MinerData:
@@ -95,7 +97,9 @@ class MinerData:
pool_1_user: str = "Unknown"
pool_2_url: str = ""
pool_2_user: str = ""
errors: list = field(default_factory=list)
errors: List[Union[WhatsminerError, BraiinsOSError, X19Error]] = field(
default_factory=list
)
fault_light: Union[bool, None] = None
def __post_init__(self):

6
pyasic/data/error_codes/X19.py
View File

@@ -17,7 +17,11 @@ from dataclasses import dataclass, asdict
@dataclass
class X19Error:
"""A Dataclass to handle error codes of X19 miners."""
"""A Dataclass to handle error codes of X19 miners.
Attributes:
error_message: The error message as a string.
"""
error_message: str

6
pyasic/data/error_codes/bos.py
View File

@@ -17,7 +17,11 @@ from dataclasses import dataclass, asdict
@dataclass
class BraiinsOSError:
"""A Dataclass to handle error codes of BraiinsOS+ miners."""
"""A Dataclass to handle error codes of BraiinsOS+ miners.
Attributes:
error_message: The error message as a string.
"""
error_message: str

7
pyasic/data/error_codes/whatsminer.py
View File

@@ -17,7 +17,12 @@ from dataclasses import dataclass, field, asdict
@dataclass
class WhatsminerError:
"""A Dataclass to handle error codes of Whatsminers."""
"""A Dataclass to handle error codes of Whatsminers.
Attributes:
error_code: The error code as an int.
error_message: The error message as a string. Automatically found from the error code.
"""
error_code: int
error_message: str = field(init=False)

62
pyasic/miners/__init__.py
View File

@@ -15,11 +15,13 @@
import asyncssh
import logging
import ipaddress
from abc import ABC, abstractmethod
from pyasic.data import MinerData
from pyasic.config import MinerConfig
class BaseMiner:
class BaseMiner(ABC):
def __init__(self, *args) -> None:
self.ip = None
self.uname = "root"
@@ -34,6 +36,11 @@ class BaseMiner:
self.fan_count = 2
self.config = None
def __new__(cls, *args, **kwargs):
if cls is BaseMiner:
raise TypeError(f"Only children of '{cls.__name__}' may be instantiated")
return object.__new__(cls)
def __repr__(self):
return f"{'' if not self.api_type else self.api_type} {'' if not self.model else self.model}: {str(self.ip)}"
@@ -75,45 +82,56 @@ class BaseMiner:
except Exception as e:
raise e
@abstractmethod
async def fault_light_on(self) -> bool:
return False
pass
@abstractmethod
async def fault_light_off(self) -> bool:
return False
pass
async def send_file(self, src, dest):
async with (await self._get_ssh_connection()) as conn:
await asyncssh.scp(src, (conn, dest))
# async def send_file(self, src, dest):
# async with (await self._get_ssh_connection()) as conn:
# await asyncssh.scp(src, (conn, dest))
async def check_light(self):
return self.light
@abstractmethod
async def check_light(self) -> bool:
pass
# @abstractmethod
async def get_board_info(self):
return None
async def get_config(self):
return None
@abstractmethod
async def get_config(self) -> MinerConfig:
pass
async def get_hostname(self):
return None
@abstractmethod
async def get_hostname(self) -> str:
pass
async def get_model(self):
return None
@abstractmethod
async def get_model(self) -> str:
pass
async def reboot(self):
return False
@abstractmethod
async def reboot(self) -> bool:
pass
async def restart_backend(self):
return False
@abstractmethod
async def restart_backend(self) -> bool:
pass
async def send_config(self, *args, **kwargs):
return None
async def get_mac(self):
return None
@abstractmethod
async def get_mac(self) -> str:
pass
async def get_errors(self):
return None
@abstractmethod
async def get_errors(self) -> list:
pass
async def get_data(self) -> MinerData:
return MinerData(ip=str(self.ip))

20
pyasic/miners/_backends/bmminer.py
View File

@@ -154,6 +154,26 @@ class BMMiner(BaseMiner):
return True
return False
async def check_light(self) -> bool:
if not self.light:
self.light = False
return self.light
async def fault_light_off(self) -> bool:
return False
async def fault_light_on(self) -> bool:
return False
async def get_errors(self) -> list:
return []
async def get_mac(self) -> str:
return "00:00:00:00:00:00"
async def restart_backend(self) -> bool:
return False
async def get_data(self) -> MinerData:
"""Get data from the miner.

43
pyasic/miners/_backends/bosminer.py
View File

@@ -240,6 +240,49 @@ class BOSMiner(BaseMiner):
logging.debug(f"{self}: Restarting BOSMiner")
await conn.run("/etc/init.d/bosminer start")
async def check_light(self) -> bool:
if not self.light:
self.light = False
return self.light
async def get_errors(self) -> list:
tunerstatus = None
errors = []
try:
tunerstatus = await self.api.tunerstatus()
except Exception as e:
logging.warning(e)
if tunerstatus:
tuner = tunerstatus[0].get("TUNERSTATUS")
if tuner:
if len(tuner) > 0:
chain_status = tuner[0].get("TunerChainStatus")
if chain_status and len(chain_status) > 0:
board_map = {
0: "Left board",
1: "Center board",
2: "Right board",
}
offset = (
6
if chain_status[0]["HashchainIndex"] in [6, 7, 8]
else chain_status[0]["HashchainIndex"]
)
for board in chain_status:
_id = board["HashchainIndex"] - offset
if board["Status"] not in [
"Stable",
"Testing performance profile",
]:
_error = board["Status"]
_error = _error[0].lower() + _error[1:]
errors.append(
BraiinsOSError(f"{board_map[_id]} {_error}")
)
return errors
async def get_data(self) -> MinerData:
"""Get data from the miner.

33
pyasic/miners/_backends/bosminer_old.py
View File

@@ -15,6 +15,7 @@
import logging
import ipaddress
from typing import Union
from pyasic.API.bosminer import BOSMinerAPI
from pyasic.miners import BaseMiner
@@ -29,7 +30,7 @@ class BOSMinerOld(BaseMiner):
self.uname = "root"
self.pwd = "admin"
async def send_ssh_command(self, cmd: str) -> str or None:
async def send_ssh_command(self, cmd: str) -> Union[str, None]:
"""Send a command to the miner over ssh.
:return: Result of the command or None.
@@ -61,3 +62,33 @@ class BOSMinerOld(BaseMiner):
async def update_to_plus(self):
result = await self.send_ssh_command("opkg update && opkg install bos_plus")
return result
async def check_light(self) -> bool:
return False
async def fault_light_on(self) -> bool:
return False
async def fault_light_off(self) -> bool:
return False
async def get_config(self) -> None:
return None
async def get_errors(self) -> list:
return []
async def get_hostname(self) -> str:
return "?"
async def get_mac(self) -> str:
return "00:00:00:00:00:00"
async def get_model(self) -> str:
return "S9"
async def reboot(self) -> bool:
return False
async def restart_backend(self) -> bool:
return False

35
pyasic/miners/_backends/btminer.py
View File

@@ -23,6 +23,7 @@ from pyasic.API import APIError
from pyasic.data import MinerData
from pyasic.data.error_codes import WhatsminerError
from pyasic.config import MinerConfig
from pyasic.settings import PyasicSettings
@@ -99,6 +100,40 @@ class BTMiner(BaseMiner):
return str(mac).upper()
async def check_light(self) -> bool:
if not self.light:
self.light = False
return self.light
async def fault_light_off(self) -> bool:
return False
async def fault_light_on(self) -> bool:
return False
async def get_errors(self) -> list:
return []
async def reboot(self) -> bool:
return False
async def restart_backend(self) -> bool:
return False
async def get_config(self) -> MinerConfig:
pools = None
cfg = MinerConfig()
try:
pools = await self.api.pools()
except APIError as e:
logging.warning(e)
if pools:
if "POOLS" in pools.keys():
cfg = cfg.from_api(pools["POOLS"])
return cfg
async def get_data(self) -> MinerData:
"""Get data from the miner.

17
pyasic/miners/_backends/cgminer.py
View File

@@ -151,6 +151,23 @@ class CGMiner(BaseMiner):
self.config = result.stdout
return self.config
async def check_light(self) -> bool:
if not self.light:
self.light = False
return self.light
async def fault_light_off(self) -> bool:
return False
async def fault_light_on(self) -> bool:
return False
async def get_errors(self) -> list:
return []
async def get_mac(self) -> str:
return "00:00:00:00:00:00"
async def get_data(self) -> MinerData:
"""Get data from the miner.

1
pyasic/miners/miner_factory.py
View File

@@ -235,6 +235,7 @@ MINER_CLASSES = {
"Default": CGMinerAvalon1066,
"CGMiner": CGMinerAvalon1066,
},
"Unknown": {"Default": UnknownMiner},
}

26
pyasic/miners/unknown.py
View File

@@ -31,3 +31,29 @@ class UnknownMiner(BaseMiner):
async def get_hostname(self):
return "Unknown"
async def check_light(self) -> bool:
if not self.light:
self.light = False
return self.light
async def fault_light_off(self) -> bool:
return False
async def fault_light_on(self) -> bool:
return False
async def get_config(self) -> None:
return None
async def get_errors(self) -> list:
return []
async def get_mac(self) -> str:
return "00:00:00:00:00:00"
async def reboot(self) -> bool:
return False
async def restart_backend(self) -> bool:
return False

2
pyasic/tests/__init__.py
View File

@@ -13,6 +13,8 @@
# limitations under the License.
import unittest
from pyasic.tests.miners_tests import MinersTest
from pyasic.tests.network_tests import NetworkTest
if __name__ == "__main__":
unittest.main()

54
pyasic/tests/miners_tests/__init__.py Normal file
View File

@@ -0,0 +1,54 @@
# 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 unittest
from pyasic.miners.miner_factory import MINER_CLASSES
import inspect
import sys
class MinersTest(unittest.TestCase):
def test_miner_model_creation(self):
for miner_model in MINER_CLASSES.keys():
for miner_api in MINER_CLASSES[miner_model].keys():
with self.subTest(miner_model=miner_model, miner_api=miner_api):
miner = MINER_CLASSES[miner_model][miner_api]("0.0.0.0")
self.assertTrue(
isinstance(miner, MINER_CLASSES[miner_model][miner_api])
)
def test_miner_backend_backup_creation(self):
backends = inspect.getmembers(
sys.modules["pyasic.miners._backends"], inspect.isclass
)
for backend in backends:
miner_class = backend[1]
with self.subTest(miner_class=miner_class):
miner = miner_class("0.0.0.0")
self.assertTrue(isinstance(miner, miner_class))
def test_miner_type_creation_failure(self):
backends = inspect.getmembers(
sys.modules["pyasic.miners._types"], inspect.isclass
)
for backend in backends:
miner_class = backend[1]
with self.subTest(miner_class=miner_class):
with self.assertRaises(TypeError):
miner_class("0.0.0.0")
if __name__ == "__main__":
unittest.main()