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

refactor: Update config to use future annotations and move merge_dicts to misc.

Browse Source
This commit is contained in:
UpstreamData
2024-01-25 11:32:03 -07:00
parent c5eed797ec
commit b328a27f04
8 changed files with 64 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
pyasic/config/__init__.py
View File

@@ -13,7 +13,6 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from copy import deepcopy
from dataclasses import asdict, dataclass, field
from pyasic.config.fans import FanModeConfig
@@ -21,6 +20,7 @@ from pyasic.config.mining import MiningModeConfig
from pyasic.config.pools import PoolConfig
from pyasic.config.power_scaling import PowerScalingConfig
from pyasic.config.temperature import TemperatureConfig
from pyasic.misc import merge_dicts
@dataclass
@@ -93,7 +93,7 @@ class MinerConfig:
def as_bosminer(self, user_suffix: str = None) -> dict:
return {
**merge(self.fan_mode.as_bosminer(), self.temperature.as_bosminer()),
**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.power_scaling.as_bosminer(),
@@ -120,8 +120,10 @@ class MinerConfig:
def as_auradine(self, user_suffix: str = None) -> dict:
return {
**self.fan_mode.as_auradine(),
**self.temperature.as_auradine(),
**self.mining_mode.as_auradine(),
**self.pools.as_auradine(user_suffix=user_suffix),
**self.power_scaling.as_auradine(),
}
@classmethod
@@ -203,14 +205,3 @@ class MinerConfig:
fan_mode=FanModeConfig.from_auradine(web_conf["fan"]),
mining_mode=MiningModeConfig.from_auradine(web_conf["mode"]),
)
def merge(a: dict, b: dict) -> dict:
result = deepcopy(a)
for b_key, b_val in b.items():
a_val = result.get(b_key)
if isinstance(a_val, dict) and isinstance(b_val, dict):
result[b_key] = merge(a_val, b_val)
else:
result[b_key] = deepcopy(b_val)
return result

7
pyasic/config/base.py
View File

@@ -13,14 +13,15 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from __future__ import annotations
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]):
def from_dict(cls, dict_conf: dict | None):
return cls.default()
def as_am_modern(self) -> dict:
@@ -67,7 +68,7 @@ class MinerConfigOption(Enum):
@dataclass
class MinerConfigValue:
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
def from_dict(cls, dict_conf: dict | None):
return cls()
def as_dict(self) -> dict:

17
pyasic/config/fans.py
View File

@@ -13,8 +13,9 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Union
from pyasic.config.base import MinerConfigOption, MinerConfigValue
@@ -26,7 +27,7 @@ class FanModeNormal(MinerConfigValue):
minimum_speed: int = 0
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "FanModeNormal":
def from_dict(cls, dict_conf: dict | None) -> "FanModeNormal":
cls_conf = {}
if dict_conf.get("minimum_fans") is not None:
cls_conf["minimum_fans"] = dict_conf["minimum_fans"]
@@ -57,7 +58,7 @@ class FanModeManual(MinerConfigValue):
minimum_fans: int = 1
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "FanModeManual":
def from_dict(cls, dict_conf: dict | None) -> "FanModeManual":
cls_conf = {}
if dict_conf.get("speed") is not None:
cls_conf["speed"] = dict_conf["speed"]
@@ -101,7 +102,7 @@ class FanModeImmersion(MinerConfigValue):
mode: str = field(init=False, default="immersion")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "FanModeImmersion":
def from_dict(cls, dict_conf: dict | None) -> "FanModeImmersion":
return cls()
def as_am_modern(self) -> dict:
@@ -124,7 +125,7 @@ class FanModeConfig(MinerConfigOption):
return cls.normal()
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
def from_dict(cls, dict_conf: dict | None):
if dict_conf is None:
return cls.default()
@@ -132,9 +133,9 @@ class FanModeConfig(MinerConfigOption):
if mode is None:
return cls.default()
clsattr = getattr(cls, mode)
if clsattr is not None:
return clsattr().from_dict(dict_conf)
cls_attr = getattr(cls, mode)
if cls_attr is not None:
return cls_attr().from_dict(dict_conf)
@classmethod
def from_am_modern(cls, web_conf: dict):

29
pyasic/config/mining.py
View File

