Merge pull request #158 from 1e9abhi1e10/update_firmware_3

feat: Add update firmware for Whatsminer

pyasic/miners/backends/btminer.py

@@ -16,6 +16,8 @@
 
 import logging
 from typing import List, Optional
+import aiofiles
+from pathlib import Path
 
 from pyasic.config import MinerConfig, MiningModeConfig
 from pyasic.data import AlgoHashRate, Fan, HashBoard, HashUnit
@@ -649,3 +651,41 @@ class BTMiner(StockFirmware):
                 return int(rpc_summary["SUMMARY"][0]["Elapsed"])
             except LookupError:
                 pass
+
+    async def upgrade_firmware(self, file: Path, token: str):
+        """
+        Upgrade the firmware of the Whatsminer device.
+
+        Args:
+            file (Path): The local file path of the firmware to be uploaded.
+            token (str): The authentication token for the firmware upgrade.
+
+        Returns:
+            str: Confirmation message after upgrading the firmware.
+        """
+        try:
+            logging.info("Starting firmware upgrade process for Whatsminer.")
+
+            if not file:
+                raise ValueError("File location must be provided for firmware upgrade.")
+
+            # Read the firmware file contents
+            async with aiofiles.open(file, "rb") as f:
+                upgrade_contents = await f.read()
+
+            result = await self.rpc.update_firmware(upgrade_contents)
+
+            logging.info("Firmware upgrade process completed successfully for Whatsminer.")
+            return result
+        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

pyasic/miners/backends/whatsminer.py

@@ -29,4 +29,4 @@ class M3X(BTMiner):
 
 
 class M2X(BTMiner):
-    pass
+    pass

pyasic/rpc/btminer.py

@@ -23,6 +23,7 @@ import json
 import logging
 import re
 from typing import Literal, Union
+import struct
 
 import httpx
 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
@@ -559,11 +560,24 @@ class BTMinerRPCAPI(BaseMinerRPCAPI):
         """
         return await self.send_privileged_command("set_normal_power")
 
-    async def update_firmware(self):  # noqa - static
-        """Not implemented."""
-        # to be determined if this will be added later
-        # requires a file stream in bytes
-        return NotImplementedError
+    async def update_firmware(self, firmware: bytes):
+        """Upgrade the firmware running on the miner and using the firmware passed in bytes.
+
+        Parameters:
+            firmware (bytes): The firmware binary data to be uploaded.
+
+        Returns:
+            bool: A boolean indicating the success of the firmware upgrade.
+
+        Raises:
+            APIError: If the miner is not ready for firmware update.
+        """
+        ready = await self.send_privileged_command("upgrade_firmware")
+        if not ready.get("Msg") == "ready":
+            raise APIError(f"Not ready for firmware update: {self}")
+        file_size = struct.pack("<I", len(firmware))
+        await self._send_bytes(file_size + firmware)
+        return True
 
     async def reboot(self, timeout: int = 10) -> dict:
         """Reboot the miner using the API.