feature: add support for elphapex configs

This commit is contained in:
Upstream Data
2025-02-05 10:45:35 -07:00
parent 7797023689
commit 36d16c7235
6 changed files with 139 additions and 1 deletions

View File

@@ -56,6 +56,16 @@ class MinerConfig(BaseModel):
**self.temperature.as_am_modern(), **self.temperature.as_am_modern(),
} }
def as_elphapex(self, user_suffix: str | None = None) -> dict:
"""Generates the configuration in the format suitable for modern Elphapex."""
return {
**self.fan_mode.as_elphapex(),
"fc-freq-level": "100",
**self.mining_mode.as_elphapex(),
**self.pools.as_elphapex(user_suffix=user_suffix),
**self.temperature.as_elphapex(),
}
def as_wm(self, user_suffix: str | None = None) -> dict: def as_wm(self, user_suffix: str | None = None) -> dict:
"""Generates the configuration in the format suitable for Whatsminers.""" """Generates the configuration in the format suitable for Whatsminers."""
return { return {
@@ -199,6 +209,15 @@ class MinerConfig(BaseModel):
fan_mode=FanModeConfig.from_am_modern(web_conf), fan_mode=FanModeConfig.from_am_modern(web_conf),
) )
@classmethod
def from_elphapex(cls, web_conf: dict) -> "MinerConfig":
"""Constructs a MinerConfig object from web configuration for modern Antminers."""
return cls(
pools=PoolConfig.from_elphapex(web_conf),
mining_mode=MiningModeConfig.from_elphapex(web_conf),
fan_mode=FanModeConfig.from_elphapex(web_conf),
)
@classmethod @classmethod
def from_am_old(cls, web_conf: dict) -> "MinerConfig": def from_am_old(cls, web_conf: dict) -> "MinerConfig":
"""Constructs a MinerConfig object from web configuration for old versions of Antminers.""" """Constructs a MinerConfig object from web configuration for old versions of Antminers."""

View File

@@ -67,6 +67,9 @@ class MinerConfigOption(Enum):
def as_luxos(self) -> dict: def as_luxos(self) -> dict:
return self.value.as_luxos() return self.value.as_luxos()
def as_elphapex(self) -> dict:
return self.value.as_elphapex()
def __call__(self, *args, **kwargs): def __call__(self, *args, **kwargs):
return self.value(*args, **kwargs) return self.value(*args, **kwargs)
@@ -131,6 +134,9 @@ class MinerConfigValue(BaseModel):
def as_luxos(self) -> dict: def as_luxos(self) -> dict:
return {} return {}
def as_elphapex(self) -> dict:
return {}
def __getitem__(self, item): def __getitem__(self, item):
try: try:
return getattr(self, item) return getattr(self, item)

View File

