class HiveonT9(HiveonOld, T9):
data_locations = HIVEON_T9_DATA_LOC
##################################################
### DATA GATHERING FUNCTIONS (get_{some_data}) ###
##################################################
async def _get_hashboards(self, rpc_stats: dict = None) -> List[HashBoard]:
hashboards = [
HashBoard(slot=board, expected_chips=self.expected_chips)
for board in range(self.expected_hashboards)
]
if rpc_stats is None:
try:
rpc_stats = self.rpc.stats()
except APIError:
return []
board_map = {
0: [2, 9, 10],
1: [3, 11, 12],
2: [4, 13, 14],
}
for board in board_map:
hashrate = 0
chips = 0
for chipset in board_map[board]:
if hashboards[board].chip_temp is None:
try:
hashboards[board].temp = rpc_stats["STATS"][1][f"temp{chipset}"]
hashboards[board].chip_temp = rpc_stats["STATS"][1][
f"temp2_{chipset}"
]
except (KeyError, IndexError):
pass
else:
hashboards[board].missing = False
try:
hashrate += rpc_stats["STATS"][1][f"chain_rate{chipset}"]
chips += rpc_stats["STATS"][1][f"chain_acn{chipset}"]
except (KeyError, IndexError):
pass
hashboards[board].hashrate = AlgoHashRate.SHA256(
rate=float(hashrate), unit=HashUnit.SHA256.GH
).into(self.algo.unit.default)
hashboards[board].chips = chips
return hashboards
async def _get_env_temp(self, rpc_stats: dict = None) -> Optional[float]:
env_temp_list = []
board_map = {
0: [2, 9, 10],
1: [3, 11, 12],
2: [4, 13, 14],
}
if not rpc_stats:
try:
rpc_stats = await self.rpc.stats()
except APIError:
pass
if rpc_stats:
for board in board_map.values():
for chipset in board:
try:
env_temp = rpc_stats["STATS"][1][f"temp3_{chipset}"]
if not env_temp == 0:
env_temp_list.append(int(env_temp))
except (KeyError, IndexError):
pass
if not env_temp_list == []:
return round(sum(env_temp_list) / len(env_temp_list))