@@ -13,8 +13,9 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Dict, Union
from pyasic.config.base import MinerConfigOption, MinerConfigValue
from pyasic.web.braiins_os.proto.braiins.bos.v1 import (
@@ -34,7 +35,7 @@ class MiningModeNormal(MinerConfigValue):
mode: str = field(init=False, default="normal")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeNormal":
def from_dict(cls, dict_conf: dict | None) -> "MiningModeNormal":
return cls()
def as_am_modern(self) -> dict:
@@ -52,7 +53,7 @@ class MiningModeSleep(MinerConfigValue):
mode: str = field(init=False, default="sleep")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeSleep":
def from_dict(cls, dict_conf: dict | None) -> "MiningModeSleep":
return cls()
def as_am_modern(self) -> dict:
@@ -70,7 +71,7 @@ class MiningModeLPM(MinerConfigValue):
mode: str = field(init=False, default="low")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeLPM":
def from_dict(cls, dict_conf: dict | None) -> "MiningModeLPM":
return cls()
def as_am_modern(self) -> dict:
@@ -88,7 +89,7 @@ class MiningModeHPM(MinerConfigValue):
mode: str = field(init=False, default="high")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeHPM":
def from_dict(cls, dict_conf: dict | None) -> "MiningModeHPM":
return cls()
def as_am_modern(self) -> dict:
@@ -107,7 +108,7 @@ class MiningModePowerTune(MinerConfigValue):
power: int = None
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModePowerTune":
def from_dict(cls, dict_conf: dict | None) -> "MiningModePowerTune":
return cls(dict_conf.get("power"))
def as_am_modern(self) -> dict:
@@ -145,7 +146,7 @@ class MiningModeHashrateTune(MinerConfigValue):
hashrate: int = None
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeHashrateTune":
def from_dict(cls, dict_conf: dict | None) -> "MiningModeHashrateTune":
return cls(dict_conf.get("hashrate"))
def as_am_modern(self) -> dict:
@@ -177,7 +178,7 @@ class ManualBoardSettings(MinerConfigValue):
volt: float
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "ManualBoardSettings":
def from_dict(cls, dict_conf: dict | None) -> "ManualBoardSettings":
return cls(freq=dict_conf["freq"], volt=dict_conf["volt"])
def as_am_modern(self) -> dict:
@@ -190,10 +191,10 @@ class MiningModeManual(MinerConfigValue):
global_freq: float
global_volt: float
boards: Dict[int, ManualBoardSettings] = field(default_factory=dict)
boards: dict[int, ManualBoardSettings] = field(default_factory=dict)
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "MiningModeManual":
def from_dict(cls, dict_conf: dict | None) -> "MiningModeManual":
return cls(
global_freq=dict_conf["global_freq"],
global_volt=dict_conf["global_volt"],
@@ -232,7 +233,7 @@ class MiningModeConfig(MinerConfigOption):
return cls.normal()
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
def from_dict(cls, dict_conf: dict | None):
if dict_conf is None:
return cls.default()
@@ -240,9 +241,9 @@ class MiningModeConfig(MinerConfigOption):
if mode is None:
return cls.default()
clsattr = getattr(cls, mode)
if clsattr is not None:
return clsattr().from_dict(dict_conf)
cls_attr = getattr(cls, mode)
if cls_attr is not None:
return cls_attr().from_dict(dict_conf)
@classmethod
def from_am_modern(cls, web_conf: dict):

14
pyasic/config/pools.py
View File

@@ -13,10 +13,12 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from __future__ import annotations
import random
import string
from dataclasses import dataclass, field
from typing import Dict, List, Union
from typing import List
from pyasic.config.base import MinerConfigValue
@@ -108,7 +110,7 @@ class Pool(MinerConfigValue):
return {"url": self.url, "user": self.user, "pass": self.password}
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "Pool":
def from_dict(cls, dict_conf: dict | None) -> "Pool":
return cls(
url=dict_conf["url"], user=dict_conf["user"], password=dict_conf["password"]
)
@@ -169,7 +171,7 @@ class Pool(MinerConfigValue):
@dataclass
class PoolGroup(MinerConfigValue):
pools: List[Pool] = field(default_factory=list)
pools: list[Pool] = field(default_factory=list)
quota: int = 1
name: str = None
@@ -254,7 +256,7 @@ class PoolGroup(MinerConfigValue):
return [p.as_auradine(user_suffix=user_suffix) for p in self.pools]
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PoolGroup":
def from_dict(cls, dict_conf: dict | None) -> "PoolGroup":
cls_conf = {}
if dict_conf.get("quota") is not None:
@@ -330,14 +332,14 @@ class PoolConfig(MinerConfigValue):
return cls()
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PoolConfig":
def from_dict(cls, dict_conf: 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":
def simple(cls, pools: list[Pool | dict[str, str]]) -> "PoolConfig":
group_pools = []
for pool in pools:
if isinstance(pool, dict):

17
pyasic/config/power_scaling.py
View File

@@ -13,8 +13,9 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Union
from pyasic.config.base import MinerConfigOption, MinerConfigValue
from pyasic.web.braiins_os.proto.braiins.bos.v1 import (
@@ -31,7 +32,7 @@ class PowerScalingShutdownEnabled(MinerConfigValue):
duration: int = None
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PowerScalingShutdownEnabled":
def from_dict(cls, dict_conf: dict | None) -> "PowerScalingShutdownEnabled":
return cls(duration=dict_conf.get("duration"))
def as_bosminer(self) -> dict:
@@ -51,7 +52,7 @@ class PowerScalingShutdownDisabled(MinerConfigValue):
mode: str = field(init=False, default="disabled")
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PowerScalingShutdownDisabled":
def from_dict(cls, dict_conf: dict | None) -> "PowerScalingShutdownDisabled":
return cls()
def as_bosminer(self) -> dict:
@@ -66,7 +67,7 @@ class PowerScalingShutdown(MinerConfigOption):
disabled = PowerScalingShutdownDisabled
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
def from_dict(cls, dict_conf: dict | None):
if dict_conf is None:
return cls.default()
@@ -107,9 +108,7 @@ class PowerScalingEnabled(MinerConfigValue):
mode: str = field(init=False, default="enabled")
power_step: int = None
minimum_power: int = None
shutdown_enabled: Union[
PowerScalingShutdownEnabled, PowerScalingShutdownDisabled
] = None
shutdown_enabled: PowerScalingShutdownEnabled | PowerScalingShutdownDisabled = None
@classmethod
def from_bosminer(cls, power_scaling_conf: dict) -> "PowerScalingEnabled":
@@ -122,7 +121,7 @@ class PowerScalingEnabled(MinerConfigValue):
)
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "PowerScalingEnabled":
def from_dict(cls, dict_conf: dict | None) -> "PowerScalingEnabled":
cls_conf = {
"power_step": dict_conf.get("power_step"),
"minimum_power": dict_conf.get("minimum_power"),
@@ -175,7 +174,7 @@ class PowerScalingConfig(MinerConfigOption):
return cls.disabled()
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]):
def from_dict(cls, dict_conf: dict | None):
if dict_conf is None:
return cls.default()

5
pyasic/config/temperature.py
View File

@@ -13,8 +13,9 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from __future__ import annotations
from dataclasses import dataclass
from typing import Union
from pyasic.config.base import MinerConfigValue
@@ -40,7 +41,7 @@ class TemperatureConfig(MinerConfigValue):
return {"temp_control": temp_cfg}
@classmethod
def from_dict(cls, dict_conf: Union[dict, None]) -> "TemperatureConfig":
def from_dict(cls, dict_conf: dict | None) -> "TemperatureConfig":
return cls(
target=dict_conf.get("target"),
hot=dict_conf.get("hot"),

13
pyasic/misc/__init__.py
View File

@@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and -
# limitations under the License. -
# ------------------------------------------------------------------------------
from copy import deepcopy
from pyasic.rpc import APIError
@@ -76,3 +78,14 @@ def api_min_version(version: str):
return inner
return decorator
def merge_dicts(a: dict, b: dict) -> dict:
result = deepcopy(a)
for b_key, b_val in b.items():
a_val = result.get(b_key)
if isinstance(a_val, dict) and isinstance(b_val, dict):
result[b_key] = merge_dicts(a_val, b_val)
else:
result[b_key] = deepcopy(b_val)
return result