Moved update_firmware from base.py to braiins_os.py

pyasic/miners/base.py

@@ -16,9 +16,6 @@
 import asyncio
 import ipaddress
 import warnings
-import httpx
-import tempfile
-import os
 from typing import List, Optional, Protocol, Tuple, Type, TypeVar, Union
 
 from pyasic.config import MinerConfig
@@ -522,58 +519,6 @@ class MinerProtocol(Protocol):
 
         return data
 
-    async def update_firmware(self, firmware_url: str) -> bool:
-        """
-        Update the firmware of the miner.
-
-        Parameters:
-            firmware_url: The URL of the firmware to download and apply to the miner.
-
-        Returns:
-            A boolean value indicating the success of the update process.
-        """
-        # Verify if the miner type is supported
-        # TODO
-
-        # Determine if a server URL is provided and query for firmware, otherwise use the direct URL
-        if firmware_url.startswith("http"):
-            latest_firmware_info = await self._fetch_firmware_info(firmware_url)
-        else:
-            latest_firmware_info = await self._query_firmware_server(firmware_url)
-        if not latest_firmware_info:
-            raise ValueError("Could not fetch firmware information.")
-
-        # Check the current firmware version on the miner
-        current_firmware_version = await self.get_fw_ver()
-        if current_firmware_version == latest_firmware_info['version']:
-            return True  # Firmware is already up to date
-
-        # Download the new firmware file if it's not a local file
-        if not firmware_url.startswith("file://"):
-            firmware_file_path = await self._download_firmware(latest_firmware_info['download_url'])
-            if not firmware_file_path:
-                raise IOError("Failed to download the firmware file.")
-        else:
-            firmware_file_path = latest_firmware_info['download_url']
-
-        # Transfer the firmware file to the miner
-        transfer_success = await self._transfer_firmware_to_miner(firmware_file_path)
-        if not transfer_success:
-            raise IOError("Failed to transfer the firmware file to the miner.")
-
-        # Apply the firmware update on the miner
-        update_success = await self._apply_firmware_update(firmware_file_path)
-        if not update_success:
-            raise IOError("Failed to apply the firmware update.")
-
-        # Reboot the miner
-        # TODO
-
-        # Verify the update success by polling the firmware version
-        # TODO
-
-        return True
-
 
 class BaseMiner(MinerProtocol):
     def __init__(self, ip: str) -> None:
@@ -593,165 +538,5 @@ class BaseMiner(MinerProtocol):
         if self._ssh_cls is not None:
             self.ssh = self._ssh_cls(ip)
 
