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

Merge branch 'master' into update_firmware_2

Browse Source
This commit is contained in:
Abhishek Patidar
2024-06-22 07:57:48 +05:30
committed by GitHub
parent 999e8ef318 30f385c2d9
commit ba69a1de2c
15 changed files with 511 additions and 388 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.board_tune(),
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)

6
pyasic/data/boards.py
View File

@@ -33,6 +33,9 @@ class HashBoard:
expected_chips: The expected chip count of the board as an int.
serial_number: The serial number of the board.
missing: Whether the board is returned from the miners data as a bool.
tuned: Whether the board is tuned as a bool.
active: Whether the board is currently tuning as a bool.
voltage: Current input voltage of the board as a float.
"""
slot: int = 0
@@ -43,6 +46,9 @@ class HashBoard:
expected_chips: int = None
serial_number: str = None
missing: bool = True
tuned: bool = None
active: bool = None
voltage: float = None
def get(self, __key: str, default: Any = None):
try:

35
pyasic/miners/backends/braiins_os.py
View File

@@ -39,6 +39,7 @@ from pyasic.rpc.bosminer import BOSMinerRPCAPI
from pyasic.ssh.braiins_os import BOSMinerSSH
from pyasic.web.braiins_os import BOSerWebAPI, BOSMinerWebAPI
from pyasic.web.braiins_os.proto.braiins.bos.v1 import SaveAction
from pyasic.data.pools import PoolMetrics
BOSMINER_DATA_LOC = DataLocations(
**{
@@ -94,6 +95,10 @@ BOSMINER_DATA_LOC = DataLocations(
"_get_uptime",
[RPCAPICommand("rpc_summary", "summary")],
),
str(DataOptions.POOLS): DataFunction(
"_get_pools",
[RPCAPICommand("rpc_pools", "pools")],
),
}
)
@@ -573,6 +578,36 @@ class BOSMiner(BraiinsOSFirmware):
except LookupError:
pass
async def _get_pools(self, rpc_pools: dict = None) -> List[PoolMetrics]:
if rpc_pools is None:
try:
rpc_pools = await self.rpc.pools()
except APIError:
pass
pools_data = []
if rpc_pools is not None:
try:
pools = rpc_pools.get("POOLS", [])
for pool_info in pools:
pool_data = PoolMetrics(
accepted=pool_info.get("Accepted"),
rejected=pool_info.get("Rejected"),
get_failures=pool_info.get("Get Failures"),
remote_failures=pool_info.get("Remote Failures"),
active=pool_info.get("Stratum Active"),
alive=pool_info.get("Status") == "Alive",
url=pool_info.get("URL"),
user=pool_info.get("User"),
index=pool_info.get("POOL"),
)
pools_data.append(pool_data)
except LookupError:
pass
return pools_data
async def upgrade_firmware(self, file: Path):
"""
Upgrade the firmware of the BOSMiner device.

40
pyasic/miners/backends/btminer.py
View File

@@ -16,6 +16,8 @@
import logging
from typing import List, Optional
import aiofiles
from pathlib import Path
from pyasic.config import MinerConfig, MiningModeConfig
from pyasic.data import AlgoHashRate, Fan, HashBoard, HashUnit
@@ -649,3 +651,41 @@ class BTMiner(StockFirmware):
return int(rpc_summary["SUMMARY"][0]["Elapsed"])
except LookupError:
pass
async def upgrade_firmware(self, file: Path, token: str):
"""
Upgrade the firmware of the Whatsminer device.
Args:
file (Path): The local file path of the firmware to be uploaded.
token (str): The authentication token for the firmware upgrade.
Returns:
str: Confirmation message after upgrading the firmware.
"""
try:
logging.info("Starting firmware upgrade process for Whatsminer.")
if not file:
raise ValueError("File location must be provided for firmware upgrade.")
# Read the firmware file contents
async with aiofiles.open(file, "rb") as f:
upgrade_contents = await f.read()
result = await self.rpc.update_firmware(upgrade_contents)
logging.info("Firmware upgrade process completed successfully for Whatsminer.")
return result
except FileNotFoundError as e:
logging.error(f"File not found during the firmware upgrade process: {e}")
raise
except ValueError as e:
logging.error(f"Validation error occurred during the firmware upgrade process: {e}")
raise
except OSError as e:
logging.error(f"OS error occurred during the firmware upgrade process: {e}")
raise
except Exception as e:
logging.error(f"An unexpected error occurred during the firmware upgrade process: {e}", exc_info=True)
raise

77
pyasic/miners/backends/epic.py
View File

