from dataclasses import dataclass, asdict from typing import List, Literal import random import string import toml import yaml import json import time @dataclass class _Pool: """A dataclass for pool information. :param url: URL of the pool. :param username: Username on the pool. :param password: Worker password on the pool. """ url: str = "" username: str = "" password: str = "" def from_dict(self, data: dict): """Convert raw pool data as a dict to usable data and save it to this class. :param data: The raw config data to convert. """ for key in data.keys(): if key == "url": self.url = data[key] if key in ["user", "username"]: self.username = data[key] if key in ["pass", "password"]: self.password = data[key] return self def as_x19(self, user_suffix: str = None): """Convert the data in this class to a dict usable by an X19 device. :param user_suffix: The suffix to append to username. """ username = self.username if user_suffix: username = f"{username}{user_suffix}" pool = {"url": self.url, "user": username, "pass": self.password} return pool def as_avalon(self, user_suffix: str = None): username = self.username if user_suffix: username = f"{username}{user_suffix}" pool = ",".join([self.url, username, self.password]) return pool def as_bos(self, user_suffix: str = None): """Convert the data in this class to a dict usable by an BOSMiner device. :param user_suffix: The suffix to append to username. """ username = self.username if user_suffix: username = f"{username}{user_suffix}" pool = {"url": self.url, "user": username, "password": self.password} return pool @dataclass class _PoolGroup: """A dataclass for pool group information. :param quota: The group quota. :param group_name: The name of the pool group. :param pools: A list of pools in this group. """ quota: int = 1 group_name: str = None pools: List[_Pool] = None def __post_init__(self): if not self.group_name: self.group_name = "".join( random.choice(string.ascii_uppercase + string.digits) for _ in range(6) ) # generate random pool group name in case it isn't set def from_dict(self, data: dict): """Convert raw pool group data as a dict to usable data and save it to this class. :param data: The raw config data to convert. """ pools = [] for key in data.keys(): if key in ["name", "group_name"]: self.group_name = data[key] if key == "quota": self.quota = data[key] if key in ["pools", "pool"]: for pool in data[key]: pools.append(_Pool().from_dict(pool)) self.pools = pools return self def as_x19(self, user_suffix: str = None): """Convert the data in this class to a dict usable by an X19 device. :param user_suffix: The suffix to append to username. """ pools = [] for pool in self.pools[:3]: pools.append(pool.as_x19(user_suffix=user_suffix)) return pools def as_avalon(self, user_suffix: str = None): pool = self.pools[0].as_avalon(user_suffix=user_suffix) return pool def as_bos(self, user_suffix: str = None): """Convert the data in this class to a dict usable by an BOSMiner device. :param user_suffix: The suffix to append to username. """ group = { "name": self.group_name, "quota": self.quota, "pool": [pool.as_bos(user_suffix=user_suffix) for pool in self.pools], } return group @dataclass class MinerConfig: """A dataclass for miner configuration information. :param pool_groups: A list of pool groups in this config. :param temp_mode: The temperature control mode. :param temp_target: The target temp. :param temp_hot: The hot temp (100% fans). :param temp_dangerous: The dangerous temp (shutdown). :param minimum_fans: The minimum numbers of fans to run the miner. :param fan_speed: Manual fan speed to run the fan at (only if temp_mode == "manual"). :param asicboost: Whether or not to enable asicboost. :param autotuning_enabled: Whether or not to enable autotuning. :param autotuning_wattage: The wattage to use when autotuning. :param dps_enabled: Whether or not to enable dynamic power scaling. :param dps_power_step: The amount of power to reduce autotuning by when the miner reaches dangerous temp. :param dps_min_power: The minimum power to reduce autotuning to. :param dps_shutdown_enabled: Whether or not to shutdown the miner when `dps_min_power` is reached. :param dps_shutdown_duration: The amount of time to shutdown for (in hours). """ pool_groups: List[_PoolGroup] = None temp_mode: Literal["auto", "manual", "disabled"] = "auto" temp_target: float = 70.0 temp_hot: float = 80.0 temp_dangerous: float = 10.0 minimum_fans: int = None fan_speed: Literal[tuple(range(101))] = None # noqa - Ignore weird Literal usage asicboost: bool = None autotuning_enabled: bool = True autotuning_wattage: int = 900 dps_enabled: bool = None dps_power_step: int = None dps_min_power: int = None dps_shutdown_enabled: bool = None dps_shutdown_duration: float = None def as_dict(self): """Convert the data in this class to a dict.""" data_dict = asdict(self) for key in asdict(self).keys(): if data_dict[key] is None: del data_dict[key] return data_dict def as_toml(self): """Convert the data in this class to toml.""" return toml.dumps(self.as_dict()) def as_yaml(self): """Convert the data in this class to yaml.""" return yaml.dump(self.as_dict(), sort_keys=False) def from_raw(self, data: dict): """Convert raw config data as a dict to usable data and save it to this class. :param data: The raw config data to convert. """ pool_groups = [] for key in data.keys(): if key == "pools": pool_groups.append(_PoolGroup().from_dict({"pools": data[key]})) elif key == "group": for group in data[key]: pool_groups.append(_PoolGroup().from_dict(group)) if key == "bitmain-fan-ctrl": if data[key]: self.temp_mode = "manual" if data.get("bitmain-fan-pwm"): self.fan_speed = int(data["bitmain-fan-pwm"]) elif key == "fan_control": for _key in data[key].keys(): if _key == "min_fans": self.minimum_fans = data[key][_key] elif _key == "speed": self.fan_speed = data[key][_key] elif key == "temp_control": for _key in data[key].keys(): if _key == "mode": self.temp_mode = data[key][_key] elif _key == "target_temp": self.temp_target = data[key][_key] elif _key == "hot_temp": self.temp_hot = data[key][_key] elif _key == "dangerous_temp": self.temp_dangerous = data[key][_key] if key == "hash_chain_global": if data[key].get("asic_boost"): self.asicboost = data[key]["asic_boost"] if key == "autotuning": for _key in data[key].keys(): if _key == "enabled": self.autotuning_enabled = data[key][_key] elif _key == "psu_power_limit": self.autotuning_wattage = data[key][_key] if key == "power_scaling": for _key in data[key].keys(): if _key == "enabled": self.dps_enabled = data[key][_key] elif _key == "power_step": self.dps_power_step = data[key][_key] elif _key == "min_psu_power_limit": self.dps_min_power = data[key][_key] elif _key == "shutdown_enabled": self.dps_shutdown_enabled = data[key][_key] elif _key == "shutdown_duration": self.dps_shutdown_duration = data[key][_key] self.pool_groups = pool_groups return self def from_dict(self, data: dict): """Convert an output dict of this class back into usable data and save it to this class. :param data: The raw config data to convert. """ pool_groups = [] for group in data["pool_groups"]: pool_groups.append(_PoolGroup().from_dict(group)) for key in data.keys(): if getattr(self, key) and not key == "pool_groups": setattr(self, key, data[key]) self.pool_groups = pool_groups return self def from_toml(self, data: str): """Convert output toml of this class back into usable data and save it to this class. :param data: The raw config data to convert. """ return self.from_dict(toml.loads(data)) def from_yaml(self, data: str): """Convert output yaml of this class back into usable data and save it to this class. :param data: The raw config data to convert. """ return self.from_dict(yaml.load(data, Loader=yaml.SafeLoader)) def as_x19(self, user_suffix: str = None) -> str: """Convert the data in this class to a config usable by an X19 device. :param user_suffix: The suffix to append to username. """ cfg = { "pools": self.pool_groups[0].as_x19(user_suffix=user_suffix), "bitmain-fan-ctrl": False, "bitmain-fan-pwn": 100, } if not self.temp_mode == "auto": cfg["bitmain-fan-ctrl"] = True if self.fan_speed: cfg["bitmain-fan-ctrl"] = str(self.fan_speed) return json.dumps(cfg) def as_avalon(self, user_suffix: str = None) -> str: cfg = self.pool_groups[0].as_avalon() return cfg def as_bos(self, model: str = "S9", user_suffix: str = None) -> str: """Convert the data in this class to a config usable by an BOSMiner device. :param model: The model of the miner to be used in the format portion of the config. :param user_suffix: The suffix to append to username. """ cfg = { "format": { "version": "1.2+", "model": f"Antminer {model}", "generator": "Upstream Config Utility", "timestamp": int(time.time()), }, "group": [ group.as_bos(user_suffix=user_suffix) for group in self.pool_groups ], "temp_control": { "mode": self.temp_mode, "target_temp": self.temp_target, "hot_temp": self.temp_hot, "dangerous_temp": self.temp_dangerous, }, } if self.autotuning_enabled or self.autotuning_wattage: cfg["autotuning"] = {} if self.autotuning_enabled: cfg["autotuning"]["enabled"] = self.autotuning_enabled if self.autotuning_wattage: cfg["autotuning"]["psu_power_limit"] = self.autotuning_wattage if self.asicboost: cfg["hash_chain_global"] = {} cfg["hash_chain_global"]["asic_boost"] = self.asicboost if any( [ getattr(self, item) for item in [ "dps_enabled", "dps_power_step", "dps_min_power", "dps_shutdown_enabled", "dps_shutdown_duration", ] ] ): cfg["power_scaling"] = {} if self.dps_enabled: cfg["power_scaling"]["enabled"] = self.dps_enabled if self.dps_power_step: cfg["power_scaling"]["power_step"] = self.dps_power_step if self.dps_min_power: cfg["power_scaling"]["min_psu_power_limit"] = self.dps_min_power if self.dps_shutdown_enabled: cfg["power_scaling"]["shutdown_enabled"] = self.dps_shutdown_enabled if self.dps_shutdown_duration: cfg["power_scaling"]["shutdown_duration"] = self.dps_shutdown_duration return toml.dumps(cfg)