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

Merge pull request #160 from UpstreamData/dev

Browse Source 
Improved config handling.
This commit is contained in:
Brett Rowan
2024-06-11 08:46:45 -06:00
committed by GitHub
parent 5a4b1b6ee1 927bbae0c1
commit 3e4b347506
8 changed files with 352 additions and 362 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
pyasic/config/__init__.py
View File

@@ -17,8 +17,8 @@ from dataclasses import asdict, dataclass, field
from pyasic.config.fans import FanModeConfig
from pyasic.config.mining import MiningModeConfig
from pyasic.config.mining.scaling import ScalingConfig
from pyasic.config.pools import PoolConfig
from pyasic.config.scaling import ScalingConfig
from pyasic.config.temperature import TemperatureConfig
from pyasic.misc import merge_dicts
@@ -32,7 +32,6 @@ class MinerConfig:
fan_mode: FanModeConfig = field(default_factory=FanModeConfig.default)
temperature: TemperatureConfig = field(default_factory=TemperatureConfig.default)
mining_mode: MiningModeConfig = field(default_factory=MiningModeConfig.default)
scaling: ScalingConfig = field(default_factory=ScalingConfig.default)
def __getitem__(self, item):
try:
@@ -52,7 +51,6 @@ class MinerConfig:
**self.mining_mode.as_am_modern(),
**self.pools.as_am_modern(user_suffix=user_suffix),
**self.temperature.as_am_modern(),
**self.scaling.as_am_modern(),
}
def as_wm(self, user_suffix: str = None) -> dict:
@@ -62,7 +60,6 @@ class MinerConfig:
**self.mining_mode.as_wm(),
**self.pools.as_wm(user_suffix=user_suffix),
**self.temperature.as_wm(),
**self.scaling.as_wm(),
}
def as_am_old(self, user_suffix: str = None) -> dict:
@@ -72,7 +69,6 @@ class MinerConfig:
**self.mining_mode.as_am_old(),
**self.pools.as_am_old(user_suffix=user_suffix),
**self.temperature.as_am_old(),
**self.scaling.as_am_old(),
}
def as_goldshell(self, user_suffix: str = None) -> dict:
@@ -82,7 +78,6 @@ class MinerConfig:
**self.mining_mode.as_goldshell(),
**self.pools.as_goldshell(user_suffix=user_suffix),
**self.temperature.as_goldshell(),
**self.scaling.as_goldshell(),
}
def as_avalon(self, user_suffix: str = None) -> dict:
@@ -92,7 +87,6 @@ class MinerConfig:
**self.mining_mode.as_avalon(),
**self.pools.as_avalon(user_suffix=user_suffix),
**self.temperature.as_avalon(),
**self.scaling.as_avalon(),
}
def as_inno(self, user_suffix: str = None) -> dict:
@@ -102,7 +96,6 @@ class MinerConfig:
**self.mining_mode.as_inno(),
**self.pools.as_inno(user_suffix=user_suffix),
**self.temperature.as_inno(),
**self.scaling.as_inno(),
}
def as_bosminer(self, user_suffix: str = None) -> dict:
@@ -111,7 +104,6 @@ class MinerConfig:
**merge_dicts(self.fan_mode.as_bosminer(), self.temperature.as_bosminer()),
**self.mining_mode.as_bosminer(),
**self.pools.as_bosminer(user_suffix=user_suffix),
**self.scaling.as_bosminer(),
}
def as_boser(self, user_suffix: str = None) -> dict:
@@ -121,7 +113,6 @@ class MinerConfig:
**self.temperature.as_boser(),
**self.mining_mode.as_boser(),
**self.pools.as_boser(user_suffix=user_suffix),
**self.scaling.as_boser(),
}
def as_epic(self, user_suffix: str = None) -> dict:
@@ -130,7 +121,6 @@ class MinerConfig:
**merge_dicts(self.fan_mode.as_epic(), self.temperature.as_epic()),
**self.mining_mode.as_epic(),
**self.pools.as_epic(user_suffix=user_suffix),
**self.scaling.as_epic(),
}
def as_auradine(self, user_suffix: str = None) -> dict:
@@ -140,7 +130,6 @@ class MinerConfig:
**self.temperature.as_auradine(),
**self.mining_mode.as_auradine(),
**self.pools.as_auradine(user_suffix=user_suffix),
**self.scaling.as_auradine(),
}
def as_mara(self, user_suffix: str = None) -> dict:
@@ -149,7 +138,6 @@ class MinerConfig:
**self.temperature.as_mara(),
**self.mining_mode.as_mara(),
**self.pools.as_mara(user_suffix=user_suffix),
**self.scaling.as_mara(),
}
@classmethod
@@ -160,7 +148,6 @@ class MinerConfig:
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")),
scaling=ScalingConfig.from_dict(dict_conf.get("scaling")),
)
@classmethod
@@ -200,7 +187,6 @@ class MinerConfig:
mining_mode=MiningModeConfig.from_bosminer(toml_conf),
fan_mode=FanModeConfig.from_bosminer(toml_conf),
temperature=TemperatureConfig.from_bosminer(toml_conf),
scaling=ScalingConfig.from_bosminer(toml_conf),
)
@classmethod
@@ -211,7 +197,6 @@ class MinerConfig:
mining_mode=MiningModeConfig.from_boser(grpc_miner_conf),
fan_mode=FanModeConfig.from_boser(grpc_miner_conf),
temperature=TemperatureConfig.from_boser(grpc_miner_conf),
scaling=ScalingConfig.from_boser(grpc_miner_conf),
)
@classmethod

