feature: add auradine control functions.

This commit is contained in:
b-rowan
2024-01-23 15:28:37 -07:00
parent eed1973345
commit b60c7a55d4
7 changed files with 206 additions and 11 deletions

View File

@@ -113,10 +113,17 @@ class MinerConfig:
**self.fan_mode.as_epic(), **self.fan_mode.as_epic(),
**self.temperature.as_epic(), **self.temperature.as_epic(),
**self.mining_mode.as_epic(), **self.mining_mode.as_epic(),
**self.pools.as_epic(user_suffix=user_suffix), **self.pools.as_epic(),
**self.power_scaling.as_epic(), **self.power_scaling.as_epic(),
} }
def as_auradine(self, user_suffix: str = None) -> dict:
return {
**self.fan_mode.as_auradine(),
**self.mining_mode.as_auradine(),
**self.pools.as_auradine(user_suffix=user_suffix),
}
@classmethod @classmethod
def from_dict(cls, dict_conf: dict) -> "MinerConfig": def from_dict(cls, dict_conf: dict) -> "MinerConfig":
return cls( return cls(
@@ -189,6 +196,14 @@ class MinerConfig:
mining_mode=MiningModeConfig.from_vnish(web_settings), mining_mode=MiningModeConfig.from_vnish(web_settings),
) )
@classmethod
def from_auradine(cls, web_conf: dict) -> "MinerConfig":
return cls(
pools=PoolConfig.from_api(web_conf["pools"][0]),
fan_mode=FanModeConfig.from_auradine(web_conf["fans"][0]),
mining_mode=MiningModeConfig.from_auradine(web_conf["mode"][0]),
)
def merge(a: dict, b: dict) -> dict: def merge(a: dict, b: dict) -> dict:
result = deepcopy(a) result = deepcopy(a)

View File

@@ -53,6 +53,9 @@ class MinerConfigOption(Enum):
def as_vnish(self) -> dict: def as_vnish(self) -> dict:
return self.value.as_vnish() return self.value.as_vnish()
def as_auradine(self) -> dict:
return self.value.as_auradine()
def __call__(self, *args, **kwargs): def __call__(self, *args, **kwargs):
return self.value(*args, **kwargs) return self.value(*args, **kwargs)
@@ -99,3 +102,6 @@ class MinerConfigValue:
def as_vnish(self) -> dict: def as_vnish(self) -> dict:
return {} return {}
def as_auradine(self) -> dict:
return {}

View File

@@ -92,6 +92,9 @@ class FanModeManual(MinerConfigValue):
"fan_control": {"min_fans": self.minimum_fans, "speed": self.speed}, "fan_control": {"min_fans": self.minimum_fans, "speed": self.speed},
} }
def as_auradine(self) -> dict:
return {"fan": {"percentage": self.speed}}
@dataclass @dataclass
class FanModeImmersion(MinerConfigValue): class FanModeImmersion(MinerConfigValue):
@@ -107,6 +110,9 @@ class FanModeImmersion(MinerConfigValue):
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
return {"temp_control": {"mode": "disabled"}} return {"temp_control": {"mode": "disabled"}}
def as_auradine(self) -> dict:
return {"fan": {"percentage": 0}}
class FanModeConfig(MinerConfigOption): class FanModeConfig(MinerConfigOption):
normal = FanModeNormal normal = FanModeNormal
@@ -202,3 +208,13 @@ class FanModeConfig(MinerConfigOption):
if "minimumRequiredFans" in keys: if "minimumRequiredFans" in keys:
conf["minimum_fans"] = int(temperature_conf["minimumRequiredFans"]) conf["minimum_fans"] = int(temperature_conf["minimumRequiredFans"])
return cls.manual(**conf) return cls.manual(**conf)
@classmethod
def from_auradine(cls, web_fan: dict):
try:
fan_data = web_fan["Fan"][0]
fan_1_max = fan_data["Max"]
fan_1_target = fan_data["Target"]
return cls.manual(speed=round((fan_1_target / fan_1_max) * 100))
except LookupError:
return cls.default()

View File

@@ -43,6 +43,9 @@ class MiningModeNormal(MinerConfigValue):
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
def as_auradine(self) -> dict:
return {"mode": {"mode": self.mode}}
@dataclass @dataclass
class MiningModeSleep(MinerConfigValue): class MiningModeSleep(MinerConfigValue):
@@ -58,6 +61,9 @@ class MiningModeSleep(MinerConfigValue):
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
def as_auradine(self) -> dict:
return {"mode": {"sleep": "on"}}
@dataclass @dataclass
class MiningModeLPM(MinerConfigValue): class MiningModeLPM(MinerConfigValue):
@@ -73,6 +79,9 @@ class MiningModeLPM(MinerConfigValue):
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
def as_auradine(self) -> dict:
return {"mode": {"mode": "eco"}}
@dataclass @dataclass
class MiningModeHPM(MinerConfigValue): class MiningModeHPM(MinerConfigValue):
@@ -88,6 +97,9 @@ class MiningModeHPM(MinerConfigValue):
def as_wm(self) -> dict: def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
def as_auradine(self) -> dict:
return {"mode": {"mode": "turbo"}}
@dataclass @dataclass
class MiningModePowerTune(MinerConfigValue): class MiningModePowerTune(MinerConfigValue):
@@ -123,6 +135,9 @@ class MiningModePowerTune(MinerConfigValue):
), ),
} }
def as_auradine(self) -> dict:
return {"mode": {"mode": "custom", "tune": "power", "power": self.power}}
@dataclass @dataclass
class MiningModeHashrateTune(MinerConfigValue): class MiningModeHashrateTune(MinerConfigValue):
@@ -152,6 +167,9 @@ class MiningModeHashrateTune(MinerConfigValue):
) )
} }
def as_auradine(self) -> dict:
return {"mode": {"mode": "custom", "tune": "ths", "ths": self.hashrate}}
@dataclass @dataclass
class ManualBoardSettings(MinerConfigValue): class ManualBoardSettings(MinerConfigValue):
@@ -330,3 +348,22 @@ class MiningModeConfig(MinerConfigOption):
return cls.hashrate_tuning( return cls.hashrate_tuning(
int(tuner_conf["hashrateTarget"]["terahashPerSecond"]) int(tuner_conf["hashrateTarget"]["terahashPerSecond"])
) )
@classmethod
def from_auradine(cls, web_mode: dict):
try:
mode_data = web_mode["Mode"][0]
if mode_data.get("Sleep") == "on":
return cls.sleep()
if mode_data.get("Mode") == "normal":
return cls.normal()
if mode_data.get("Mode") == "eco":
return cls.low()
if mode_data.get("Mode") == "turbo":
return cls.high()
if mode_data.get("Ths") is not None:
return cls.hashrate_tuning(mode_data["Ths"])
if mode_data.get("Power") is not None:
return cls.power_tuning(mode_data["Power"])
except LookupError:
return cls.default()

View File

@@ -27,7 +27,7 @@ class Pool(MinerConfigValue):
user: str user: str
password: str password: str
def as_am_modern(self, user_suffix: str = None): def as_am_modern(self, user_suffix: str = None) -> dict:
if user_suffix is not None: if user_suffix is not None:
return { return {
"url": self.url, "url": self.url,
@@ -36,7 +36,7 @@ class Pool(MinerConfigValue):
} }
return {"url": self.url, "user": self.user, "pass": self.password} return {"url": self.url, "user": self.user, "pass": self.password}
def as_wm(self, idx: int = 1, user_suffix: str = None): def as_wm(self, idx: int = 1, user_suffix: str = None) -> dict:
if user_suffix is not None: if user_suffix is not None:
return { return {
f"pool_{idx}": self.url, f"pool_{idx}": self.url,
@@ -49,7 +49,7 @@ class Pool(MinerConfigValue):
f"passwd_{idx}": self.password, f"passwd_{idx}": self.password,
} }
def as_am_old(self, idx: int = 1, user_suffix: str = None): def as_am_old(self, idx: int = 1, user_suffix: str = None) -> dict:
if user_suffix is not None: if user_suffix is not None:
return { return {
f"_ant_pool{idx}url": self.url, f"_ant_pool{idx}url": self.url,
@@ -62,7 +62,7 @@ class Pool(MinerConfigValue):
f"_ant_pool{idx}pw": self.password, f"_ant_pool{idx}pw": self.password,
} }
def as_goldshell(self, user_suffix: str = None): def as_goldshell(self, user_suffix: str = None) -> dict:
if user_suffix is not None: if user_suffix is not None:
return { return {
"url": self.url, "url": self.url,
@@ -71,12 +71,12 @@ class Pool(MinerConfigValue):
} }
return {"url": self.url, "user": self.user, "pass": self.password} return {"url": self.url, "user": self.user, "pass": self.password}
def as_avalon(self, user_suffix: str = None): def as_avalon(self, user_suffix: str = None) -> str:
if user_suffix is not None: if user_suffix is not None:
return ",".join([self.url, f"{self.user}{user_suffix}", self.password]) return ",".join([self.url, f"{self.user}{user_suffix}", self.password])
return ",".join([self.url, self.user, self.password]) return ",".join([self.url, self.user, self.password])
def as_inno(self, idx: int = 1, user_suffix: str = None): def as_inno(self, idx: int = 1, user_suffix: str = None) -> dict:
if user_suffix is not None: if user_suffix is not None:
return { return {
f"Pool{idx}": self.url, f"Pool{idx}": self.url,
@@ -89,7 +89,7 @@ class Pool(MinerConfigValue):
f"Password{idx}": self.password, f"Password{idx}": self.password,
} }
def as_bosminer(self, user_suffix: str = None): def as_bosminer(self, user_suffix: str = None) -> dict:
if user_suffix is not None: if user_suffix is not None:
return { return {
"url": self.url, "url": self.url,
@@ -98,6 +98,15 @@ class Pool(MinerConfigValue):
} }
return {"url": self.url, "user": self.user, "password": self.password} return {"url": self.url, "user": self.user, "password": self.password}
def as_auradine(self, user_suffix: str = None) -> dict:
if user_suffix is not None:
return {
"url": self.url,
"user": f"{self.user}{user_suffix}",
"pass": self.password,
}
return {"url": self.url, "user": self.user, "pass": self.password}
@classmethod @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "Pool": def from_dict(cls, dict_conf: Union[dict, None]) -> "Pool":
return cls( return cls(
@@ -241,6 +250,9 @@ class PoolGroup(MinerConfigValue):
return conf return conf
return {"name": "Group", "pool": []} return {"name": "Group", "pool": []}
def as_auradine(self, user_suffix: str = None) -> list:
return [p.as_auradine(user_suffix=user_suffix) for p in self.pools]
@classmethod @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PoolGroup": def from_dict(cls, dict_conf: Union[dict, None]) -> "PoolGroup":
cls_conf = {} cls_conf = {}
@@ -296,7 +308,7 @@ class PoolGroup(MinerConfigValue):
return cls([Pool.from_vnish(p) for p in web_settings_pools]) return cls([Pool.from_vnish(p) for p in web_settings_pools])
@classmethod @classmethod
def from_boser(cls, grpc_pool_group: dict): def from_boser(cls, grpc_pool_group: dict) -> "PoolGroup":
try: try:
return cls( return cls(
pools=[Pool.from_boser(p) for p in grpc_pool_group["pools"]], pools=[Pool.from_boser(p) for p in grpc_pool_group["pools"]],
@@ -373,6 +385,15 @@ class PoolConfig(MinerConfigValue):
def as_boser(self, user_suffix: str = None) -> dict: def as_boser(self, user_suffix: str = None) -> dict:
return {} return {}
def as_auradine(self, user_suffix: str = None) -> dict:
if len(self.groups) > 0:
return {
"updatepools": {
"pools": self.groups[0].as_auradine(user_suffix=user_suffix)
}
}
return {"updatepools": {"pools": PoolGroup().as_auradine()}}
@classmethod @classmethod
def from_api(cls, api_pools: dict) -> "PoolConfig": def from_api(cls, api_pools: dict) -> "PoolConfig":
try: try:
@@ -417,7 +438,7 @@ class PoolConfig(MinerConfigValue):
return cls() return cls()
@classmethod @classmethod
def from_boser(cls, grpc_miner_conf: dict): def from_boser(cls, grpc_miner_conf: dict) -> "PoolConfig":
try: try:
return cls( return cls(
groups=[ groups=[

View File

@@ -13,7 +13,10 @@
# See the License for the specific language governing permissions and - # See the License for the specific language governing permissions and -
# limitations under the License. - # limitations under the License. -
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
import logging
from enum import Enum
from pyasic import APIError, MinerConfig
from pyasic.miners.base import BaseMiner, DataLocations from pyasic.miners.base import BaseMiner, DataLocations
from pyasic.rpc.gcminer import GCMinerRPCAPI from pyasic.rpc.gcminer import GCMinerRPCAPI
from pyasic.web.auradine import FluxWebAPI from pyasic.web.auradine import FluxWebAPI
@@ -21,6 +24,35 @@ from pyasic.web.auradine import FluxWebAPI
AURADINE_DATA_LOC = DataLocations(**{}) AURADINE_DATA_LOC = DataLocations(**{})
class AuradineLEDColors(Enum):
OFF = 0
GREEN = 1
RED = 2
YELLOW = 3
GREEN_FLASHING = 4
RED_FLASHING = 5
YELLOW_FLASHING = 6
def __int__(self):
return self.value
class AuradineLEDCodes(Enum):
NO_POWER = 1
NORMAL = 2
LOCATE_MINER = 3
TEMPERATURE = 4
POOL_CONFIG = 5
NETWORK = 6
CONTROL_BOARD = 7
HASH_RATE_LOW = 8
CUSTOM1 = 101
CUSTOM2 = 102
def __int__(self):
return self.value
class Auradine(BaseMiner): class Auradine(BaseMiner):
"""Base handler for Auradine miners""" """Base handler for Auradine miners"""
@@ -30,3 +62,68 @@ class Auradine(BaseMiner):
web: FluxWebAPI web: FluxWebAPI
data_locations = AURADINE_DATA_LOC data_locations = AURADINE_DATA_LOC
supports_shutdown = True
supports_autotuning = True
async def fault_light_on(self) -> bool:
return await self.web.set_led(code=int(AuradineLEDCodes.LOCATE_MINER))
async def fault_light_off(self) -> bool:
return await self.web.set_led(code=int(AuradineLEDCodes.NORMAL))
async def reboot(self) -> bool:
try:
await self.web.reboot()
except APIError:
return False
return True
async def restart_backend(self) -> bool:
try:
await self.web.restart_gcminer()
except APIError:
return False
return True
async def stop_mining(self) -> bool:
try:
await self.web.set_mode(sleep="on")
except APIError:
return False
return True
async def resume_mining(self) -> bool:
try:
await self.web.set_mode(sleep="off")
except APIError:
return False
return True
async def set_power_limit(self, wattage: int) -> bool:
try:
await self.web.set_mode(mode="custom", tune="power", power=wattage)
except APIError:
return False
return True
async def get_config(self) -> MinerConfig:
try:
web_conf = await self.web.multicommand("pools", "mode", "fan")
return MinerConfig.from_auradine(web_conf=web_conf)
except APIError as e:
logging.warning(e)
except LookupError:
pass
return MinerConfig()
async def send_config(self, config: MinerConfig, user_suffix: str = None) -> None:
self.config = config
conf = config.as_auradine(user_suffix=user_suffix)
for key in conf.keys():
await self.web.send_command(command=key, **conf[key])
##################################################
### DATA GATHERING FUNCTIONS (get_{some_data}) ###
##################################################

View File

@@ -192,7 +192,10 @@ class FluxWebAPI(BaseWebAPI):
async def get_led(self): async def get_led(self):
return await self.send_command("led") return await self.send_command("led")
async def set_led(self, code: int, led_1: int, led_2: int, msg: str = ""): async def set_led(self, code: int):
return await self.send_command("led", code=code)
async def set_led_custom(self, code: int, led_1: int, led_2: int, msg: str):
return await self.send_command( return await self.send_command(
"led", code=code, led1=led_1, led2=led_2, msg=msg "led", code=code, led1=led_1, led2=led_2, msg=msg
) )