real_miner.py

import socket
import json
import hashlib
import struct
import threading
import time
import queue
import logging
import random
from datetime import datetime
import customtkinter as ctk
from typing import Optional, List, Dict, Any
import binascii
import psutil
from collections import deque

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - [%(threadName)s] - %(message)s',
    handlers=[
        logging.FileHandler('miner.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("BTC_Miner")

class StratumClient:
    """
    Implements Stratum mining protocol client for connecting to Bitcoin mining pools.
    Handles pool communication, job management, and share submission.
    """
    
    def __init__(self, pool_host: str, pool_port: int, wallet_address: str, worker_name: str = "worker1", password: str = "x"):
        """
        Initialize Stratum client with connection details.
        
        Args:
            pool_host: Mining pool hostname/IP
            pool_port: Mining pool port number
            wallet_address: Bitcoin wallet address
            worker_name: Worker identifier (default: worker1)
            password: Worker password (default: x)
        """
        self.pool_host = pool_host
        self.pool_port = pool_port
        self.wallet_address = wallet_address
        self.worker_name = worker_name
        self.password = password
        self.socket: Optional[socket.socket] = None
        self.is_connected = False
        self.job_id: Optional[str] = None
        self.prevhash: Optional[str] = None
        self.coinb1: Optional[str] = None
        self.coinb2: Optional[str] = None
        self.merkle_branches: List[str] = []
        self.version: Optional[str] = None
        self.nbits: Optional[str] = None
        self.ntime: Optional[str] = None
        self.clean_jobs = False
        self.subscription_id: Optional[str] = None
        self.extranonce1: Optional[str] = None
        self.extranonce2_size: Optional[int] = None
        self.difficulty = 1
        self.target: Optional[int] = None
        self.message_id = 0
        self.responses: queue.Queue = queue.Queue()
        self.reconnect_delay = 5
        self.max_reconnect_delay = 60
        self.connection_attempts = 0
        self.max_connection_attempts = 5
        self.response_buffer = ""
        self.response_lock = threading.Lock()
        self.pool_username: Optional[str] = None

    def _send_request(self, request: dict) -> bool:
        """Send JSON-RPC request to pool"""
        try:
            data = json.dumps(request) + "\n"
            self.socket.send(data.encode())
            logging.debug(f"Sent request: {data.strip()}")
            return True
        except Exception as e:
            logging.error(f"Send error: {e}")
            return False

    def _handle_responses(self):
        """Handle incoming messages from pool"""
        while self.is_connected:
            try:
                data = self.socket.recv(4096).decode()
                if not data:
                    logging.error("Connection closed by pool")
                    break

                with self.response_lock:
                    self.response_buffer += data
                    while "\n" in self.response_buffer:
                        line, self.response_buffer = self.response_buffer.split("\n", 1)
                        try:
                            message = json.loads(line)
                            logging.debug(f"Received message: {message}")
                            
                            if "method" in message:
                                if message["method"] == "mining.notify":
                                    self._handle_notify(message["params"])
                                elif message["method"] == "mining.set_difficulty":
                                    self._handle_difficulty(message["params"])
                            elif "result" in message:
                                self.responses.put(message)
                            elif "error" in message and message["error"]:
                                logging.error(f"Pool error: {message['error']}")
                                
                        except json.JSONDecodeError as e:
                            logging.error(f"Invalid JSON received: {line}")
                            continue
                            
            except socket.timeout:
                continue
            except Exception as e:
                logging.error(f"Response handling error: {e}")
                break
                
        self.is_connected = False

    def _subscribe(self) -> bool:
        """Subscribe to mining notifications"""
        subscribe_request = {
            "id": self._get_message_id(),
            "method": "mining.subscribe",
            "params": ["xRaisen-Miner/1.0.0"]
        }
        
        if not self._send_request(subscribe_request):
            return False
            
        try:
            response = self.responses.get(timeout=10)
            if response.get("error"):
                logging.error(f"Subscription error: {response['error']}")
                return False
                
            if "result" not in response or not response["result"]:
                logging.error("Invalid subscription response")
                return False
                
            result = response["result"]
            if not isinstance(result, list) or len(result) < 3:
                logging.error(f"Invalid subscription result format: {result}")
                return False
                
            self.subscription_id = result[0]
            self.extranonce1 = result[1]
            self.extranonce2_size = result[2]
            logging.info(f"Successfully subscribed: {self.subscription_id}")
            return True
            
        except queue.Empty:
            logging.error("Subscription response timeout")
            return False
        except Exception as e:
            logging.error(f"Subscription error: {e}")
            return False

    def _authorize(self) -> bool:
        """Authorize worker"""
        # For F2Pool, username is account_name.worker_id
        self.pool_username = f"{self.wallet_address}.{self.worker_name}"
        
        auth_request = {
            "id": self._get_message_id(),
            "method": "mining.authorize",
            "params": [self.pool_username, self.password]
        }
        
        logging.info(f"Authorizing worker: {self.pool_username}")
        
        if not self._send_request(auth_request):
            return False
            
        try:
            response = self.responses.get(timeout=10)
            if response.get("error"):
                error = response["error"]
                if isinstance(error, list) and len(error) >= 2:
                    error_code, error_msg = error[0], error[1]
                    logging.error(f"Authorization error [{error_code}]: {error_msg}")
                else:
                    logging.error(f"Authorization error: {error}")
                return False
                
            if "result" not in response:
                logging.error("Invalid authorization response")
                return False
                
            result = response["result"]
            if result:
                logging.info(f"Worker {self.pool_username} successfully authorized")
            else:
                logging.error(f"Worker {self.pool_username} authorization failed")
                
            return result
            
        except queue.Empty:
            logging.error("Authorization response timeout")
            return False
        except Exception as e:
            logging.error(f"Authorization error: {e}")
            return False

    def connect(self) -> bool:
        """Connect to mining pool with retry logic"""
        while self.connection_attempts < self.max_connection_attempts:
            try:
                logging.info(f"Attempting to connect to {self.pool_host}:{self.pool_port}")
                self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                self.socket.settimeout(10)
                
                # Parse pool address
                if "://" in self.pool_host:
                    self.pool_host = self.pool_host.split("://")[1]
                    
                self.socket.connect((self.pool_host, self.pool_port))
                self.is_connected = True
                self.connection_attempts = 0
                
                # Start response handler thread
                threading.Thread(target=self._handle_responses, daemon=True).start()
                
                # Subscribe to mining notifications
                if not self._subscribe():
                    raise Exception("Failed to subscribe to mining notifications")
                    
                # Authorize worker
                if not self._authorize():
                    raise Exception("Failed to authorize worker")
                    
                logging.info("Successfully connected to mining pool")
                return True
                
            except Exception as e:
                self.connection_attempts += 1
                delay = min(self.reconnect_delay * (2 ** (self.connection_attempts - 1)), self.max_reconnect_delay)
                logging.error(f"Connection attempt {self.connection_attempts} failed: {e}")
                logging.info(f"Retrying in {delay} seconds...")
                
                if self.socket:
                    try:
                        self.socket.close()
                    except:
                        pass
                    
                if self.connection_attempts < self.max_connection_attempts:
                    time.sleep(delay)
                else:
                    logging.error("Maximum connection attempts reached")
                    return False
                    
        return False

    def _handle_notify(self, params):
        """Handle new mining job notification"""
        self.job_id = params[0]
        self.prevhash = params[1]
        self.coinb1 = params[2]
        self.coinb2 = params[3]
        self.merkle_branches = params[4]
        self.version = params[5]
        self.nbits = params[6]
        self.ntime = params[7]
        self.clean_jobs = params[8]
        
        # Calculate target based on nbits
        self._calculate_target()

    def _handle_difficulty(self, params):
        """Handle difficulty change notification"""
        self.difficulty = params[0]
        self._calculate_target()

    def _calculate_target(self):
        """Calculate target hash based on difficulty"""
        # Maximum target (difficulty 1)
        max_target = 0x00000000ffff0000000000000000000000000000000000000000000000000000
        
        # Adjust for current difficulty
        self.target = int(max_target / self.difficulty)

    def _get_message_id(self) -> int:
        """Get unique message ID"""
        self.message_id += 1
        return self.message_id

    def submit_share(self, job_id: str, extranonce2: str, ntime: str, nonce: str) -> bool:
        """Submit found share to pool"""
        submit_request = {
            "id": self._get_message_id(),
            "method": "mining.submit",
            "params": [
                self.pool_username,  # Use full worker name (account_name.worker_id)
                job_id,
                extranonce2,
                ntime,
                nonce
            ]
        }
        
        if not self._send_request(submit_request):
            return False
            
        try:
            response = self.responses.get(timeout=10)
            if response.get("error"):
                logging.error(f"Share submission error: {response['error']}")
                return False
                
            return response["result"]
            
        except Exception as e:
            logging.error(f"Share submission error: {e}")
            return False

    def disconnect(self):
        """Disconnect from pool"""
        self.is_connected = False
        if self.socket:
            try:
                self.socket.close()
            except:
                pass

class Miner:
    """
    Bitcoin miner implementation that manages mining threads and hashrate calculations.
    Coordinates with StratumClient for pool communication.
    """
    
    def __init__(self, host: str, port: int, wallet_address: str, worker_name: str = "worker1", password: str = "x"):
        """
        Initialize miner with connection details.
        
        Args:
            host: Mining pool hostname/IP
            port: Mining pool port number
            wallet_address: Bitcoin wallet address
            worker_name: Worker identifier (default: worker1)
            password: Worker password (default: x)
        """
        self.stratum_client = StratumClient(host, port, wallet_address, worker_name, password)
        self.is_mining = False
        self.mining_threads: List[threading.Thread] = []
        self.shares_submitted = 0
        self.shares_accepted = 0
        self.start_time: Optional[float] = None
        self.hashrate = 0
        self.total_hashes = 0
        
    def start_mining(self, num_threads: int = 1):
        """Start mining with specified number of threads"""
        if not self.stratum_client.connect():
            raise Exception("Failed to connect to mining pool")
            
        self.is_mining = True
        self.start_time = time.time()
        
        # Start mining threads
        for _ in range(num_threads):
            thread = threading.Thread(target=self._mining_thread, daemon=True)
            thread.start()
            self.mining_threads.append(thread)
            
        # Start hashrate calculator thread
        threading.Thread(target=self._calculate_hashrate, daemon=True).start()

    def _mining_thread(self):
        """Main mining loop"""
        while self.is_mining:
            if not self.stratum_client.job_id:
                time.sleep(0.1)
                continue
                
            # Generate unique extranonce2
            extranonce2 = hex(random.getrandbits(self.stratum_client.extranonce2_size * 8))[2:].zfill(self.stratum_client.extranonce2_size * 2)
            
            # Construct coinbase transaction
            coinbase = (self.stratum_client.coinb1 +
                       self.stratum_client.extranonce1 +
                       extranonce2 +
                       self.stratum_client.coinb2)
                       
            # Calculate merkle root
            merkle_root = hashlib.sha256(hashlib.sha256(binascii.unhexlify(coinbase)).digest()).digest()
            
            for branch in self.stratum_client.merkle_branches:
                merkle_root = hashlib.sha256(hashlib.sha256(merkle_root + binascii.unhexlify(branch)).digest()).digest()
                
            # Construct block header
            header = struct.pack("<L", int(self.stratum_client.version, 16))
            header += binascii.unhexlify(self.stratum_client.prevhash)[::-1]
            header += merkle_root[::-1]
            header += struct.pack("<L", int(self.stratum_client.ntime, 16))
            header += binascii.unhexlify(self.stratum_client.nbits)[::-1]
            
            # Try nonces
            for nonce in range(0, 0x100000000):
                if not self.is_mining:
                    return
                    
                self.total_hashes += 1
                
                work = header + struct.pack("<L", nonce)
                hash_result = hashlib.sha256(hashlib.sha256(work).digest()).digest()
                
                if int.from_bytes(hash_result[::-1], byteorder="big") <= self.stratum_client.target:
                    # Found a valid share
                    if self.stratum_client.submit_share(
                        self.stratum_client.job_id,
                        extranonce2,
                        self.stratum_client.ntime,
                        hex(nonce)[2:].zfill(8)
                    ):
                        self.shares_accepted += 1
                    self.shares_submitted += 1

    def _calculate_hashrate(self):
        """Calculate and update hashrate"""
        last_total = self.total_hashes
        while self.is_mining:
            time.sleep(1)
            current_total = self.total_hashes
            self.hashrate = current_total - last_total
            last_total = current_total

    def stop_mining(self):
        """Stop mining"""
        self.is_mining = False
        for thread in self.mining_threads:
            thread.join()
        self.mining_threads = []
        self.stratum_client.disconnect()

class MinerGUI(ctk.CTk):
    """
    Graphical user interface for the Bitcoin miner.
    Provides controls for mining operations and displays mining statistics.
    """
    
    def __init__(self):
        """Initialize the mining GUI with default settings."""
        super().__init__()
        
        # Basic setup
        ctk.set_appearance_mode("dark")
        self.title("xRaisen Bitcoin Miner")
        self.geometry("800x600")
        self.minsize(800, 600)
        
        # State variables
        self.is_mining = False
        self.is_optimizing = False
        self.is_scanning = False
        self.hashrate = 0
        
        # Create main layout
        self.setup_gui()
        
        # Start stats update
        self.update_stats()
        
        # Initialize logger
        self.logger = logging.getLogger("BTC_Miner.GUI")

    def setup_gui(self):
        # Main container
        container = ctk.CTkFrame(self)
        container.pack(fill="both", expand=True, padx=10, pady=10)
        
        # Left panel
        left_panel = ctk.CTkFrame(container, width=300)
        left_panel.pack(side="left", fill="y", padx=5, pady=5)
        left_panel.pack_propagate(False)
        
        # Stats
        stats_frame = ctk.CTkFrame(left_panel)
        stats_frame.pack(fill="x", padx=10, pady=5)
        
        self.hashrate_label = ctk.CTkLabel(
            stats_frame,
            text="Hashrate: 0 H/s",
            font=("Consolas", 14, "bold"),
            text_color="#00ff00"
        )
        self.hashrate_label.pack(pady=5)
        
        # Control buttons
        button_frame = ctk.CTkFrame(left_panel)
        button_frame.pack(fill="x", padx=10, pady=5)
        
        # Mining button
        self.mine_button = ctk.CTkButton(
            button_frame,
            text="▶ Start Mining",
            command=self.toggle_mining,
            font=("Consolas", 14, "bold"),
            fg_color="#00aa00",
            hover_color="#008800",
            height=40
        )
        self.mine_button.pack(pady=5, fill="x")
        
        # Optimize button
        self.optimize_button = ctk.CTkButton(
            button_frame,
            text="⚡ Optimize",
            command=self.toggle_optimize,
            font=("Consolas", 14, "bold"),
            fg_color="#007acc",
            hover_color="#005c99",
            height=40
        )
        self.optimize_button.pack(pady=5, fill="x")
        
        # Scan button
        self.scan_button = ctk.CTkButton(
            button_frame,
            text="🔍 Scan Wallets",
            command=self.toggle_scan,
            font=("Consolas", 14, "bold"),
            fg_color="#7a00cc",
            hover_color="#5c0099",
            height=40
        )
        self.scan_button.pack(pady=5, fill="x")
        
        # Console
        self.console = ConsoleDisplay(container)
        self.console.pack(side="left", fill="both", expand=True, padx=5, pady=5)
        
    def toggle_mining(self):
        self.is_mining = not self.is_mining
        if self.is_mining:
            self.mine_button.configure(
                text="⏹ Stop Mining",
                fg_color="#ff0000",
                hover_color="#cc0000"
            )
            self.console.add_line("Mining started", "success")
            threading.Thread(target=self._mining_process, daemon=True).start()
        else:
            self.mine_button.configure(
                text="▶ Start Mining",
                fg_color="#00aa00",
                hover_color="#008800"
            )
            self.console.add_line("Mining stopped", "info")
            
    def toggle_optimize(self):
        self.is_optimizing = not self.is_optimizing
        if self.is_optimizing:
            self.optimize_button.configure(
                text="⏹ Stop Optimize",
                fg_color="#ff0000",
                hover_color="#cc0000"
            )
            self.console.add_line("Optimization started", "info")
            threading.Thread(target=self._optimize_process, daemon=True).start()
        else:
            self.optimize_button.configure(
                text="⚡ Optimize",
                fg_color="#007acc",
                hover_color="#005c99"
            )
            self.console.add_line("Optimization stopped", "info")
            
    def toggle_scan(self):
        self.is_scanning = not self.is_scanning
        if self.is_scanning:
            self.scan_button.configure(
                text="⏹ Stop Scan",
                fg_color="#ff0000",
                hover_color="#cc0000"
            )
            self.console.add_line("Wallet scan started", "info")
            threading.Thread(target=self._scan_process, daemon=True).start()
        else:
            self.scan_button.configure(
                text="🔍 Scan Wallets",
                fg_color="#7a00cc",
                hover_color="#5c0099"
            )
            self.console.add_line("Wallet scan stopped", "info")
            
    def _mining_process(self):
        while self.is_mining:
            self.hashrate += 100000  # Simulated hashrate increase
            time.sleep(1)
            
    def _optimize_process(self):
        while self.is_optimizing:
            self.console.add_line("Optimizing...", "info")
            time.sleep(2)
            
    def _scan_process(self):
        while self.is_scanning:
            self.console.add_line("Scanning for wallets...", "info")
            time.sleep(2)
            
    def update_stats(self):
        if self.is_mining:
            if self.hashrate >= 1_000_000:
                text = f"Hashrate: {self.hashrate/1_000_000:.2f} MH/s"
            elif self.hashrate >= 1_000:
                text = f"Hashrate: {self.hashrate/1_000:.2f} KH/s"
            else:
                text = f"Hashrate: {self.hashrate:.2f} H/s"
            self.hashrate_label.configure(text=text)
        self.after(1000, self.update_stats)

class ConsoleDisplay(ctk.CTkTextbox):
    """
    Custom console widget for displaying mining operation logs.
    Implements a circular buffer for log messages with color coding.
    """
    
    def __init__(self, *args, **kwargs):
        """Initialize console display with custom styling."""
        super().__init__(*args, **kwargs)
        
        self.configure(
            font=("Consolas", 12),
            text_color="#00ff00",
            fg_color="#000000",
            border_color="#00ff00",
            border_width=2
        )
        
        self.line_queue = deque(maxlen=100)
        
        # Configure colors for different message types
        self.tag_config("error", foreground="#ff3333")    # Red for errors
        self.tag_config("success", foreground="#33ff33")  # Green for success
        self.tag_config("info", foreground="#00ffff")     # Cyan for info
        self.tag_config("warning", foreground="#ffff00")  # Yellow for warnings
        
        self.logger = logging.getLogger("BTC_Miner.Console")

    def add_line(self, text: str, tag: str = "info"):
        timestamp = datetime.now().strftime("%H:%M:%S")
        line = f"[{timestamp}] {text}\n"
        
        self.line_queue.append((line, tag))
        
        self.configure(state="normal")
        self.delete("1.0", "end")
        
        for line_text, line_tag in self.line_queue:
            self.insert("end", line_text, line_tag)
        
        self.configure(state="disabled")
        self.see("end")

if __name__ == "__main__":
    try:
        logger.info("Starting Bitcoin Miner application")
        app = MinerGUI()
        app.mainloop()
    except Exception as e:
        logger.critical(f"Application crashed: {e}", exc_info=True)