used web attribute and removed some redundant changes from the firmware.py

This commit is contained in:
1e9abhi1e10
2024-06-27 03:43:33 +05:30
parent ba58e80ec3
commit e649348af2
2 changed files with 17 additions and 31 deletions

View File

@@ -15,7 +15,7 @@
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
import logging import logging
from enum import Enum from enum import Enum
import httpx import base64
import aiofiles import aiofiles
from typing import List, Optional from typing import List, Optional
@@ -195,7 +195,7 @@ class Auradine(StockFirmware):
for key in conf.keys(): for key in conf.keys():
await self.web.send_command(command=key, **conf[key]) await self.web.send_command(command=key, **conf[key])
async def upgrade_firmware(self, url: str = None, file_path: str = None, version: str = "latest") -> bool: async def upgrade_firmware(self, *, url: str = None, file_path: str = None, version: str = "latest") -> bool:
""" """
Upgrade the firmware of the Auradine device. Upgrade the firmware of the Auradine device.
@@ -215,16 +215,23 @@ class Auradine(StockFirmware):
if url: if url:
# Download the firmware file from the URL # Download the firmware file from the URL
async with httpx.AsyncClient() as client: response = await self.web.get(url)
response = await client.get(url) if response.status_code != 200:
if response.status_code != 200: raise ValueError(f"Failed to download firmware from URL: {url}")
raise ValueError(f"Failed to download firmware from URL: {url}") upgrade_contents = response.content
upgrade_contents = response.content
else: else:
# read the fimware file contents from the local file path # Read the firmware file contents from the local file path
async with aiofiles.open(file_path, "rb") as f: async with aiofiles.open(file_path, "rb") as f:
upgrade_contents = await f.read() upgrade_contents = await f.read()
# Encode the firmware contents in base64
encoded_contents = base64.b64encode(upgrade_contents).decode("utf-8")
# Upload the firmware file to the Auradine miner device
await self.ssh.send_command(
f"echo {encoded_contents} | base64 -d > /tmp/firmware.tar && sysupgrade /tmp/firmware.tar"
)
logging.info("Firmware upgrade process completed successfully.") logging.info("Firmware upgrade process completed successfully.")
return True return True
except Exception as e: except Exception as e:
@@ -423,4 +430,4 @@ class Auradine(StockFirmware):
try: try:
return rpc_summary["SUMMARY"][0]["Elapsed"] return rpc_summary["SUMMARY"][0]["Elapsed"]
except LookupError: except LookupError:
pass pass

View File

@@ -21,47 +21,26 @@ from pyasic.miners.base import BaseMiner
class StockFirmware(BaseMiner): class StockFirmware(BaseMiner):
firmware = MinerFirmware.STOCK firmware = MinerFirmware.STOCK
async def upgrade_firmware(self, *, url: str = None, version: str = "latest") -> bool:
return await super().upgrade_firmware(url=url, version=version)
class BraiinsOSFirmware(BaseMiner): class BraiinsOSFirmware(BaseMiner):
firmware = MinerFirmware.BRAIINS_OS firmware = MinerFirmware.BRAIINS_OS
async def upgrade_firmware(self, *, url: str = None, version: str = "latest") -> bool:
return await super().upgrade_firmware(url=url, version=version)
class VNishFirmware(BaseMiner): class VNishFirmware(BaseMiner):
firmware = MinerFirmware.VNISH firmware = MinerFirmware.VNISH
async def upgrade_firmware(self, *, url: str = None, version: str = "latest") -> bool:
return await super().upgrade_firmware(url=url, version=version)
class ePICFirmware(BaseMiner): class ePICFirmware(BaseMiner):
firmware = MinerFirmware.EPIC firmware = MinerFirmware.EPIC
async def upgrade_firmware(self, *, url: str = None, version: str = "latest") -> bool:
return await super().upgrade_firmware(url=url, version=version)
class HiveonFirmware(BaseMiner): class HiveonFirmware(BaseMiner):
firmware = MinerFirmware.HIVEON firmware = MinerFirmware.HIVEON
async def upgrade_firmware(self, *, url: str = None, version: str = "latest") -> bool:
return await super().upgrade_firmware(url=url, version=version)
class LuxOSFirmware(BaseMiner): class LuxOSFirmware(BaseMiner):
firmware = MinerFirmware.LUXOS firmware = MinerFirmware.LUXOS
async def upgrade_firmware(self, *, url: str = None, version: str = "latest") -> bool:
return await super().upgrade_firmware(url=url, version=version)
class MaraFirmware(BaseMiner): class MaraFirmware(BaseMiner):
firmware = MinerFirmware.MARATHON firmware = MinerFirmware.MARATHON
async def upgrade_firmware(self, *, url: str = None, version: str = "latest") -> bool:
return await super().upgrade_firmware(url=url, version=version)