test_real_ecg.py

import requests
import numpy as np
import pandas as pd
import json

# Your deployed Hugging Face Space URL
SPACE_URL = "https://mystic-cbk-ecg-fm-api.hf.space"

def load_ecg_data(csv_file):
    """Load ECG data from CSV file"""
    print(f"📁 Loading ECG data from: {csv_file}")
    
    # Read the CSV file
    df = pd.read_csv(csv_file)
    
    print(f"📊 ECG Data Shape: {df.shape}")
    print(f"📊 Leads: {list(df.columns)}")
    print(f"📊 Samples per lead: {len(df)}")
    
    # Convert to the format expected by the API
    # Each lead should be a list of float values
    ecg_data = []
    for lead in df.columns:
        ecg_data.append(df[lead].astype(float).tolist())
    
    print(f"✅ ECG data loaded successfully!")
    print(f"📊 Data format: {len(ecg_data)} leads × {len(ecg_data[0])} samples")
    
    return ecg_data

def test_api_with_real_ecg():
    """Test the deployed ECG-FM API with real ECG data"""
    
    print("🧪 Testing ECG-FM API with Real ECG Data")
    print("=" * 60)
    print(f"🌐 API URL: {SPACE_URL}")
    print()
    
    # Load the real ECG data
    try:
        ecg_data = load_ecg_data("ecg_d2c4fdc0-01ee-4210-a027-54a527329cd8.csv")
    except Exception as e:
        print(f"❌ Error loading ECG data: {e}")
        return
    
    print()
    
    # Test 1: Health check
    print("1️⃣ Testing health endpoint...")
    try:
        r = requests.get(f"{SPACE_URL}/healthz", timeout=30)
        print(f"   Status: {r.status_code}")
        if r.status_code == 200:
            response = r.json()
            print(f"   Response: {response}")
            print(f"   Mode: {response.get('mode', 'unknown')}")
        else:
            print(f"   Error: {r.text}")
    except Exception as e:
        print(f"   ❌ Error: {e}")
    
    print()
    
    # Test 2: Root endpoint
    print("2️⃣ Testing root endpoint...")
    try:
        r = requests.get(f"{SPACE_URL}/", timeout=30)
        print(f"   Status: {r.status_code}")
        if r.status_code == 200:
            response = r.json()
            print(f"   Response: {response}")
        else:
            print(f"   Error: {r.text}")
    except Exception as e:
        print(f"   ❌ Error: {e}")
    
    print()
    
    # Test 3: Predict endpoint with REAL ECG data
    print("3️⃣ Testing predict endpoint with REAL ECG data...")
    
    # Prepare the payload with real ECG data
    payload = {
        "signal": ecg_data,
        "fs": 500  # 500 Hz sampling rate
    }
    
    try:
        print("   📤 Sending REAL ECG data...")
        print(f"   📊 Input: 12 leads × {len(ecg_data[0])} samples")
        print(f"   📊 Sampling rate: 500 Hz")
        print(f"   📊 Duration: {len(ecg_data[0])/500:.1f} seconds")
        print("   ⏳ Waiting for inference...")
        
        r = requests.post(f"{SPACE_URL}/predict", json=payload, timeout=180)  # 3 min timeout
        print(f"   Status: {r.status_code}")
        
        if r.status_code == 200:
            result = r.json()
            print(f"   ✅ Success!")
            print(f"   📊 Mode: {result.get('mode', 'unknown')}")
            print(f"   📊 Output shape: {result.get('output_shape', 'unknown')}")
            print(f"   📊 Input shape: {result.get('input_shape', 'unknown')}")
            print(f"   📊 Output length: {len(result.get('output', []))}")
            
            if result.get('mode') == 'fallback':
                print(f"   💡 Note: {result.get('note', 'Running in fallback mode')}")
            else:
                print(f"   🎯 Full ECG-FM model inference completed!")
                
        else:
            print(f"   ❌ Error: {r.text}")
            
    except requests.exceptions.Timeout:
        print("   ⏰ Timeout - Inference taking longer than expected")
    except Exception as e:
        print(f"   ❌ Error: {e}")
    
    print()
    print("=" * 60)
    print("🏁 Real ECG Testing Complete!")
    print()
    print("📊 Results Summary:")
    print("   - If you see 200 status codes: API is working with real data!")
    print("   - If you see fallback mode: API works but model not fully loaded")
    print("   - If you see full mode: Complete ECG-FM inference working!")
    print()
    print("🔗 Your API is live at:")
    print(f"   {SPACE_URL}")
    print()
    print("📱 Test manually:")
    print(f"   Health: {SPACE_URL}/healthz")
    print(f"   Root: {SPACE_URL}/")

if __name__ == "__main__":
    test_api_with_real_ecg()