diff --git a/pyasic/miners/avalonminer/cgminer/nano/nano3.py b/pyasic/miners/avalonminer/cgminer/nano/nano3.py index a1ba3f47..3d1688d4 100644 --- a/pyasic/miners/avalonminer/cgminer/nano/nano3.py +++ b/pyasic/miners/avalonminer/cgminer/nano/nano3.py @@ -13,10 +13,91 @@ # See the License for the specific language governing permissions and - # limitations under the License. - # ------------------------------------------------------------------------------ +from typing import Optional +from pyasic import APIError from pyasic.miners.backends import AvalonMiner +from pyasic.miners.data import ( + DataFunction, + DataLocations, + DataOptions, + RPCAPICommand, + WebAPICommand, +) from pyasic.miners.device.models import AvalonNano3 +from pyasic.web.avalonminer import AvalonMinerWebAPI + +AVALON_NANO_DATA_LOC = DataLocations( + **{ + str(DataOptions.MAC): DataFunction( + "_get_mac", + [WebAPICommand("web_minerinfo", "get_minerinfo")], + ), + str(DataOptions.API_VERSION): DataFunction( + "_get_api_ver", + [RPCAPICommand("rpc_version", "version")], + ), + str(DataOptions.FW_VERSION): DataFunction( + "_get_fw_ver", + [RPCAPICommand("rpc_version", "version")], + ), + str(DataOptions.HASHRATE): DataFunction( + "_get_hashrate", + [RPCAPICommand("rpc_devs", "devs")], + ), + str(DataOptions.EXPECTED_HASHRATE): DataFunction( + "_get_expected_hashrate", + [RPCAPICommand("rpc_stats", "stats")], + ), + str(DataOptions.HASHBOARDS): DataFunction( + "_get_hashboards", + [RPCAPICommand("rpc_stats", "stats")], + ), + str(DataOptions.ENVIRONMENT_TEMP): DataFunction( + "_get_env_temp", + [RPCAPICommand("rpc_stats", "stats")], + ), + str(DataOptions.WATTAGE_LIMIT): DataFunction( + "_get_wattage_limit", + [RPCAPICommand("rpc_stats", "stats")], + ), + str(DataOptions.FANS): DataFunction( + "_get_fans", + [RPCAPICommand("rpc_stats", "stats")], + ), + str(DataOptions.FAULT_LIGHT): DataFunction( + "_get_fault_light", + [RPCAPICommand("rpc_stats", "stats")], + ), + str(DataOptions.UPTIME): DataFunction( + "_get_uptime", + [RPCAPICommand("rpc_stats", "stats")], + ), + str(DataOptions.POOLS): DataFunction( + "_get_pools", + [RPCAPICommand("rpc_pools", "pools")], + ), + } +) class CGMinerAvalonNano3(AvalonMiner, AvalonNano3): - pass + _web_cls = AvalonMinerWebAPI + web: AvalonMinerWebAPI + + data_locations = AVALON_NANO_DATA_LOC + + async def _get_mac(self, web_minerinfo: dict) -> Optional[dict]: + if web_minerinfo is None: + try: + web_minerinfo = await self.web.minerinfo() + except APIError: + pass + + if web_minerinfo is not None: + try: + mac = web_minerinfo.get("mac") + if mac is not None: + return mac.upper() + except (KeyError, ValueError): + pass diff --git a/pyasic/web/avalonminer.py b/pyasic/web/avalonminer.py new file mode 100644 index 00000000..43ca87ce --- /dev/null +++ b/pyasic/web/avalonminer.py @@ -0,0 +1,110 @@ +# ------------------------------------------------------------------------------ +# 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 asyncio +import hashlib +import json +from pathlib import Path +from typing import Any + +import aiofiles +import httpx + +from pyasic import settings +from pyasic.web.base import BaseWebAPI + + +class AvalonMinerWebAPI(BaseWebAPI): + def __init__(self, ip: str) -> None: + """Initialize the modern Avalonminer API client with a specific IP address. + + Args: + ip (str): IP address of the Avalonminer device. + """ + super().__init__(ip) + self.username = "root" + self.pwd = settings.get("default_avalonminer_web_password", "root") + + async def send_command( + self, + command: str | bytes, + ignore_errors: bool = False, + allow_warning: bool = True, + **parameters: Any, + ) -> dict: + """Send a command to the Avalonminer device using HTTP digest authentication. + + Args: + command (str | bytes): The CGI command to send. + ignore_errors (bool): If True, ignore any HTTP errors. + allow_warning (bool): If True, proceed with warnings. + **parameters: Arbitrary keyword arguments to be sent as parameters in the request. + + Returns: + dict: The JSON response from the device or an empty dictionary if an error occurs. + """ + cookie_data = "ff0000ff" + hashlib.sha256(self.pwd.encode()).hexdigest()[:24] + + url = f"http://{self.ip}:{self.port}/{command}.cgi" + try: + async with httpx.AsyncClient(transport=settings.transport()) as client: + client.cookies.set("auth", cookie_data) + resp = await client.get(url) + raw_data = resp.text.replace("minerinfoCallback(", "").replace(");", "") + return json.loads(raw_data) + except httpx.HTTPError: + pass + return {} + + async def multicommand( + self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True + ) -> dict: + async with httpx.AsyncClient(transport=settings.transport()) as client: + cookie_data = ( + "ff0000ff" + hashlib.sha256(self.pwd.encode()).hexdigest()[:24] + ) + client.cookies.set("auth", cookie_data) + tasks = [ + asyncio.create_task(self._handle_multicommand(client, command)) + for command in commands + ] + all_data = await asyncio.gather(*tasks) + + data = {} + for item in all_data: + data.update(item) + + data["multicommand"] = True + return data + + async def _handle_multicommand( + self, client: httpx.AsyncClient, command: str + ) -> dict: + try: + url = f"http://{self.ip}:{self.port}/{command}.cgi" + resp = await client.get(url) + raw_data = resp.text.replace("minerinfoCallback(", "").replace(");", "") + return json.loads(raw_data) + except httpx.HTTPError: + pass + return {} + + async def minerinfo(self): + return await self.send_command("get_minerinfo") + + async def home(self): + return await self.send_command("get_home")