feat: Add update firmware for VNish miner

pyasic/miners/backends/vnish.py

@@ -15,6 +15,8 @@
 # ------------------------------------------------------------------------------
 
 from typing import Optional
+import logging
+from pathlib import Path
 
 from pyasic import MinerConfig
 from pyasic.data import AlgoHashRate, HashUnit
@@ -256,3 +258,27 @@ async def get_config(self) -> MinerConfig:
             return self.config
         self.config = MinerConfig.from_vnish(web_settings)
         return self.config
+
+    async def upgrade_firmware(self, file: Path, keep_settings: bool = True) -> str:
+        """
+        Upgrade the firmware of the VNish Miner device.
+        Args:
+            file (Path): Path to the firmware file.
+            keep_settings (bool): Whether to keep the current settings after the update.
+        Returns:
+            str: Result of the upgrade process.
+        """
+        if not file:
+            raise ValueError("File location must be provided for firmware upgrade.")
+        try:
+            result = await self.web.update_firmware(file=file, keep_settings=keep_settings)
+            if result.get("success"):
+                logging.info("Firmware upgrade process completed successfully for VNish Miner.")
+                return "Firmware upgrade completed successfully."
+            else:
+                error_message = result.get("message", "Unknown error")
+                logging.error(f"Firmware upgrade failed. Response: {error_message}")
+                return f"Firmware upgrade failed. Response: {error_message}"
+        except Exception as e:
+            logging.error(f"An error occurred during the firmware upgrade process: {e}", exc_info=True)
+            raise

pyasic/web/vnish.py

@@ -18,7 +18,8 @@
 import json
 import warnings
 from typing import Any
-
+import aiofiles
+from pathlib import Path
 import httpx
 
 from pyasic import settings
@@ -69,18 +70,21 @@ async def send_command(
                     if command.startswith("system"):
                         auth = "Bearer " + self.token
 
+                    url = f"http://{self.ip}:{self.port}/api/v1/{command}"
+                    headers = {"Authorization": auth}
+
                     if post:
                         response = await client.post(
-                            f"http://{self.ip}:{self.port}/api/v1/{command}",
-                            headers={"Authorization": auth},
-                            timeout=settings.get("api_function_timeout", 5),
+                            url,
+                            headers=headers,
+                            timeout=settings.get("api_function_timeout", 30),
                             json=parameters,
                         )
                     else:
                         response = await client.get(
-                            f"http://{self.ip}:{self.port}/api/v1/{command}",
-                            headers={"Authorization": auth},
-                            timeout=settings.get("api_function_timeout", 5),
+                            url,
+                            headers=headers,
+                            timeout=settings.get("api_function_timeout", 30),
                         )
                     if not response.status_code == 200:
                         # refresh the token, retry
@@ -91,7 +95,10 @@ async def send_command(
                         return json_data
                     return {"success": True}
                 except (httpx.HTTPError, json.JSONDecodeError, AttributeError):
-                    pass
+                    if not ignore_errors:
+                        raise
+
+        return {"success": False, "message": "Command failed after retries"}
 
     async def multicommand(
         self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
@@ -143,3 +150,16 @@ async def autotune_presets(self) -> dict:
 
     async def find_miner(self) -> dict:
         return await self.send_command("find-miner", privileged=True)
+
+    async def update_firmware(self, file: Path, keep_settings: bool = True) -> dict:
+        """Perform a system update by uploading a firmware file and sending a command to initiate the update."""
+        async with aiofiles.open(file, "rb") as firmware:
+            file_content = await firmware.read()
+        parameters = {
+            "file": (file.name, file_content, "application/octet-stream"),
+            "keep_settings": keep_settings
+        }
+        return await self.send_command(
+            command="system/upgrade",
+            **parameters
+        )