feature: allow config conversion to and from dict.

This commit is contained in:
UpstreamData
2023-12-11 13:40:10 -07:00
parent f162529883
commit c919b00312
7 changed files with 274 additions and 76 deletions

View File

@@ -13,14 +13,14 @@
# 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. -
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
from dataclasses import dataclass from dataclasses import asdict, dataclass
import toml import toml
from pyasic.config.fans import FanModeConfig from pyasic.config.fans import FanModeConfig
from pyasic.config.mining import MiningModeConfig from pyasic.config.mining import MiningModeConfig
from pyasic.config.pools import PoolConfig from pyasic.config.pools import PoolConfig
from pyasic.config.power_scaling import PowerScalingConfig from pyasic.config.power_scaling import PowerScalingConfig, PowerScalingShutdown
from pyasic.config.temperature import TemperatureConfig from pyasic.config.temperature import TemperatureConfig
@@ -32,7 +32,10 @@ class MinerConfig:
temperature: TemperatureConfig = TemperatureConfig.default() temperature: TemperatureConfig = TemperatureConfig.default()
power_scaling: PowerScalingConfig = PowerScalingConfig.default() power_scaling: PowerScalingConfig = PowerScalingConfig.default()
def as_am_modern(self, user_suffix: str = None): def as_dict(self) -> dict:
return asdict(self)
def as_am_modern(self, user_suffix: str = None) -> dict:
return { return {
**self.fan_mode.as_am_modern(), **self.fan_mode.as_am_modern(),
"freq-level": "100", "freq-level": "100",
@@ -42,7 +45,7 @@ class MinerConfig:
**self.power_scaling.as_am_modern(), **self.power_scaling.as_am_modern(),
} }
def as_wm(self, user_suffix: str = None): def as_wm(self, user_suffix: str = None) -> dict:
return { return {
**self.fan_mode.as_wm(), **self.fan_mode.as_wm(),
**self.mining_mode.as_wm(), **self.mining_mode.as_wm(),
@@ -51,7 +54,7 @@ class MinerConfig:
**self.power_scaling.as_wm(), **self.power_scaling.as_wm(),
} }
def as_am_old(self, user_suffix: str = None): def as_am_old(self, user_suffix: str = None) -> dict:
return { return {
**self.fan_mode.as_am_old(), **self.fan_mode.as_am_old(),
**self.mining_mode.as_am_old(), **self.mining_mode.as_am_old(),
@@ -60,7 +63,7 @@ class MinerConfig:
**self.power_scaling.as_am_old(), **self.power_scaling.as_am_old(),
} }
def as_goldshell(self, user_suffix: str = None): def as_goldshell(self, user_suffix: str = None) -> dict:
return { return {
**self.fan_mode.as_goldshell(), **self.fan_mode.as_goldshell(),
**self.mining_mode.as_goldshell(), **self.mining_mode.as_goldshell(),
@@ -69,7 +72,7 @@ class MinerConfig:
**self.power_scaling.as_goldshell(), **self.power_scaling.as_goldshell(),
} }
def as_avalon(self, user_suffix: str = None): def as_avalon(self, user_suffix: str = None) -> dict:
return { return {
**self.fan_mode.as_avalon(), **self.fan_mode.as_avalon(),
**self.mining_mode.as_avalon(), **self.mining_mode.as_avalon(),
@@ -78,7 +81,7 @@ class MinerConfig:
**self.power_scaling.as_avalon(), **self.power_scaling.as_avalon(),
} }
def as_inno(self, user_suffix: str = None): def as_inno(self, user_suffix: str = None) -> dict:
return { return {
**self.fan_mode.as_inno(), **self.fan_mode.as_inno(),
**self.mining_mode.as_inno(), **self.mining_mode.as_inno(),
@@ -87,7 +90,7 @@ class MinerConfig:
**self.power_scaling.as_inno(), **self.power_scaling.as_inno(),
} }
def as_bosminer(self, user_suffix: str = None): def as_bosminer(self, user_suffix: str = None) -> dict:
return { return {
**merge(self.fan_mode.as_bosminer(), self.temperature.as_bosminer()), **merge(self.fan_mode.as_bosminer(), self.temperature.as_bosminer()),
**self.mining_mode.as_bosminer(), **self.mining_mode.as_bosminer(),
@@ -95,7 +98,7 @@ class MinerConfig:
**self.power_scaling.as_bosminer(), **self.power_scaling.as_bosminer(),
} }
def as_bos_grpc(self, user_suffix: str = None): def as_bos_grpc(self, user_suffix: str = None) -> dict:
return { return {
**self.fan_mode.as_bos_grpc(), **self.fan_mode.as_bos_grpc(),
**self.temperature.as_bos_grpc(), **self.temperature.as_bos_grpc(),
@@ -105,11 +108,21 @@ class MinerConfig:
} }
@classmethod @classmethod
def from_api(cls, api_pools: dict): def from_dict(cls, dict_conf: dict) -> "MinerConfig":
return cls(
pools=PoolConfig.from_dict(dict_conf.get("pools")),
mining_mode=MiningModeConfig.from_dict(dict_conf.get("mining_mode")),
fan_mode=FanModeConfig.from_dict(dict_conf.get("fan_mode")),
temperature=TemperatureConfig.from_dict(dict_conf.get("temperature")),
power_scaling=PowerScalingConfig.from_dict(dict_conf.get("power_scaling")),
)
@classmethod
def from_api(cls, api_pools: dict) -> "MinerConfig":
return cls(pools=PoolConfig.from_api(api_pools)) return cls(pools=PoolConfig.from_api(api_pools))
@classmethod @classmethod
def from_am_modern(cls, web_conf: dict): def from_am_modern(cls, web_conf: dict) -> "MinerConfig":
return cls( return cls(
pools=PoolConfig.from_am_modern(web_conf), pools=PoolConfig.from_am_modern(web_conf),
mining_mode=MiningModeConfig.from_am_modern(web_conf), mining_mode=MiningModeConfig.from_am_modern(web_conf),
@@ -117,19 +130,19 @@ class MinerConfig:
) )
@classmethod @classmethod
def from_am_old(cls, web_conf: dict): def from_am_old(cls, web_conf: dict) -> "MinerConfig":
return cls.from_am_modern(web_conf) return cls.from_am_modern(web_conf)
@classmethod @classmethod
def from_goldshell(cls, web_conf: dict): def from_goldshell(cls, web_conf: dict) -> "MinerConfig":
return cls(pools=PoolConfig.from_am_modern(web_conf)) return cls(pools=PoolConfig.from_am_modern(web_conf))
@classmethod @classmethod
def from_inno(cls, web_pools: list): def from_inno(cls, web_pools: list) -> "MinerConfig":
return cls(pools=PoolConfig.from_inno(web_pools)) return cls(pools=PoolConfig.from_inno(web_pools))
@classmethod @classmethod
def from_bosminer(cls, toml_conf: dict): def from_bosminer(cls, toml_conf: dict) -> "MinerConfig":
return cls( return cls(
pools=PoolConfig.from_bosminer(toml_conf), pools=PoolConfig.from_bosminer(toml_conf),
mining_mode=MiningModeConfig.from_bosminer(toml_conf), mining_mode=MiningModeConfig.from_bosminer(toml_conf),
@@ -139,7 +152,7 @@ class MinerConfig:
) )
def merge(a: dict, b: dict): def merge(a: dict, b: dict) -> dict:
ret = {} ret = {}
for k in a: for k in a:
v = a[k] v = a[k]
@@ -173,20 +186,26 @@ if __name__ == "__main__":
mining_mode=MiningModeConfig.power_tuning(3000), mining_mode=MiningModeConfig.power_tuning(3000),
temperature=TemperatureConfig(hot=100, danger=110), temperature=TemperatureConfig(hot=100, danger=110),
fan_mode=FanModeConfig.manual(minimum_fans=2, speed=70), fan_mode=FanModeConfig.manual(minimum_fans=2, speed=70),
power_scaling=PowerScalingConfig.enabled(power_step=100, minimum_power=2400), power_scaling=PowerScalingConfig.enabled(
power_step=100,
minimum_power=2400,
shutdown_enabled=PowerScalingShutdown.enabled(duration=3),
),
) )
print(config) print(config)
print("WM:", config.as_wm()) # print("WM:", config.as_wm())
print("AM Modern:", config.as_am_modern()) # print("AM Modern:", config.as_am_modern())
print("AM Old:", config.as_am_old()) # print("AM Old:", config.as_am_old())
print("GS:", config.as_goldshell()) # print("GS:", config.as_goldshell())
print("Avalon:", config.as_avalon()) # print("Avalon:", config.as_avalon())
print("Inno:", config.as_inno()) # print("Inno:", config.as_inno())
print("BOS+ .toml:", config.as_bosminer()) # print("BOS+ .toml:", config.as_bosminer())
print("BOS+ .toml as toml:") # print("BOS+ .toml as toml:")
print(toml.dumps(config.as_bosminer())) # print(toml.dumps(config.as_bosminer()))
print(config.as_bos_grpc()) # print(config.as_bos_grpc())
dict_config = config.as_dict()
bos_parsed = MinerConfig.from_bosminer(config.as_bosminer()) parsed_conf = MinerConfig.from_dict(dict_config)
print(bos_parsed) print(parsed_conf)
print(bos_parsed == config) # bos_parsed = MinerConfig.from_bosminer(config.as_bosminer())
# print(bos_parsed)
# print(bos_parsed == config)