@@ -55,6 +55,9 @@ class FanModeNormal(MinerConfigValue):
def as_am_modern(self) -> dict: def as_am_modern(self) -> dict:
return {"bitmain-fan-ctrl": False, "bitmain-fan-pwn": "100"} return {"bitmain-fan-ctrl": False, "bitmain-fan-pwn": "100"}
def as_elphapex(self) -> dict:
return {"fc-fan-ctrl": False, "fc-fan-pwn": "100"}
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
return { return {
"temp_control": {"mode": "auto"}, "temp_control": {"mode": "auto"},
@@ -135,6 +138,9 @@ class FanModeManual(MinerConfigValue):
def as_am_modern(self) -> dict: def as_am_modern(self) -> dict:
return {"bitmain-fan-ctrl": True, "bitmain-fan-pwm": str(self.speed)} return {"bitmain-fan-ctrl": True, "bitmain-fan-pwm": str(self.speed)}
def as_elphapex(self) -> dict:
return {"fc-fan-ctrl": True, "fc-fan-pwm": str(self.speed)}
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
return { return {
"temp_control": {"mode": "manual"}, "temp_control": {"mode": "manual"},
@@ -185,6 +191,9 @@ class FanModeImmersion(MinerConfigValue):
def as_am_modern(self) -> dict: def as_am_modern(self) -> dict:
return {"bitmain-fan-ctrl": True, "bitmain-fan-pwm": "0"} return {"bitmain-fan-ctrl": True, "bitmain-fan-pwm": "0"}
def as_elphapex(self) -> dict:
return {"fc-fan-ctrl": True, "fc-fan-pwm": "0"}
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
return { return {
"fan_control": {"min_fans": 0}, "fan_control": {"min_fans": 0},
@@ -239,6 +248,20 @@ class FanModeConfig(MinerConfigOption):
else: else:
return cls.default() return cls.default()
@classmethod
def from_elphapex(cls, web_conf: dict):
if web_conf.get("fc-fan-ctrl") is not None:
fan_manual = web_conf["fc-fan-ctrl"]
if fan_manual:
speed = int(web_conf["fc-fan-pwm"])
if speed == 0:
return cls.immersion()
return cls.manual(speed=speed)
else:
return cls.normal()
else:
return cls.default()
@classmethod @classmethod
def from_epic(cls, web_conf: dict): def from_epic(cls, web_conf: dict):
try: try:

View File

@@ -52,6 +52,9 @@ class MiningModeNormal(MinerConfigValue):
return {"miner-mode": "0"} return {"miner-mode": "0"}
return {"miner-mode": 0} return {"miner-mode": 0}
def as_elphapex(self) -> dict:
return {"miner-mode": 0}
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -87,6 +90,9 @@ class MiningModeSleep(MinerConfigValue):
return {"miner-mode": "1"} return {"miner-mode": "1"}
return {"miner-mode": 1} return {"miner-mode": 1}
def as_elphapex(self) -> dict:
return {"miner-mode": 1}
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -119,6 +125,9 @@ class MiningModeLPM(MinerConfigValue):
return {"miner-mode": "3"} return {"miner-mode": "3"}
return {"miner-mode": 3} return {"miner-mode": 3}
def as_elphapex(self) -> dict:
return {"miner-mode": 3}
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -141,6 +150,9 @@ class MiningModeHPM(MinerConfigValue):
return {"miner-mode": "0"} return {"miner-mode": "0"}
return {"miner-mode": 0} return {"miner-mode": 0}
def as_elphapex(self) -> dict:
return {"miner-mode": 0}
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -174,6 +186,9 @@ class MiningModePowerTune(MinerConfigValue):
return {"miner-mode": "0"} return {"miner-mode": "0"}
return {"miner-mode": 0} return {"miner-mode": 0}
def as_elphapex(self) -> dict:
return {"miner-mode": 0}
def as_wm(self) -> dict: def as_wm(self) -> dict:
if self.power is not None: if self.power is not None:
return {"mode": self.mode, self.mode: {"wattage": self.power}} return {"mode": self.mode, self.mode: {"wattage": self.power}}
@@ -273,6 +288,9 @@ class MiningModeHashrateTune(MinerConfigValue):
return {"miner-mode": "0"} return {"miner-mode": "0"}
return {"miner-mode": 0} return {"miner-mode": 0}
def as_elphapex(self) -> dict:
return {"miner-mode": 0}
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
conf = {"enabled": True, "mode": "hashrate_target"} conf = {"enabled": True, "mode": "hashrate_target"}
if self.hashrate is not None: if self.hashrate is not None:
@@ -404,6 +422,9 @@ class ManualBoardSettings(MinerConfigValue):
return {"miner-mode": "0"} return {"miner-mode": "0"}
return {"miner-mode": 0} return {"miner-mode": 0}
def as_elphapex(self) -> dict:
return {"miner-mode": 0}
def as_vnish(self) -> dict: def as_vnish(self) -> dict:
return {"freq": self.freq} return {"freq": self.freq}
@@ -428,6 +449,9 @@ class MiningModeManual(MinerConfigValue):
return {"miner-mode": "0"} return {"miner-mode": "0"}
return {"miner-mode": 0} return {"miner-mode": 0}
def as_elphapex(self) -> dict:
return {"miner-mode": 0}
def as_vnish(self) -> dict: def as_vnish(self) -> dict:
chains = [b.as_vnish() for b in self.boards.values() if b.freq != 0] chains = [b.as_vnish() for b in self.boards.values() if b.freq != 0]
return { return {
@@ -525,6 +549,20 @@ class MiningModeConfig(MinerConfigOption):
return cls.low() return cls.low()
return cls.default() return cls.default()
@classmethod
def from_elphapex(cls, web_conf: dict):
if web_conf.get("fc-work-mode") is not None:
work_mode = web_conf["fc-work-mode"]
if work_mode == "":
return cls.default()
if int(work_mode) == 0:
return cls.normal()
elif int(work_mode) == 1:
return cls.sleep()
elif int(work_mode) == 3:
return cls.low()
return cls.default()
@classmethod @classmethod
def from_epic(cls, web_conf: dict): def from_epic(cls, web_conf: dict):
try: try:

View File

@@ -43,6 +43,13 @@ class Pool(MinerConfigValue):
"pass": self.password, "pass": self.password,
} }
def as_elphapex(self, user_suffix: str | None = None) -> dict:
return {
"url": self.url,
"user": f"{self.user}{user_suffix or ''}",
"pass": self.password,
}
def as_wm(self, idx: int = 1, user_suffix: str | None = None) -> dict: def as_wm(self, idx: int = 1, user_suffix: str | None = None) -> dict:
return { return {
f"pool_{idx}": self.url, f"pool_{idx}": self.url,
@@ -146,6 +153,12 @@ class Pool(MinerConfigValue):
url=web_pool["url"], user=web_pool["user"], password=web_pool["pass"] url=web_pool["url"], user=web_pool["user"], password=web_pool["pass"]
) )
@classmethod
def from_elphapex(cls, web_pool: dict) -> "Pool":
return cls(
url=web_pool["url"], user=web_pool["user"], password=web_pool["pass"]
)
# TODO: check if this is accurate, user/username, pass/password # TODO: check if this is accurate, user/username, pass/password
@classmethod @classmethod
def from_goldshell(cls, web_pool: dict) -> "Pool": def from_goldshell(cls, web_pool: dict) -> "Pool":
@@ -235,6 +248,17 @@ class PoolGroup(MinerConfigValue):
idx += 1 idx += 1
return pools return pools
def as_elphapex(self, user_suffix: str | None = None) -> list:
pools = []
idx = 0
while idx < 3:
if len(self.pools) > idx:
pools.append(self.pools[idx].as_elphapex(user_suffix=user_suffix))
else:
pools.append(Pool(url="", user="", password="").as_elphapex())
idx += 1
return pools
def as_wm(self, user_suffix: str | None = None) -> dict: def as_wm(self, user_suffix: str | None = None) -> dict:
pools = {} pools = {}
idx = 0 idx = 0
@@ -351,6 +375,13 @@ class PoolGroup(MinerConfigValue):
pools.append(Pool.from_am_modern(pool)) pools.append(Pool.from_am_modern(pool))
return cls(pools=pools) return cls(pools=pools)
@classmethod
def from_elphapex(cls, web_pool_list: list) -> "PoolGroup":
pools = []
for pool in web_pool_list:
pools.append(Pool.from_elphapex(pool))
return cls(pools=pools)
@classmethod @classmethod
def from_goldshell(cls, web_pools: list) -> "PoolGroup": def from_goldshell(cls, web_pools: list) -> "PoolGroup":
return cls(pools=[Pool.from_goldshell(p) for p in web_pools]) return cls(pools=[Pool.from_goldshell(p) for p in web_pools])
@@ -436,6 +467,11 @@ class PoolConfig(MinerConfigValue):
return {"pools": self.groups[0].as_am_modern(user_suffix=user_suffix)} return {"pools": self.groups[0].as_am_modern(user_suffix=user_suffix)}
return {"pools": PoolGroup().as_am_modern()} return {"pools": PoolGroup().as_am_modern()}
def as_elphapex(self, user_suffix: str | None = None) -> dict:
if len(self.groups) > 0:
return {"pools": self.groups[0].as_elphapex(user_suffix=user_suffix)}
return {"pools": PoolGroup().as_elphapex()}
def as_wm(self, user_suffix: str | None = None) -> dict: def as_wm(self, user_suffix: str | None = None) -> dict:
if len(self.groups) > 0: if len(self.groups) > 0:
return {"pools": self.groups[0].as_wm(user_suffix=user_suffix)} return {"pools": self.groups[0].as_wm(user_suffix=user_suffix)}
@@ -537,6 +573,12 @@ class PoolConfig(MinerConfigValue):
return cls(groups=[PoolGroup.from_am_modern(pool_data)]) return cls(groups=[PoolGroup.from_am_modern(pool_data)])
@classmethod
def from_elphapex(cls, web_conf: dict) -> "PoolConfig":
pool_data = web_conf["pools"]
return cls(groups=[PoolGroup.from_elphapex(pool_data)])
@classmethod @classmethod
def from_goldshell(cls, web_pools: list) -> "PoolConfig": def from_goldshell(cls, web_pools: list) -> "PoolConfig":
return cls(groups=[PoolGroup.from_goldshell(web_pools)]) return cls(groups=[PoolGroup.from_goldshell(web_pools)])

View File

@@ -15,7 +15,7 @@
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
from typing import List, Optional from typing import List, Optional
from pyasic import APIError from pyasic import APIError, MinerConfig
from pyasic.data import Fan, HashBoard, X19Error from pyasic.data import Fan, HashBoard, X19Error
from pyasic.data.error_codes import MinerErrorData from pyasic.data.error_codes import MinerErrorData
from pyasic.data.pools import PoolMetrics, PoolUrl from pyasic.data.pools import PoolMetrics, PoolUrl
@@ -91,6 +91,16 @@ class ElphapexMiner(StockFirmware):
data_locations = ELPHAPEX_DATA_LOC data_locations = ELPHAPEX_DATA_LOC
async def get_config(self) -> MinerConfig:
data = await self.web.get_miner_conf()
if data:
self.config = MinerConfig.from_elphapex(data)
return self.config
async def send_config(self, config: MinerConfig, user_suffix: str = None) -> None:
self.config = config
await self.web.set_miner_conf(config.as_elphapex(user_suffix=user_suffix))
async def fault_light_on(self) -> bool: async def fault_light_on(self) -> bool:
data = await self.web.blink(blink=True) data = await self.web.blink(blink=True)
if data: if data: