"""
FastAPI server for Bitcoin mining dashboard - Complete Integration
"""
from fastapi import FastAPI, HTTPException, BackgroundTasks
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
import threading
import time
import json
import asyncio
from typing import Dict, Optional, List
import logging
import sys
import os
# Add the current directory to Python path to import your miner
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
app = FastAPI(title="Bitcoin Mining Dashboard", version="1.0.0")
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Mount static files
app.mount("/static", StaticFiles(directory="static"), name="static")
# Global mining state
class MiningState:
def __init__(self):
self.is_mining = False
self.miner_instance = None
self.mining_thread = None
self.stats = {
"status": "Stopped",
"hashrate": "0 H/s",
"total_hashes": "0",
"blocks_found": "0",
"best_hash": "None",
"difficulty": "0",
"network_difficulty": "0",
"block_alert": "Ready to start mining",
"mining_time": "0s",
"cores_active": "0",
"wallet_address": "1Ks4WtCEK96BaBF7HSuCGt3rEpVKPqcJKf"
}
self.start_time = None
mining_state = MiningState()
@app.get("/", response_class=HTMLResponse)
async def get_index():
"""Serve the dashboard HTML"""
try:
return FileResponse("static/index.html")
except:
# Fallback minimal HTML
return """
Bitcoin Mining Dashboard
⛏️ Bitcoin Mining Dashboard
Ready to start mining
"""
def mining_worker(duration=None):
"""Worker function to run mining in background thread"""
try:
# Import and initialize the miner
from parallel_miner_v3 import ParallelMiner
mining_state.miner_instance = ParallelMiner(
num_cores=7,
wallet_address="1Ks4WtCEK96BaBF7HSuCGt3rEpVKPqcJKf"
)
mining_state.start_time = time.time()
# Start mining
mining_state.miner_instance.start_mining(duration=duration)
except Exception as e:
logging.error(f"Mining worker error: {e}")
finally:
mining_state.is_mining = False
mining_state.mining_thread = None
@app.post("/start_mining")
async def start_mining(background_tasks: BackgroundTasks):
"""Start the mining process"""
global mining_state
if mining_state.is_mining:
raise HTTPException(status_code=400, detail="Mining is already running")
try:
mining_state.is_mining = True
mining_state.stats["status"] = "Starting..."
mining_state.stats["block_alert"] = "🔄 Starting mining process..."
# Start mining in background thread
mining_state.mining_thread = threading.Thread(
target=mining_worker,
kwargs={"duration": None} # Mine indefinitely
)
mining_state.mining_thread.daemon = True
mining_state.mining_thread.start()
logging.info("✅ Mining started via API")
return {"message": "Mining started successfully", "status": "started"}
except Exception as e:
mining_state.is_mining = False
logging.error(f"Error starting mining: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/stop_mining")
async def stop_mining():
"""Stop the mining process"""
global mining_state
if not mining_state.is_mining:
raise HTTPException(status_code=400, detail="Mining is not running")
try:
if mining_state.miner_instance:
mining_state.miner_instance.mining = False
mining_state.is_mining = False
# Log final statistics
if hasattr(mining_state.miner_instance, 'total_hashes'):
total_hashes = mining_state.miner_instance.total_hashes
blocks_found = mining_state.miner_instance.blocks_found
mining_time = time.time() - mining_state.start_time
logging.info("\n" + "="*50)
logging.info("⛏️ FINAL MINING STATISTICS")
logging.info("="*50)
logging.info(f"⏱️ Total mining time: {mining_time:.2f} seconds")
logging.info(f"🔢 Total hashes: {total_hashes:,}")
logging.info(f"💰 Blocks found: {blocks_found}")
logging.info(f"⚡ Average hash rate: {total_hashes/max(mining_time,1)/1000:.2f} KH/s")
if hasattr(mining_state.miner_instance, 'cores'):
for core_idx, core in enumerate(mining_state.miner_instance.cores):
logging.info(f"🔩 Core {core_idx}: {core.total_hashes:,} hashes")
logging.info("="*50)
mining_state.stats["status"] = "Stopped"
mining_state.stats["block_alert"] = "🛑 Mining stopped"
return {"message": "Mining stopped successfully", "status": "stopped"}
raise HTTPException(status_code=400, detail="No active mining instance")
except Exception as e:
logging.error(f"Error stopping mining: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.get("/get_stats")
async def get_mining_stats():
"""Get current mining statistics"""
global mining_state
if not mining_state.is_mining or not mining_state.miner_instance:
return mining_state.stats
try:
miner = mining_state.miner_instance
# Calculate mining time
mining_time = time.time() - mining_state.start_time
time_str = f"{int(mining_time//3600)}h {int((mining_time%3600)//60)}m {int(mining_time%60)}s"
# Create block alert message
if miner.blocks_found > 0:
block_alert = f"🎉 FOUND {miner.blocks_found} BLOCK(S)! 🎉"
if miner.best_hash:
block_alert += f" Hash: {miner.best_hash.hex()[:16]}..."
else:
progress = (miner.best_hash_difficulty / max(miner.network_difficulty, 1)) * 100 if miner.best_hash_difficulty else 0
block_alert = f"⛏️ Mining... Progress: {progress:.8f}%"
# Update stats
mining_state.stats.update({
"status": "Running",
"hashrate": f"{miner.current_hashrate/1000:.2f} KH/s",
"total_hashes": f"{miner.total_hashes:,}",
"blocks_found": str(miner.blocks_found),
"best_hash": miner.best_hash.hex()[:32] + "..." if miner.best_hash else "None",
"difficulty": f"{miner.best_hash_difficulty:.8f}" if miner.best_hash_difficulty else "0",
"network_difficulty": f"{miner.network_difficulty:,.2f}" if hasattr(miner, 'network_difficulty') else "0",
"block_alert": block_alert,
"mining_time": time_str,
"cores_active": f"{len(miner.cores)}" if hasattr(miner, 'cores') else "0"
})
return mining_state.stats
except Exception as e:
logging.error(f"Error getting mining stats: {e}")
return {
"status": "Error",
"hashrate": "0 H/s",
"total_hashes": "0",
"blocks_found": "0",
"best_hash": "Error",
"difficulty": "0",
"block_alert": f"Error: {str(e)}",
"mining_time": "0s",
"cores_active": "0"
}
@app.get("/get_detailed_stats")
async def get_detailed_stats():
"""Get detailed mining statistics including per-core info"""
global mining_state
if not mining_state.is_mining or not mining_state.miner_instance:
return {"error": "Mining not active"}
try:
miner = mining_state.miner_instance
detailed_stats = {
"overview": {
"status": "Running",
"total_hashes": miner.total_hashes,
"blocks_found": miner.blocks_found,
"current_hashrate": miner.current_hashrate,
"best_hash_difficulty": miner.best_hash_difficulty,
"network_difficulty": miner.network_difficulty,
"mining_time": time.time() - mining_state.start_time
},
"cores": []
}
if hasattr(miner, 'cores'):
for core in miner.cores:
core_info = {
"core_id": core.core_id,
"total_hashes": core.total_hashes,
"blocks_found": core.blocks_found,
"units": []
}
for unit in core.units:
core_info["units"].append({
"unit_id": unit.unit_id,
"total_hashes": unit.total_hashes,
"blocks_found": unit.blocks_found
})
detailed_stats["cores"].append(core_info)
return detailed_stats
except Exception as e:
return {"error": str(e)}
@app.get("/health")
async def health_check():
"""Health check endpoint"""
return {
"status": "healthy",
"mining_active": mining_state.is_mining,
"timestamp": time.time()
}
# Background task to periodically update stats
@app.on_event("startup")
async def startup_event():
"""Initialize on startup"""
logging.info("🚀 Bitcoin Mining Dashboard starting up...")
@app.on_event("shutdown")
async def shutdown_event():
"""Cleanup on shutdown"""
global mining_state
if mining_state.is_mining and mining_state.miner_instance:
mining_state.miner_instance.mining = False
mining_state.is_mining = False
logging.info("🛑 Bitcoin Mining Dashboard shutting down...")
if __name__ == "__main__":
uvicorn.run(
"app:app",
host="0.0.0.0",
port=7868,
reload=False,
log_level="info"
)