feat: Add update firmware for Avalon miner

pyasic/miners/backends/avalonminer.py

@@ -16,11 +16,14 @@
 
 import re
 from typing import List, Optional
+import logging
+from pathlib import Path
 
 from pyasic.data import AlgoHashRate, Fan, HashBoard, HashUnit
 from pyasic.errors import APIError
 from pyasic.miners.backends.cgminer import CGMiner
 from pyasic.miners.data import DataFunction, DataLocations, DataOptions, RPCAPICommand
+from pyasic.web.avalon import AvalonWebAPI
 
 AVALON_DATA_LOC = DataLocations(
     **{
@@ -79,6 +82,8 @@
 class AvalonMiner(CGMiner):
     """Handler for Avalon Miners"""
 
+    _web_cls = AvalonWebAPI
+    web: AvalonWebAPI
     data_locations = AVALON_DATA_LOC
 
     async def fault_light_on(self) -> bool:
@@ -112,6 +117,36 @@ async def reboot(self) -> bool:
             return False
         return False
 
+    async def upgrade_firmware(self, file: Path) -> str:
+        """
+        Upgrade the firmware of an Avalon Miner.
+
+        Args:
+            file (Path): Path to the firmware file to be uploaded.
+
+        Returns:
+            str: Result of the upgrade process.
+        """
+        try:
+            if not file:
+                raise ValueError("File location must be provided for firmware upgrade.")
+
+            result = await self.web.update_firmware(file=file)
+
+            if 'Success' in result:
+                logging.info("Firmware upgrade process completed successfully for Avalon Miner.")
+                return "Firmware upgrade successful."
+            else:
+                logging.error(f"Firmware upgrade failed. Response: {result}")
+                return f"Firmware upgrade failed. Response: {result}"
+
+        except ValueError as e:
+            logging.error(f"Validation 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)

pyasic/web/avalon.py

@@ -0,0 +1,87 @@
+# ------------------------------------------------------------------------------
+#  Copyright 2022 Upstream Data Inc                                            -
+#                                                                              -
+#  Licensed under the Apache License, Version 2.0 (the "License");             -
+#  you may not use this file except in compliance with the License.            -
+#  You may obtain a copy of the License at                                     -
+#                                                                              -
+#      http://www.apache.org/licenses/LICENSE-2.0                              -
+#                                                                              -
+#  Unless required by applicable law or agreed to in writing, software         -
+#  distributed under the License is distributed on an "AS IS" BASIS,           -
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.    -
+#  See the License for the specific language governing permissions and         -
+#  limitations under the License.                                              -
+# ------------------------------------------------------------------------------
+
+from __future__ import annotations
+
+import aiofiles
+import httpx
+import json
+from pathlib import Path
+from typing import Any
+
+from pyasic import settings
+from pyasic.web.base import BaseWebAPI
+
+
+class AvalonWebAPI(BaseWebAPI):
+    def __init__(self, ip: str) -> None:
+        super().__init__(ip)
+        self.username = "admin"
+        self.pwd = settings.get("default_avalon_web_password", "admin")
+        self.port = 80
+        self.token = None
+
+    async def auth(self) -> None:
+        """Authenticate and get the token. Implement the actual authentication logic here."""
+        self.token = "Success"
+
+    async def send_command(
+        self,
+        command: str,
+        ignore_errors: bool = False,
+        allow_warning: bool = True,
+        privileged: bool = False,
+        **parameters: Any,
+    ) -> dict:
+        if self.token is None:
+            await self.auth()
+
+        async with httpx.AsyncClient(transport=settings.transport()) as client:
+            try:
+                url = f"http://{self.ip}:{self.port}/{command}"
+
+                if 'file' in parameters:
+                    file = parameters.pop('file')
+                    async with aiofiles.open(file, "rb") as f:
+                        file_contents = await f.read()
+                    files = {'firmware': file_contents}
+                    response = await client.post(
+                        url,
+                        files=files,
+                        data=parameters,
+                        auth=(self.username, self.pwd),
+                        timeout=60
+                    )
+                else:
+                    response = await client.get(
+                        url,
+                        params=parameters,
+                        auth=(self.username, self.pwd),
+                        timeout=5,
+                    )
+
+                return response.json()
+            except (httpx.HTTPError, json.JSONDecodeError):
+                if not ignore_errors:
+                    raise
+        return {}
+
+    async def update_firmware(self, file: Path) -> dict:
+        """Perform a system update by uploading a firmware file and sending a command to initiate the update."""
+        return await self.send_command(
+            command="cgi-bin/upgrade",
+            file=file,
+        )