cpu-miner/miner.py

import random
import struct
import time
import hashlib
import logging
from binascii import hexlify, unhexlify
from threading import Event
from typing import Callable

import config

log = logging.getLogger(__name__)

# Seconds between hashrate logs
_RATE_INT = 2


def _compute_hash_batch(
    header_76: bytes,
    start_nonce: int,
    batch_size: int,
    target_be: bytes,
):
    """
    Compute hashes for a nonce batch and return the first valid nonce found.

    Optimizations:
    - Precompute first chunk (64 bytes) with sha256.copy() to avoid
      re-updating the invariant portion on each nonce.
    - Preallocate a 16-byte tail and update only the nonce field
      with struct.pack_into to minimize allocations.
    - Local function binding to reduce lookups in the hot loop.
    """
    first_chunk  = header_76[:64]
    tail_static  = header_76[64:]   # 12 bytes: merkle[28:32] + ts(4) + bits(4)

    sha_base = hashlib.sha256()
    sha_base.update(first_chunk)

    tail       = bytearray(16)
    tail[0:12] = tail_static

    sha_copy  = sha_base.copy
    sha256    = hashlib.sha256
    pack_into = struct.pack_into

    for i in range(batch_size):
        n = (start_nonce + i) & 0xFFFFFFFF
        pack_into("<I", tail, 12, n)

        ctx = sha_copy()
        ctx.update(tail)
        digest = sha256(ctx.digest()).digest()

        if digest[::-1] < target_be:
            return n, digest

    return None, None


def mine_block(
    header_hex: str,
    target_hex: str,
    nonce_mode: str = "incremental",
    stop_event: Event | None = None,
    status_callback: Callable | None = None,
):
    """
    Run mining by iterating nonces until a valid hash is found or stop_event is received.

    Calls status_callback(attempts, hashrate_hz) every ~2 seconds if provided.
    Returns (mined_header_hex, nonce, hashrate) or (None, None, None) if stopped.
    """
    log.info("Starting mining - mode %s", nonce_mode)

    if nonce_mode not in ("incremental", "random", "mixed"):
        raise ValueError(f"Invalid mining mode: {nonce_mode!r}")

    version   = unhexlify(header_hex[0:8])
    prev_hash = unhexlify(header_hex[8:72])
    merkle    = unhexlify(header_hex[72:136])
    ts_bytes  = unhexlify(header_hex[136:144])
    bits      = unhexlify(header_hex[144:152])

    header_76 = version + prev_hash + merkle + ts_bytes + bits
    target_be = int(target_hex, 16).to_bytes(32, "big")

    nonce = 0 if nonce_mode == "incremental" else random.randint(0, 0xFFFFFFFF)

    # Read configuration once before entering the loop
    batch_size  = config.BATCH
    ts_interval = config.TIMESTAMP_UPDATE_INTERVAL

    attempts    = 0
    start_t     = time.time()
    last_rate_t = start_t
    last_rate_n = 0
    last_tsu    = start_t

    while True:
        if stop_event is not None and stop_event.is_set():
            log.info("Mining stopped: stop_event received")
            return None, None, None

        now = time.time()

        # Periodic timestamp update in block header
        if ts_interval and (now - last_tsu) >= ts_interval:
            ts_bytes   = struct.pack("<I", int(now))
            header_76  = version + prev_hash + merkle + ts_bytes + bits
            last_tsu   = now

        found_nonce, digest = _compute_hash_batch(header_76, nonce, batch_size, target_be)

        if found_nonce is not None:
            total       = time.time() - start_t
            hashrate    = (attempts + batch_size) / total if total else 0
            full_header = header_76 + struct.pack("<I", found_nonce)
            log.info(
                "Block found - nonce=%d attempts=%d time=%.2fs hashrate=%.2f kH/s",
                found_nonce, attempts + batch_size, total, hashrate / 1000,
            )
            log.info("Valid hash: %s", digest[::-1].hex())
            return hexlify(full_header).decode(), found_nonce, hashrate

        attempts += batch_size
        nonce     = (nonce + batch_size) & 0xFFFFFFFF

        # Periodic logs and callback
        now = time.time()
        if now - last_rate_t >= _RATE_INT:
            hashrate    = (attempts - last_rate_n) / (now - last_rate_t)
            last_rate_t = now
            last_rate_n = attempts
            log.info(
                "Mining status - hashrate=%.2f kH/s attempts=%d nonce=%d",
                hashrate / 1000, attempts, nonce,
            )
            if status_callback:
                status_callback(attempts, hashrate)