Spaces:
Sleeping
Sleeping
File size: 4,920 Bytes
b7c2c9d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | """
QCrypt RNG API - Quantum Endpoints
API endpoints for quantum backend operations and analysis
"""
from fastapi import APIRouter, HTTPException
from app.quantum.qrng import get_quantum_rng
from app.api.v2.models.responses import (
EntropyStatusResponse,
SystemStatsResponse,
ResponseStatus
)
from app.utils.logging import logger
from app.config import settings
import time
router = APIRouter()
@router.get("/entropy", response_model=EntropyStatusResponse)
async def get_entropy_status() -> EntropyStatusResponse:
"""
Get entropy pool status
Returns comprehensive entropy analysis including:
- Shannon entropy
- Min entropy
- Statistical tests (Chi-square, autocorrelation)
- Bit balance
- Overall health status
"""
try:
qrng = get_quantum_rng()
analysis = qrng.analyze_entropy()
return EntropyStatusResponse(
status=ResponseStatus.SUCCESS,
request_id=f"req_{int(time.time()*1000000)}",
data={
"shannon_entropy": round(analysis.shannon_entropy, 4),
"min_entropy": round(analysis.min_entropy, 4),
"chi_square_p_value": round(analysis.chi_square_p_value, 4),
"autocorrelation": round(analysis.autocorrelation, 4),
"bit_balance": round(analysis.bit_balance, 4),
"health_status": analysis.health_status,
"pool_size": analysis.pool_size,
"passed_tests": analysis.passed_tests
}
)
except Exception as e:
logger.error(f"Error getting entropy status: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.get("/stats", response_model=SystemStatsResponse)
async def get_system_stats() -> SystemStatsResponse:
"""
Get system statistics
Returns operational metrics including:
- Total bytes generated
- Generation count
- Average generation time
- Backend status
"""
try:
qrng = get_quantum_rng()
stats = qrng.get_statistics()
# Calculate uptime (simplified - would track actual start time in production)
import os
uptime = time.time() - os.path.getmtime(__file__)
return SystemStatsResponse(
status=ResponseStatus.SUCCESS,
request_id=f"req_{int(time.time()*1000000)}",
data={
"total_bytes_generated": stats["total_bytes_generated"],
"total_generations": stats["total_generations"],
"average_generation_time_ms": round(stats["average_generation_time_ms"], 2),
"entropy_pool_size": stats["entropy_pool_size"],
"backend": stats["backend"],
"backend_status": stats["backend_status"],
"uptime_seconds": int(uptime),
"api_version": settings.app_version
}
)
except Exception as e:
logger.error(f"Error getting system stats: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.post("/reseed")
async def reseed_entropy_pool():
"""
Reseed entropy pool
Forces regeneration of the entropy pool with fresh quantum measurements.
This operation may take several seconds.
"""
try:
qrng = get_quantum_rng()
# Clear existing pool
qrng.entropy_pool = []
# Generate fresh entropy
for _ in range(100):
await qrng.generate_bytes(32, 16, "hex")
return {
"status": "success",
"message": "Entropy pool reseeded",
"new_pool_size": len(qrng.entropy_pool)
}
except Exception as e:
logger.error(f"Error reseeding entropy pool: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error")
@router.get("/backend")
async def get_backend_info():
"""
Get quantum backend information
Returns details about the current quantum backend configuration
"""
try:
return {
"backend": settings.quantum_backend,
"configuration": settings.quantum_backend_config,
"available_backends": ["qrisp_simulator", "ibm_quantum", "iqm_quantum", "rigetti"],
"max_qubits": settings.max_qubits,
"default_qubits": settings.default_qubits,
"features": {
"superposition": True,
"entanglement": True,
"measurement": True,
"post_processing": True,
"entropy_validation": True
}
}
except Exception as e:
logger.error(f"Error getting backend info: {str(e)}", exc_info=True)
raise HTTPException(status_code=500, detail="Internal server error") |