diff --git a/AI/microsoft_phi-silica-3.6_v1/azure_foundry_deploy.py b/AI/microsoft_phi-silica-3.6_v1/azure_foundry_deploy.py index dd4a54e70..40bbeb3f8 100644 --- a/AI/microsoft_phi-silica-3.6_v1/azure_foundry_deploy.py +++ b/AI/microsoft_phi-silica-3.6_v1/azure_foundry_deploy.py @@ -1,4 +1,4 @@ -""" +r""" Azure AI Foundry (Managed Online Endpoint) deployment for Phi-3.x + LoRA adapter. - Registers LoRA adapter as a model asset (if given a local path) - Creates a managed online endpoint and GPU deployment diff --git a/AI/microsoft_phi-silica-3.6_v1/python mcp.py b/AI/microsoft_phi-silica-3.6_v1/python mcp.py index 2ab80f288..6b1724e8b 100644 --- a/AI/microsoft_phi-silica-3.6_v1/python mcp.py +++ b/AI/microsoft_phi-silica-3.6_v1/python mcp.py @@ -32,7 +32,6 @@ def __init__(self): endpoint = "https://models.github.ai/inference", credential = AzureKeyCredential(os.environ["GITHUB_TOKEN"]), api_version = "2024-12-01-preview", - api_version = "2024-08-01-preview", ) async def connect_stdio_server(self, server_id: str, command: str, args: list[str], env: Dict[str, str]): diff --git a/AI/microsoft_phi-silica-3.6_v1/scripts/train_lora.py b/AI/microsoft_phi-silica-3.6_v1/scripts/train_lora.py index 6322650dc..284540d23 100644 --- a/AI/microsoft_phi-silica-3.6_v1/scripts/train_lora.py +++ b/AI/microsoft_phi-silica-3.6_v1/scripts/train_lora.py @@ -680,6 +680,8 @@ def on_evaluate(self, args, state, control, metrics=None, **kwargs): pass except Exception: # Swallow unexpected callback errors + pass + is_streaming = is_iterable_dataset(train_ds) # Remove 'messages' column so only tokenized output is kept # Note: IterableDataset.map() has limited parameter support compared to Dataset.map() diff --git a/quantum-ai/production/banknote_api.py b/quantum-ai/production/banknote_api.py index 61640b674..51e45be0f 100644 --- a/quantum-ai/production/banknote_api.py +++ b/quantum-ai/production/banknote_api.py @@ -1,618 +1,309 @@ - -REST API for quantum-powered banknote fraud detection. -Achieves 100% accuracy on validation data. - -Endpoints: - POST /api/predict - Classify single banknote - POST /api/predict_batch - Classify multiple banknotes - GET /api/health - Service health check - GET /api/model_info - Model metadata - -Usage: - python banknote_api.py - - # Test prediction: - curl -X POST http://localhost:8080/api/predict \ - -H "Content-Type: application/json" \ - -d '{"features": [3.5, 0.5, -1.2, 0.8]}' - -Author: Quantum AI System -Date: November 16, 2025 -""" - -from flask import Flask, request, jsonify -from flask_cors import CORS -import numpy as np -import torch -import joblib -from pathlib import Path -import sys -import json -from datetime import datetime - -# Add src to path -src_path = Path(__file__).parent.parent / "src" -sys.path.insert(0, str(src_path)) - -from src.hybrid_qnn import HybridQNN - -app = Flask(__name__) -CORS(app) # Enable CORS for web clients - -# Global model variables -model = None -scaler = None -model_metadata = {} - -def load_model(): - """Load trained model and preprocessors""" - global model, scaler, model_metadata - - model_dir = Path(__file__).parent.parent / "results" - - print("šŸ”„ Loading model artifacts...") - - # Load scaler - scaler_path = model_dir / "custom_scaler.pkl" - if not scaler_path.exists(): - raise FileNotFoundError(f"Scaler not found: {scaler_path}") - scaler = joblib.load(scaler_path) - print(f" āœ… Loaded scaler from {scaler_path}") - - # Create model architecture - model = HybridQNN( - input_dim=4, - hidden_dim=16, - n_qubits=4, - n_quantum_layers=2, - output_dim=2, - dropout=0.2 - ) - - # Load trained weights safely with weights_only=True to prevent arbitrary code execution - model_path = model_dir / "custom_model.pt" - if not model_path.exists(): - raise FileNotFoundError(f"Model not found: {model_path}") - model.load_state_dict(torch.load(model_path, weights_only=True)) - model.eval() # Set to evaluation mode - print(f" āœ… Loaded model weights from {model_path}") - - # Load metadata - summary_path = model_dir / "custom_training_summary.json" - if summary_path.exists(): - with open(summary_path) as f: - model_metadata = json.load(f) - print(f" āœ… Loaded metadata from {summary_path}") - - print("āœ… Model loaded successfully!") - print(f" Validation Accuracy: {model_metadata.get('metrics', {}).get('val_acc_best', 'N/A')}") - -@app.route('/api/health', methods=['GET']) -def health_check(): - """Health check endpoint""" - return jsonify({ - 'status': 'healthy', - 'service': 'banknote-fraud-detector', - 'version': '1.0.0', - 'model_loaded': model is not None, - 'timestamp': datetime.now().isoformat() - }) - -@app.route('/api/model_info', methods=['GET']) -def model_info(): - """Return model information""" - return jsonify({ - 'model': 'Hybrid Quantum-Classical Neural Network', - 'task': 'Binary classification: Genuine vs Forged banknotes', - 'architecture': { - 'qubits': 4, - 'quantum_layers': 2, - 'hidden_dim': 16, - 'output_classes': 2 - }, - 'performance': { - 'validation_accuracy': model_metadata.get('metrics', {}).get('val_acc_best', 'N/A'), - 'training_loss': model_metadata.get('metrics', {}).get('train_loss_last', 'N/A') - }, - 'features': { - 'names': ['variance', 'skewness', 'curtosis', 'entropy'], - 'count': 4, - 'preprocessing': 'StandardScaler normalization' - }, - 'training': { - 'dataset': model_metadata.get('dataset', 'banknote authentication'), - 'samples': model_metadata.get('data', {}).get('n_train', 'N/A'), - 'epochs': model_metadata.get('params', {}).get('epochs', 'N/A') - } - }) - -@app.route('/api/predict', methods=['POST']) -def predict(): - """Classify a single banknote""" - try: - data = request.get_json() - - # Validate input - if 'features' not in data: - return jsonify({ - 'error': 'Missing required field: features', - 'expected_format': { - 'features': [variance, skewness, curtosis, entropy] - } - }), 400 - - features = np.array(data['features']) - - # Validate feature count - if features.shape[0] != 4: - return jsonify({ - 'error': f'Expected 4 features, got {features.shape[0]}', - 'features_required': ['variance', 'skewness', 'curtosis', 'entropy'] - }), 400 - - # Preprocess - features_scaled = scaler.transform(features.reshape(1, -1)) - features_tensor = torch.FloatTensor(features_scaled) - - # Predict - with torch.no_grad(): - outputs = model(features_tensor) - probabilities = torch.softmax(outputs, dim=1) - prediction = torch.argmax(outputs, dim=1).item() - - # Interpret result - label = "GENUINE" if prediction == 0 else "FORGED" - confidence = float(probabilities[0][prediction]) - - return jsonify({ - 'prediction': label, - 'confidence': round(confidence, 4), - 'probabilities': { - 'genuine': round(float(probabilities[0][0]), 4), - 'forged': round(float(probabilities[0][1]), 4) - }, - 'features_received': features.tolist(), - 'timestamp': datetime.now().isoformat() - }) - - except Exception as e: - return jsonify({ - 'error': str(e), - 'type': type(e).__name__ - }), 500 - -@app.route('/api/predict_batch', methods=['POST']) -def predict_batch(): - """Classify multiple banknotes""" - try: - data = request.get_json() - - # Validate input - if 'batch' not in data: - return jsonify({ - 'error': 'Missing required field: batch', - 'expected_format': { - 'batch': [ - {'features': [variance, skewness, curtosis, entropy]}, - {'features': [variance, skewness, curtosis, entropy]} - ] - } - }), 400 - - batch = data['batch'] - results = [] - - for i, item in enumerate(batch): - if 'features' not in item: - results.append({ - 'index': i, - 'error': 'Missing features field' - }) - continue - - features = np.array(item['features']) - - if features.shape[0] != 4: - results.append({ - 'index': i, - 'error': f'Expected 4 features, got {features.shape[0]}' - }) - continue - - # Preprocess and predict - features_scaled = scaler.transform(features.reshape(1, -1)) - features_tensor = torch.FloatTensor(features_scaled) - - with torch.no_grad(): - outputs = model(features_tensor) - probabilities = torch.softmax(outputs, dim=1) - prediction = torch.argmax(outputs, dim=1).item() - - label = "GENUINE" if prediction == 0 else "FORGED" - confidence = float(probabilities[0][prediction]) - - results.append({ - 'index': i, - 'prediction': label, - 'confidence': round(confidence, 4), - 'probabilities': { - 'genuine': round(float(probabilities[0][0]), 4), - 'forged': round(float(probabilities[0][1]), 4) - } - }) - - # Summary statistics - valid_predictions = [r for r in results if 'prediction' in r] - genuine_count = sum(1 for r in valid_predictions if r['prediction'] == 'GENUINE') - forged_count = len(valid_predictions) - genuine_count - - return jsonify({ - 'results': results, - 'summary': { - 'total_processed': len(batch), - 'successful': len(valid_predictions), - 'errors': len(batch) - len(valid_predictions), - 'genuine_count': genuine_count, - 'forged_count': forged_count - }, - 'timestamp': datetime.now().isoformat() - }) - - except Exception as e: - return jsonify({ - 'error': str(e), - 'type': type(e).__name__ - }), 500 - -def main(): - """Start the API server""" - print("=" * 70) - print(" BANKNOTE FRAUD DETECTOR - PRODUCTION API") - print("=" * 70) - print("\nšŸ”¬ Model: Hybrid Quantum-Classical Neural Network") - print("šŸŽÆ Accuracy: 100% on validation data") - print("āš” Backend: Quantum simulation (PennyLane)") - - # Load model - try: - load_model() - except Exception as e: - print(f"\nāŒ Failed to load model: {e}") - print("\nšŸ’” Make sure you've trained the model first:") - print(" python train_custom_dataset.py --preset banknote --epochs 25") - sys.exit(1) - - print("\n" + "=" * 70) - print(" šŸš€ STARTING API SERVER") - print("=" * 70) - print("\nšŸ“” Endpoints:") - print(" POST http://localhost:8080/api/predict") - print(" POST http://localhost:8080/api/predict_batch") - print(" GET http://localhost:8080/api/health") - print(" GET http://localhost:8080/api/model_info") - - print("\nšŸ“ Example curl command:") - print(' curl -X POST http://localhost:8080/api/predict \\') - print(' -H "Content-Type: application/json" \\') - print(' -d \'{"features": [3.5, 0.5, -1.2, 0.8]}\'') - - print("\nā³ Server starting on http://localhost:8080 ...") - print(" Press Ctrl+C to stop") - print("=" * 70 + "\n") - - # Start server - default to localhost for security - # Use BANKNOTE_API_HOST environment variable to override if needed - host = os.environ.get('BANKNOTE_API_HOST', '127.0.0.1') - app.run(host=host, port=8080, debug=False) - -if __name__ == '__main__': - main() -""" -Banknote Fraud Detector - Production API - -REST API for quantum-powered banknote fraud detection. -Achieves 100% accuracy on validation data. - -Endpoints: - POST /api/predict - Classify single banknote - POST /api/predict_batch - Classify multiple banknotes - GET /api/health - Service health check - GET /api/model_info - Model metadata - -Usage: - python banknote_api.py - - # Test prediction: - curl -X POST http://localhost:8080/api/predict \ - -H "Content-Type: application/json" \ - -d '{"features": [3.5, 0.5, -1.2, 0.8]}' - -Author: Quantum AI System -Date: November 16, 2025 -""" - -from flask import Flask, request, jsonify -from flask_cors import CORS -import numpy as np -import torch -import joblib -from pathlib import Path -import sys -import json -from datetime import datetime - -# Add src to path -src_path = Path(__file__).parent.parent / "src" -sys.path.insert(0, str(src_path)) - -from src.hybrid_qnn import HybridQNN - -app = Flask(__name__) -CORS(app) # Enable CORS for web clients - -# Global model variables -model = None -scaler = None -model_metadata = {} - -def load_model(): - """Load trained model and preprocessors""" - global model, scaler, model_metadata - - model_dir = Path(__file__).parent.parent / "results" - - print("šŸ”„ Loading model artifacts...") - - # Load scaler - scaler_path = model_dir / "custom_scaler.pkl" - if not scaler_path.exists(): - raise FileNotFoundError(f"Scaler not found: {scaler_path}") - scaler = joblib.load(scaler_path) - print(f" āœ… Loaded scaler from {scaler_path}") - - # Create model architecture - model = HybridQNN( - input_dim=4, - hidden_dim=16, - n_qubits=4, - n_quantum_layers=2, - output_dim=2, - dropout=0.2 - ) - - # Load trained weights - model_path = model_dir / "custom_model.pt" - if not model_path.exists(): - raise FileNotFoundError(f"Model not found: {model_path}") - model.load_state_dict(torch.load(model_path, weights_only=True)) - model.eval() # Set to evaluation mode - print(f" āœ… Loaded model weights from {model_path}") - - # Load metadata - summary_path = model_dir / "custom_training_summary.json" - if summary_path.exists(): - with open(summary_path) as f: - model_metadata = json.load(f) - print(f" āœ… Loaded metadata from {summary_path}") - - print("āœ… Model loaded successfully!") - print(f" Validation Accuracy: {model_metadata.get('metrics', {}).get('val_acc_best', 'N/A')}") - -@app.route('/api/health', methods=['GET']) -def health_check(): - """Health check endpoint""" - return jsonify({ - 'status': 'healthy', - 'service': 'banknote-fraud-detector', - 'version': '1.0.0', - 'model_loaded': model is not None, - 'timestamp': datetime.now().isoformat() - }) - -@app.route('/api/model_info', methods=['GET']) -def model_info(): - """Return model information""" - return jsonify({ - 'model': 'Hybrid Quantum-Classical Neural Network', - 'task': 'Binary classification: Genuine vs Forged banknotes', - 'architecture': { - 'qubits': 4, - 'quantum_layers': 2, - 'hidden_dim': 16, - 'output_classes': 2 - }, - 'performance': { - 'validation_accuracy': model_metadata.get('metrics', {}).get('val_acc_best', 'N/A'), - 'training_loss': model_metadata.get('metrics', {}).get('train_loss_last', 'N/A') - }, - 'features': { - 'names': ['variance', 'skewness', 'curtosis', 'entropy'], - 'count': 4, - 'preprocessing': 'StandardScaler normalization' - }, - 'training': { - 'dataset': model_metadata.get('dataset', 'banknote authentication'), - 'samples': model_metadata.get('data', {}).get('n_train', 'N/A'), - 'epochs': model_metadata.get('params', {}).get('epochs', 'N/A') - } - }) - -@app.route('/api/predict', methods=['POST']) -def predict(): - """Classify a single banknote""" - try: - data = request.get_json() - - # Validate input - if 'features' not in data: - return jsonify({ - 'error': 'Missing required field: features', - 'expected_format': { - 'features': [variance, skewness, curtosis, entropy] - } - }), 400 - - features = np.array(data['features']) - - # Validate feature count - if features.shape[0] != 4: - return jsonify({ - 'error': f'Expected 4 features, got {features.shape[0]}', - 'features_required': ['variance', 'skewness', 'curtosis', 'entropy'] - }), 400 - - # Preprocess - features_scaled = scaler.transform(features.reshape(1, -1)) - features_tensor = torch.FloatTensor(features_scaled) - - # Predict - with torch.no_grad(): - outputs = model(features_tensor) - probabilities = torch.softmax(outputs, dim=1) - prediction = torch.argmax(outputs, dim=1).item() - - # Interpret result - label = "GENUINE" if prediction == 0 else "FORGED" - confidence = float(probabilities[0][prediction]) - - return jsonify({ - 'prediction': label, - 'confidence': round(confidence, 4), - 'probabilities': { - 'genuine': round(float(probabilities[0][0]), 4), - 'forged': round(float(probabilities[0][1]), 4) - }, - 'features_received': features.tolist(), - 'timestamp': datetime.now().isoformat() - }) - - except Exception as e: - return jsonify({ - 'error': str(e), - 'type': type(e).__name__ - }), 500 - -@app.route('/api/predict_batch', methods=['POST']) -def predict_batch(): - """Classify multiple banknotes""" - try: - data = request.get_json() - - # Validate input - if 'batch' not in data: - return jsonify({ - 'error': 'Missing required field: batch', - 'expected_format': { - 'batch': [ - {'features': [variance, skewness, curtosis, entropy]}, - {'features': [variance, skewness, curtosis, entropy]} - ] - } - }), 400 - - batch = data['batch'] - results = [] - - for i, item in enumerate(batch): - if 'features' not in item: - results.append({ - 'index': i, - 'error': 'Missing features field' - }) - continue - - features = np.array(item['features']) - - if features.shape[0] != 4: - results.append({ - 'index': i, - 'error': f'Expected 4 features, got {features.shape[0]}' - }) - continue - - # Preprocess and predict - features_scaled = scaler.transform(features.reshape(1, -1)) - features_tensor = torch.FloatTensor(features_scaled) - - with torch.no_grad(): - outputs = model(features_tensor) - probabilities = torch.softmax(outputs, dim=1) - prediction = torch.argmax(outputs, dim=1).item() - - label = "GENUINE" if prediction == 0 else "FORGED" - confidence = float(probabilities[0][prediction]) - - results.append({ - 'index': i, - 'prediction': label, - 'confidence': round(confidence, 4), - 'probabilities': { - 'genuine': round(float(probabilities[0][0]), 4), - 'forged': round(float(probabilities[0][1]), 4) - } - }) - - # Summary statistics - valid_predictions = [r for r in results if 'prediction' in r] - genuine_count = sum(1 for r in valid_predictions if r['prediction'] == 'GENUINE') - forged_count = len(valid_predictions) - genuine_count - - return jsonify({ - 'results': results, - 'summary': { - 'total_processed': len(batch), - 'successful': len(valid_predictions), - 'errors': len(batch) - len(valid_predictions), - 'genuine_count': genuine_count, - 'forged_count': forged_count - }, - 'timestamp': datetime.now().isoformat() - }) - - except Exception as e: - return jsonify({ - 'error': str(e), - 'type': type(e).__name__ - }), 500 - -def main(): - """Start the API server""" - print("=" * 70) - print(" BANKNOTE FRAUD DETECTOR - PRODUCTION API") - print("=" * 70) - print("\nšŸ”¬ Model: Hybrid Quantum-Classical Neural Network") - print("šŸŽÆ Accuracy: 100% on validation data") - print("āš” Backend: Quantum simulation (PennyLane)") - - # Load model - try: - load_model() - except Exception as e: - print(f"\nāŒ Failed to load model: {e}") - print("\nšŸ’” Make sure you've trained the model first:") - print(" python train_custom_dataset.py --preset banknote --epochs 25") - sys.exit(1) - - print("\n" + "=" * 70) - print(" šŸš€ STARTING API SERVER") - print("=" * 70) - print("\nšŸ“” Endpoints:") - print(" POST http://localhost:8080/api/predict") - print(" POST http://localhost:8080/api/predict_batch") - print(" GET http://localhost:8080/api/health") - print(" GET http://localhost:8080/api/model_info") - - print("\nšŸ“ Example curl command:") - print(' curl -X POST http://localhost:8080/api/predict \\') - print(' -H "Content-Type: application/json" \\') - print(' -d \'{"features": [3.5, 0.5, -1.2, 0.8]}\'') - - print("\nā³ Server starting on http://localhost:8080 ...") - print(" Press Ctrl+C to stop") - print("=" * 70 + "\n") - - # Start server - app.run(host='0.0.0.0', port=8080, debug=False) - -if __name__ == '__main__': - main() +""" +REST API for quantum-powered banknote fraud detection. +Achieves 100% accuracy on validation data. + +Endpoints: + POST /api/predict - Classify single banknote + POST /api/predict_batch - Classify multiple banknotes + GET /api/health - Service health check + GET /api/model_info - Model metadata + +Usage: + python banknote_api.py + + # Test prediction: + curl -X POST http://localhost:8080/api/predict \ + -H "Content-Type: application/json" \ + -d '{"features": [3.5, 0.5, -1.2, 0.8]}' + +Author: Quantum AI System +Date: November 16, 2025 +""" + +from flask import Flask, request, jsonify +from flask_cors import CORS +import numpy as np +import torch +import joblib +from pathlib import Path +import sys +import json +from datetime import datetime + +# Add src to path +src_path = Path(__file__).parent.parent / "src" +sys.path.insert(0, str(src_path)) + +from src.hybrid_qnn import HybridQNN + +app = Flask(__name__) +CORS(app) # Enable CORS for web clients + +# Global model variables +model = None +scaler = None +model_metadata = {} + +def load_model(): + """Load trained model and preprocessors""" + global model, scaler, model_metadata + + model_dir = Path(__file__).parent.parent / "results" + + print("šŸ”„ Loading model artifacts...") + + # Load scaler + scaler_path = model_dir / "custom_scaler.pkl" + if not scaler_path.exists(): + raise FileNotFoundError(f"Scaler not found: {scaler_path}") + scaler = joblib.load(scaler_path) + print(f" āœ… Loaded scaler from {scaler_path}") + + # Create model architecture + model = HybridQNN( + input_dim=4, + hidden_dim=16, + n_qubits=4, + n_quantum_layers=2, + output_dim=2, + dropout=0.2 + ) + + # Load trained weights safely with weights_only=True to prevent arbitrary code execution + model_path = model_dir / "custom_model.pt" + if not model_path.exists(): + raise FileNotFoundError(f"Model not found: {model_path}") + model.load_state_dict(torch.load(model_path, weights_only=True)) + model.eval() # Set to evaluation mode + print(f" āœ… Loaded model weights from {model_path}") + + # Load metadata + summary_path = model_dir / "custom_training_summary.json" + if summary_path.exists(): + with open(summary_path) as f: + model_metadata = json.load(f) + print(f" āœ… Loaded metadata from {summary_path}") + + print("āœ… Model loaded successfully!") + print(f" Validation Accuracy: {model_metadata.get('metrics', {}).get('val_acc_best', 'N/A')}") + +@app.route('/api/health', methods=['GET']) +def health_check(): + """Health check endpoint""" + return jsonify({ + 'status': 'healthy', + 'service': 'banknote-fraud-detector', + 'version': '1.0.0', + 'model_loaded': model is not None, + 'timestamp': datetime.now().isoformat() + }) + +@app.route('/api/model_info', methods=['GET']) +def model_info(): + """Return model information""" + return jsonify({ + 'model': 'Hybrid Quantum-Classical Neural Network', + 'task': 'Binary classification: Genuine vs Forged banknotes', + 'architecture': { + 'qubits': 4, + 'quantum_layers': 2, + 'hidden_dim': 16, + 'output_classes': 2 + }, + 'performance': { + 'validation_accuracy': model_metadata.get('metrics', {}).get('val_acc_best', 'N/A'), + 'training_loss': model_metadata.get('metrics', {}).get('train_loss_last', 'N/A') + }, + 'features': { + 'names': ['variance', 'skewness', 'curtosis', 'entropy'], + 'count': 4, + 'preprocessing': 'StandardScaler normalization' + }, + 'training': { + 'dataset': model_metadata.get('dataset', 'banknote authentication'), + 'samples': model_metadata.get('data', {}).get('n_train', 'N/A'), + 'epochs': model_metadata.get('params', {}).get('epochs', 'N/A') + } + }) + +@app.route('/api/predict', methods=['POST']) +def predict(): + """Classify a single banknote""" + try: + data = request.get_json() + + # Validate input + if 'features' not in data: + return jsonify({ + 'error': 'Missing required field: features', + 'expected_format': { + 'features': [variance, skewness, curtosis, entropy] + } + }), 400 + + features = np.array(data['features']) + + # Validate feature count + if features.shape[0] != 4: + return jsonify({ + 'error': f'Expected 4 features, got {features.shape[0]}', + 'features_required': ['variance', 'skewness', 'curtosis', 'entropy'] + }), 400 + + # Preprocess + features_scaled = scaler.transform(features.reshape(1, -1)) + features_tensor = torch.FloatTensor(features_scaled) + + # Predict + with torch.no_grad(): + outputs = model(features_tensor) + probabilities = torch.softmax(outputs, dim=1) + prediction = torch.argmax(outputs, dim=1).item() + + # Interpret result + label = "GENUINE" if prediction == 0 else "FORGED" + confidence = float(probabilities[0][prediction]) + + return jsonify({ + 'prediction': label, + 'confidence': round(confidence, 4), + 'probabilities': { + 'genuine': round(float(probabilities[0][0]), 4), + 'forged': round(float(probabilities[0][1]), 4) + }, + 'features_received': features.tolist(), + 'timestamp': datetime.now().isoformat() + }) + + except Exception as e: + return jsonify({ + 'error': str(e), + 'type': type(e).__name__ + }), 500 + +@app.route('/api/predict_batch', methods=['POST']) +def predict_batch(): + """Classify multiple banknotes""" + try: + data = request.get_json() + + # Validate input + if 'batch' not in data: + return jsonify({ + 'error': 'Missing required field: batch', + 'expected_format': { + 'batch': [ + {'features': [variance, skewness, curtosis, entropy]}, + {'features': [variance, skewness, curtosis, entropy]} + ] + } + }), 400 + + batch = data['batch'] + results = [] + + for i, item in enumerate(batch): + if 'features' not in item: + results.append({ + 'index': i, + 'error': 'Missing features field' + }) + continue + + features = np.array(item['features']) + + if features.shape[0] != 4: + results.append({ + 'index': i, + 'error': f'Expected 4 features, got {features.shape[0]}' + }) + continue + + # Preprocess and predict + features_scaled = scaler.transform(features.reshape(1, -1)) + features_tensor = torch.FloatTensor(features_scaled) + + with torch.no_grad(): + outputs = model(features_tensor) + probabilities = torch.softmax(outputs, dim=1) + prediction = torch.argmax(outputs, dim=1).item() + + label = "GENUINE" if prediction == 0 else "FORGED" + confidence = float(probabilities[0][prediction]) + + results.append({ + 'index': i, + 'prediction': label, + 'confidence': round(confidence, 4), + 'probabilities': { + 'genuine': round(float(probabilities[0][0]), 4), + 'forged': round(float(probabilities[0][1]), 4) + } + }) + + # Summary statistics + valid_predictions = [r for r in results if 'prediction' in r] + genuine_count = sum(1 for r in valid_predictions if r['prediction'] == 'GENUINE') + forged_count = len(valid_predictions) - genuine_count + + return jsonify({ + 'results': results, + 'summary': { + 'total_processed': len(batch), + 'successful': len(valid_predictions), + 'errors': len(batch) - len(valid_predictions), + 'genuine_count': genuine_count, + 'forged_count': forged_count + }, + 'timestamp': datetime.now().isoformat() + }) + + except Exception as e: + return jsonify({ + 'error': str(e), + 'type': type(e).__name__ + }), 500 + +def main(): + """Start the API server""" + print("=" * 70) + print(" BANKNOTE FRAUD DETECTOR - PRODUCTION API") + print("=" * 70) + print("\nšŸ”¬ Model: Hybrid Quantum-Classical Neural Network") + print("šŸŽÆ Accuracy: 100% on validation data") + print("āš” Backend: Quantum simulation (PennyLane)") + + # Load model + try: + load_model() + except Exception as e: + print(f"\nāŒ Failed to load model: {e}") + print("\nšŸ’” Make sure you've trained the model first:") + print(" python train_custom_dataset.py --preset banknote --epochs 25") + sys.exit(1) + + print("\n" + "=" * 70) + print(" šŸš€ STARTING API SERVER") + print("=" * 70) + print("\nšŸ“” Endpoints:") + print(" POST http://localhost:8080/api/predict") + print(" POST http://localhost:8080/api/predict_batch") + print(" GET http://localhost:8080/api/health") + print(" GET http://localhost:8080/api/model_info") + + print("\nšŸ“ Example curl command:") + print(' curl -X POST http://localhost:8080/api/predict \\') + print(' -H "Content-Type: application/json" \\') + print(' -d \'{"features": [3.5, 0.5, -1.2, 0.8]}\'') + + print("\nā³ Server starting on http://localhost:8080 ...") + print(" Press Ctrl+C to stop") + print("=" * 70 + "\n") + + # Start server - default to localhost for security + # Use BANKNOTE_API_HOST environment variable to override if needed + host = os.environ.get('BANKNOTE_API_HOST', '127.0.0.1') + app.run(host=host, port=8080, debug=False) + +if __name__ == '__main__': + main() diff --git a/quantum-ai/production/test_api.py b/quantum-ai/production/test_api.py index 6b0cbf7a9..7341ebfbe 100644 --- a/quantum-ai/production/test_api.py +++ b/quantum-ai/production/test_api.py @@ -1,578 +1,292 @@ - -Comprehensive test suite for the production API. - -Usage: - python test_api.py - -Author: Quantum AI System -Date: November 16, 2025 -""" - -import requests -import json -import time -from datetime import datetime - -BASE_URL = "http://localhost:8080" -# Use timeout for all requests to prevent hanging -REQUEST_TIMEOUT = 30 # seconds - -def test_health(): - """Test health endpoint""" - print("\n" + "="*70) - print("TEST 1: Health Check") - print("="*70) - - response = requests.get(f"{BASE_URL}/api/health", timeout=REQUEST_TIMEOUT, timeout=REQUEST_TIMEOUT) - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - assert response.json()['status'] == 'healthy' - print("āœ… PASSED") - -def test_model_info(): - """Test model info endpoint""" - print("\n" + "="*70) - print("TEST 2: Model Info") - print("="*70) - - response = requests.get(f"{BASE_URL}/api/model_info", timeout=REQUEST_TIMEOUT) - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - assert 'architecture' in response.json() - print("āœ… PASSED") - -def test_single_prediction_genuine(): - """Test single prediction - genuine banknote""" - print("\n" + "="*70) - print("TEST 3: Single Prediction (Genuine)") - print("="*70) - - # Example genuine banknote features - data = { - "features": [3.6216, 8.6661, -2.8073, -0.44699] - } - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"}, - timeout=REQUEST_TIMEOUT - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - result = response.json() - assert 'prediction' in result - assert 'confidence' in result - assert result['confidence'] > 0.5 - print("āœ… PASSED") - -def test_single_prediction_forged(): - """Test single prediction - forged banknote""" - print("\n" + "="*70) - print("TEST 4: Single Prediction (Forged)") - print("="*70) - - # Example forged banknote features - data = { - "features": [-4.5459, 8.1674, -2.4586, -1.4621] - } - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"}, - timeout=REQUEST_TIMEOUT - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - result = response.json() - assert 'prediction' in result - assert 'confidence' in result - print("āœ… PASSED") - -def test_batch_prediction(): - """Test batch prediction""" - print("\n" + "="*70) - print("TEST 5: Batch Prediction") - print("="*70) - - data = { - "batch": [ - {"features": [3.6216, 8.6661, -2.8073, -0.44699]}, # Genuine - {"features": [-4.5459, 8.1674, -2.4586, -1.4621]}, # Forged - {"features": [2.5, 1.2, -0.8, 0.3]}, # Test - ] - } - - print(f"Input: {len(data['batch'])} banknotes") - - response = requests.post( - f"{BASE_URL}/api/predict_batch", - json=data, - headers={"Content-Type": "application/json"}, - timeout=REQUEST_TIMEOUT - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - result = response.json() - assert 'results' in result - assert 'summary' in result - assert result['summary']['total_processed'] == 3 - print("āœ… PASSED") - -def test_invalid_input(): - """Test error handling with invalid input""" - print("\n" + "="*70) - print("TEST 6: Invalid Input Handling") - print("="*70) - - # Missing features field - data = {"wrong_field": [1, 2, 3, 4]} - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"}, - timeout=REQUEST_TIMEOUT - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 400 - assert 'error' in response.json() - print("āœ… PASSED") - -def test_wrong_feature_count(): - """Test error handling with wrong feature count""" - print("\n" + "="*70) - print("TEST 7: Wrong Feature Count") - print("="*70) - - # Only 3 features instead of 4 - data = {"features": [1.0, 2.0, 3.0]} - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"}, - timeout=REQUEST_TIMEOUT - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 400 - assert 'error' in response.json() - print("āœ… PASSED") - -def test_performance(): - """Test API performance""" - print("\n" + "="*70) - print("TEST 8: Performance Benchmark") - print("="*70) - - data = {"features": [3.6216, 8.6661, -2.8073, -0.44699]} - - n_requests = 10 - times = [] - - print(f"Making {n_requests} requests...") - - for i in range(n_requests): - start = time.time() - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"}, - timeout=REQUEST_TIMEOUT - ) - elapsed = time.time() - start - times.append(elapsed * 1000) # Convert to ms - - assert response.status_code == 200 - print(f" Request {i+1}/{n_requests}: {elapsed*1000:.2f}ms", end='\r') - - avg_time = sum(times) / len(times) - min_time = min(times) - max_time = max(times) - - print(f"\n\nPerformance Results:") - print(f" Average: {avg_time:.2f}ms") - print(f" Min: {min_time:.2f}ms") - print(f" Max: {max_time:.2f}ms") - - assert avg_time < 200 # Should be under 200ms - print("āœ… PASSED") - -def main(): - """Run all tests""" - print("=" * 70) - print(" BANKNOTE FRAUD DETECTOR API - TEST SUITE") - print("=" * 70) - print(f"\nTarget: {BASE_URL}") - print(f"Time: {datetime.now().isoformat()}") - - # Check if API is running - try: - response = requests.get(f"{BASE_URL}/api/health", timeout=5, timeout=REQUEST_TIMEOUT) - if response.status_code != 200: - print("\nāŒ API is not responding correctly") - print(" Make sure the API is running:") - print(" python banknote_api.py") - return - except requests.exceptions.RequestException: - print("\nāŒ Cannot connect to API") - print(" Make sure the API is running:") - print(" python banknote_api.py") - return - - # Run tests - tests = [ - test_health, - test_model_info, - test_single_prediction_genuine, - test_single_prediction_forged, - test_batch_prediction, - test_invalid_input, - test_wrong_feature_count, - test_performance - ] - - passed = 0 - failed = 0 - - for test in tests: - try: - test() - passed += 1 - except Exception as e: - print(f"āŒ FAILED: {e}") - failed += 1 - - # Summary - print("\n" + "=" * 70) - print(" TEST SUMMARY") - print("=" * 70) - print(f"\nTotal Tests: {len(tests)}") - print(f"āœ… Passed: {passed}") - print(f"āŒ Failed: {failed}") - - if failed == 0: - print("\nšŸŽ‰ ALL TESTS PASSED!") - print("āœ… API is production-ready!") - else: - print(f"\nāš ļø {failed} test(s) failed") - print(" Review the output above for details") - - print("=" * 70) - -if __name__ == "__main__": - main() -""" -Test the Banknote Fraud Detector API - -Comprehensive test suite for the production API. - -Usage: - python test_api.py - -Author: Quantum AI System -Date: November 16, 2025 -""" - -import requests -import json -import time -from datetime import datetime - -BASE_URL = "http://localhost:8080" - -def test_health(): - """Test health endpoint""" - print("\n" + "="*70) - print("TEST 1: Health Check") - print("="*70) - - response = requests.get(f"{BASE_URL}/api/health") - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - assert response.json()['status'] == 'healthy' - print("āœ… PASSED") - -def test_model_info(): - """Test model info endpoint""" - print("\n" + "="*70) - print("TEST 2: Model Info") - print("="*70) - - response = requests.get(f"{BASE_URL}/api/model_info") - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - assert 'architecture' in response.json() - print("āœ… PASSED") - -def test_single_prediction_genuine(): - """Test single prediction - genuine banknote""" - print("\n" + "="*70) - print("TEST 3: Single Prediction (Genuine)") - print("="*70) - - # Example genuine banknote features - data = { - "features": [3.6216, 8.6661, -2.8073, -0.44699] - } - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"} - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - result = response.json() - assert 'prediction' in result - assert 'confidence' in result - assert result['confidence'] > 0.5 - print("āœ… PASSED") - -def test_single_prediction_forged(): - """Test single prediction - forged banknote""" - print("\n" + "="*70) - print("TEST 4: Single Prediction (Forged)") - print("="*70) - - # Example forged banknote features - data = { - "features": [-4.5459, 8.1674, -2.4586, -1.4621] - } - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"} - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - result = response.json() - assert 'prediction' in result - assert 'confidence' in result - print("āœ… PASSED") - -def test_batch_prediction(): - """Test batch prediction""" - print("\n" + "="*70) - print("TEST 5: Batch Prediction") - print("="*70) - - data = { - "batch": [ - {"features": [3.6216, 8.6661, -2.8073, -0.44699]}, # Genuine - {"features": [-4.5459, 8.1674, -2.4586, -1.4621]}, # Forged - {"features": [2.5, 1.2, -0.8, 0.3]}, # Test - ] - } - - print(f"Input: {len(data['batch'])} banknotes") - - response = requests.post( - f"{BASE_URL}/api/predict_batch", - json=data, - headers={"Content-Type": "application/json"} - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 200 - result = response.json() - assert 'results' in result - assert 'summary' in result - assert result['summary']['total_processed'] == 3 - print("āœ… PASSED") - -def test_invalid_input(): - """Test error handling with invalid input""" - print("\n" + "="*70) - print("TEST 6: Invalid Input Handling") - print("="*70) - - # Missing features field - data = {"wrong_field": [1, 2, 3, 4]} - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"} - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 400 - assert 'error' in response.json() - print("āœ… PASSED") - -def test_wrong_feature_count(): - """Test error handling with wrong feature count""" - print("\n" + "="*70) - print("TEST 7: Wrong Feature Count") - print("="*70) - - # Only 3 features instead of 4 - data = {"features": [1.0, 2.0, 3.0]} - - print(f"Input: {json.dumps(data, indent=2)}") - - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"} - ) - - print(f"Status Code: {response.status_code}") - print(f"Response: {json.dumps(response.json(), indent=2)}") - - assert response.status_code == 400 - assert 'error' in response.json() - print("āœ… PASSED") - -def test_performance(): - """Test API performance""" - print("\n" + "="*70) - print("TEST 8: Performance Benchmark") - print("="*70) - - data = {"features": [3.6216, 8.6661, -2.8073, -0.44699]} - - n_requests = 10 - times = [] - - print(f"Making {n_requests} requests...") - - for i in range(n_requests): - start = time.time() - response = requests.post( - f"{BASE_URL}/api/predict", - json=data, - headers={"Content-Type": "application/json"} - ) - elapsed = time.time() - start - times.append(elapsed * 1000) # Convert to ms - - assert response.status_code == 200 - print(f" Request {i+1}/{n_requests}: {elapsed*1000:.2f}ms", end='\r') - - avg_time = sum(times) / len(times) - min_time = min(times) - max_time = max(times) - - print(f"\n\nPerformance Results:") - print(f" Average: {avg_time:.2f}ms") - print(f" Min: {min_time:.2f}ms") - print(f" Max: {max_time:.2f}ms") - - assert avg_time < 200 # Should be under 200ms - print("āœ… PASSED") - -def main(): - """Run all tests""" - print("=" * 70) - print(" BANKNOTE FRAUD DETECTOR API - TEST SUITE") - print("=" * 70) - print(f"\nTarget: {BASE_URL}") - print(f"Time: {datetime.now().isoformat()}") - - # Check if API is running - try: - response = requests.get(f"{BASE_URL}/api/health", timeout=5) - if response.status_code != 200: - print("\nāŒ API is not responding correctly") - print(" Make sure the API is running:") - print(" python banknote_api.py") - return - except requests.exceptions.RequestException: - print("\nāŒ Cannot connect to API") - print(" Make sure the API is running:") - print(" python banknote_api.py") - return - - # Run tests - tests = [ - test_health, - test_model_info, - test_single_prediction_genuine, - test_single_prediction_forged, - test_batch_prediction, - test_invalid_input, - test_wrong_feature_count, - test_performance - ] - - passed = 0 - failed = 0 - - for test in tests: - try: - test() - passed += 1 - except Exception as e: - print(f"āŒ FAILED: {e}") - failed += 1 - - # Summary - print("\n" + "=" * 70) - print(" TEST SUMMARY") - print("=" * 70) - print(f"\nTotal Tests: {len(tests)}") - print(f"āœ… Passed: {passed}") - print(f"āŒ Failed: {failed}") - - if failed == 0: - print("\nšŸŽ‰ ALL TESTS PASSED!") - print("āœ… API is production-ready!") - else: - print(f"\nāš ļø {failed} test(s) failed") - print(" Review the output above for details") - - print("=" * 70) - -if __name__ == "__main__": - main() +""" +Comprehensive test suite for the production API. + +Usage: + python test_api.py + +Author: Quantum AI System +Date: November 16, 2025 +""" + +import requests +import json +import time +from datetime import datetime + +BASE_URL = "http://localhost:8080" +# Use timeout for all requests to prevent hanging +REQUEST_TIMEOUT = 30 # seconds + +def test_health(): + """Test health endpoint""" + print("\n" + "="*70) + print("TEST 1: Health Check") + print("="*70) + + response = requests.get(f"{BASE_URL}/api/health", timeout=REQUEST_TIMEOUT) + print(f"Status Code: {response.status_code}") + print(f"Response: {json.dumps(response.json(), indent=2)}") + + assert response.status_code == 200 + assert response.json()['status'] == 'healthy' + print("āœ… PASSED") + +def test_model_info(): + """Test model info endpoint""" + print("\n" + "="*70) + print("TEST 2: Model Info") + print("="*70) + + response = requests.get(f"{BASE_URL}/api/model_info", timeout=REQUEST_TIMEOUT) + print(f"Status Code: {response.status_code}") + print(f"Response: {json.dumps(response.json(), indent=2)}") + + assert response.status_code == 200 + assert 'architecture' in response.json() + print("āœ… PASSED") + +def test_single_prediction_genuine(): + """Test single prediction - genuine banknote""" + print("\n" + "="*70) + print("TEST 3: Single Prediction (Genuine)") + print("="*70) + + # Example genuine banknote features + data = { + "features": [3.6216, 8.6661, -2.8073, -0.44699] + } + + print(f"Input: {json.dumps(data, indent=2)}") + + response = requests.post( + f"{BASE_URL}/api/predict", + json=data, + headers={"Content-Type": "application/json"}, + timeout=REQUEST_TIMEOUT + ) + + print(f"Status Code: {response.status_code}") + print(f"Response: {json.dumps(response.json(), indent=2)}") + + assert response.status_code == 200 + result = response.json() + assert 'prediction' in result + assert 'confidence' in result + assert result['confidence'] > 0.5 + print("āœ… PASSED") + +def test_single_prediction_forged(): + """Test single prediction - forged banknote""" + print("\n" + "="*70) + print("TEST 4: Single Prediction (Forged)") + print("="*70) + + # Example forged banknote features + data = { + "features": [-4.5459, 8.1674, -2.4586, -1.4621] + } + + print(f"Input: {json.dumps(data, indent=2)}") + + response = requests.post( + f"{BASE_URL}/api/predict", + json=data, + headers={"Content-Type": "application/json"}, + timeout=REQUEST_TIMEOUT + ) + + print(f"Status Code: {response.status_code}") + print(f"Response: {json.dumps(response.json(), indent=2)}") + + assert response.status_code == 200 + result = response.json() + assert 'prediction' in result + assert 'confidence' in result + print("āœ… PASSED") + +def test_batch_prediction(): + """Test batch prediction""" + print("\n" + "="*70) + print("TEST 5: Batch Prediction") + print("="*70) + + data = { + "batch": [ + {"features": [3.6216, 8.6661, -2.8073, -0.44699]}, # Genuine + {"features": [-4.5459, 8.1674, -2.4586, -1.4621]}, # Forged + {"features": [2.5, 1.2, -0.8, 0.3]}, # Test + ] + } + + print(f"Input: {len(data['batch'])} banknotes") + + response = requests.post( + f"{BASE_URL}/api/predict_batch", + json=data, + headers={"Content-Type": "application/json"}, + timeout=REQUEST_TIMEOUT + ) + + print(f"Status Code: {response.status_code}") + print(f"Response: {json.dumps(response.json(), indent=2)}") + + assert response.status_code == 200 + result = response.json() + assert 'results' in result + assert 'summary' in result + assert result['summary']['total_processed'] == 3 + print("āœ… PASSED") + +def test_invalid_input(): + """Test error handling with invalid input""" + print("\n" + "="*70) + print("TEST 6: Invalid Input Handling") + print("="*70) + + # Missing features field + data = {"wrong_field": [1, 2, 3, 4]} + + print(f"Input: {json.dumps(data, indent=2)}") + + response = requests.post( + f"{BASE_URL}/api/predict", + json=data, + headers={"Content-Type": "application/json"}, + timeout=REQUEST_TIMEOUT + ) + + print(f"Status Code: {response.status_code}") + print(f"Response: {json.dumps(response.json(), indent=2)}") + + assert response.status_code == 400 + assert 'error' in response.json() + print("āœ… PASSED") + +def test_wrong_feature_count(): + """Test error handling with wrong feature count""" + print("\n" + "="*70) + print("TEST 7: Wrong Feature Count") + print("="*70) + + # Only 3 features instead of 4 + data = {"features": [1.0, 2.0, 3.0]} + + print(f"Input: {json.dumps(data, indent=2)}") + + response = requests.post( + f"{BASE_URL}/api/predict", + json=data, + headers={"Content-Type": "application/json"}, + timeout=REQUEST_TIMEOUT + ) + + print(f"Status Code: {response.status_code}") + print(f"Response: {json.dumps(response.json(), indent=2)}") + + assert response.status_code == 400 + assert 'error' in response.json() + print("āœ… PASSED") + +def test_performance(): + """Test API performance""" + print("\n" + "="*70) + print("TEST 8: Performance Benchmark") + print("="*70) + + data = {"features": [3.6216, 8.6661, -2.8073, -0.44699]} + + n_requests = 10 + times = [] + + print(f"Making {n_requests} requests...") + + for i in range(n_requests): + start = time.time() + response = requests.post( + f"{BASE_URL}/api/predict", + json=data, + headers={"Content-Type": "application/json"}, + timeout=REQUEST_TIMEOUT + ) + elapsed = time.time() - start + times.append(elapsed * 1000) # Convert to ms + + assert response.status_code == 200 + print(f" Request {i+1}/{n_requests}: {elapsed*1000:.2f}ms", end='\r') + + avg_time = sum(times) / len(times) + min_time = min(times) + max_time = max(times) + + print(f"\n\nPerformance Results:") + print(f" Average: {avg_time:.2f}ms") + print(f" Min: {min_time:.2f}ms") + print(f" Max: {max_time:.2f}ms") + + assert avg_time < 200 # Should be under 200ms + print("āœ… PASSED") + +def main(): + """Run all tests""" + print("=" * 70) + print(" BANKNOTE FRAUD DETECTOR API - TEST SUITE") + print("=" * 70) + print(f"\nTarget: {BASE_URL}") + print(f"Time: {datetime.now().isoformat()}") + + # Check if API is running + try: + response = requests.get(f"{BASE_URL}/api/health", timeout=5) + if response.status_code != 200: + print("\nāŒ API is not responding correctly") + print(" Make sure the API is running:") + print(" python banknote_api.py") + return + except requests.exceptions.RequestException: + print("\nāŒ Cannot connect to API") + print(" Make sure the API is running:") + print(" python banknote_api.py") + return + + # Run tests + tests = [ + test_health, + test_model_info, + test_single_prediction_genuine, + test_single_prediction_forged, + test_batch_prediction, + test_invalid_input, + test_wrong_feature_count, + test_performance + ] + + passed = 0 + failed = 0 + + for test in tests: + try: + test() + passed += 1 + except Exception as e: + print(f"āŒ FAILED: {e}") + failed += 1 + + # Summary + print("\n" + "=" * 70) + print(" TEST SUMMARY") + print("=" * 70) + print(f"\nTotal Tests: {len(tests)}") + print(f"āœ… Passed: {passed}") + print(f"āŒ Failed: {failed}") + + if failed == 0: + print("\nšŸŽ‰ ALL TESTS PASSED!") + print("āœ… API is production-ready!") + else: + print(f"\nāš ļø {failed} test(s) failed") + print(" Review the output above for details") + + print("=" * 70) + +if __name__ == "__main__": + main() diff --git a/quantum-ai/train_custom_dataset.py b/quantum-ai/train_custom_dataset.py index 150008c7a..ebe9a6d59 100644 --- a/quantum-ai/train_custom_dataset.py +++ b/quantum-ai/train_custom_dataset.py @@ -1,4 +1,4 @@ -""" +r""" Train Quantum AI on Your Custom Dataset ====================================== diff --git a/scripts/repo_automation.py b/scripts/repo_automation.py index c667ab5af..fee6cf64e 100644 --- a/scripts/repo_automation.py +++ b/scripts/repo_automation.py @@ -342,7 +342,7 @@ def start_component(self, name: str) -> bool: def stop_component(self, name: str): """Stop a single component""" - """ + if name not in self.components or name not in self.processes: return component = self.components[name] @@ -588,6 +588,16 @@ def show_status(self): ) and p.status() != psutil.STATUS_ZOMBIE except Exception: dynamic_running[name] = False + + # Fallback: if PID not recorded, try discovering existing processes + if psutil is not None: + for name, component in self.components.items(): + if name not in dynamic_running: + try: + proc = self._find_existing_process(component) + dynamic_running[name] = proc is not None + except Exception: + dynamic_running[name] = False # Prefer dynamic running info; fall back to status file content components_running = status.get( @@ -599,14 +609,6 @@ def show_status(self): if component: status_icon = "āœ…" if running else "āŒ" dep_ok = (status.get("dependency_status", {}).get( - # Fallback: if PID not recorded, try discovering existing processes - for name, component in self.components.items(): - if name not in dynamic_running: - try: - proc=self._find_existing_process(component) - dynamic_running[name]=proc is not None - except Exception: - dynamic_running[name]=False name, True) if status else True) dep_icon = "šŸ§©" if dep_ok else "āš ļø" pid_info = f" (PID {pid_map.get(name)})" if name in pid_map else "" diff --git a/scripts/validate_datasets.py b/scripts/validate_datasets.py index 13260905f..2dd34ee85 100644 --- a/scripts/validate_datasets.py +++ b/scripts/validate_datasets.py @@ -150,6 +150,7 @@ def validate_jsonl(self, filepath: Path, verbose: bool = False) -> Dict: stats["samples"] += 1 except json.JSONDecodeError as e: + stats["format_errors"].append(f"Line {i}: Invalid JSON - {str(e)}") # Check if file was empty or contained only blank lines if stats["samples"] == 0 and not stats["format_errors"]: