refactor: improve settings handling to not use a dataclass, and not use singleton.

pyasic/API/btminer.py

@@ -27,10 +27,10 @@ from typing import Literal, Union
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
 from passlib.handlers.md5_crypt import md5_crypt
 
+from pyasic import settings
 from pyasic.API import BaseMinerAPI
 from pyasic.errors import APIError
 from pyasic.misc import api_min_version
-from pyasic.settings import PyasicSettings
 
 ### IMPORTANT ###
 # you need to change the password of the miners using the Whatsminer
@@ -192,7 +192,7 @@ class BTMinerAPI(BaseMinerAPI):
         ip: str,
         api_ver: str = "0.0.0",
         port: int = 4028,
-        pwd: str = PyasicSettings().global_whatsminer_password,
+        pwd: str = settings.get("default_whatsminer_password", "admin"),
     ):
         super().__init__(ip, port)
         self.pwd = pwd

pyasic/__init__.py

@@ -13,6 +13,7 @@
 #  See the License for the specific language governing permissions and         -
 #  limitations under the License.                                              -
 # ------------------------------------------------------------------------------
+from pyasic import settings
 from pyasic.API.bmminer import BMMinerAPI
 from pyasic.API.bosminer import BOSMinerAPI
 from pyasic.API.btminer import BTMinerAPI
@@ -32,7 +33,6 @@ from pyasic.miners.base import AnyMiner
 from pyasic.miners.miner_factory import MinerFactory
 from pyasic.miners.miner_listener import MinerListener
 from pyasic.network import MinerNetwork
