ha-nicehash/custom_components/nicehash/nicehash.py

"""
NiceHash API interface

References:
 - https://docs.nicehash.com/main/index.html
 - https://github.com/nicehash/rest-clients-demo/blob/master/python/nicehash.py
"""
from datetime import datetime
from hashlib import sha256
import hmac
import httpx
import json
import sys
from time import mktime
import uuid

from .const import NICEHASH_API_URL


class MiningRigDevice:
    def __init__(self, data: dict):
        self.device_id = data.get("id")
        self.name = data.get("name")
        self.status = data.get("status").get("description")
        self.temperature = int(data.get("temperature"))
        self.load = float(data.get("load"))
        self.rpm = float(data.get("revolutionsPerMinute"))
        self.speeds = data.get("speeds")


class MiningRig:
    def __init__(self, data: dict):
        self.rig_id = data.get("rig_id")
        self.name = data.get("name")
        self.status = data.get("minerStatus")
        self.status_time = data.get("statusTime")
        self.devices = dict()
        self.temperatures = []
        self.profitability = data.get("profitability")
        self.unpaid_amount = data.get("unpaidAmount")
        devices = data.get("devices")
        self.num_devices = len(devices)
        for raw_device in devices:
            device = MiningRigDevice(raw_device)
            self.devices[f"{device.device_id}"] = device
            self.temperatures.append(device.temperature)


class Payout:
    def __init__(self, data: dict):
        self.id = data.get("id")
        self.currency = "Unknown"
        self.created = data.get("created")
        self.amount = data.get("amount")
        self.fee = data.get("feeAmount")
        self.account_type = "Unknown"
        # Currency
        currency = data.get("currency")
        if currency:
            self.currency = currency.get("enumName")
        # Account Type
        account_type = data.get("accountType")
        if account_type:
            self.account_type = account_type.get("enumName")


class NiceHashPublicClient:
    async def get_exchange_rates(self):
        exchange_data = await self.request("GET", "/main/api/v2/exchangeRate/list")
        return exchange_data["list"]

    async def request(self, method, path, query=None, body=None):
        url = NICEHASH_API_URL + path

        if query is not None:
            url += f"?{query}"

        async with httpx.AsyncClient() as client:
            if body:
                data = json.dumps(body)
                response = await client.request(method, url, data=data)
            else:
                response = await client.request(method, url)

            if response.status_code == 200:
                return response.json()
            else:
                err_messages = [str(response.status_code), response.reason]
                if response.content:
                    err_messages.append(str(response.content))
                    raise Exception(": ".join(err_messages))


class NiceHashPrivateClient:
    def __init__(self, organization_id, key, secret, verbose=False):
        self.organization_id = organization_id
        self.key = key
        self.secret = secret
        self.verbose = verbose

    async def get_accounts(self):
        return await self.request("GET", "/main/api/v2/accounting/accounts2")

    async def get_mining_rigs(self):
        return await self.request("GET", "/main/api/v2/mining/rigs2")

    async def get_mining_rig(self, rig_id):
        return await self.request("GET", f"/main/api/v2/mining/rig2/{rig_id}")

    async def get_rig_payouts(self, size=84):
        query = f"size={size}"
        return await self.request("GET", "/main/api/v2/mining/rigs/payouts", query)

    async def request(self, method, path, query="", body=None):
        xtime = self.get_epoch_ms_from_now()
        xnonce = str(uuid.uuid4())

        message = f"{self.key}\00{str(xtime)}\00{xnonce}\00\00{self.organization_id}\00\00{method}\00{path}\00{query}"

        data = None
        if body:
            data = json.dumps(body)
            message += f"\00{data}"

        digest = hmac.new(self.secret.encode(), message.encode(), sha256).hexdigest()
        xauth = f"{self.key}:{digest}"

        headers = {
            "X-Time": str(xtime),
            "X-Nonce": xnonce,
            "X-Auth": xauth,
            "Content-Type": "application/json",
            "X-Organization-Id": self.organization_id,
            "X-Request-Id": str(uuid.uuid4()),
        }

        async with httpx.AsyncClient() as client:
            client.headers = headers

            url = NICEHASH_API_URL + path
            if query:
                url += "?" + query

            if self.verbose:
                print(method, url)

            if data:
                response = await client.request(method, url, data=data)
            else:
                response = await client.request(method, url)

            if response.status_code == 200:
                return response.json()
            else:
                err_messages = [str(response.status_code), response.reason]
                if response.content:
                    err_messages.append(str(response.content))
                    raise Exception(": ".join(err_messages))

    def get_epoch_ms_from_now(self):
        now = datetime.now()
        now_ec_since_epoch = mktime(now.timetuple()) + now.microsecond / 1000000.0
        return int(now_ec_since_epoch * 1000)