3
pyasic/config/base.py
View File

@@ -46,7 +46,7 @@ class MinerConfigOption(Enum):
return self.value.as_bosminer()
def as_boser(self) -> dict:
return self.value.as_boser()
return self.value.as_boser
def as_epic(self) -> dict:
return self.value.as_epic()
@@ -74,7 +74,6 @@ class MinerConfigOption(Enum):
raise KeyError
@dataclass
class MinerConfigValue:
@classmethod

229
pyasic/config/mining.py → pyasic/config/mining/__init__.py
View File

@@ -20,16 +20,23 @@ from dataclasses import dataclass, field
from pyasic import settings
from pyasic.config.base import MinerConfigOption, MinerConfigValue
from pyasic.web.braiins_os.proto.braiins.bos.v1 import (
DpsHashrateTarget,
DpsPowerTarget,
DpsTarget,
HashrateTargetMode,
PerformanceMode,
Power,
PowerTargetMode,
SaveAction,
SetDpsRequest,
SetPerformanceModeRequest,
TeraHashrate,
TunerPerformanceMode,
)
from .algo import TunerAlgo
from .scaling import ScalingConfig
@dataclass
class MiningModeNormal(MinerConfigValue):
@@ -140,56 +147,12 @@ class MiningModeHPM(MinerConfigValue):
return {"mode": {"mode": "turbo"}}
@dataclass
class StandardTuneAlgo(MinerConfigValue):
mode: str = field(init=False, default="standard")
def as_epic(self) -> str:
return VOptAlgo().as_epic()
@dataclass
class VOptAlgo(MinerConfigValue):
mode: str = field(init=False, default="voltage_optimizer")
def as_epic(self) -> str:
return "VoltageOptimizer"
@dataclass
class ChipTuneAlgo(MinerConfigValue):
mode: str = field(init=False, default="chip_tune")
def as_epic(self) -> str:
return "ChipTune"
@dataclass
class TunerAlgo(MinerConfigOption):
standard = StandardTuneAlgo
voltage_optimizer = VOptAlgo
chip_tune = ChipTuneAlgo
@classmethod
def default(cls):
return cls.standard()
@classmethod
def from_dict(cls, dict_conf: dict | None):
mode = dict_conf.get("mode")
if mode is None:
return cls.default()
cls_attr = getattr(cls, mode)
if cls_attr is not None:
return cls_attr().from_dict(dict_conf)
@dataclass
class MiningModePowerTune(MinerConfigValue):
mode: str = field(init=False, default="power_tuning")
power: int = None
algo: TunerAlgo = field(default_factory=TunerAlgo.default)
scaling: ScalingConfig = None
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "MiningModePowerTune":
@@ -198,6 +161,8 @@ class MiningModePowerTune(MinerConfigValue):
cls_conf["power"] = dict_conf["power"]
if dict_conf.get("algo"):
cls_conf["algo"] = TunerAlgo.from_dict(dict_conf["algo"])
if dict_conf.get("scaling"):
cls_conf["scaling"] = ScalingConfig.from_dict(dict_conf["scaling"])
return cls(**cls_conf)
@@ -212,13 +177,26 @@ class MiningModePowerTune(MinerConfigValue):
return {}
def as_bosminer(self) -> dict:
conf = {"enabled": True, "mode": "power_target"}
tuning_cfg = {"enabled": True, "mode": "power_target"}
if self.power is not None:
conf["power_target"] = self.power
return {"autotuning": conf}
tuning_cfg["power_target"] = self.power
cfg = {"autotuning": tuning_cfg}
if self.scaling is not None:
scaling_cfg = {"enabled": True}
if self.scaling.step is not None:
scaling_cfg["power_step"] = self.scaling.step
if self.scaling.minimum is not None:
scaling_cfg["min_power_target"] = self.scaling.minimum
if self.scaling.shutdown is not None:
scaling_cfg = {**scaling_cfg, **self.scaling.shutdown.as_bosminer()}
cfg["performance_scaling"] = scaling_cfg
return cfg
def as_boser(self) -> dict:
return {
cfg = {
"set_performance_mode": SetPerformanceModeRequest(
save_action=SaveAction.SAVE_ACTION_SAVE_AND_APPLY,
mode=PerformanceMode(
@@ -230,6 +208,23 @@ class MiningModePowerTune(MinerConfigValue):
),
),
}
if self.scaling is not None:
sd_cfg = {}
if self.scaling.shutdown is not None:
sd_cfg = self.scaling.shutdown.as_boser()
cfg["set_dps"] = (
SetDpsRequest(
enable=True,
**sd_cfg,
target=DpsTarget(
power_target=DpsPowerTarget(
power_step=Power(self.scaling.step),
min_power_target=Power(self.scaling.minimum),
)
),
),
)
return cfg
def as_auradine(self) -> dict:
return {"mode": {"mode": "custom", "tune": "power", "power": self.power}}
@@ -250,21 +245,18 @@ class MiningModePowerTune(MinerConfigValue):
class MiningModeHashrateTune(MinerConfigValue):
mode: str = field(init=False, default="hashrate_tuning")
hashrate: int = None
throttle_limit: int = None
throttle_step: int = None
algo: TunerAlgo = field(default_factory=TunerAlgo.default)
scaling: ScalingConfig = None
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "MiningModeHashrateTune":
cls_conf = {}
if dict_conf.get("hashrate"):
cls_conf["hashrate"] = dict_conf["hashrate"]
if dict_conf.get("throttle_limit"):
cls_conf["throttle_limit"] = dict_conf["throttle_limit"]
if dict_conf.get("throttle_step"):
cls_conf["throttle_step"] = dict_conf["throttle_step"]
if dict_conf.get("algo"):
cls_conf["algo"] = TunerAlgo.from_dict(dict_conf["algo"])
if dict_conf.get("scaling"):
cls_conf["scaling"] = ScalingConfig.from_dict(dict_conf["scaling"])
return cls(**cls_conf)
@@ -279,8 +271,9 @@ class MiningModeHashrateTune(MinerConfigValue):
conf["hashrate_target"] = self.hashrate
return {"autotuning": conf}
@property
def as_boser(self) -> dict:
return {
cfg = {
"set_performance_mode": SetPerformanceModeRequest(
save_action=SaveAction.SAVE_ACTION_SAVE_AND_APPLY,
mode=PerformanceMode(
@@ -294,6 +287,23 @@ class MiningModeHashrateTune(MinerConfigValue):
),
)
}
if self.scaling is not None:
sd_cfg = {}
if self.scaling.shutdown is not None:
sd_cfg = self.scaling.shutdown.as_boser()
cfg["set_dps"] = (
SetDpsRequest(
enable=True,
**sd_cfg,
target=DpsTarget(
hashrate_target=DpsHashrateTarget(
hashrate_step=TeraHashrate(self.scaling.step),
min_hashrate_target=TeraHashrate(self.scaling.minimum),
)
),
),
)
return cfg
def as_auradine(self) -> dict:
return {"mode": {"mode": "custom", "tune": "ths", "ths": self.hashrate}}
@@ -305,10 +315,11 @@ class MiningModeHashrateTune(MinerConfigValue):
"target": self.hashrate,
}
}
if self.throttle_limit is not None:
mode["ptune"]["min_throttle"] = self.throttle_limit
if self.throttle_step is not None:
mode["ptune"]["throttle_step"] = self.throttle_step
if self.scaling is not None:
if self.scaling.minimum is not None:
mode["ptune"]["min_throttle"] = self.scaling.minimum
if self.scaling.step is not None:
mode["ptune"]["throttle_step"] = self.scaling.step
return mode
def as_mara(self) -> dict:
@@ -373,6 +384,24 @@ class MiningModeManual(MinerConfigValue):
}
return cls(global_freq=freq, global_volt=voltage, boards=boards)
@classmethod
def from_epic(cls, epic_conf: dict) -> "MiningModeManual":
voltage = 0
freq = 0
if epic_conf.get("HwConfig") is not None:
freq = epic_conf["HwConfig"]["Boards Target Clock"][0]["Data"]
if epic_conf.get("Power Supply Stats") is not None:
voltage = epic_conf["Power Supply Stats"]["Target Voltage"]
boards = {}
if epic_conf.get("HBs") is not None:
boards = {
board["Index"]: ManualBoardSettings(
freq=board["Core Clock Avg"], volt=board["Input Voltage"]
)
for board in epic_conf["HBs"]
}
return cls(global_freq=freq, global_volt=voltage, boards=boards)
def as_mara(self) -> dict:
return {
"mode": {
@@ -432,15 +461,32 @@ class MiningModeConfig(MinerConfigOption):
if tuner_running:
algo_info = web_conf["PerpetualTune"]["Algorithm"]
if algo_info.get("VoltageOptimizer") is not None:
scaling_cfg = None
if "Throttle Step" in algo_info["VoltageOptimizer"]:
scaling_cfg = ScalingConfig(
minimum=algo_info["VoltageOptimizer"].get(
"Min Throttle Target"
),
step=algo_info["VoltageOptimizer"].get("Throttle Step"),
)
return cls.hashrate_tuning(
hashrate=algo_info["VoltageOptimizer"].get("Target"),
throttle_limit=algo_info["VoltageOptimizer"].get(
"Min Throttle Target"
),
throttle_step=algo_info["VoltageOptimizer"].get(
"Throttle Step"
),
algo=TunerAlgo.voltage_optimizer(),
scaling=scaling_cfg,
)
elif algo_info.get("BoardTune") is not None:
scaling_cfg = None
if "Throttle Step" in algo_info["BoardTune"]:
scaling_cfg = ScalingConfig(
minimum=algo_info["BoardTune"].get("Min Throttle Target"),
step=algo_info["BoardTune"].get("Throttle Step"),
)
return cls.hashrate_tuning(
hashrate=algo_info["BoardTune"].get("Target"),
algo=TunerAlgo.voltage_optimizer(),
scaling=scaling_cfg,
)
else:
return cls.hashrate_tuning(
@@ -448,7 +494,7 @@ class MiningModeConfig(MinerConfigOption):
algo=TunerAlgo.chip_tune(),
)
else:
return cls.normal()
return MiningModeManual.from_epic(web_conf)
except KeyError:
return cls.default()
@@ -465,18 +511,31 @@ class MiningModeConfig(MinerConfigOption):
if autotuning_conf.get("psu_power_limit") is not None:
# old autotuning conf
return cls.power_tuning(autotuning_conf["psu_power_limit"])
return cls.power_tuning(
autotuning_conf["psu_power_limit"],
scaling=ScalingConfig.from_bosminer(toml_conf, mode="power"),
)
if autotuning_conf.get("mode") is not None:
# new autotuning conf
mode = autotuning_conf["mode"]
if mode == "power_target":
if autotuning_conf.get("power_target") is not None:
return cls.power_tuning(autotuning_conf["power_target"])
return cls.power_tuning()
return cls.power_tuning(
autotuning_conf["power_target"],
scaling=ScalingConfig.from_bosminer(toml_conf, mode="power"),
)
return cls.power_tuning(
scaling=ScalingConfig.from_bosminer(toml_conf, mode="power"),
)
if mode == "hashrate_target":
if autotuning_conf.get("hashrate_target") is not None:
return cls.hashrate_tuning(autotuning_conf["hashrate_target"])
return cls.hashrate_tuning()
return cls.hashrate_tuning(
autotuning_conf["hashrate_target"],
scaling=ScalingConfig.from_bosminer(toml_conf, mode="hashrate"),
)
return cls.hashrate_tuning(
scaling=ScalingConfig.from_bosminer(toml_conf, mode="hashrate"),
)
@classmethod
def from_vnish(cls, web_settings: dict):
@@ -502,22 +561,36 @@ class MiningModeConfig(MinerConfigOption):
if tuner_conf.get("tunerMode") is not None:
if tuner_conf["tunerMode"] == 1:
if tuner_conf.get("powerTarget") is not None:
return cls.power_tuning(tuner_conf["powerTarget"]["watt"])
return cls.power_tuning()
return cls.power_tuning(
tuner_conf["powerTarget"]["watt"],
scaling=ScalingConfig.from_boser(grpc_miner_conf, mode="power"),
)
return cls.power_tuning(
scaling=ScalingConfig.from_boser(grpc_miner_conf, mode="power")
)
if tuner_conf["tunerMode"] == 2:
if tuner_conf.get("hashrateTarget") is not None:
return cls.hashrate_tuning(
int(tuner_conf["hashrateTarget"]["terahashPerSecond"])
int(tuner_conf["hashrateTarget"]["terahashPerSecond"]),
scaling=ScalingConfig.from_boser(
grpc_miner_conf, mode="hashrate"
),
)
return cls.hashrate_tuning()
return cls.hashrate_tuning(
scaling=ScalingConfig.from_boser(grpc_miner_conf, mode="hashrate"),
)
if tuner_conf.get("powerTarget") is not None:
return cls.power_tuning(tuner_conf["powerTarget"]["watt"])
return cls.power_tuning(
tuner_conf["powerTarget"]["watt"],
scaling=ScalingConfig.from_boser(grpc_miner_conf, mode="power"),
)
if tuner_conf.get("hashrateTarget") is not None:
return cls.hashrate_tuning(
int(tuner_conf["hashrateTarget"]["terahashPerSecond"])
int(tuner_conf["hashrateTarget"]["terahashPerSecond"]),
scaling=ScalingConfig.from_boser(grpc_miner_conf, mode="hashrate"),
)
@classmethod

57
pyasic/config/mining/algo.py Normal file
View File

@@ -0,0 +1,57 @@
from dataclasses import dataclass, field
from pyasic.config.base import MinerConfigOption, MinerConfigValue
@dataclass
class StandardTuneAlgo(MinerConfigValue):
mode: str = field(init=False, default="standard")
def as_epic(self) -> str:
return VOptAlgo().as_epic()
@dataclass
class VOptAlgo(MinerConfigValue):
mode: str = field(init=False, default="voltage_optimizer")
def as_epic(self) -> str:
return "VoltageOptimizer"
@dataclass
class BoardTuneAlgo(MinerConfigValue):
mode: str = field(init=False, default="board_tune")
def as_epic(self) -> str:
return "BoardTune"
@dataclass
class ChipTuneAlgo(MinerConfigValue):
mode: str = field(init=False, default="chip_tune")
def as_epic(self) -> str:
return "ChipTune"
@dataclass
class TunerAlgo(MinerConfigOption):
standard = StandardTuneAlgo
voltage_optimizer = VOptAlgo
board_tune = BoardTuneAlgo
chip_tune = ChipTuneAlgo
@classmethod
def default(cls):
return cls.standard()
@classmethod
def from_dict(cls, dict_conf: dict | None):
mode = dict_conf.get("mode")
if mode is None:
return cls.default()
cls_attr = getattr(cls, mode)
if cls_attr is not None:
return cls_attr().from_dict(dict_conf)

128
pyasic/config/mining/scaling.py Normal file
View File

@@ -0,0 +1,128 @@
# ------------------------------------------------------------------------------
# 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
from dataclasses import dataclass
from pyasic.config.base import MinerConfigValue
@dataclass
class ScalingShutdown(MinerConfigValue):
enabled: bool = False
duration: int = None
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "ScalingShutdown":
return cls(
enabled=dict_conf.get("enabled", False), duration=dict_conf.get("duration")
)
@classmethod
def from_bosminer(cls, power_scaling_conf: dict):
sd_enabled = power_scaling_conf.get("shutdown_enabled")
if sd_enabled is not None:
return cls(sd_enabled, power_scaling_conf.get("shutdown_duration"))
return None
@classmethod
def from_boser(cls, power_scaling_conf: dict):
sd_enabled = power_scaling_conf.get("shutdownEnabled")
if sd_enabled is not None:
try:
return cls(sd_enabled, power_scaling_conf["shutdownDuration"]["hours"])
except KeyError:
return cls(sd_enabled)
return None
def as_bosminer(self) -> dict:
cfg = {"shutdown_enabled": self.enabled}
if self.duration is not None:
cfg["shutdown_duration"] = self.duration
return cfg
def as_boser(self) -> dict:
return {"enable_shutdown": self.enabled, "shutdown_duration": self.duration}
@dataclass
class ScalingConfig(MinerConfigValue):
step: int = None
minimum: int = None
shutdown: ScalingShutdown = None
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "ScalingConfig":
cls_conf = {
"step": dict_conf.get("step"),
"minimum": dict_conf.get("minimum"),
}
shutdown = dict_conf.get("shutdown")
if shutdown is not None:
cls_conf["shutdown"] = ScalingShutdown.from_dict(shutdown)
return cls(**cls_conf)
@classmethod
def from_bosminer(cls, toml_conf: dict, mode: str = None):
if mode == "power":
return cls._from_bosminer_power(toml_conf)
if mode == "hashrate":
# not implemented yet
pass
@classmethod
def _from_bosminer_power(cls, toml_conf: dict):
power_scaling = toml_conf.get("power_scaling")
if power_scaling is None:
power_scaling = toml_conf.get("performance_scaling")
if power_scaling is not None:
enabled = power_scaling.get("enabled")
if not enabled:
return None
power_step = power_scaling.get("power_step")
min_power = power_scaling.get("min_psu_power_limit")
if min_power is None:
min_power = power_scaling.get("min_power_target")
sd_mode = ScalingShutdown.from_bosminer(power_scaling)
return cls(step=power_step, minimum=min_power, shutdown=sd_mode)
@classmethod
def from_boser(cls, grpc_miner_conf: dict, mode: str = None):
if mode == "power":
return cls._from_boser_power(grpc_miner_conf)
if mode == "hashrate":
# not implemented yet
pass
@classmethod
def _from_boser_power(cls, grpc_miner_conf: dict):
try:
dps_conf = grpc_miner_conf["dps"]
if not dps_conf.get("enabled", False):
return None
except LookupError:
return None
conf = {"shutdown": ScalingShutdown.from_boser(dps_conf)}
if dps_conf.get("minPowerTarget") is not None:
conf["minimum"] = dps_conf["minPowerTarget"]["watt"]
if dps_conf.get("powerStep") is not None:
conf["step"] = dps_conf["powerStep"]["watt"]
return cls(**conf)

253
pyasic/config/scaling.py
View File

@@ -1,253 +0,0 @@
# ------------------------------------------------------------------------------
# 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
from dataclasses import dataclass, field
from pyasic.config.base import MinerConfigOption, MinerConfigValue
from pyasic.web.braiins_os.proto.braiins.bos.v1 import (
DpsHashrateTarget,
DpsPowerTarget,
DpsTarget,
Power,
SetDpsRequest,
TeraHashrate,
)
@dataclass
class ScalingShutdownEnabled(MinerConfigValue):
mode: str = field(init=False, default="enabled")
duration: int = None
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "ScalingShutdownEnabled":
return cls(duration=dict_conf.get("duration"))
def as_bosminer(self) -> dict:
cfg = {"shutdown_enabled": True}
if self.duration is not None:
cfg["shutdown_duration"] = self.duration
return cfg
def as_boser(self) -> dict:
return {"enable_shutdown": True, "shutdown_duration": self.duration}
@dataclass
class ScalingShutdownDisabled(MinerConfigValue):
mode: str = field(init=False, default="disabled")
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "ScalingShutdownDisabled":
return cls()
def as_bosminer(self) -> dict:
return {"shutdown_enabled": False}
def as_boser(self) -> dict:
return {"enable_shutdown ": False}
class ScalingShutdown(MinerConfigOption):
enabled = ScalingShutdownEnabled
disabled = ScalingShutdownDisabled
@classmethod
def from_dict(cls, dict_conf: 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")
if sd_enabled is not None:
if sd_enabled:
return cls.enabled(power_scaling_conf.get("shutdown_duration"))
else:
return cls.disabled()
return None
@classmethod
def from_boser(cls, power_scaling_conf: dict):
sd_enabled = power_scaling_conf.get("shutdownEnabled")
if sd_enabled is not None:
if sd_enabled:
try:
return cls.enabled(power_scaling_conf["shutdownDuration"]["hours"])
except KeyError:
return cls.enabled()
else:
return cls.disabled()
return None
@dataclass
class PowerScaling(MinerConfigValue):
mode: str = field(init=False, default="power")
step: int = None
minimum: int = None
shutdown: ScalingShutdownEnabled | ScalingShutdownDisabled = None
@classmethod
def from_bosminer(cls, power_scaling_conf: dict) -> "PowerScaling":
power_step = power_scaling_conf.get("power_step")
min_power = power_scaling_conf.get("min_psu_power_limit")
if min_power is None:
min_power = power_scaling_conf.get("min_power_target")
sd_mode = ScalingShutdown.from_bosminer(power_scaling_conf)
return cls(step=power_step, minimum=min_power, shutdown=sd_mode)
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "PowerScaling":
cls_conf = {
"step": dict_conf.get("step"),
"minimum": dict_conf.get("minimum"),
}
shutdown = dict_conf.get("shutdown")
if shutdown is not None:
cls_conf["shutdown"] = ScalingShutdown.from_dict(shutdown)
return cls(**cls_conf)
def as_bosminer(self) -> dict:
cfg = {"enabled": True}
if self.step is not None:
cfg["power_step"] = self.step
if self.minimum is not None:
cfg["min_power_target"] = self.minimum
if self.shutdown is not None:
cfg = {**cfg, **self.shutdown.as_bosminer()}
return {"performance_scaling": cfg}
def as_boser(self) -> dict:
return {
"set_dps": SetDpsRequest(
enable=True,
**self.shutdown.as_boser(),
target=DpsTarget(
power_target=DpsPowerTarget(
power_step=Power(self.step),
min_power_target=Power(self.minimum),
)
),
),
}
@dataclass
class HashrateScaling(MinerConfigValue):
mode: str = field(init=False, default="hashrate")
step: int = None
minimum: int = None
shutdown: ScalingShutdownEnabled | ScalingShutdownDisabled = None
@classmethod
def from_dict(cls, dict_conf: dict | None) -> "HashrateScaling":
cls_conf = {
"step": dict_conf.get("step"),
"minimum": dict_conf.get("minimum"),
}
shutdown = dict_conf.get("shutdown")
if shutdown is not None:
cls_conf["shutdown"] = ScalingShutdown.from_dict(shutdown)
return cls(**cls_conf)
def as_boser(self) -> dict:
return {
"set_dps": SetDpsRequest(
enable=True,
**self.shutdown.as_boser(),
target=DpsTarget(
hashrate_target=DpsHashrateTarget(
hashrate_step=TeraHashrate(self.step),
min_hashrate_target=TeraHashrate(self.minimum),
)
),
),
}
@dataclass
class ScalingDisabled(MinerConfigValue):
mode: str = field(init=False, default="disabled")
class ScalingConfig(MinerConfigOption):
power = PowerScaling
hashrate = HashrateScaling
disabled = ScalingDisabled
@classmethod
def default(cls):
return cls.disabled()
@classmethod
def from_dict(cls, dict_conf: 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")
if power_scaling is None:
power_scaling = toml_conf.get("performance_scaling")
if power_scaling is not None:
enabled = power_scaling.get("enabled")
if enabled is not None:
if enabled:
return cls.power().from_bosminer(power_scaling)
else:
return cls.disabled()
return cls.default()
@classmethod
def from_boser(cls, grpc_miner_conf: dict):
try:
dps_conf = grpc_miner_conf["dps"]
if not dps_conf.get("enabled", False):
return cls.disabled()
except LookupError:
return cls.default()
conf = {"shutdown_enabled": ScalingShutdown.from_boser(dps_conf)}
if dps_conf.get("minPowerTarget") is not None:
conf["minimum_power"] = dps_conf["minPowerTarget"]["watt"]
if dps_conf.get("powerStep") is not None:
conf["power_step"] = dps_conf["powerStep"]["watt"]
return cls.power(**conf)

25
tests/config_tests/__init__.py
View File

@@ -23,7 +23,7 @@ from pyasic.config import (
ScalingConfig,
TemperatureConfig,
)
from pyasic.config.scaling import ScalingShutdown
from pyasic.config.mining.scaling import ScalingShutdown
class TestConfig(unittest.TestCase):
@@ -40,11 +40,13 @@ class TestConfig(unittest.TestCase):
),
fan_mode=FanModeConfig.manual(speed=90, minimum_fans=2),
temperature=TemperatureConfig(target=70, danger=120),
mining_mode=MiningModeConfig.power_tuning(power=3000),
scaling=ScalingConfig.power(
step=100,
minimum=2000,
shutdown=ScalingShutdown.enabled(duration=3),
mining_mode=MiningModeConfig.power_tuning(
power=3000,
scaling=ScalingConfig(
step=100,
minimum=2000,
shutdown=ScalingShutdown(enabled=True, duration=3),
),
),
)
@@ -76,12 +78,11 @@ class TestConfig(unittest.TestCase):
"mode": "power_tuning",
"power": 3000,
"algo": {"mode": "standard"},
},
"scaling": {
"mode": "power",
"step": 100,
"minimum": 2000,
"shutdown": {"mode": "enabled", "duration": 3},
"scaling": {
"step": 100,
"minimum": 2000,
"shutdown": {"enabled": True, "duration": 3},
},
},
}

2
tests/config_tests/fans.py
View File

@@ -71,5 +71,5 @@ class TestFanConfig(unittest.TestCase):
fan_mode=fan_mode,
):
conf = fan_mode()
boser_conf = conf.as_boser()
boser_conf = conf.as_boser
self.assertEqual(conf, FanModeConfig.from_boser(boser_conf))