From c919b0031243a18622aad8b54a040d13ceaa85de Mon Sep 17 00:00:00 2001 From: UpstreamData Date: Mon, 11 Dec 2023 13:40:10 -0700 Subject: [PATCH] feature: allow config conversion to and from dict. --- pyasic/config/__init__.py | 83 +++++++++++++++++++++------------- pyasic/config/base.py | 19 ++++++++ pyasic/config/fans.py | 47 +++++++++++++++---- pyasic/config/mining.py | 75 ++++++++++++++++++++++++------ pyasic/config/pools.py | 63 ++++++++++++++++++-------- pyasic/config/power_scaling.py | 52 ++++++++++++++++++++- pyasic/config/temperature.py | 11 ++++- 7 files changed, 274 insertions(+), 76 deletions(-) diff --git a/pyasic/config/__init__.py b/pyasic/config/__init__.py index bfdc6970..9fb75a2a 100644 --- a/pyasic/config/__init__.py +++ b/pyasic/config/__init__.py @@ -13,14 +13,14 @@ # See the License for the specific language governing permissions and - # limitations under the License. - # ------------------------------------------------------------------------------ -from dataclasses import dataclass +from dataclasses import asdict, dataclass import toml from pyasic.config.fans import FanModeConfig from pyasic.config.mining import MiningModeConfig 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 @@ -32,7 +32,10 @@ class MinerConfig: temperature: TemperatureConfig = TemperatureConfig.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 { **self.fan_mode.as_am_modern(), "freq-level": "100", @@ -42,7 +45,7 @@ class MinerConfig: **self.power_scaling.as_am_modern(), } - def as_wm(self, user_suffix: str = None): + def as_wm(self, user_suffix: str = None) -> dict: return { **self.fan_mode.as_wm(), **self.mining_mode.as_wm(), @@ -51,7 +54,7 @@ class MinerConfig: **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 { **self.fan_mode.as_am_old(), **self.mining_mode.as_am_old(), @@ -60,7 +63,7 @@ class MinerConfig: **self.power_scaling.as_am_old(), } - def as_goldshell(self, user_suffix: str = None): + def as_goldshell(self, user_suffix: str = None) -> dict: return { **self.fan_mode.as_goldshell(), **self.mining_mode.as_goldshell(), @@ -69,7 +72,7 @@ class MinerConfig: **self.power_scaling.as_goldshell(), } - def as_avalon(self, user_suffix: str = None): + def as_avalon(self, user_suffix: str = None) -> dict: return { **self.fan_mode.as_avalon(), **self.mining_mode.as_avalon(), @@ -78,7 +81,7 @@ class MinerConfig: **self.power_scaling.as_avalon(), } - def as_inno(self, user_suffix: str = None): + def as_inno(self, user_suffix: str = None) -> dict: return { **self.fan_mode.as_inno(), **self.mining_mode.as_inno(), @@ -87,7 +90,7 @@ class MinerConfig: **self.power_scaling.as_inno(), } - def as_bosminer(self, user_suffix: str = None): + def as_bosminer(self, user_suffix: str = None) -> dict: return { **merge(self.fan_mode.as_bosminer(), self.temperature.as_bosminer()), **self.mining_mode.as_bosminer(), @@ -95,7 +98,7 @@ class MinerConfig: **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 { **self.fan_mode.as_bos_grpc(), **self.temperature.as_bos_grpc(), @@ -105,11 +108,21 @@ class MinerConfig: } @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)) @classmethod - def from_am_modern(cls, web_conf: dict): + def from_am_modern(cls, web_conf: dict) -> "MinerConfig": return cls( pools=PoolConfig.from_am_modern(web_conf), mining_mode=MiningModeConfig.from_am_modern(web_conf), @@ -117,19 +130,19 @@ class MinerConfig: ) @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) @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)) @classmethod - def from_inno(cls, web_pools: list): + def from_inno(cls, web_pools: list) -> "MinerConfig": return cls(pools=PoolConfig.from_inno(web_pools)) @classmethod - def from_bosminer(cls, toml_conf: dict): + def from_bosminer(cls, toml_conf: dict) -> "MinerConfig": return cls( pools=PoolConfig.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 = {} for k in a: v = a[k] @@ -173,20 +186,26 @@ if __name__ == "__main__": mining_mode=MiningModeConfig.power_tuning(3000), temperature=TemperatureConfig(hot=100, danger=110), 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("WM:", config.as_wm()) - print("AM Modern:", config.as_am_modern()) - print("AM Old:", config.as_am_old()) - print("GS:", config.as_goldshell()) - print("Avalon:", config.as_avalon()) - print("Inno:", config.as_inno()) - print("BOS+ .toml:", config.as_bosminer()) - print("BOS+ .toml as toml:") - print(toml.dumps(config.as_bosminer())) - print(config.as_bos_grpc()) - - bos_parsed = MinerConfig.from_bosminer(config.as_bosminer()) - print(bos_parsed) - print(bos_parsed == config) + # print("WM:", config.as_wm()) + # print("AM Modern:", config.as_am_modern()) + # print("AM Old:", config.as_am_old()) + # print("GS:", config.as_goldshell()) + # print("Avalon:", config.as_avalon()) + # print("Inno:", config.as_inno()) + # print("BOS+ .toml:", config.as_bosminer()) + # print("BOS+ .toml as toml:") + # print(toml.dumps(config.as_bosminer())) + # print(config.as_bos_grpc()) + dict_config = config.as_dict() + parsed_conf = MinerConfig.from_dict(dict_config) + print(parsed_conf) + # bos_parsed = MinerConfig.from_bosminer(config.as_bosminer()) + # print(bos_parsed) + # print(bos_parsed == config) diff --git a/pyasic/config/base.py b/pyasic/config/base.py index e43b851f..b082228f 100644 --- a/pyasic/config/base.py +++ b/pyasic/config/base.py @@ -13,10 +13,17 @@ # See the License for the specific language governing permissions and - # limitations under the License. - # ------------------------------------------------------------------------------ +from abc import ABC, abstractmethod +from dataclasses import asdict, dataclass from enum import Enum +from typing import Union class MinerConfigOption(Enum): + @classmethod + def from_dict(cls, dict_conf: Union[dict, None]): + return cls.default() + def as_am_modern(self) -> dict: return self.value.as_am_modern() @@ -44,8 +51,20 @@ class MinerConfigOption(Enum): def __call__(self, *args, **kwargs): return self.value(*args, **kwargs) + @classmethod + def default(cls): + pass + +@dataclass 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: return {} diff --git a/pyasic/config/fans.py b/pyasic/config/fans.py index 1e628f01..ca91972b 100644 --- a/pyasic/config/fans.py +++ b/pyasic/config/fans.py @@ -14,6 +14,7 @@ # limitations under the License. - # ------------------------------------------------------------------------------ from dataclasses import dataclass, field +from typing import Union from pyasic.config.base import MinerConfigOption, MinerConfigValue @@ -22,10 +23,14 @@ from pyasic.config.base import MinerConfigOption, MinerConfigValue class FanModeNormal(MinerConfigValue): 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"} - def as_bosminer(self): + def as_bosminer(self) -> dict: return {"temp_control": {"mode": "auto"}} @@ -36,7 +41,16 @@ class FanModeManual(MinerConfigValue): speed: int = 100 @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 = {} if toml_fan_conf.get("min_fans") is not None: cls_conf["minimum_fans"] = toml_fan_conf["min_fans"] @@ -44,12 +58,10 @@ class FanModeManual(MinerConfigValue): cls_conf["speed"] = toml_fan_conf["speed"] return cls(**cls_conf) - - - def as_am_modern(self): + def as_am_modern(self) -> dict: return {"bitmain-fan-ctrl": True, "bitmain-fan-pwn": str(self.speed)} - def as_bosminer(self): + def as_bosminer(self) -> dict: return { "temp_control": {"mode": "manual"}, "fan_control": {"min_fans": self.minimum_fans, "speed": self.speed}, @@ -60,10 +72,14 @@ class FanModeManual(MinerConfigValue): class FanModeImmersion(MinerConfigValue): 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"} - def as_bosminer(self): + def as_bosminer(self) -> dict: return {"temp_control": {"mode": "disabled"}} @@ -76,6 +92,19 @@ class FanModeConfig(MinerConfigOption): def default(cls): 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 def from_am_modern(cls, web_conf: dict): if web_conf.get("bitmain-fan-ctrl") is not None: diff --git a/pyasic/config/mining.py b/pyasic/config/mining.py index dcbc3bb6..16525768 100644 --- a/pyasic/config/mining.py +++ b/pyasic/config/mining.py @@ -14,6 +14,7 @@ # limitations under the License. - # ------------------------------------------------------------------------------ from dataclasses import dataclass, field +from typing import Union from pyasic.config.base import MinerConfigOption, MinerConfigValue @@ -22,10 +23,14 @@ from pyasic.config.base import MinerConfigOption, MinerConfigValue class MiningModeNormal(MinerConfigValue): 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"} - def as_wm(self): + def as_wm(self) -> dict: return {"mode": self.mode} @@ -33,10 +38,14 @@ class MiningModeNormal(MinerConfigValue): class MiningModeSleep(MinerConfigValue): 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"} - def as_wm(self): + def as_wm(self) -> dict: return {"mode": self.mode} @@ -44,10 +53,14 @@ class MiningModeSleep(MinerConfigValue): class MiningModeLPM(MinerConfigValue): 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"} - def as_wm(self): + def as_wm(self) -> dict: return {"mode": self.mode} @@ -55,10 +68,14 @@ class MiningModeLPM(MinerConfigValue): class MiningModeHPM(MinerConfigValue): 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): return {"miner-mode": "0"} - def as_wm(self): + def as_wm(self) -> dict: return {"mode": self.mode} @@ -67,10 +84,14 @@ class MiningModePowerTune(MinerConfigValue): mode: str = field(init=False, default="power_tuning") 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"} - def as_wm(self): + def as_wm(self) -> dict: if self.power is not None: return {"mode": self.mode, self.mode: {"wattage": self.power}} return {} @@ -84,7 +105,11 @@ class MiningModeHashrateTune(MinerConfigValue): mode: str = field(init=False, default="hashrate_tuning") 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"} @@ -93,7 +118,11 @@ class ManualBoardSettings(MinerConfigValue): freq: 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"} @@ -105,7 +134,15 @@ class MiningModeManual(MinerConfigValue): global_volt: float 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"} @@ -122,6 +159,19 @@ class MiningModeConfig(MinerConfigOption): def default(cls): 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 def from_am_modern(cls, web_conf: dict): 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: return cls.hashrate_tuning(autotuning_conf["hashrate_target"]) return cls.hashrate_tuning() - diff --git a/pyasic/config/pools.py b/pyasic/config/pools.py index d9ccc826..d2835a91 100644 --- a/pyasic/config/pools.py +++ b/pyasic/config/pools.py @@ -94,36 +94,42 @@ class Pool(MinerConfigValue): return { "url": self.url, "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 - 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") @classmethod - def from_am_modern(cls, web_pool: dict): + def from_am_modern(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 @classmethod - def from_goldshell(cls, web_pool: dict): + def from_goldshell(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 @classmethod - def from_inno(cls, web_pool: dict): + def from_inno(cls, web_pool: dict) -> "Pool": return cls( url=web_pool["url"], user=web_pool["user"], password=web_pool["pass"] ) @classmethod - def from_bosminer(cls, toml_pool_conf: dict): + def from_bosminer(cls, toml_pool_conf: dict) -> "Pool": return cls( url=toml_pool_conf["url"], user=toml_pool_conf["user"], @@ -211,32 +217,44 @@ class PoolGroup(MinerConfigValue): } if self.quota is not None: conf["quota"] = self.quota + return conf return {"name": "Group", "pool": []} @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 = [] for pool in api_pool_list: pools.append(Pool.from_api(pool)) return cls(pools=pools) @classmethod - def from_am_modern(cls, web_pool_list: list): + def from_am_modern(cls, web_pool_list: list) -> "PoolGroup": pools = [] for pool in web_pool_list: pools.append(Pool.from_am_modern(pool)) return cls(pools=pools) @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]) @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]) @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: return cls( name=toml_group_conf["name"], @@ -251,11 +269,18 @@ class PoolConfig(MinerConfigValue): groups: list[PoolGroup] = field(default_factory=list) @classmethod - def default(cls): + def default(cls) -> "PoolConfig": return cls() @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 = [] for pool in pools: if isinstance(pool, dict): @@ -304,28 +329,28 @@ class PoolConfig(MinerConfigValue): return {} @classmethod - def from_api(cls, api_pools: dict): + def from_api(cls, api_pools: dict) -> "PoolConfig": pool_data = api_pools["POOLS"] pool_data = sorted(pool_data, key=lambda x: int(x["POOL"])) return cls([PoolGroup.from_api(pool_data)]) @classmethod - def from_am_modern(cls, web_conf: dict): + def from_am_modern(cls, web_conf: dict) -> "PoolConfig": pool_data = web_conf["pools"] return cls([PoolGroup.from_am_modern(pool_data)]) @classmethod - def from_goldshell(cls, web_pools: list): + def from_goldshell(cls, web_pools: list) -> "PoolConfig": return cls([PoolGroup.from_goldshell(web_pools)]) @classmethod - def from_inno(cls, web_pools: list): + def from_inno(cls, web_pools: list) -> "PoolConfig": return cls([PoolGroup.from_inno(web_pools)]) @classmethod - def from_bosminer(cls, toml_conf: dict): + def from_bosminer(cls, toml_conf: dict) -> "PoolConfig": if toml_conf.get("group") is None: return cls() diff --git a/pyasic/config/power_scaling.py b/pyasic/config/power_scaling.py index 0ba31122..e776b49c 100644 --- a/pyasic/config/power_scaling.py +++ b/pyasic/config/power_scaling.py @@ -14,6 +14,7 @@ # limitations under the License. - # ------------------------------------------------------------------------------ from dataclasses import dataclass, field +from typing import Union from pyasic.config.base import MinerConfigOption, MinerConfigValue 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") 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: cfg = {"shutdown_enabled": True} @@ -45,6 +50,10 @@ class PowerScalingShutdownEnabled(MinerConfigValue): class PowerScalingShutdownDisabled(MinerConfigValue): 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: return {"shutdown_enabled": False} @@ -56,6 +65,19 @@ class PowerScalingShutdown(MinerConfigOption): enabled = PowerScalingShutdownEnabled 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 def from_bosminer(cls, power_scaling_conf: dict): sd_enabled = power_scaling_conf.get("shutdown_enabled") @@ -75,7 +97,7 @@ class PowerScalingEnabled(MinerConfigValue): shutdown_enabled: PowerScalingShutdown = None @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") min_power = power_scaling_conf.get("min_psu_power_limit") 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 ) + @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: cfg = {"enabled": True} if self.power_step is not None: @@ -96,7 +131,7 @@ class PowerScalingEnabled(MinerConfigValue): return {"power_scaling": cfg} - def as_bos_grpc(self): + def as_bos_grpc(self) -> dict: cfg = {"enable": True} target_conf = {} if self.power_step is not None: @@ -125,6 +160,19 @@ class PowerScalingConfig(MinerConfigOption): def default(cls): 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 def from_bosminer(cls, toml_conf: dict): power_scaling = toml_conf.get("power_scaling") diff --git a/pyasic/config/temperature.py b/pyasic/config/temperature.py index f353e507..d875104b 100644 --- a/pyasic/config/temperature.py +++ b/pyasic/config/temperature.py @@ -14,6 +14,7 @@ # limitations under the License. - # ------------------------------------------------------------------------------ from dataclasses import dataclass +from typing import Union from pyasic.config.base import MinerConfigValue @@ -39,7 +40,15 @@ class TemperatureConfig(MinerConfigValue): return {"temp_control": temp_cfg} @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") if temp_control is not None: return cls(