CAT_TOM/pyasic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature: add support for Elphapex DG1+

Browse Source
This commit is contained in:
Upstream Data
2025-02-04 10:19:45 -07:00
committed by Brett Rowan
parent 90d8a795e6
commit a5c42c9c2b
24 changed files with 891 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
pyasic/device/makes.py
View File

@@ -30,6 +30,7 @@ class MinerMake(str, Enum):
ICERIVER = "IceRiver"
HAMMER = "Hammer"
VOLCMINER = "VolcMiner"
ELPHAPEX = "Elphapex"
BRAIINS = "Braiins"
def __str__(self):

5
pyasic/device/models.py
View File

@@ -550,6 +550,10 @@ class BraiinsModels(MinerModelType):
BMM101 = "BMM101"
class ElphapexModels(MinerModelType):
DG1Plus = "DG1+"
class MinerModel:
ANTMINER = AntminerModels
WHATSMINER = WhatsminerModels
@@ -563,4 +567,5 @@ class MinerModel:
ICERIVER = IceRiverModels
HAMMER = HammerModels
VOLCMINER = VolcMinerModels
ELPHAPEX = ElphapexModels
BRAIINS = BraiinsModels

1
pyasic/miners/backends/__init__.py
View File

@@ -22,6 +22,7 @@ from .bmminer import BMMiner
from .braiins_os import BOSer, BOSMiner
from .btminer import BTMiner
from .cgminer import CGMiner
from .elphapex import ElphapexMiner
from .epic import ePIC
from .goldshell import GoldshellMiner
from .hammer import BlackMiner

4
pyasic/miners/backends/antminer.py
View File

