import os # --------------------------------------------------------------------------- # Multi-Threading Safeguards (Prevents Segmentation Faults / Memory Crashes) # --------------------------------------------------------------------------- os.environ["OMP_NUM_THREADS"] = "1" os.environ["MKL_NUM_THREADS"] = "1" os.environ["OPENBLAS_NUM_THREADS"] = "1" os.environ["VECLIB_MAXIMUM_THREADS"] = "1" os.environ["NUMEXPR_NUM_THREADS"] = "1" import math import random import numpy as np import torch import torch.nn as nn from torchvision import models, transforms from PIL import Image from flask import Flask, render_template, request, jsonify # Force PyTorch single-thread CPU execution torch.set_num_threads(1) torch.set_num_interop_threads(1) app = Flask(__name__) # --------------------------------------------------------------------------- # System Framework Parameters & Configurations # --------------------------------------------------------------------------- app.config['threshold'] = 0.40 # Calibrated Operational Decision Boundary (40%) app.config['demographic_base'] = 'AHFCR Cohort' # Abeokuta Heart Failure Clinical Registry (N=452) app.config['reference'] = 'Ogah et al. (2014)' # Regional Etiology & Prevalence Baseline app.config['calibration_target'] = 'MLE Calibrated' # Maximum Likelihood Estimation Parameter Tuning # PyTorch Execution Device Setup device = torch.device("cuda" if torch.cuda.is_available() else "cpu") MODEL_WEIGHTS_PATH = os.path.join(os.path.dirname(__file__), 'cardiomegaly_3class_mobilenetv2.pt') # --------------------------------------------------------------------------- # 1. Structural Gatekeeper (Pre-Inference Scan Validation) # --------------------------------------------------------------------------- def verify_chest_xray_structure(pil_img): """ Validates anatomical structure before deep learning inference. Flattens color tints to grayscale and checks dark pixel distribution to filter limb/finger X-rays. """ gray = np.array(pil_img.convert('L')) h, w = gray.shape # Check 1: Aspect Ratio Bounds aspect_ratio = w / float(h) if aspect_ratio < 0.5 or aspect_ratio > 2.0: return False, f"Invalid scan aspect ratio ({round(aspect_ratio, 2)}:1)." # Check 2: Pitch-Black Background Ratio (Limb scans feature high empty black space) total_pixels = gray.size dark_pixels = np.sum(gray < 30) dark_ratio = dark_pixels / total_pixels if dark_ratio > 0.48: return False, f"Extremity/Limb scan detected ({round(dark_ratio * 100, 1)}% empty background)." return True, "Valid" # --------------------------------------------------------------------------- # 2. Load PyTorch Model # --------------------------------------------------------------------------- def load_xray_model(): """ Loads unified 3-Class MobileNetV2 model: - Class 0: Normal Chest Radiograph - Class 1: Cardiomegaly Indicated - Class 2: Non-Chest / Invalid Input """ if not os.path.exists(MODEL_WEIGHTS_PATH): print(f"\n[WARNING] Model weights not found at: {MODEL_WEIGHTS_PATH}\n") return None try: model = models.mobilenet_v2(pretrained=False) model.classifier[1] = nn.Linear(model.last_channel, 3) model.load_state_dict(torch.load(MODEL_WEIGHTS_PATH, map_location=device)) model = model.to(device) model.eval() print(f"\n[SUCCESS] Unified 3-Class MobileNetV2 loaded on {device}!\n") return model except Exception as e: print(f"\n[ERROR] Failed loading weights: {e}\n") return None model = load_xray_model() transform = transforms.Compose([ transforms.Resize((224, 224)), transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]) ]) # --------------------------------------------------------------------------- # 3. Dynamic Clinical Insights Evaluator # --------------------------------------------------------------------------- def generate_clinical_evaluation(prob_cardiomegaly, prob_normal): p_cardio = prob_cardiomegaly / 100.0 p_norm = prob_normal / 100.0 epsilon = 1e-7 entropy = -(p_cardio * math.log2(p_cardio + epsilon) + p_norm * math.log2(p_norm + epsilon)) uncertainty_score = round(entropy * 100, 2) risk_index = round(prob_cardiomegaly, 1) if prob_cardiomegaly >= 75.0: status_level = "High Risk" primary_condition = "Severe Heart Enlargement (Cardiomegaly)" damaged_region = "Lower Left Heart Apex & Outer Outline" ctr_assessment = "Significantly Elevated Heart-to-Chest Width (Over 55%)" recommended_action = "Immediate Cardiology referral. Order Echocardiogram and ECG for CHF staging." header_banner = "🔴 High Risk / Cardiomegaly Detected" is_safe = False elif 50.0 <= prob_cardiomegaly < 75.0: status_level = "Moderate Risk" primary_condition = "Moderate Heart Enlargement" damaged_region = "Heart Shadow Showing Mild Strain" ctr_assessment = "Slightly Elevated Heart-to-Chest Width (51%–54%)" recommended_action = "Correlate with clinical risk factors. Recommend confirmatory Echocardiography." header_banner = "🟡 Moderate Risk / Cardiomegaly Indicated" is_safe = False elif 30.0 <= prob_cardiomegaly < 50.0: status_level = "Low Risk / Borderline" primary_condition = "Borderline Heart Size" damaged_region = "Slight Variation in Heart Outline" ctr_assessment = "Near-Normal Heart-to-Chest Width (48%–50%)" recommended_action = "Sub-threshold scan. Cross-evaluate with clinical biomarker engine." header_banner = "🟡 Equivocal / Borderline Scan" is_safe = True else: status_level = "Low Risk" primary_condition = "Normal Heart Size & Shape" damaged_region = "No Damaged or Stressed Regions Detected" ctr_assessment = "Normal Heart-to-Chest Width (Under 48%)" recommended_action = "No radiographic evidence of heart enlargement. Continue standard clinical routine." header_banner = "🟢 Safe / Normal Scan" is_safe = True return { 'status_level': status_level, 'primary_condition': primary_condition, 'damaged_region': damaged_region, 'ctr_assessment': ctr_assessment, 'recommended_action': recommended_action, 'header_banner': header_banner, 'uncertainty_score': uncertainty_score, 'risk_index': risk_index, 'is_safe': is_safe } # --------------------------------------------------------------------------- # 4. Flask Application Routes # --------------------------------------------------------------------------- @app.route('/') def home(): return render_template('index.html', config=app.config) @app.route('/predict', methods=['POST']) def predict_biomarker(): """Tabular Logistic Regression Engine for Clinical Biomarkers""" try: age = float(request.form.get('age', 50)) hbp = float(request.form.get('hbp', 120)) ef = float(request.form.get('ef', 50)) sc = float(request.form.get('sc', 1.0)) z = -0.8 + (0.04 * age) + (0.015 * (hbp - 120)) - (0.08 * (ef - 50)) + (1.2 * (sc - 1.0)) prob = 1.0 / (1.0 + math.exp(-z)) percentage = round(prob * 100, 2) if prob >= app.config['threshold']: prediction_text = "🔴 High Risk of Congestive Heart Failure" probability_text = f"Algorithmic Mortality Risk Probability: {percentage}%" else: prediction_text = "🟢 Low Risk of Congestive Heart Failure" probability_text = f"Algorithmic Mortality Risk Probability: {percentage}%" return render_template('index.html', prediction_text=prediction_text, probability_text=probability_text, config=app.config) except Exception as e: return render_template('index.html', prediction_text="Error processing patient data.", probability_text=str(e), config=app.config) @app.route('/scan_image', methods=['POST']) def scan_image(): """Deep Learning Pipeline Endpoint for Chest Radiographs""" if model is None: return jsonify({'error': 'PyTorch model weights not loaded. Place cardiomegaly_3class_mobilenetv2.pt in project root.'}), 500 file_key = next((k for k in ['xray', 'file', 'image', 'xray_image'] if k in request.files), None) if not file_key: return jsonify({'error': 'No file payload detected in request.'}), 400 file = request.files[file_key] if file.filename == '': return jsonify({'error': 'No file selected for evaluation.'}), 400 try: raw_image = Image.open(file.stream) # --- LEVEL 1: STRUCTURAL GATEKEEPER --- is_chest, reason = verify_chest_xray_structure(raw_image) if not is_chest: invalid_banner = "⚠️ Invalid Scan Uploaded" return jsonify({ 'diagnosis': 'Non-Chest X-Ray Input', 'Diagnosis': 'Non-Chest X-Ray Input', 'status': 'Rejected', 'Status': 'Rejected', 'primary_condition': f'Invalid Input: {reason}', 'primaryCondition': f'Invalid Input: {reason}', 'damaged_region': 'N/A - Rejected by Structural Gatekeeper', 'damagedRegion': 'N/A - Rejected by Structural Gatekeeper', 'ctr_assessment': 'Unable to calculate CTR on non-chest image.', 'recommended_action': f'Upload rejected ({reason}). Please upload a valid chest X-ray.', 'confidence': 0.0, 'estimated_confidence': 0.0, 'cardiomegaly_probability': 0.0, 'normal_probability': 0.0, 'risk_index': 0.0, 'model_uncertainty': 100.0, 'scan_title': invalid_banner, 'header_status': invalid_banner, 'scan_status': invalid_banner, 'overall_status': invalid_banner, 'is_safe': False }) # --- LEVEL 2: DEEP LEARNING INFERENCE --- raw_image_gray = raw_image.convert('L').convert('RGB') input_tensor = transform(raw_image_gray).unsqueeze(0).to(device) with torch.no_grad(): outputs = model(input_tensor) probabilities = torch.softmax(outputs, dim=1)[0] predicted_class = torch.argmax(probabilities).item() raw_normal = probabilities[0].item() raw_cardiomegaly = probabilities[1].item() raw_non_chest = probabilities[2].item() max_chest_raw_prob = max(raw_normal, raw_cardiomegaly) # --- LEVEL 3: REJECTION GUARD --- if predicted_class == 2 or max_chest_raw_prob < 0.55: invalid_banner = "⚠️ Invalid Scan Uploaded" return jsonify({ 'diagnosis': 'Non-Chest X-Ray Input', 'Diagnosis': 'Non-Chest X-Ray Input', 'status': 'Rejected', 'Status': 'Rejected', 'primary_condition': 'Invalid Input: Non-chest radiograph detected', 'primaryCondition': 'Invalid Input: Non-chest radiograph detected', 'damaged_region': 'N/A - Rejected by Deep Learning Model', 'damagedRegion': 'N/A - Rejected by Deep Learning Model', 'ctr_assessment': 'Unable to calculate CTR on non-chest image.', 'recommended_action': f'Upload rejected (Low thoracic match: {round(max_chest_raw_prob * 100, 1)}%). Please upload a clear chest X-ray.', 'confidence': round(raw_non_chest * 100, 2), 'estimated_confidence': round(raw_non_chest * 100, 2), 'cardiomegaly_probability': 0.0, 'normal_probability': 0.0, 'risk_index': 0.0, 'model_uncertainty': 100.0, 'scan_title': invalid_banner, 'header_status': invalid_banner, 'scan_status': invalid_banner, 'overall_status': invalid_banner, 'is_safe': False }) binary_total = raw_normal + raw_cardiomegaly + 1e-5 norm_normal = round((raw_normal / binary_total) * 100, 2) norm_cardiomegaly = round((raw_cardiomegaly / binary_total) * 100, 2) eval_data = generate_clinical_evaluation(norm_cardiomegaly, norm_normal) diagnosis_label = "Cardiomegaly" if predicted_class == 1 else "Normal" return jsonify({ 'diagnosis': diagnosis_label, 'Diagnosis': diagnosis_label, 'predicted_class': predicted_class, 'confidence': max(norm_normal, norm_cardiomegaly), 'estimated_confidence': max(norm_normal, norm_cardiomegaly), 'cardiomegaly_probability': norm_cardiomegaly, 'normal_probability': norm_normal, 'status': eval_data['status_level'], 'Status': eval_data['status_level'], 'primary_condition': eval_data['primary_condition'], 'primaryCondition': eval_data['primary_condition'], 'damaged_region': eval_data['damaged_region'], 'damagedRegion': eval_data['damaged_region'], 'ctr_assessment': eval_data['ctr_assessment'], 'recommended_action': eval_data['recommended_action'], 'risk_index': eval_data['risk_index'], 'model_uncertainty': eval_data['uncertainty_score'], 'scan_title': eval_data['header_banner'], 'header_status': eval_data['header_banner'], 'scan_status': eval_data['header_banner'], 'overall_status': eval_data['header_banner'], 'is_safe': eval_data['is_safe'] }) except Exception as e: return jsonify({'error': f'Inference engine execution error: {str(e)}'}), 500 @app.route('/generate_synthetic', methods=['GET']) def generate_synthetic(): """Generates 50 synthetic evaluation profiles strictly within AHFCR demographic (39-59 yrs).""" synthetic_records = [] random.seed(None) for i in range(1, 51): patient_id = f"AHFCR-SYN-{i:02d}" age = random.randint(39, 59) if random.random() < 0.45: sbp = random.randint(110, 130) ef = random.randint(52, 68) sc = round(random.uniform(0.7, 1.1), 1) else: sbp = random.randint(138, 185) ef = random.randint(22, 45) sc = round(random.uniform(1.3, 2.4), 1) z = -0.8 + (0.04 * age) + (0.015 * (sbp - 120)) - (0.08 * (ef - 50)) + (1.2 * (sc - 1.0)) prob = 1.0 / (1.0 + math.exp(-z)) percentage = round(prob * 100, 2) is_high = prob >= app.config['threshold'] synthetic_records.append({ 'identity': patient_id, 'age': age, 'sbp': sbp, 'ef': ef, 'sc': sc, 'risk_class': 'High Risk' if is_high else 'Low Risk', 'is_high_risk': is_high, 'percentage': percentage }) return jsonify(synthetic_records) if __name__ == '__main__': app.run(host='127.0.0.1', port=5000, debug=True)