feat: Add update firmware for Avalon miner

pyasic/miners/backends/avalonminer.py

@@ -16,6 +16,9 @@
 
 import re
 from typing import List, Optional
+import httpx
+import aiofiles
+import logging
 
 from pyasic.data import AlgoHashRate, Fan, HashBoard, HashUnit
 from pyasic.errors import APIError
@@ -108,6 +111,58 @@ async def reboot(self) -> bool:
             return False
         return False
 
+    async def upgrade_firmware(self, ip: str, port: int, file: str) -> str:
+        """
+        Upgrade the firmware of an Avalon Miner.
+
+        Parameters:
+        ip (str): The IP address of the Avalon Miner.
+        port (int): The port number of the Avalon Miner's web interface.
+        file (str): Path to the firmware file to be uploaded.
+
+        Returns:
+        str: Result of the upgrade process.
+        """
+        if not file:
+            raise ValueError("File location must be provided for firmware upgrade.")
+
+        async with aiofiles.open(file, "rb") as f:
+            upgrade_contents = await f.read()
+
+        url = f"http://{ip}:{port}/cgi-bin/upgrade"
+        data = {'version': self.get_fw_ver()}
+
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    url,
+                    files={'firmware': upgrade_contents},
+                    data=data,
+                    auth=('root', 'root'),
+                    timeout=60
+                )
+
+                response_text = response.text
+
+                if 'Upgrade success' in response_text:
+                    logging.info("Firmware upgrade process completed successfully for Avalon Miner.")
+                    return "Firmware upgrade to version {} successful.".format(self.get_fw_ver())
+                else:
+                    return f"Firmware upgrade failed. Response: {response_text}"
+
+        except FileNotFoundError as e:
+            logging.error(f"File not found during the firmware upgrade process: {e}")
+            raise
+        except ValueError as e:
+            logging.error(f"Validation error occurred during the firmware upgrade process: {e}")
+            raise
+        except OSError as e:
+            logging.error(f"OS error occurred during the firmware upgrade process: {e}")
+            raise
+        except Exception as e:
+            logging.error(f"An unexpected error occurred during the firmware upgrade process: {e}", exc_info=True)
+            raise
+
     @staticmethod
     def parse_stats(stats):
         _stats_items = re.findall(".+?\\[*?]", stats)