Python/miner-kaspa.py

#!/usr/bin/env python3
# Description: Kaspa Mining Script - Supports both solo mining and pool mining with JSON configuration
# Usage: python3 miner-kaspa.py
# Author: Justin Oros
# Source: https://github.com/JustinOros 

import json
import subprocess
import sys
import os
import time
import signal
from pathlib import Path
from typing import Dict, Optional

class KaspaMiner:
    def __init__(self, config_file: str = "miner-kaspa.json"):
        self.config_file = config_file
        self.config = self.load_config()
        self.process: Optional[subprocess.Popen] = None
        
    def load_config(self) -> Dict:
        """Load configuration from JSON file"""
        if not os.path.exists(self.config_file):
            print(f"Error: Configuration file '{self.config_file}' not found.")
            print("Creating example configuration file...")
            self.create_example_config()
            sys.exit(1)
            
        try:
            with open(self.config_file, 'r') as f:
                config = json.load(f)
            self.validate_config(config)
            return config
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON in configuration file: {e}")
            sys.exit(1)
        except Exception as e:
            print(f"Error loading configuration: {e}")
            sys.exit(1)
    
    def validate_config(self, config: Dict):
        """Validate required configuration fields"""
        required_fields = ['wallet_address', 'mining_mode']
        
        for field in required_fields:
            if field not in config:
                raise ValueError(f"Missing required field: {field}")
        
        if config['mining_mode'] not in ['solo', 'pool']:
            raise ValueError("mining_mode must be 'solo' or 'pool'")
        
        if config['mining_mode'] == 'pool' and 'pool_address' not in config:
            raise ValueError("pool_address is required for pool mining")
    
    def create_example_config(self):
        """Create an example configuration file"""
        example_config = {
            "wallet_address": "kaspa:qz1234567890abcdefghijklmnopqrstuvwxyz",
            "mining_mode": "solo",
            "miner_executable": "kaspa-miner",
            "kaspad_address": "127.0.0.1:16110",
            "pool_address": "pool.example.com:5555",
            "worker_name": "worker1",
            "threads": 0,
            "gpu_enable": True,
            "gpu_devices": [0],
            "intensity": 20,
            "extra_args": [],
            "log_file": "kaspa-miner.log"
        }
        
        with open(self.config_file, 'w') as f:
            json.dump(example_config, f, indent=4)
        
        print(f"Created example configuration file: {self.config_file}")
        print("Please edit it with your mining details.")
    
    def build_command(self) -> list:
        """Build the mining command based on configuration"""
        cmd = [self.config.get('miner_executable', 'kaspa-miner')]
        
        # Add wallet address
        cmd.extend(['--mining-address', self.config['wallet_address']])
        
        # Configure based on mining mode
        if self.config['mining_mode'] == 'solo':
            # Solo mining - connect to local kaspad
            kaspad_addr = self.config.get('kaspad_address', '127.0.0.1:16110')
            cmd.extend(['--kaspad-address', kaspad_addr])
            print(f"Solo mining mode - connecting to kaspad at {kaspad_addr}")
        else:
            # Pool mining
            pool_addr = self.config['pool_address']
            cmd.extend(['--pool', pool_addr])
            
            # Add worker name if specified
            worker = self.config.get('worker_name')
            if worker:
                cmd.extend(['--worker-name', worker])
            
            print(f"Pool mining mode - connecting to pool at {pool_addr}")
        
        # Thread configuration
        threads = self.config.get('threads', 0)
        if threads > 0:
            cmd.extend(['--threads', str(threads)])
        
        # GPU configuration
        if self.config.get('gpu_enable', True):
            gpu_devices = self.config.get('gpu_devices', [0])
            if gpu_devices:
                cmd.extend(['--cuda-devices', ','.join(map(str, gpu_devices))])
            
            intensity = self.config.get('intensity')
            if intensity:
                cmd.extend(['--intensity', str(intensity)])
        else:
            cmd.append('--no-cuda')
        
        # Add any extra arguments
        extra_args = self.config.get('extra_args', [])
        if extra_args:
            cmd.extend(extra_args)
        
        return cmd
    
    def start_mining(self):
        """Start the mining process"""
        cmd = self.build_command()
        
        print("\n" + "="*60)
        print("Kaspa Miner Starting")
        print("="*60)
        print(f"Wallet: {self.config['wallet_address']}")
        print(f"Mode: {self.config['mining_mode'].upper()}")
        print(f"Command: {' '.join(cmd)}")
        print("="*60 + "\n")
        
        # Setup logging if specified
        log_file = self.config.get('log_file')
        log_handle = None
        
        try:
            if log_file:
                log_handle = open(log_file, 'a')
                log_handle.write(f"\n{'='*60}\n")
                log_handle.write(f"Mining started at {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
                log_handle.write(f"Command: {' '.join(cmd)}\n")
                log_handle.write(f"{'='*60}\n\n")
                log_handle.flush()
            
            # Start the miner process
            self.process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE if log_handle else None,
                stderr=subprocess.STDOUT if log_handle else None,
                universal_newlines=True,
                bufsize=1
            )
            
            # Monitor output if logging
            if log_handle and self.process.stdout:
                try:
                    for line in self.process.stdout:
                        print(line, end='')
                        log_handle.write(line)
                        log_handle.flush()
                except KeyboardInterrupt:
                    print("\n\nReceived interrupt signal. Stopping miner...")
                    self.stop_mining()
            else:
                # Wait for process to complete
                self.process.wait()
                
        except FileNotFoundError:
            print(f"Error: Miner executable '{cmd[0]}' not found.")
            print("Please ensure the miner is installed and the path is correct.")
            sys.exit(1)
        except Exception as e:
            print(f"Error starting miner: {e}")
            sys.exit(1)
        finally:
            if log_handle:
                log_handle.close()
    
    def stop_mining(self):
        """Stop the mining process"""
        if self.process:
            print("Stopping miner...")
            self.process.terminate()
            try:
                self.process.wait(timeout=10)
            except subprocess.TimeoutExpired:
                print("Force killing miner process...")
                self.process.kill()
            print("Miner stopped.")

def signal_handler(signum, frame):
    """Handle interrupt signals"""
    print("\n\nShutdown signal received...")
    sys.exit(0)

def main():
    # Register signal handlers
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    
    # Parse command line arguments
    config_file = "miner-kaspa.json"
    if len(sys.argv) > 1:
        config_file = sys.argv[1]
    
    # Create and start miner
    miner = KaspaMiner(config_file)
    
    try:
        miner.start_mining()
    except KeyboardInterrupt:
        print("\n\nInterrupted by user")
        miner.stop_mining()
    except Exception as e:
        print(f"\nError: {e}")
        miner.stop_mining()
        sys.exit(1)

if __name__ == "__main__":
    main()