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

@@ -16,9 +16,11 @@
 from __future__ import annotations
 
 import json
+import time
 import warnings
 from typing import Any
-
+import aiofiles
+from pathlib import Path
 import httpx
 
 from pyasic import settings
@@ -57,9 +59,11 @@ async def send_command(
         ignore_errors: bool = False,
         allow_warning: bool = True,
         privileged: bool = False,
+        custom_data: bytes = None,
+        custom_headers: dict = None,
         **parameters: Any,
     ) -> dict:
-        post = privileged or not parameters == {}
+        post = privileged or bool(parameters) or custom_data is not None
         if self.token is None:
             await self.auth()
         async with httpx.AsyncClient(transport=settings.transport()) as client:
@@ -69,18 +73,30 @@ async def send_command(
                     if command.startswith("system"):
                         auth = "Bearer " + self.token
 
+                    url = f"http://{self.ip}:{self.port}/api/v1/{command}"
+                    headers = custom_headers or {}
+                    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),
-                            json=parameters,
-                        )
+                        if custom_data is not None:
+                            response = await client.post(
+                                url,
+                                headers=headers,
+                                content=custom_data,
+                                timeout=settings.get("api_function_timeout", 30),
+                            )
+                        else:
+                            response = await client.post(
+                                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 +107,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 +162,34 @@ 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()
+
+        boundary = f"-----------------------VNishTools{int(time.time())}"
+
+        data = b''
+        data += f'--{boundary}\r\n'.encode('utf-8')
+        data += f'Content-Disposition: form-data; name="file"; filename="{file.name}"\r\n'.encode('utf-8')
+        data += b'Content-Type: application/octet-stream\r\n\r\n'
+        data += file_content
+        data += b'\r\n'
+        data += f'--{boundary}\r\n'.encode('utf-8')
+        data += f'Content-Disposition: form-data; name="keep_settings"\r\n\r\n'.encode('utf-8')
+        data += f'{"true" if keep_settings else "false"}'.encode('utf-8')
+        data += b'\r\n'
+        data += f'--{boundary}--\r\n'.encode('utf-8')
+
+        headers = {
+            "Content-Type": f"multipart/form-data; boundary={boundary}"
+        }
+
+        return await self.send_command(
+            command="system/upgrade",
+            post=True,
+            privileged=True,
+            custom_data=data,
+            custom_headers=headers
+        )