xmrig_server.py

#!/usr/bin/env python3
"""

XMRig FastAPI Server - Web interface for RandomX benchmarking

Spawns xmrig_launcher.py as subprocess for mining operations

"""

import os
import sys
import subprocess
import platform
import ctypes
from pathlib import Path
from typing import Optional
from datetime import datetime
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
import uvicorn
import threading

# ==================== Admin Detection ====================

def is_admin():
    """Check if running with admin privileges"""
    try:
        return ctypes.windll.shell.IsUserAnAdmin()
    except:
        return False


# ==================== Pydantic Models ====================

class BenchmarkRequest(BaseModel):
    benchmark: str = "1M"
    threads: Optional[int] = None


# ==================== Mining Manager Class ====================

class MiningManager:
    def __init__(self, base_dir=None):
        if base_dir is None:
            base_dir = Path(__file__).parent.absolute()
        
        self.base_dir = Path(base_dir)
        self.launcher_script = self.base_dir / "xmrig_launcher.py"
        self.process = None
        self.benchmark_config = {}
        self.logs = []
        self.log_reader_thread = None
        
    def start_mining(self, benchmark="1M", threads=None):
        """Start mining by spawning xmrig_launcher.py"""
        if self.process and self.process.poll() is None:
            raise HTTPException(status_code=400, detail="Mining already running")
        
        if threads is None:
            threads = os.cpu_count() or 4
        
        # Clear previous logs
        self.logs = []
        
        # Build command to run launcher script
        cmd = [
            sys.executable,
            str(self.launcher_script),
            "--bench", benchmark,
            "--threads", str(threads)
        ]
        
        # Store config
        self.benchmark_config = {
            "benchmark": benchmark,
            "threads": threads,
            "started_at": datetime.now().isoformat(),
            "status": "running"
        }
        
        try:
            # Start process with output capture
            self.process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
                text=True,
                bufsize=1,
                universal_newlines=True
            )
            
            # Start log reader thread
            self.log_reader_thread = threading.Thread(target=self._read_logs, daemon=True)
            self.log_reader_thread.start()
            
            return True
        except Exception as e:
            raise HTTPException(status_code=500, detail=f"Failed to start mining: {e}")
    
    def _read_logs(self):
        """Read logs from process in real-time (daemon thread)"""
        try:
            if self.process and self.process.stdout:
                # This runs in a daemon thread, so blocking readline is fine
                while self.process and self.process.poll() is None:
                    try:
                        line = self.process.stdout.readline()
                        if line:
                            log_entry = {
                                "timestamp": datetime.now().isoformat(),
                                "message": line.rstrip()
                            }
                            self.logs.append(log_entry)
                            print(line.rstrip())
                        else:
                            # EOF reached
                            break
                    except Exception as e:
                        print(f"[!] Error reading line: {e}")
                        break
        except Exception as e:
            print(f"[!] Error in log reader: {e}")
    
    def stop_mining(self):
        """Stop running mining"""
        if self.process and self.process.poll() is None:
            self.process.terminate()
            try:
                self.process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                self.process.kill()
            self.benchmark_config["status"] = "stopped"
            return True
        return False
    
    def is_running(self):
        """Check if mining is running"""
        return self.process and self.process.poll() is None
    
    def get_xmrig_status(self):
        """Check if launcher script is available"""
        return {
            "launcher_available": self.launcher_script.exists(),
            "launcher_path": str(self.launcher_script)
        }


# Global manager instance
manager = MiningManager()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    print("\n[*] Server starting...")
    
    # Check if launcher script exists
    if not manager.launcher_script.exists():
        print(f"[!] Warning: xmrig_launcher.py not found at {manager.launcher_script}")
    else:
        print(f"[+] Launcher script found")
    
    print("[+] Server ready - click Start Mining button to begin")
    
    yield
    
    # Shutdown
    print("\n[*] Server shutting down...")
    manager.stop_mining()


# ==================== FastAPI App ====================

app = FastAPI(
    title="XMRig Mining Server",
    description="FastAPI server for RandomX benchmarking with XMRig launcher",
    version="1.0.0",
    lifespan=lifespan
)

# Mount static files (if directory exists)
static_dir = Path(__file__).parent / "static"
if static_dir.exists():
    app.mount("/static", StaticFiles(directory="static"), name="static")


@app.get("/api")
async def api_info():
    """API endpoints info"""
    return {
        "name": "XMRig Mining Server",
        "version": "1.0.0",
        "endpoints": {
            "GET /health": "Health check",
            "GET /status": "Get system and mining status",
            "POST /mining/start": "Start mining with XMRig launcher",
            "POST /mining/stop": "Stop mining",
            "GET /mining/status": "Get mining status",
            "GET /logs": "Get mining logs"
        }
    }


@app.get("/")
async def root():
    """Root endpoint - serve static HTML UI"""
    try:
        with open("static/index.html", "r", encoding="utf-8") as f:
            return HTMLResponse(content=f.read())
    except FileNotFoundError:
        return HTMLResponse(
            content="<h1>404</h1><p>static/index.html not found</p>",
            status_code=404
        )


@app.get("/health")
async def health():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat(),
        "admin": is_admin()
    }


@app.get("/status")
async def status():
    """Get system and mining status"""
    return {
        "system": {
            "platform": platform.system(),
            "admin": is_admin(),
            "cpu_count": os.cpu_count() or 4
        },
        "xmrig": manager.get_xmrig_status(),
        "mining": {
            "running": manager.is_running(),
            "config": manager.benchmark_config if manager.benchmark_config else None
        }
    }


@app.post("/mining/start")
async def start_mining(request: BenchmarkRequest):
    """Start mining using xmrig_launcher.py"""
    if manager.is_running():
        raise HTTPException(status_code=400, detail="Mining already running")
    
    manager.start_mining(request.benchmark, request.threads)
    
    return {
        "status": "started",
        "benchmark": request.benchmark,
        "threads": request.threads or os.cpu_count() or 4,
        "admin": is_admin(),
        "admin_warning": "Run with admin for 2-3x better performance" if not is_admin() else None
    }


@app.post("/mining/stop")
async def stop_mining():
    """Stop running mining"""
    if manager.stop_mining():
        return {"status": "stopped"}
    else:
        raise HTTPException(status_code=400, detail="No mining running")


@app.get("/mining/status")
async def mining_status():
    """Get mining status"""
    return {
        "running": manager.is_running(),
        "config": manager.benchmark_config if manager.benchmark_config else None,
        "pid": manager.process.pid if manager.process else None
    }


@app.get("/logs")
async def get_logs():
    """Get mining logs"""
    return {
        "logs": manager.logs[-100:] if manager.logs else [],  # Last 100 logs
        "total_logs": len(manager.logs)
    }


# ==================== Main ====================

def main():
    """Run the FastAPI server"""
    print("\n" + "="*60)
    print("XMRig Mining Server")
    print("="*60)
    print("[*] Starting server on http://127.0.0.1:8000")
    print("[*] Web UI: http://127.0.0.1:8000/")
    print("[*] API docs: http://127.0.0.1:8000/docs")
    
    if is_admin():
        print("[+] Running as Administrator ✓")
    else:
        print("[!] NOT running as Administrator")
        print("    → Run as Admin for 2-3x better performance")
    
    print("="*60 + "\n")
    
    uvicorn.run(app, host="127.0.0.1", port=8000, log_level="info")


if __name__ == "__main__":
    main()