organized better, started the structure for different miner types

miners/__init__.py

@@ -1,11 +1,12 @@
 from API.bmminer import BMMinerAPI
 from API.bosminer import BOSMinerAPI
 from API.cgminer import CGMinerAPI
+from API.btminer import BTMinerAPI
 from API.unknown import UnknownAPI
 import ipaddress
 
 
 class BaseMiner:
-    def __init__(self, ip: str, api: BMMinerAPI or BOSMinerAPI or CGMinerAPI or UnknownAPI) -> None:
+    def __init__(self, ip: str, api: BMMinerAPI | BOSMinerAPI | CGMinerAPI | BTMinerAPI | UnknownAPI) -> None:
         self.ip = ipaddress.ip_address(ip)
         self.api = api

miners/antminer/S9/bos.py

@@ -0,0 +1,118 @@
+from miners import BaseMiner
+from API.bosminer import BOSMinerAPI
+import asyncssh
+import toml
+from config.bos import bos_config_convert, general_config_convert_bos
+
+
+class BOSminer(BaseMiner):
+    def __init__(self, ip: str) -> None:
+        api = BOSMinerAPI(ip)
+        super().__init__(ip, api)
+        self.config = None
+        self.uname = 'root'
+        self.pwd = 'admin'
+
+    def __repr__(self) -> str:
+        return f"BOSminer: {str(self.ip)}"
+
+    async def _get_ssh_connection(self) -> asyncssh.connect:
+        """Create a new asyncssh connection"""
+        conn = await asyncssh.connect(str(self.ip), known_hosts=None, username=self.uname, password=self.pwd,
+                                      server_host_key_algs=['ssh-rsa'])
+        # return created connection
+        return conn
+
+    async def send_ssh_command(self, cmd: str) -> None:
+        """Sends SSH command to miner."""
+        # creates result variable
+        result = None
+
+        # runs the command on the miner
+        async with (await self._get_ssh_connection()) as conn:
+            # attempt to run command up to 3 times
+            for i in range(3):
+                try:
+                    # save result of the command
+                    result = await conn.run(cmd)
+                except Exception as e:
+                    print(f"{cmd} error: {e}")
+                    if i == 3:
+                        return
+                    continue
+
+        # let the user know the result of the command
+        if result is not None:
+            if result.stdout != "":
+                print(result.stdout)
+                if result.stderr != "":
+                    print("ERROR: " + result.stderr)
+            elif result.stderr != "":
+                print("ERROR: " + result.stderr)
+            else:
+                print(cmd)
+
+    async def fault_light_on(self) -> None:
+        """Sends command to turn on fault light on the miner."""
+        await self.send_ssh_command('miner fault_light on')
+
+    async def fault_light_off(self) -> None:
+        """Sends command to turn off fault light on the miner."""
+        await self.send_ssh_command('miner fault_light off')
+
+    async def restart_backend(self) -> None:
+        """Restart bosminer hashing process."""
+        await self.send_ssh_command('/etc/init.d/bosminer restart')
+
+    async def reboot(self) -> None:
+        """Reboots power to the physical miner."""
+        await self.send_ssh_command('/sbin/reboot')
+
+    async def get_config(self) -> None:
+        async with (await self._get_ssh_connection()) as conn:
+            async with conn.start_sftp_client() as sftp:
+                async with sftp.open('/etc/bosminer.toml') as file:
+                    toml_data = toml.loads(await file.read())
+        cfg = await bos_config_convert(toml_data)
+        self.config = cfg
+
+    async def get_hostname(self) -> str:
+        """Attempts to get hostname from miner."""
+        try:
+            async with (await self._get_ssh_connection()) as conn:
+                data = await conn.run('cat /proc/sys/kernel/hostname')
+            return data.stdout.strip()
+        except Exception as e:
+            print(self.ip, e)
+            return "BOSMiner Unknown"
+
+    async def send_config(self, yaml_config) -> None:
+        """Configures miner with yaml config."""
+        toml_conf = await general_config_convert_bos(yaml_config)
+        async with (await self._get_ssh_connection()) as conn:
+            async with conn.start_sftp_client() as sftp:
+                async with sftp.open('/etc/bosminer.toml', 'w+') as file:
+                    await file.write(toml_conf)
+            await conn.run("/etc/init.d/bosminer restart")
+
+    async def get_bad_boards(self) -> list:
+        """Checks for and provides list of non working boards."""
+        devs = await self.api.devdetails()
+        bad = 0
+        chains = devs['DEVDETAILS']
+        for chain in chains:
+            if chain['Chips'] == 0:
+                bad += 1
+        if bad > 0:
+            return [str(self.ip), bad]
+
+    async def check_good_boards(self) -> str:
+        """Checks for and provides list for working boards."""
+        devs = await self.api.devdetails()
+        bad = 0
+        chains = devs['DEVDETAILS']
+        for chain in chains:
+            if chain['Chips'] == 0:
+                bad += 1
+        if not bad > 0:
+            return str(self.ip)