-    async def _fetch_firmware_info(self, firmware_url: str) -> dict:
-        """
-        Fetch the latest firmware information from the given URL.
 
-        Parameters:
+AnyMiner = TypeVar("AnyMiner", bound=BaseMiner)
-            firmware_url: The URL to fetch the firmware information from.
-
-        Returns:
-            A dictionary containing the firmware version and download URL.
-        """
-        async with httpx.AsyncClient() as client:
-            response = await client.get(firmware_url)
-            if response.status_code != 200:
-                raise ConnectionError(f"Failed to fetch firmware info, status code: {response.status_code}")
-            firmware_info = response.json()
-            return {
-                'version': firmware_info['version'],
-                'download_url': firmware_info['download_url']
-            }
-
-    async def _download_firmware(self, download_url: str) -> str:
-        """
-        Download the firmware file from the given URL and save it to a temporary location.
-
-        Parameters:
-            download_url: The URL to download the firmware from.
-
-        Returns:
-            The file path to the downloaded firmware file.
-        """
-        async with httpx.AsyncClient() as client:
-            response = await client.get(download_url)
-            if response.status_code != 200:
-                raise ConnectionError(f"Failed to download firmware, status code: {response.status_code}")
-            
-            # Assuming the Content-Disposition header contains the filename
-            content_disposition = response.headers.get('Content-Disposition')
-            if not content_disposition:
-                raise ValueError("Could not determine the filename of the firmware download.")
-            
-            filename = content_disposition.split("filename=")[-1].strip().strip('"')
-            temp_dir = tempfile.gettempdir()
-            file_path = os.path.join(temp_dir, filename)
-            
-            with open(file_path, 'wb') as file:
-                file.write(response.content)
-            
-            return file_path
-
-    async def _query_firmware_server(self, server_url: str) -> dict:
-        """
-        Query the firmware server for available firmware files.
-
-        Parameters:
-            server_url: The URL of the server to query for firmware files.
-
-        Returns:
-            A dictionary containing the firmware version and download URL.
-        """
-        async with httpx.AsyncClient() as client:
-            response = await client.get(server_url)
-            if response.status_code != 200:
-                raise ConnectionError(f"Failed to query firmware server, status code: {response.status_code}")
-            firmware_files = response.json()
-            for firmware_file in firmware_files:
-                if self._is_appropriate_firmware(firmware_file['version']):
-                    return {
-                        'version': firmware_file['version'],
-                        'download_url': firmware_file['url']
-                    }
-            raise ValueError("No appropriate firmware file found on the server.")
-
-    async def _transfer_firmware_to_miner(self, firmware_file_path: str) -> bool:
-        """
-        Transfer the firmware file to the miner using SSH.
-
-        Parameters:
-            firmware_file_path: The file path to the firmware file that needs to be transferred.
-
-        Returns:
-            A boolean value indicating the success of the transfer.
-        """
-        import paramiko
-        from paramiko import SSHClient
-        from scp import SCPClient
-
-        # Retrieve SSH credentials from environment variables or secure storage
-        ssh_username = os.getenv('MINER_SSH_USERNAME')
-        ssh_password = os.getenv('MINER_SSH_PASSWORD')
-        ssh_key_path = os.getenv('MINER_SSH_KEY_PATH')
-
-        try:
-            # Establish an SSH client session using credentials
-            with SSHClient() as ssh_client:
-                ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
-
-                if ssh_key_path:
-                    ssh_client.connect(self.ip, username=ssh_username, key_filename=ssh_key_path)
-                else:
-                    ssh_client.connect(self.ip, username=ssh_username, password=ssh_password)
-
-                # Use SCPClient to transfer the firmware file to the miner
-                with SCPClient(ssh_client.get_transport()) as scp_client:
-                    scp_client.put(firmware_file_path, remote_path='/tmp/')
-
-                return True
-        except Exception as e:
-            logger.error(f"Failed to transfer firmware to miner {self.ip}: {e}")
-            return False
-
-    async def _apply_firmware_update(self, firmware_file_path: str) -> bool:
-        """
-        Apply the firmware update to the miner.
-
-        Parameterrs:
-            firmware_file_path: The path to the firmware file to be uploaded.
-        
-        Returns:
-            True if the firmware update was successful, False otherwise.
-        """
-        try:
-            # Identify the miner using the get_miner function
-            miner = await self.get_miner(self.ip)
-            if not miner:
-                logger.error(f"Failed to identify miner at {self.ip}")
-                return False
-
-            # Gather necessary data from the miner
-            miner_data = await miner.get_data()
-
-            # Logic to apply firmware update using SSH
-            if miner.ssh:
-                async with miner.ssh as ssh_client:
-                    await ssh_client.upload_file(firmware_file_path, "/tmp/firmware.bin")
-                    await ssh_client.execute_command("upgrade_firmware /tmp/firmware.bin")
-                    return True
-            # Logic to apply firmware update using RPC
-            elif miner.rpc:
-                await miner.rpc.upload_firmware(firmware_file_path)
-                await miner.rpc.execute_command("upgrade_firmware")
-                return True
-            else:
-                logger.error("No valid interface available to apply firmware update.")
-                return False
-        except Exception as e:
-            logger.error(f"Error applying firmware update: {e}")
-            return False
-
-    async def _is_appropriate_firmware(self, firmware_version: str) -> bool:
-            """
-            Determine if the given firmware version is appropriate for the miner.
-
-            Parameters:
-                firmware_version: The version of the firmware to evaluate.
-
-            Returns:
-                A boolean indicating if the firmware version is appropriate.
-            """
-            # TODO
-            return True
-
-AnyMiner = TypeVar("AnyMiner", bound=BaseMiner)

pyasic/ssh/braiins_os.py

@@ -93,3 +93,15 @@ class BOSMinerSSH(BaseSSH):
             str: Status of the LED.
         """
         return await self.send_command("cat /sys/class/leds/'Red LED'/delay_off")
+
+    async def upgrade_firmware(self, file: str):
+        """Upgrade the firmware of the BrainOS miner using a specified file.
+
+        This function upgrades the firmware of the BrainOS miner using a specified file.
+        It downloads the firmware file to a temporary location and performs the upgrade.
+
+        Args:
+            file (str): The URL or local path of the firmware file.
+
+        """
+        pass

tests/miners_tests/tests_base.py

@@ -1 +0,0 @@
-# Tests for update_firmware