hosting/lib/hive_miner_interface.py

import aiofiles
import asyncio
import json
import time
import os
import re

def extract_json_with_key(text, key):
    json_pattern = r'({.*?})'
    json_strings = re.findall(json_pattern, text, re.DOTALL)
    json_objects_with_key = []
    for json_str in json_strings:
        try:
            json_obj = json.loads(json_str)
            if key in json.dumps(json_obj):
                json_objects_with_key.append(json_obj)
        except json.JSONDecodeError:
            continue
    return json_objects_with_key

async def async_run_bash_command(command):
    process = await asyncio.create_subprocess_exec(
        '/bin/bash', '-c', command,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    stdout, stderr = await process.communicate()

    return stdout.decode().strip(), stderr.decode().strip(), process.returncode

async def check_and_read_file(file_path):
    try:
        if os.path.exists(file_path):
            async with aiofiles.open(file_path, mode='r') as file:
                contents = await file.read()
                return contents
        else:
            return "fail"
    except Exception as e:
        return "fail"
    
async def get_session_start_time(pid):
    try:
        async with aiofiles.open(f'/proc/{pid}/stat', 'r') as file:
            stat_info = (await file.read()).split()
            start_time_ticks = int(stat_info[21])

        clock_ticks_per_sec = os.sysconf(os.sysconf_names['SC_CLK_TCK'])
        start_time_seconds = start_time_ticks / clock_ticks_per_sec

        boot_time = None
        async with aiofiles.open('/proc/stat', 'r') as file:
            async for line in file:
                if line.startswith('btime'):
                    boot_time = int(line.split()[1])
                    break

        if boot_time is None:
            raise ValueError("Failed to find boot time in /proc/stat")

        start_time = (boot_time + start_time_seconds)
        return start_time

    except FileNotFoundError:
        return None
    except Exception as e:
        print(f"Error retrieving session start time: {e}")
        return None

async def get_miner_stats(miner_dir, api_timeout=15):
    stdout, stderr, code = await async_run_bash_command(f'export PATH=$PATH:/hive/sbin:/hive/bin && export API_TIMEOUT={api_timeout}'+''' && read -t $((API_TIMEOUT + 5)) -d "" pid khs stats < <(function run_miner_scripts { echo "$BASHPID"; cd '''+miner_dir+''' || exit; cd '''+miner_dir+'''; source h-manifest.conf; cd '''+miner_dir+'''; source h-config.sh; cd '''+miner_dir+'''; source h-stats.sh; printf "%q\n" "$khs"; echo "$stats"; }; run_miner_scripts) && [[ $? -ge 128 ]] && [[ ! -z "$pid" && "$pid" -ne $$ ]] && kill -9 "$pid" 2>/dev/null ; echo $stats''')
    try:
        if stdout and not stderr and code == 0:
            stats = json.loads(stdout)
            return stats
        else:
            return 'fail'
    except Exception as e:
        try:
            miner_stat_jsons = extract_json_with_key(stdout, "hs_units")
            return miner_stat_jsons[0] if len(miner_stat_jsons) > 0 else 'fail'
        except Exception :
            return 'fail'

async def get_miner_uptime_stats():
    stdout, stderr, code = await async_run_bash_command("screen -ls")
    if not stderr and code == 0:
        for line in stdout.split('\n'):
            if line[:1]=='\t':
                if line.split('\t')[1].endswith(".miner"):
                    miner_screen_pid = line.split('\t')[1].split('.')[0]
                    if miner_screen_pid.isdigit():
                        miner_start_time = await get_session_start_time(miner_screen_pid)
                        return miner_start_time
    return None
    
def extract_miner_names(rig_conf):
    miner_names=[]
    for line in rig_conf.split('\n'):
        if '=' in line:
            key, value = line.split('=', 1)
            if key[:5]=="MINER" and (len(key)==5 or str(key[5:]).isdigit()):
                if value.startswith('"') and value.endswith('"'):
                    value = value.strip('"')
                if len(value)>0:
                    miner_names.append(value)
    return miner_names

def extract_miner_config(miner_names, wallet_conf):
    lines = wallet_conf.split('\n')
    meta_config = {}
    miners = {}
    for miner_idx, miner_name in enumerate(miner_names):
        remaining_lines = []
        algos = []
        for line in lines:
            if not line.startswith('#') and '=' in line:
                key, value = line.split('=', 1)
                if value.startswith('"') and value.endswith('"'):
                    value = value.strip('"')
                if value.startswith("'") and value.endswith("'"):
                    value = value.strip("'")
                if key[:len(miner_name.replace('-','_'))+1].lower() == f"{miner_name.replace('-','_')}_".lower():
                    if key.split('_')[-1][:4].lower()=="algo":
                        algos.append(value)
                    elif miner_name.lower()=="custom" and key.lower()=="custom_miner":
                        miner_names[miner_idx] = os.path.join(miner_names[miner_idx],value)
                elif key.lower()=="meta":
                    try:
                        meta_config=json.loads(value)
                    except Exception as e:
                        pass
                else:
                    remaining_lines.append(line)
        lines = remaining_lines
        miners[miner_name]={
            "algos":algos,
            "coins":[]
        }
    for miner_name in miner_names:
        if "custom/" in miner_name.lower():
            miner_name = "custom"
        if miner_name in meta_config and type(meta_config[miner_name]) == dict:
            for key in meta_config[miner_name].keys():
                if "coin" in key:
                    miners[miner_name]["coins"].append(meta_config[miner_name][key])
    return miner_names, miners

def sum_numbers_in_list(lst):
    if all(isinstance(i, (int, float)) for i in lst):
        return sum(lst)
    else:
        return "The list contains non-numeric elements."

class hive_interface:
    def __init__(self):
        self.hive_miners_dir = "/hive/miners"
        self.hive_rig_config = "/hive-config/rig.conf"
        self.hive_wallet_config = "/hive-config/wallet.conf"
    
    async def get_miners_stats(self, miner_names):
        scrape_tasks = []
        for miner_name in miner_names:
            scrape_tasks.append(get_miner_stats(os.path.join(self.hive_miners_dir, miner_name)))
        results = await asyncio.gather(*scrape_tasks)
        return results

    async def get_configured_miners(self):
        rig_conf, wallet_conf = await asyncio.gather(*[check_and_read_file(self.hive_rig_config), check_and_read_file(self.hive_wallet_config)])
        miner_names = extract_miner_names(rig_conf)
        miner_names, miner_config = extract_miner_config(miner_names, wallet_conf)
        return miner_names, miner_config

    async def export_miner_stats(self, get_hashrates=False):
        output = {
            "miner_uptime": 0
        }
        miner_start_ts = await get_miner_uptime_stats()
        if miner_start_ts:
            output["miner_uptime"] = int(time.time()-miner_start_ts)
        miner_names, miner_config = await self.get_configured_miners()
        output["miners"]=miner_config
        if get_hashrates:
            miners_stats = await self.get_miners_stats(miner_names)
            for idx, miner_name in enumerate(miner_names):
                miner_names[idx] = "custom" if "custom/" in miner_name.lower() else miner_name
            for idx, miner_stats in enumerate(miners_stats):
                if type(miner_stats) == dict:
                    for key in miner_stats.keys():
                        if key[:2]=="hs" and (key=="hs" or key[2:].isdigit()):
                            all_hs = sum_numbers_in_list(miner_stats[key])
                            try:
                                if "hs_units" in miner_stats:
                                    if miner_stats["hs_units"]=="hs":
                                        all_hs = all_hs/1000
                                if not "hashrates" in output['miners'][miner_names[idx]]:
                                    output['miners'][miner_names[idx]]["hashrates"]=[]
                                if isinstance(all_hs, (float, int)):
                                    output['miners'][miner_names[idx]]["hashrates"].append(all_hs)
                                else:
                                    output['miners'][miner_names[idx]]["hashrates"].append(0)
                            except Exception as e:
                                pass
        return output