@@ -21,6 +21,7 @@ from pyasic.data import AlgoHashRate, Fan, HashBoard, HashUnit
from pyasic.data.error_codes import MinerErrorData, X19Error
from pyasic.errors import APIError
from pyasic.logger import logger
from pyasic.data.pools import PoolMetrics
from pyasic.miners.data import DataFunction, DataLocations, DataOptions, WebAPICommand
from pyasic.miners.device.firmware import ePICFirmware
from pyasic.web.epic import ePICWebAPI
@@ -58,10 +59,6 @@ EPIC_DATA_LOC = DataLocations(
"_get_wattage",
[WebAPICommand("web_summary", "summary")],
),
str(DataOptions.VOLTAGE): DataFunction(
"_get_voltage",
[WebAPICommand("web_summary", "summary")],
),
str(DataOptions.FANS): DataFunction(
"_get_fans",
[WebAPICommand("web_summary", "summary")],
@@ -82,6 +79,10 @@ EPIC_DATA_LOC = DataLocations(
"_is_mining",
[WebAPICommand("web_summary", "summary")],
),
str(DataOptions.POOLS): DataFunction(
"_get_pools",
[WebAPICommand("web_summary", "summary")],
),
}
)
@@ -219,20 +220,6 @@ class ePIC(ePICFirmware):
except KeyError:
pass
async def _get_voltage(self, web_summary: dict = None) -> Optional[float]:
if web_summary is None:
try:
web_summary = await self.web.summary()
except APIError:
pass
if web_summary is not None:
try:
voltage = web_summary["Power Supply Stats"]["Output Voltage"]
return voltage
except KeyError:
pass
async def _get_hashrate(self, web_summary: dict = None) -> Optional[float]:
if web_summary is None:
try:
@@ -323,6 +310,24 @@ class ePIC(ePICFirmware):
except APIError:
pass
tuned = True
active = False
if web_summary is not None:
tuner_running = web_summary["PerpetualTune"]["Running"]
if tuner_running:
active = True
algo_info = web_summary["PerpetualTune"]["Algorithm"]
if algo_info.get("VoltageOptimizer") is not None:
tuned = algo_info["VoltageOptimizer"].get("Optimized")
elif algo_info.get("BoardTune") is not None:
tuned = algo_info["BoardTune"].get("Optimized")
else:
tuned = algo_info["ChipTune"].get("Optimized")
# To be extra detailed, also ensure the miner is in "Mining" state
tuned = tuned and web_summary["Status"]["Operating State"] == "Mining"
active = active and web_summary["Status"]["Operating State"] == "Mining"
hb_list = [
HashBoard(slot=i, expected_chips=self.expected_chips)
for i in range(self.expected_hashboards)
@@ -349,6 +354,9 @@ class ePIC(ePICFirmware):
).into(self.algo.unit.default)
hb_list[hb["Index"]].chips = num_of_chips
hb_list[hb["Index"]].temp = hb["Temperature"]
hb_list[hb["Index"]].tuned = tuned
hb_list[hb["Index"]].active = active
hb_list[hb["Index"]].voltage = hb["Input Voltage"]
return hb_list
async def _is_mining(self, web_summary, *args, **kwargs) -> Optional[bool]:
@@ -410,4 +418,35 @@ class ePIC(ePICFirmware):
return errors
except KeyError:
pass
return errors
return errors
async def _get_pools(self, web_summary: dict = None) -> List[PoolMetrics]:
if web_summary is None:
try:
web_summary = await self.web.summary()
except APIError:
pass
pool_data = []
try:
if web_summary is not None:
if (
web_summary.get("Session") is not None
and web_summary.get("Stratum") is not None
):
pool_data.append(
PoolMetrics(
accepted=web_summary["Session"].get("Accepted"),
rejected=web_summary["Session"].get("Rejected"),
get_failures=0,
remote_failures=0,
active=web_summary["Stratum"].get("IsPoolConnected"),
alive=web_summary["Stratum"].get("IsPoolConnected"),
url=web_summary["Stratum"].get("Current Pool"),
user=web_summary["Stratum"].get("Current User"),
index=web_summary["Stratum"].get("Config Id"),
)
)
return pool_data
except LookupError:
pass

2
pyasic/miners/backends/whatsminer.py
View File

@@ -29,4 +29,4 @@ class M3X(BTMiner):
class M2X(BTMiner):
pass
pass

1
pyasic/miners/data.py
View File

@@ -37,7 +37,6 @@ class DataOptions(Enum):
IS_MINING = "is_mining"
UPTIME = "uptime"
CONFIG = "config"
VOLTAGE = "voltage"
POOLS = "pools"
def __str__(self):

24
pyasic/rpc/btminer.py
View File

@@ -23,6 +23,7 @@ import json
import logging
import re
from typing import Literal, Union
import struct
import httpx
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
@@ -559,11 +560,24 @@ class BTMinerRPCAPI(BaseMinerRPCAPI):
"""
return await self.send_privileged_command("set_normal_power")
async def update_firmware(self): # noqa - static
"""Not implemented."""
# to be determined if this will be added later
# requires a file stream in bytes
return NotImplementedError
async def update_firmware(self, firmware: bytes):
"""Upgrade the firmware running on the miner and using the firmware passed in bytes.
Parameters:
firmware (bytes): The firmware binary data to be uploaded.
Returns:
bool: A boolean indicating the success of the firmware upgrade.
Raises:
APIError: If the miner is not ready for firmware update.
"""
ready = await self.send_privileged_command("upgrade_firmware")
if not ready.get("Msg") == "ready":
raise APIError(f"Not ready for firmware update: {self}")
file_size = struct.pack("<I", len(firmware))
await self._send_bytes(file_size + firmware)
return True
async def reboot(self, timeout: int = 10) -> dict:
"""Reboot the miner using the API.

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))