miners/antminer/X17/bos.py

@@ -0,0 +1,118 @@
+from miners import BaseMiner
+from API.bosminer import BOSMinerAPI
+import asyncssh
+import toml
+from config.bos import bos_config_convert, general_config_convert_bos
+
+
+class BOSminer(BaseMiner):
+    def __init__(self, ip: str) -> None:
+        api = BOSMinerAPI(ip)
+        super().__init__(ip, api)
+        self.config = None
+        self.uname = 'root'
+        self.pwd = 'admin'
+
+    def __repr__(self) -> str:
+        return f"BOSminer: {str(self.ip)}"
+
+    async def _get_ssh_connection(self) -> asyncssh.connect:
+        """Create a new asyncssh connection"""
+        conn = await asyncssh.connect(str(self.ip), known_hosts=None, username=self.uname, password=self.pwd,
+                                      server_host_key_algs=['ssh-rsa'])
+        # return created connection
+        return conn
+
+    async def send_ssh_command(self, cmd: str) -> None:
+        """Sends SSH command to miner."""
+        # creates result variable
+        result = None
+
+        # runs the command on the miner
+        async with (await self._get_ssh_connection()) as conn:
+            # attempt to run command up to 3 times
+            for i in range(3):
+                try:
+                    # save result of the command
+                    result = await conn.run(cmd)
+                except Exception as e:
+                    print(f"{cmd} error: {e}")
+                    if i == 3:
+                        return
+                    continue
+
+        # let the user know the result of the command
+        if result is not None:
+            if result.stdout != "":
+                print(result.stdout)
+                if result.stderr != "":
+                    print("ERROR: " + result.stderr)
+            elif result.stderr != "":
+                print("ERROR: " + result.stderr)
+            else:
+                print(cmd)
+
+    async def fault_light_on(self) -> None:
+        """Sends command to turn on fault light on the miner."""
+        await self.send_ssh_command('miner fault_light on')
+
+    async def fault_light_off(self) -> None:
+        """Sends command to turn off fault light on the miner."""
+        await self.send_ssh_command('miner fault_light off')
+
+    async def restart_backend(self) -> None:
+        """Restart bosminer hashing process."""
+        await self.send_ssh_command('/etc/init.d/bosminer restart')
+
+    async def reboot(self) -> None:
+        """Reboots power to the physical miner."""
+        await self.send_ssh_command('/sbin/reboot')
+
+    async def get_config(self) -> None:
+        async with (await self._get_ssh_connection()) as conn:
+            async with conn.start_sftp_client() as sftp:
+                async with sftp.open('/etc/bosminer.toml') as file:
+                    toml_data = toml.loads(await file.read())
+        cfg = await bos_config_convert(toml_data)
+        self.config = cfg
+
+    async def get_hostname(self) -> str:
+        """Attempts to get hostname from miner."""
+        try:
+            async with (await self._get_ssh_connection()) as conn:
+                data = await conn.run('cat /proc/sys/kernel/hostname')
+            return data.stdout.strip()
+        except Exception as e:
+            print(self.ip, e)
+            return "BOSMiner Unknown"
+
+    async def send_config(self, yaml_config) -> None:
+        """Configures miner with yaml config."""
+        toml_conf = await general_config_convert_bos(yaml_config)
+        async with (await self._get_ssh_connection()) as conn:
+            async with conn.start_sftp_client() as sftp:
+                async with sftp.open('/etc/bosminer.toml', 'w+') as file:
+                    await file.write(toml_conf)
+            await conn.run("/etc/init.d/bosminer restart")
+
+    async def get_bad_boards(self) -> list:
+        """Checks for and provides list of non working boards."""
+        devs = await self.api.devdetails()
+        bad = 0
+        chains = devs['DEVDETAILS']
+        for chain in chains:
+            if chain['Chips'] == 0:
+                bad += 1
+        if bad > 0:
+            return [str(self.ip), bad]
+
+    async def check_good_boards(self) -> str:
+        """Checks for and provides list for working boards."""
+        devs = await self.api.devdetails()
+        bad = 0
+        chains = devs['DEVDETAILS']
+        for chain in chains:
+            if chain['Chips'] == 0:
+                bad += 1
+        if not bad > 0:
+            return str(self.ip)