@@ -75,6 +75,10 @@ ANTMINER_MODERN_DATA_LOC = DataLocations(
"_get_fault_light",
[WebAPICommand("web_get_blink_status", "get_blink_status")],
),
str(DataOptions.HASHBOARDS): DataFunction(
"_get_hashboards",
[],
),
str(DataOptions.IS_MINING): DataFunction(
"_is_mining",
[WebAPICommand("web_get_conf", "get_miner_conf")],

319
pyasic/miners/backends/elphapex.py Normal file
View File

@@ -0,0 +1,319 @@
# ------------------------------------------------------------------------------
# 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 typing import List, Optional
from pyasic import APIError
from pyasic.data import Fan, HashBoard, X19Error
from pyasic.data.error_codes import MinerErrorData
from pyasic.device.algorithm import AlgoHashRate
from pyasic.miners.data import (
DataFunction,
DataLocations,
DataOptions,
WebAPICommand,
)
from pyasic.miners.device.firmware import StockFirmware
from pyasic.web.elphapex import ElphapexWebAPI
ELPHAPEX_DATA_LOC = DataLocations(
**{
str(DataOptions.MAC): DataFunction(
"_get_mac",
[WebAPICommand("web_get_system_info", "get_system_info")],
),
str(DataOptions.API_VERSION): DataFunction(
"_get_api_ver",
[WebAPICommand("web_summary", "summary")],
),
str(DataOptions.FW_VERSION): DataFunction(
"_get_fw_ver",
[WebAPICommand("web_get_system_info", "get_system_info")],
),
str(DataOptions.HOSTNAME): DataFunction(
"_get_hostname",
[WebAPICommand("web_get_system_info", "get_system_info")],
),
str(DataOptions.HASHBOARDS): DataFunction(
"_get_hashboards",
[WebAPICommand("web_stats", "stats")],
),
str(DataOptions.EXPECTED_HASHRATE): DataFunction(
"_get_expected_hashrate",
[WebAPICommand("web_stats", "stats")],
),
str(DataOptions.FANS): DataFunction(
"_get_fans",
[WebAPICommand("web_stats", "stats")],
),
str(DataOptions.ERRORS): DataFunction(
"_get_errors",
[WebAPICommand("web_summary", "summary")],
),
str(DataOptions.FAULT_LIGHT): DataFunction(
"_get_fault_light",
[WebAPICommand("web_get_blink_status", "get_blink_status")],
),
str(DataOptions.IS_MINING): DataFunction(
"_is_mining",
[WebAPICommand("web_get_miner_conf", "get_miner_conf")],
),
str(DataOptions.UPTIME): DataFunction(
"_get_uptime",
[WebAPICommand("web_summary", "summary")],
),
}
)
class ElphapexMiner(StockFirmware):
"""Handler for Elphapex miners."""
_web_cls = ElphapexWebAPI
web: ElphapexWebAPI
data_locations = ELPHAPEX_DATA_LOC
async def fault_light_on(self) -> bool:
data = await self.web.blink(blink=True)
if data:
if data.get("code") == "B000":
self.light = True
return self.light
async def fault_light_off(self) -> bool:
data = await self.web.blink(blink=False)
if data:
if data.get("code") == "B100":
self.light = False
return self.light
async def reboot(self) -> bool:
data = await self.web.reboot()
if data:
return True
return False
async def _get_api_ver(self, web_summary: dict = None) -> Optional[str]:
if web_summary is None:
try:
web_summary = await self.web.summary()
except APIError:
pass
if web_summary is not None:
try:
self.api_ver = web_summary["STATUS"]["api_version"]
except LookupError:
pass
return self.api_ver
async def _get_fw_ver(self, web_get_system_info: dict = None) -> Optional[str]:
if web_get_system_info is None:
try:
web_get_system_info = await self.web.get_system_info()
except APIError:
pass
if web_get_system_info is not None:
try:
self.fw_ver = (
web_get_system_info["system_filesystem_version"]
.upper()
.split("V")[-1]
)
except LookupError:
pass
return self.fw_ver
async def _get_hostname(self, web_get_system_info: dict = None) -> Optional[str]:
if web_get_system_info is None:
try:
web_get_system_info = await self.web.get_system_info()
except APIError:
pass
if web_get_system_info is not None:
try:
return web_get_system_info["hostname"]
except KeyError:
pass
async def _get_mac(self, web_get_system_info: dict = None) -> Optional[str]:
if web_get_system_info is None:
try:
web_get_system_info = await self.web.get_system_info()
except APIError:
pass
if web_get_system_info is not None:
try:
return web_get_system_info["macaddr"]
except KeyError:
pass
try:
data = await self.web.get_network_info()
if data:
return data["macaddr"]
except KeyError:
pass
async def _get_errors(self, web_summary: dict = None) -> List[MinerErrorData]:
if web_summary is None:
try:
web_summary = await self.web.summary()
except APIError:
pass
errors = []
if web_summary is not None:
try:
for item in web_summary["SUMMARY"][0]["status"]:
try:
if not item["status"] == "s":
errors.append(X19Error(error_message=item["msg"]))
except KeyError:
continue
except LookupError:
pass
return errors
async def _get_hashboards(self, web_stats: dict | None = None) -> List[HashBoard]:
hashboards = [
HashBoard(slot=idx, expected_chips=self.expected_chips)
for idx in range(self.expected_hashboards)
]
if web_stats is None:
try:
web_stats = await self.web.stats()
except APIError:
return hashboards
if web_stats is not None:
try:
for board in web_stats["STATS"][0]["chain"]:
hashboards[board["index"]].hashrate = self.algo.hashrate(
rate=board["rate_real"], unit=self.algo.unit.MH
).into(self.algo.unit.default)
hashboards[board["index"]].chips = board["asic_num"]
board_temp_data = list(
filter(lambda x: not x == 0, board["temp_pcb"])
)
hashboards[board["index"]].temp = sum(board_temp_data) / len(
board_temp_data
)
chip_temp_data = list(
filter(lambda x: not x == "", board["temp_chip"])
)
hashboards[board["index"]].chip_temp = sum(
[int(i) / 1000 for i in chip_temp_data]
) / len(chip_temp_data)
hashboards[board["index"]].serial_number = board["sn"]
hashboards[board["index"]].missing = False
except LookupError:
pass
return hashboards
async def _get_fault_light(
self, web_get_blink_status: dict = None
) -> Optional[bool]:
if self.light:
return self.light
if web_get_blink_status is None:
try:
web_get_blink_status = await self.web.get_blink_status()
except APIError:
pass
if web_get_blink_status is not None:
try:
self.light = web_get_blink_status["blink"]
except KeyError:
pass
return self.light
async def _get_expected_hashrate(
self, web_stats: dict = None
) -> Optional[AlgoHashRate]:
if web_stats is None:
try:
web_stats = await self.web.stats()
except APIError:
pass
if web_stats is not None:
try:
expected_rate = web_stats["STATS"][1]["total_rateideal"]
try:
rate_unit = web_stats["STATS"][1]["rate_unit"]
except KeyError:
rate_unit = "MH"
return self.algo.hashrate(
rate=float(expected_rate), unit=self.algo.unit.from_str(rate_unit)
).into(self.algo.unit.default)
except LookupError:
pass
async def _is_mining(self, web_get_miner_conf: dict = None) -> Optional[bool]:
if web_get_miner_conf is None:
try:
web_get_miner_conf = await self.web.get_miner_conf()
except APIError:
pass
if web_get_miner_conf is not None:
try:
if str(web_get_miner_conf["fc-work-mode"]).isdigit():
return (
False if int(web_get_miner_conf["fc-work-mode"]) == 1 else True
)
return False
except LookupError:
pass
async def _get_uptime(self, web_summary: dict = None) -> Optional[int]:
if web_summary is None:
try:
web_summary = await self.web.summary()
except APIError:
pass
if web_summary is not None:
try:
return int(web_summary["SUMMARY"][1]["elapsed"])
except LookupError:
pass
async def _get_fans(self, web_stats: dict = None) -> List[Fan]:
if web_stats is None:
try:
web_stats = await self.web.stats()
except APIError:
pass
fans = [Fan() for _ in range(self.expected_fans)]
if web_stats is not None:
for fan_n in range(self.expected_fans):
try:
fans[fan_n].speed = int(web_stats["STATS"][0]["fan"][fan_n])
except LookupError:
pass
return fans

4
pyasic/miners/device/makes.py
View File

@@ -68,3 +68,7 @@ class VolcMinerMake(BaseMiner):
class BraiinsMake(BaseMiner):
make = MinerMake.BRAIINS
class ElphapexMake(BaseMiner):
make = MinerMake.ELPHAPEX

1
pyasic/miners/device/models/__init__.py
View File

@@ -18,6 +18,7 @@ from .antminer import *
from .auradine import *
from .avalonminer import *
from .braiins import *
from .elphapex import *
from .epic import *
from .goldshell import *
from .hammer import *

27
pyasic/miners/device/models/elphapex/DGX/DG1.py Normal file
View File

@@ -0,0 +1,27 @@
# ------------------------------------------------------------------------------
# Copyright 2024 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 pyasic.device.algorithm import MinerAlgo
from pyasic.device.models import MinerModel
from pyasic.miners.device.makes import ElphapexMake, HammerMake
class DG1Plus(ElphapexMake):
raw_model = MinerModel.ELPHAPEX.DG1Plus
expected_chips = 204
expected_hashboards = 4
expected_fans = 4
algo = MinerAlgo.SCRYPT

1
pyasic/miners/device/models/elphapex/DGX/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .DG1 import DG1Plus

1
pyasic/miners/device/models/elphapex/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .DGX import *

1
pyasic/miners/elphapex/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .daoge import *

6
pyasic/miners/elphapex/daoge/DGX/DG1.py Normal file
View File

@@ -0,0 +1,6 @@
from pyasic.miners.backends.elphapex import ElphapexMiner
from pyasic.miners.device.models import DG1Plus
class ElphapexDG1Plus(ElphapexMiner, DG1Plus):
pass

1
pyasic/miners/elphapex/daoge/DGX/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .DG1 import ElphapexDG1Plus

1
pyasic/miners/elphapex/daoge/__init__.py Normal file
View File

@@ -0,0 +1 @@
from .DGX import *

26
pyasic/miners/factory.py
View File

@@ -37,6 +37,7 @@ from pyasic.miners.bitaxe import *
from pyasic.miners.blockminer import *
from pyasic.miners.braiins import *
from pyasic.miners.device.makes import *
from pyasic.miners.elphapex import *
from pyasic.miners.goldshell import *
from pyasic.miners.hammer import *
from pyasic.miners.iceriver import *
@@ -64,6 +65,7 @@ class MinerTypes(enum.Enum):
HAMMER = 14
VOLCMINER = 15
LUCKYMINER = 16
ELPHAPEX = 17
MINER_CLASSES = {
@@ -664,6 +666,10 @@ MINER_CLASSES = {
None: type("VolcMinerUnknown", (BlackMiner, VolcMinerMake), {}),
"VOLCMINER D1": VolcMinerD1,
},
MinerTypes.ELPHAPEX: {
None: type("ElphapexUnknown", (ElphapexMiner, ElphapexMake), {}),
"DG1+1": ElphapexDG1Plus,
},
}
@@ -745,6 +751,7 @@ class MinerFactory:
MinerTypes.ICERIVER: self.get_miner_model_iceriver,
MinerTypes.HAMMER: self.get_miner_model_hammer,
MinerTypes.VOLCMINER: self.get_miner_model_volcminer,
MinerTypes.ELPHAPEX: self.get_miner_model_elphapex,
}
fn = miner_model_fns.get(miner_type)
@@ -828,6 +835,10 @@ class MinerFactory:
"www-authenticate", ""
):
return MinerTypes.HAMMER
if web_resp.status_code == 401 and 'realm="Daoge' in web_resp.headers.get(
"www-authenticate", ""
):
return MinerTypes.ELPHAPEX
if len(web_resp.history) > 0:
history_resp = web_resp.history[0]
if (
@@ -1384,6 +1395,21 @@ class MinerFactory:
except (TypeError, LookupError):
pass
async def get_miner_model_elphapex(self, ip: str) -> str | None:
auth = httpx.DigestAuth(
"root", settings.get("default_elphapex_web_password", "root")
)
web_json_data = await self.send_web_command(
ip, "/cgi-bin/get_system_info.cgi", auth=auth
)
try:
miner_model = web_json_data["minertype"]
return miner_model
except (TypeError, LookupError):
pass
miner_factory = MinerFactory()

1
pyasic/settings/__init__.py
View File

@@ -40,6 +40,7 @@ _settings = { # defaults
"default_epic_web_password": "letmein",
"default_hive_web_password": "root",
"default_iceriver_web_password": "12345678",
"default_elphapex_web_password": "root",
"default_antminer_ssh_password": "miner",
"default_bosminer_ssh_password": "root",
}

216
pyasic/web/elphapex.py Normal file
View File

@@ -0,0 +1,216 @@
# ------------------------------------------------------------------------------
# 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 json
from typing import Any
import httpx
from pyasic import settings
from pyasic.web.base import BaseWebAPI
class ElphapexWebAPI(BaseWebAPI):
def __init__(self, ip: str) -> None:
"""Initialize the modern Elphapex API client with a specific IP address.
Args:
ip (str): IP address of the Elphapex device.
"""
super().__init__(ip)
self.username = "root"
self.pwd = settings.get("default_elphapex_web_password", "root")
async def send_command(
self,
command: str | bytes,
ignore_errors: bool = False,
allow_warning: bool = True,
privileged: bool = False,
**parameters: Any,
) -> dict:
"""Send a command to the Elphapex 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.
privileged (bool): If set to True, requires elevated privileges.
**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.
"""
url = f"http://{self.ip}:{self.port}/cgi-bin/{command}.cgi"
auth = httpx.DigestAuth(self.username, self.pwd)
try:
async with httpx.AsyncClient(transport=settings.transport()) as client:
if parameters:
data = await client.post(
url,
auth=auth,
timeout=settings.get("api_function_timeout", 3),
json=parameters,
)
else:
data = await client.get(url, auth=auth)
except httpx.HTTPError as e:
return {"success": False, "message": f"HTTP error occurred: {str(e)}"}
else:
if data.status_code == 200:
try:
return data.json()
except json.decoder.JSONDecodeError:
return {"success": False, "message": "Failed to decode JSON"}
return {"success": False, "message": "Unknown error occurred"}
async def multicommand(
self, *commands: str, ignore_errors: bool = False, allow_warning: bool = True
) -> dict:
"""Execute multiple commands simultaneously.
Args:
*commands (str): Multiple command strings to be executed.
ignore_errors (bool): If True, ignore any HTTP errors.
allow_warning (bool): If True, proceed with warnings.
Returns:
dict: A dictionary containing the results of all commands executed.
"""
async with httpx.AsyncClient(transport=settings.transport()) as client:
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:
"""Helper function for handling individual commands in a multicommand execution.
Args:
client (httpx.AsyncClient): The HTTP client to use for the request.
command (str): The command to be executed.
Returns:
dict: A dictionary containing the response of the executed command.
"""
auth = httpx.DigestAuth(self.username, self.pwd)
try:
url = f"http://{self.ip}/cgi-bin/{command}.cgi"
ret = await client.get(url, auth=auth)
except httpx.HTTPError:
pass
else:
if ret.status_code == 200:
try:
json_data = ret.json()
return {command: json_data}
except json.decoder.JSONDecodeError:
pass
return {command: {}}
async def get_miner_conf(self) -> dict:
"""Retrieve the miner configuration from the Elphapex device.
Returns:
dict: A dictionary containing the current configuration of the miner.
"""
return await self.send_command("get_miner_conf")
async def set_miner_conf(self, conf: dict) -> dict:
"""Set the configuration for the miner.
Args:
conf (dict): A dictionary of configuration settings to apply to the miner.
Returns:
dict: A dictionary response from the device after setting the configuration.
"""
return await self.send_command("set_miner_conf", **conf)
async def blink(self, blink: bool) -> dict:
"""Control the blinking of the LED on the miner device.
Args:
blink (bool): True to start blinking, False to stop.
Returns:
dict: A dictionary response from the device after the command execution.
"""
if blink:
return await self.send_command("blink", blink="true")
return await self.send_command("blink", blink="false")
async def reboot(self) -> dict:
"""Reboot the miner device.
Returns:
dict: A dictionary response from the device confirming the reboot command.
"""
return await self.send_command("reboot")
async def get_system_info(self) -> dict:
"""Retrieve system information from the miner.
Returns:
dict: A dictionary containing system information of the miner.
"""
return await self.send_command("get_system_info")
async def get_network_info(self) -> dict:
"""Retrieve network configuration information from the miner.
Returns:
dict: A dictionary containing the network configuration of the miner.
"""
return await self.send_command("get_network_info")
async def summary(self) -> dict:
"""Get a summary of the miner's status and performance.
Returns:
dict: A summary of the miner's current operational status.
"""
return await self.send_command("summary")
async def stats(self) -> dict:
"""Get miners stats.
Returns:
dict: A summary of the miner's current operational status.
"""
return await self.send_command("stats")
async def get_blink_status(self) -> dict:
"""Check the status of the LED blinking on the miner.
Returns:
dict: A dictionary indicating whether the LED is currently blinking.
"""
return await self.send_command("get_blink_status")