View File

@@ -13,10 +13,17 @@
# 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. -
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
from abc import ABC, abstractmethod
from dataclasses import asdict, dataclass
from enum import Enum from enum import Enum
from typing import Union
class MinerConfigOption(Enum): class MinerConfigOption(Enum):
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
return cls.default()
def as_am_modern(self) -> dict: def as_am_modern(self) -> dict:
return self.value.as_am_modern() return self.value.as_am_modern()
@@ -44,8 +51,20 @@ class MinerConfigOption(Enum):
def __call__(self, *args, **kwargs): def __call__(self, *args, **kwargs):
return self.value(*args, **kwargs) return self.value(*args, **kwargs)
@classmethod
def default(cls):
pass
@dataclass
class MinerConfigValue: class MinerConfigValue:
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
return cls()
def as_dict(self):
return asdict(self)
def as_am_modern(self) -> dict: def as_am_modern(self) -> dict:
return {} return {}

View File

@@ -14,6 +14,7 @@
# limitations under the License. - # limitations under the License. -
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Union
from pyasic.config.base import MinerConfigOption, MinerConfigValue from pyasic.config.base import MinerConfigOption, MinerConfigValue
@@ -22,10 +23,14 @@ from pyasic.config.base import MinerConfigOption, MinerConfigValue
class FanModeNormal(MinerConfigValue): class FanModeNormal(MinerConfigValue):
mode: str = field(init=False, default="auto") mode: str = field(init=False, default="auto")
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "FanModeNormal":
return cls()
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_bosminer(self): def as_bosminer(self) -> dict:
return {"temp_control": {"mode": "auto"}} return {"temp_control": {"mode": "auto"}}
@@ -36,7 +41,16 @@ class FanModeManual(MinerConfigValue):
speed: int = 100 speed: int = 100
@classmethod @classmethod
def from_bosminer(cls, toml_fan_conf: dict): def from_dict(cls, dict_conf: Union[dict, None]) -> "FanModeManual":
cls_conf = {}
if dict_conf.get("min_fans") is not None:
cls_conf["minimum_fans"] = dict_conf["minimum_fans"]
if dict_conf.get("speed") is not None:
cls_conf["speed"] = dict_conf["speed"]
return cls(**cls_conf)
@classmethod
def from_bosminer(cls, toml_fan_conf: dict) -> "FanModeManual":
cls_conf = {} cls_conf = {}
if toml_fan_conf.get("min_fans") is not None: if toml_fan_conf.get("min_fans") is not None:
cls_conf["minimum_fans"] = toml_fan_conf["min_fans"] cls_conf["minimum_fans"] = toml_fan_conf["min_fans"]
@@ -44,12 +58,10 @@ class FanModeManual(MinerConfigValue):
cls_conf["speed"] = toml_fan_conf["speed"] cls_conf["speed"] = toml_fan_conf["speed"]
return cls(**cls_conf) return cls(**cls_conf)
def as_am_modern(self) -> dict:
def as_am_modern(self):
return {"bitmain-fan-ctrl": True, "bitmain-fan-pwn": str(self.speed)} return {"bitmain-fan-ctrl": True, "bitmain-fan-pwn": str(self.speed)}
def as_bosminer(self): def as_bosminer(self) -> dict:
return { return {
"temp_control": {"mode": "manual"}, "temp_control": {"mode": "manual"},
"fan_control": {"min_fans": self.minimum_fans, "speed": self.speed}, "fan_control": {"min_fans": self.minimum_fans, "speed": self.speed},
@@ -60,10 +72,14 @@ class FanModeManual(MinerConfigValue):
class FanModeImmersion(MinerConfigValue): class FanModeImmersion(MinerConfigValue):
mode: str = field(init=False, default="immersion") mode: str = field(init=False, default="immersion")
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "FanModeImmersion":
return cls()
def as_am_modern(self) -> dict:
return {"bitmain-fan-ctrl": True, "bitmain-fan-pwn": "0"} return {"bitmain-fan-ctrl": True, "bitmain-fan-pwn": "0"}
def as_bosminer(self): def as_bosminer(self) -> dict:
return {"temp_control": {"mode": "disabled"}} return {"temp_control": {"mode": "disabled"}}
@@ -76,6 +92,19 @@ class FanModeConfig(MinerConfigOption):
def default(cls): def default(cls):
return cls.normal() return cls.normal()
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
if dict_conf is None:
return cls.default()
mode = dict_conf.get("mode")
if mode is None:
return cls.default()
clsattr = getattr(cls, mode)
if clsattr is not None:
return clsattr().from_dict(dict_conf)
@classmethod @classmethod
def from_am_modern(cls, web_conf: dict): def from_am_modern(cls, web_conf: dict):
if web_conf.get("bitmain-fan-ctrl") is not None: if web_conf.get("bitmain-fan-ctrl") is not None:

View File

@@ -14,6 +14,7 @@
# limitations under the License. - # limitations under the License. -
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Union
from pyasic.config.base import MinerConfigOption, MinerConfigValue from pyasic.config.base import MinerConfigOption, MinerConfigValue
@@ -22,10 +23,14 @@ from pyasic.config.base import MinerConfigOption, MinerConfigValue
class MiningModeNormal(MinerConfigValue): class MiningModeNormal(MinerConfigValue):
mode: str = field(init=False, default="normal") mode: str = field(init=False, default="normal")
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeNormal":
return cls()
def as_am_modern(self) -> dict:
return {"miner-mode": "0"} return {"miner-mode": "0"}
def as_wm(self): def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -33,10 +38,14 @@ class MiningModeNormal(MinerConfigValue):
class MiningModeSleep(MinerConfigValue): class MiningModeSleep(MinerConfigValue):
mode: str = field(init=False, default="sleep") mode: str = field(init=False, default="sleep")
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeSleep":
return cls()
def as_am_modern(self) -> dict:
return {"miner-mode": "1"} return {"miner-mode": "1"}
def as_wm(self): def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -44,10 +53,14 @@ class MiningModeSleep(MinerConfigValue):
class MiningModeLPM(MinerConfigValue): class MiningModeLPM(MinerConfigValue):
mode: str = field(init=False, default="low") mode: str = field(init=False, default="low")
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeLPM":
return cls()
def as_am_modern(self) -> dict:
return {"miner-mode": "3"} return {"miner-mode": "3"}
def as_wm(self): def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -55,10 +68,14 @@ class MiningModeLPM(MinerConfigValue):
class MiningModeHPM(MinerConfigValue): class MiningModeHPM(MinerConfigValue):
mode: str = field(init=False, default="high") mode: str = field(init=False, default="high")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeHPM":
return cls()
def as_am_modern(self): def as_am_modern(self):
return {"miner-mode": "0"} return {"miner-mode": "0"}
def as_wm(self): def as_wm(self) -> dict:
return {"mode": self.mode} return {"mode": self.mode}
@@ -67,10 +84,14 @@ class MiningModePowerTune(MinerConfigValue):
mode: str = field(init=False, default="power_tuning") mode: str = field(init=False, default="power_tuning")
power: int = None power: int = None
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModePowerTune":
return cls(dict_conf.get("power"))
def as_am_modern(self) -> dict:
return {"miner-mode": "0"} return {"miner-mode": "0"}
def as_wm(self): 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}}
return {} return {}
@@ -84,7 +105,11 @@ class MiningModeHashrateTune(MinerConfigValue):
mode: str = field(init=False, default="hashrate_tuning") mode: str = field(init=False, default="hashrate_tuning")
hashrate: int = None hashrate: int = None
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeHashrateTune":
return cls(dict_conf.get("hashrate"))
def as_am_modern(self) -> dict:
return {"miner-mode": "0"} return {"miner-mode": "0"}
@@ -93,7 +118,11 @@ class ManualBoardSettings(MinerConfigValue):
freq: float freq: float
volt: float volt: float
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "ManualBoardSettings":
return cls(freq=dict_conf["freq"], volt=dict_conf["volt"])
def as_am_modern(self) -> dict:
return {"miner-mode": "0"} return {"miner-mode": "0"}
@@ -105,7 +134,15 @@ class MiningModeManual(MinerConfigValue):
global_volt: float global_volt: float
boards: dict[int, ManualBoardSettings] = field(default_factory=dict) boards: dict[int, ManualBoardSettings] = field(default_factory=dict)
def as_am_modern(self): @classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeManual":
return cls(
global_freq=dict_conf["global_freq"],
global_volt=dict_conf["global_volt"],
boards={i: ManualBoardSettings.from_dict(dict_conf[i]) for i in dict_conf},
)
def as_am_modern(self) -> dict:
return {"miner-mode": "0"} return {"miner-mode": "0"}
@@ -122,6 +159,19 @@ class MiningModeConfig(MinerConfigOption):
def default(cls): def default(cls):
return cls.normal() return cls.normal()
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
if dict_conf is None:
return cls.default()
mode = dict_conf.get("mode")
if mode is None:
return cls.default()
clsattr = getattr(cls, mode)
if clsattr is not None:
return clsattr().from_dict(dict_conf)
@classmethod @classmethod
def from_am_modern(cls, web_conf: dict): def from_am_modern(cls, web_conf: dict):
if web_conf.get("bitmain-work-mode") is not None: if web_conf.get("bitmain-work-mode") is not None:
@@ -159,4 +209,3 @@ class MiningModeConfig(MinerConfigOption):
if autotuning_conf.get("hashrate_target") is not None: if autotuning_conf.get("hashrate_target") is not None:
return cls.hashrate_tuning(autotuning_conf["hashrate_target"]) return cls.hashrate_tuning(autotuning_conf["hashrate_target"])
return cls.hashrate_tuning() return cls.hashrate_tuning()