-from pyasic.settings import PyasicSettings
 
 __all__ = [
     "BMMinerAPI",
@@ -53,5 +53,5 @@ __all__ = [
     "MinerFactory",
     "MinerListener",
     "MinerNetwork",
-    "PyasicSettings",
+    "settings",
 ]

pyasic/logger/__init__.py

@@ -16,31 +16,29 @@
 
 import logging
 
-from pyasic.settings import PyasicSettings
-
 
 def init_logger():
-    if PyasicSettings().logfile:
+    # if PyasicSettings().logfile:
-        logging.basicConfig(
+    #     logging.basicConfig(
-            filename="logfile.txt",
+    #         filename="logfile.txt",
-            filemode="a",
+    #         filemode="a",
-            format="%(pathname)s:%(lineno)d in %(funcName)s\n[%(levelname)s][%(asctime)s](%(name)s) - %(message)s",
+    #         format="%(pathname)s:%(lineno)d in %(funcName)s\n[%(levelname)s][%(asctime)s](%(name)s) - %(message)s",
-            datefmt="%x %X",
+    #         datefmt="%x %X",
-        )
+    #     )
-    else:
+    # else:
-        logging.basicConfig(
+    logging.basicConfig(
-            format="%(pathname)s:%(lineno)d in %(funcName)s\n[%(levelname)s][%(asctime)s](%(name)s) - %(message)s",
+        format="%(pathname)s:%(lineno)d in %(funcName)s\n[%(levelname)s][%(asctime)s](%(name)s) - %(message)s",
-            datefmt="%x %X",
+        datefmt="%x %X",
-        )
+    )
 
     _logger = logging.getLogger()
 
-    if PyasicSettings().debug:
+    # if PyasicSettings().debug:
-        _logger.setLevel(logging.DEBUG)
+    #     _logger.setLevel(logging.DEBUG)
-        logging.getLogger("asyncssh").setLevel(logging.DEBUG)
+    #     logging.getLogger("asyncssh").setLevel(logging.DEBUG)
-    else:
+    # else:
-        _logger.setLevel(logging.WARNING)
+    _logger.setLevel(logging.WARNING)
-        logging.getLogger("asyncssh").setLevel(logging.WARNING)
+    logging.getLogger("asyncssh").setLevel(logging.WARNING)
 
     return _logger
 

pyasic/network/__init__.py

@@ -19,9 +19,9 @@ import ipaddress
 import logging
 from typing import AsyncIterator, List, Union
 
+from pyasic import settings
 from pyasic.miners.miner_factory import AnyMiner, miner_factory
 from pyasic.network.net_range import MinerNetworkRange
-from pyasic.settings import PyasicSettings
 
 
 class MinerNetwork:
@@ -108,7 +108,7 @@ class MinerNetwork:
         # clear cached miners
         miner_factory.clear_cached_miners()
 
-        limit = asyncio.Semaphore(PyasicSettings().network_scan_threads)
+        limit = asyncio.Semaphore(settings.get("network_scan_threads", 300))
         miners = await asyncio.gather(
             *[self.ping_and_get_miner(host, limit) for host in local_network.hosts()]
         )
@@ -136,7 +136,7 @@ class MinerNetwork:
         local_network = self.get_network()
 
         # create a list of scan tasks
-        limit = asyncio.Semaphore(PyasicSettings().network_scan_threads)
+        limit = asyncio.Semaphore(settings.get("network_scan_threads", 300))
         miners = asyncio.as_completed(
             [
                 loop.create_task(self.ping_and_get_miner(host, limit))
@@ -191,12 +191,12 @@ class MinerNetwork:
 async def ping_miner(
     ip: ipaddress.ip_address, port=4028
 ) -> Union[None, ipaddress.ip_address]:
-    for i in range(PyasicSettings().network_ping_retries):
+    for i in range(settings.get("network_ping_retries", 1)):
         try:
             connection_fut = asyncio.open_connection(str(ip), port)
             # get the read and write streams from the connection
             reader, writer = await asyncio.wait_for(
-                connection_fut, timeout=PyasicSettings().network_ping_timeout
+                connection_fut, timeout=settings.get("network_ping_timeout", 3)
             )
             # immediately close connection, we know connection happened
             writer.close()
@@ -220,12 +220,12 @@ async def ping_miner(
 async def ping_and_get_miner(
     ip: ipaddress.ip_address, port=4028
 ) -> Union[None, AnyMiner]:
-    for i in range(PyasicSettings().network_ping_retries):
+    for i in range(settings.get("network_ping_retries", 1)):
         try:
             connection_fut = asyncio.open_connection(str(ip), port)
             # get the read and write streams from the connection
             reader, writer = await asyncio.wait_for(
-                connection_fut, timeout=PyasicSettings().network_ping_timeout
+                connection_fut, timeout=settings.get("network_ping_timeout", 3)
             )
             # immediately close connection, we know connection happened
             writer.close()

pyasic/settings/__init__.py

@@ -14,27 +14,26 @@
 #  limitations under the License.                                              -
 # ------------------------------------------------------------------------------
 
-from dataclasses import dataclass
+from typing import Any
 
-from pyasic.misc import Singleton
+_settings = {  # defaults
+    "network_ping_retries": 1,
+    "network_ping_timeout": 3,
+    "network_scan_threads": 300,
+    "factory_get_retries": 1,
+    "get_data_retries": 1,
+    "default_whatsminer_password": "admin",
+    "default_innosilicon_password": "admin",
+    "default_antminer_password": "root",
+    "default_bosminer_password": "root",
+    "default_vnish_password": "admin",
+    "default_goldshell_password": "123456789",
+}
 
 
-@dataclass
+def get(key: str, other: Any = None) -> Any:
-class PyasicSettings(metaclass=Singleton):
+    return _settings.get(key, other)
-    network_ping_retries: int = 1
-    network_ping_timeout: int = 3
-    network_scan_threads: int = 300
 
-    miner_factory_get_version_retries: int = 1
 
-    miner_get_data_retries: int = 1
+def update(key: str, val: Any) -> Any:
-
+    _settings[key] = val
-    global_whatsminer_password = "admin"
-    global_innosilicon_password = "admin"
-    global_antminer_password = "root"
-    global_bosminer_password = "root"
-    global_vnish_password = "admin"
-    global_goldshell_password = "123456789"
-
-    debug: bool = False
-    logfile: bool = False

pyasic/web/antminer.py

@@ -19,14 +19,14 @@ from typing import Union
 
 import httpx
 
-from pyasic.settings import PyasicSettings
+from pyasic import settings
 from pyasic.web import BaseWebAPI
 
 
 class AntminerModernWebAPI(BaseWebAPI):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
-        self.pwd = PyasicSettings().global_antminer_password
+        self.pwd = settings.get("default_antminer_password", "root")
 
     async def send_command(
         self,
@@ -137,7 +137,7 @@ class AntminerModernWebAPI(BaseWebAPI):
 class AntminerOldWebAPI(BaseWebAPI):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
-        self.pwd = PyasicSettings().global_antminer_password
+        self.pwd = settings.get("default_antminer_password", "root")
 
     async def send_command(
         self,

pyasic/web/bosminer.py

@@ -18,15 +18,14 @@ from typing import Union
 
 import httpx
 
-from pyasic import APIError
+from pyasic import APIError, settings
-from pyasic.settings import PyasicSettings
 from pyasic.web import BaseWebAPI
 
 
 class BOSMinerWebAPI(BaseWebAPI):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
-        self.pwd = PyasicSettings().global_bosminer_password
+        self.pwd = settings.get("default_bosminer_password", "root")
 
     async def send_command(
         self,

pyasic/web/goldshell.py

@@ -19,7 +19,7 @@ from typing import Union
 
 import httpx
 
-from pyasic.settings import PyasicSettings
+from pyasic import settings
 from pyasic.web import BaseWebAPI
 
 
@@ -27,7 +27,7 @@ class GoldshellWebAPI(BaseWebAPI):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.username = "admin"
-        self.pwd = PyasicSettings().global_goldshell_password
+        self.pwd = settings.get("default_goldshell_password", "123456789")
         self.jwt = None
 
     async def auth(self):
@@ -72,7 +72,7 @@ class GoldshellWebAPI(BaseWebAPI):
         if not self.jwt:
             await self.auth()
         async with httpx.AsyncClient() as client:
-            for i in range(PyasicSettings().miner_get_data_retries):
+            for i in range(settings.get("get_data_retries", 1)):
                 try:
                     if parameters:
                         response = await client.put(

pyasic/web/inno.py

@@ -19,8 +19,8 @@ from typing import Union
 
 import httpx
 
+from pyasic import settings
 from pyasic.errors import APIError
-from pyasic.settings import PyasicSettings
 from pyasic.web import BaseWebAPI
 
 
@@ -28,7 +28,7 @@ class InnosiliconWebAPI(BaseWebAPI):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.username = "admin"
-        self.pwd = PyasicSettings().global_innosilicon_password
+        self.pwd = settings.get("default_innosilicon_password", "admin")
         self.jwt = None
 
     async def auth(self):
@@ -55,7 +55,7 @@ class InnosiliconWebAPI(BaseWebAPI):
         if not self.jwt:
             await self.auth()
         async with httpx.AsyncClient() as client:
-            for i in range(PyasicSettings().miner_get_data_retries):
+            for i in range(settings.get("get_data_retries", 1)):
                 try:
                     response = await client.post(
                         f"http://{self.ip}/api/{command}",

pyasic/web/vnish.py

@@ -19,7 +19,7 @@ from typing import Union
 
 import httpx
 
-from pyasic.settings import PyasicSettings
+from pyasic import settings
 from pyasic.web import BaseWebAPI
 
 
@@ -27,7 +27,7 @@ class VNishWebAPI(BaseWebAPI):
     def __init__(self, ip: str) -> None:
         super().__init__(ip)
         self.username = "admin"
-        self.pwd = PyasicSettings().global_vnish_password
+        self.pwd = settings.get("default_vnish_password", "admin")
         self.token = None
 
     async def auth(self):
@@ -59,7 +59,7 @@ class VNishWebAPI(BaseWebAPI):
         if not self.token:
             await self.auth()
         async with httpx.AsyncClient() as client:
-            for i in range(PyasicSettings().miner_get_data_retries):
+            for i in range(settings.get("get_data_retries", 1)):
                 try:
                     auth = self.token
                     if command.startswith("system"):