miners/btminer.py

@@ -0,0 +1,26 @@
+from API.btminer import BTMinerAPI
+from miners import BaseMiner
+
+
+class BTMiner(BaseMiner):
+    def __init__(self, ip: str) -> None:
+        api = BTMinerAPI(ip)
+        super().__init__(ip, api)
+
+    def __repr__(self) -> str:
+        return f"BTMiner: {str(self.ip)}"
+
+    async def get_hostname(self) -> str:
+        return "BTMiner Unknown"
+
+    async def send_config(self):
+        return None  # ignore for now
+
+    async def restart_backend(self) -> None:
+        return None
+
+    async def reboot(self) -> None:
+        return None
+
+    async def get_config(self) -> None:
+        return None

miners/miner_factory.py

@@ -1,6 +1,7 @@
 from miners.bosminer import BOSminer
 from miners.bmminer import BMMiner
 from miners.cgminer import CGMiner
+from miners.btminer import BTMiner
 from miners.unknown import UnknownMiner
 from API import APIError
 import asyncio
@@ -13,31 +14,42 @@ class MinerFactory:
         self.miners = {}
 
     async def get_miner(self, ip: ipaddress.ip_address) -> BOSminer or CGMiner or BMMiner or UnknownMiner:
+        """Decide a miner type using the IP address of the miner."""
+        # check if the miner already exists in cache
         if ip in self.miners:
             return self.miners[ip]
+        # get the version data
         version_data = await self._get_version_data(ip)
         version = None
         if version_data:
+            # if we got version data, get a list of the keys so we can check type of miner
             version = list(version_data['VERSION'][0].keys())
         if version:
+            # check version against different return miner types
             if "BOSminer" in version or "BOSminer+" in version:
                 miner = BOSminer(str(ip))
             elif "CGMiner" in version:
                 miner = CGMiner(str(ip))
             elif "BMMiner" in version:
                 miner = BMMiner(str(ip))
+            elif "BTMiner" in version:
+                miner = BTMiner(str(ip))
             else:
                 miner = UnknownMiner(str(ip))
         else:
+            # if we don't get version, miner type is unknown
             miner = UnknownMiner(str(ip))
+        # save the miner in cache
         self.miners[ip] = miner
         return miner
 
     def clear_cached_miners(self):
+        """Clear the miner factory cache."""
         self.miners = {}
 
     @staticmethod
     async def _get_version_data(ip: ipaddress.ip_address) -> dict or None:
+        """Get data on the version of the miner to return the right miner."""
         for i in range(3):
             try:
                 # open a connection to the miner
@@ -116,7 +128,7 @@ class MinerFactory:
                             raise APIError(data["Msg"])
                         else:
                             if "whatsminer" in data["Description"]:
-                                return {'STATUS': [{'STATUS': 'S', 'When': 1635435132, 'Code': 22, 'Msg': 'CGMiner versions', 'Description': 'cgminer 4.9.2'}], 'VERSION': [{'CGMiner': '4.9.2', 'API': '3.7'}], 'id': 1}
+                                return {"VERSION": [{"BTMiner": data["Description"]}]}
                     # make sure the command succeeded
                     elif data["STATUS"][0]["STATUS"] not in ("S", "I"):
                         # this is an error