View File

@@ -94,36 +94,42 @@ class Pool(MinerConfigValue):
return { return {
"url": self.url, "url": self.url,
"user": f"{self.user}{user_suffix}", "user": f"{self.user}{user_suffix}",
"pass": self.password, "password": self.password,
} }
return {"url": self.url, "user": self.user, "pass": self.password} return {"url": self.url, "user": self.user, "password": self.password}
@classmethod @classmethod
def from_api(cls, api_pool: dict): def from_dict(cls, dict_conf: Union[dict, None]) -> "Pool":
return cls(
url=dict_conf["url"], user=dict_conf["user"], password=dict_conf["password"]
)
@classmethod
def from_api(cls, api_pool: dict) -> "Pool":
return cls(url=api_pool["URL"], user=api_pool["User"], password="x") return cls(url=api_pool["URL"], user=api_pool["User"], password="x")
@classmethod @classmethod
def from_am_modern(cls, web_pool: dict): def from_am_modern(cls, web_pool: dict) -> "Pool":
return cls( return cls(
url=web_pool["url"], user=web_pool["user"], password=web_pool["pass"] 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): def from_goldshell(cls, web_pool: dict) -> "Pool":
return cls( return cls(
url=web_pool["url"], user=web_pool["user"], password=web_pool["pass"] 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_inno(cls, web_pool: dict): def from_inno(cls, web_pool: dict) -> "Pool":
return cls( return cls(
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 @classmethod
def from_bosminer(cls, toml_pool_conf: dict): def from_bosminer(cls, toml_pool_conf: dict) -> "Pool":
return cls( return cls(
url=toml_pool_conf["url"], url=toml_pool_conf["url"],
user=toml_pool_conf["user"], user=toml_pool_conf["user"],
@@ -211,32 +217,44 @@ class PoolGroup(MinerConfigValue):
} }
if self.quota is not None: if self.quota is not None:
conf["quota"] = self.quota conf["quota"] = self.quota
return conf
return {"name": "Group", "pool": []} return {"name": "Group", "pool": []}
@classmethod @classmethod
def from_api(cls, api_pool_list: list): def from_dict(cls, dict_conf: Union[dict, None]) -> "PoolGroup":
cls_conf = {}
if dict_conf.get("quota") is not None:
cls_conf["quota"] = dict_conf["quota"]
if dict_conf.get("name") is not None:
cls_conf["name"] = dict_conf["name"]
cls_conf["pools"] = [Pool.from_dict(p) for p in dict_conf["pools"]]
return cls(**cls_conf)
@classmethod
def from_api(cls, api_pool_list: list) -> "PoolGroup":
pools = [] pools = []
for pool in api_pool_list: for pool in api_pool_list:
pools.append(Pool.from_api(pool)) pools.append(Pool.from_api(pool))
return cls(pools=pools) return cls(pools=pools)
@classmethod @classmethod
def from_am_modern(cls, web_pool_list: list): def from_am_modern(cls, web_pool_list: list) -> "PoolGroup":
pools = [] pools = []
for pool in web_pool_list: for pool in web_pool_list:
pools.append(Pool.from_am_modern(pool)) pools.append(Pool.from_am_modern(pool))
return cls(pools=pools) return cls(pools=pools)
@classmethod @classmethod
def from_goldshell(cls, web_pools: list): def from_goldshell(cls, web_pools: list) -> "PoolGroup":
return cls([Pool.from_goldshell(p) for p in web_pools]) return cls([Pool.from_goldshell(p) for p in web_pools])
@classmethod @classmethod
def from_inno(cls, web_pools: list): def from_inno(cls, web_pools: list) -> "PoolGroup":
return cls([Pool.from_inno(p) for p in web_pools]) return cls([Pool.from_inno(p) for p in web_pools])
@classmethod @classmethod
def from_bosminer(cls, toml_group_conf: dict): def from_bosminer(cls, toml_group_conf: dict) -> "PoolGroup":
if toml_group_conf.get("pool") is not None: if toml_group_conf.get("pool") is not None:
return cls( return cls(
name=toml_group_conf["name"], name=toml_group_conf["name"],
@@ -251,11 +269,18 @@ class PoolConfig(MinerConfigValue):
groups: list[PoolGroup] = field(default_factory=list) groups: list[PoolGroup] = field(default_factory=list)
@classmethod @classmethod
def default(cls): def default(cls) -> "PoolConfig":
return cls() return cls()
@classmethod @classmethod
def simple(cls, pools: list[Union[Pool, dict[str, str]]]): def from_dict(cls, dict_conf: Union[dict, None]) -> "PoolConfig":
if dict_conf is None:
return cls.default()
return cls(groups=[PoolGroup.from_dict(g) for g in dict_conf["groups"]])
@classmethod
def simple(cls, pools: list[Union[Pool, dict[str, str]]]) -> "PoolConfig":
group_pools = [] group_pools = []
for pool in pools: for pool in pools:
if isinstance(pool, dict): if isinstance(pool, dict):
@@ -304,28 +329,28 @@ class PoolConfig(MinerConfigValue):
return {} return {}
@classmethod @classmethod
def from_api(cls, api_pools: dict): def from_api(cls, api_pools: dict) -> "PoolConfig":
pool_data = api_pools["POOLS"] pool_data = api_pools["POOLS"]
pool_data = sorted(pool_data, key=lambda x: int(x["POOL"])) pool_data = sorted(pool_data, key=lambda x: int(x["POOL"]))
return cls([PoolGroup.from_api(pool_data)]) return cls([PoolGroup.from_api(pool_data)])
@classmethod @classmethod
def from_am_modern(cls, web_conf: dict): def from_am_modern(cls, web_conf: dict) -> "PoolConfig":
pool_data = web_conf["pools"] pool_data = web_conf["pools"]
return cls([PoolGroup.from_am_modern(pool_data)]) return cls([PoolGroup.from_am_modern(pool_data)])
@classmethod @classmethod
def from_goldshell(cls, web_pools: list): def from_goldshell(cls, web_pools: list) -> "PoolConfig":
return cls([PoolGroup.from_goldshell(web_pools)]) return cls([PoolGroup.from_goldshell(web_pools)])
@classmethod @classmethod
def from_inno(cls, web_pools: list): def from_inno(cls, web_pools: list) -> "PoolConfig":
return cls([PoolGroup.from_inno(web_pools)]) return cls([PoolGroup.from_inno(web_pools)])
@classmethod @classmethod
def from_bosminer(cls, toml_conf: dict): def from_bosminer(cls, toml_conf: dict) -> "PoolConfig":
if toml_conf.get("group") is None: if toml_conf.get("group") is None:
return cls() return cls()

View File

@@ -14,6 +14,7 @@
# limitations under the License. - # limitations under the License. -
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Union
from pyasic.config.base import MinerConfigOption, MinerConfigValue from pyasic.config.base import MinerConfigOption, MinerConfigValue
from pyasic.web.bosminer.proto.braiins.bos.v1 import DpsPowerTarget, DpsTarget, Hours from pyasic.web.bosminer.proto.braiins.bos.v1 import DpsPowerTarget, DpsTarget, Hours
@@ -24,6 +25,10 @@ class PowerScalingShutdownEnabled(MinerConfigValue):
mode: str = field(init=False, default="enabled") mode: str = field(init=False, default="enabled")
duration: int = None duration: int = None
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PowerScalingShutdownEnabled":
return cls(duration=dict_conf.get("duration"))
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
cfg = {"shutdown_enabled": True} cfg = {"shutdown_enabled": True}
@@ -45,6 +50,10 @@ class PowerScalingShutdownEnabled(MinerConfigValue):
class PowerScalingShutdownDisabled(MinerConfigValue): class PowerScalingShutdownDisabled(MinerConfigValue):
mode: str = field(init=False, default="disabled") mode: str = field(init=False, default="disabled")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PowerScalingShutdownDisabled":
return cls()
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
return {"shutdown_enabled": False} return {"shutdown_enabled": False}
@@ -56,6 +65,19 @@ class PowerScalingShutdown(MinerConfigOption):
enabled = PowerScalingShutdownEnabled enabled = PowerScalingShutdownEnabled
disabled = PowerScalingShutdownDisabled disabled = PowerScalingShutdownDisabled
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
if dict_conf is None:
return cls.default()
mode = dict_conf.get("mode")
if mode is None:
return cls.default()
clsattr = getattr(cls, mode)
if clsattr is not None:
return clsattr().from_dict(dict_conf)
@classmethod @classmethod
def from_bosminer(cls, power_scaling_conf: dict): def from_bosminer(cls, power_scaling_conf: dict):
sd_enabled = power_scaling_conf.get("shutdown_enabled") sd_enabled = power_scaling_conf.get("shutdown_enabled")
@@ -75,7 +97,7 @@ class PowerScalingEnabled(MinerConfigValue):
shutdown_enabled: PowerScalingShutdown = None shutdown_enabled: PowerScalingShutdown = None
@classmethod @classmethod
def from_bosminer(cls, power_scaling_conf: dict): def from_bosminer(cls, power_scaling_conf: dict) -> "PowerScalingEnabled":
power_step = power_scaling_conf.get("power_step") power_step = power_scaling_conf.get("power_step")
min_power = power_scaling_conf.get("min_psu_power_limit") min_power = power_scaling_conf.get("min_psu_power_limit")
sd_mode = PowerScalingShutdown.from_bosminer(power_scaling_conf) sd_mode = PowerScalingShutdown.from_bosminer(power_scaling_conf)
@@ -84,6 +106,19 @@ class PowerScalingEnabled(MinerConfigValue):
power_step=power_step, minimum_power=min_power, shutdown_enabled=sd_mode power_step=power_step, minimum_power=min_power, shutdown_enabled=sd_mode
) )
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PowerScalingEnabled":
cls_conf = {
"power_step": dict_conf.get("power_step"),
"minimum_power": dict_conf.get("minimum_power"),
}
shutdown_enabled = dict_conf.get("shutdown_enabled")
if shutdown_enabled is not None:
cls_conf["shutdown_enabled"] = PowerScalingShutdown.from_dict(
shutdown_enabled
)
return cls(**cls_conf)
def as_bosminer(self) -> dict: def as_bosminer(self) -> dict:
cfg = {"enabled": True} cfg = {"enabled": True}
if self.power_step is not None: if self.power_step is not None:
@@ -96,7 +131,7 @@ class PowerScalingEnabled(MinerConfigValue):
return {"power_scaling": cfg} return {"power_scaling": cfg}
def as_bos_grpc(self): def as_bos_grpc(self) -> dict:
cfg = {"enable": True} cfg = {"enable": True}
target_conf = {} target_conf = {}
if self.power_step is not None: if self.power_step is not None:
@@ -125,6 +160,19 @@ class PowerScalingConfig(MinerConfigOption):
def default(cls): def default(cls):
return cls.disabled() return cls.disabled()
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
if dict_conf is None:
return cls.default()
mode = dict_conf.get("mode")
if mode is None:
return cls.default()
clsattr = getattr(cls, mode)
if clsattr is not None:
return clsattr().from_dict(dict_conf)
@classmethod @classmethod
def from_bosminer(cls, toml_conf: dict): def from_bosminer(cls, toml_conf: dict):
power_scaling = toml_conf.get("power_scaling") power_scaling = toml_conf.get("power_scaling")

View File

@@ -14,6 +14,7 @@
# limitations under the License. - # limitations under the License. -
# ------------------------------------------------------------------------------ # ------------------------------------------------------------------------------
from dataclasses import dataclass from dataclasses import dataclass
from typing import Union
from pyasic.config.base import MinerConfigValue from pyasic.config.base import MinerConfigValue
@@ -39,7 +40,15 @@ class TemperatureConfig(MinerConfigValue):
return {"temp_control": temp_cfg} return {"temp_control": temp_cfg}
@classmethod @classmethod
def from_bosminer(cls, toml_conf: dict): def from_dict(cls, dict_conf: Union[dict, None]) -> "TemperatureConfig":
return cls(
target=dict_conf.get("target"),
hot=dict_conf.get("hot"),
danger=dict_conf.get("danger"),
)
@classmethod
def from_bosminer(cls, toml_conf: dict) -> "TemperatureConfig":
temp_control = toml_conf.get("temp_control") temp_control = toml_conf.get("temp_control")
if temp_control is not None: if temp_control is not None:
return cls( return cls(