Ngks03/PPG-VascularAge

VascularAge Model

Background

Photoplethysmography (PPG) has emerged as a non-invasive method for monitoring cardiovascular health. This model estimates vascular age (AI-vascular age) from PPG signals, offering insights into an individual's cardiovascular health and associated risks.

The model was developed and evaluated using data from the UK Biobank (UKBB) cohort, where PPG signals were collected in a standardized format. For more details, please refer to our research article.

Use Case

The VascularAge model is designed to estimate vascular age, which can be used for:

• Risk Stratification: Identifying individuals at higher risk for cardiovascular events.
• Health Monitoring: Tracking cardiovascular health over time to support personalized interventions.

This model provides a non-invasive, scalable approach for real-time cardiovascular health assessment using PPG signals. It is specifically trained to process PPG data from the UKBB dataset, ensuring its effectiveness in research and clinical settings that use similar data.

Data Format

To ensure compatibility, PPG signals should be preprocessed and formatted in the same way as in the UKBB dataset. The expected input format is a 3D array with the shape (batch_size, num_channels, num_samples), where each sample corresponds to an averaged waveform of a single PPG beat.

Usage (Pytorch)

import json
import torch
import numpy as np
from net1d import Net1D  

# Function to normalize PPG data (Z-score along the last dimension)
def normalize_ppg(parsed_ppg: np.ndarray) -> np.ndarray:
    mean_ppg = parsed_ppg.mean(axis=-1, keepdims=True)  # Compute mean along the last dimension
    std_ppg = parsed_ppg.std(axis=-1, keepdims=True)    # Compute standard deviation along the last dimension
    normalized_ppg = (parsed_ppg - mean_ppg) / (std_ppg + 1e-8)  # Z-score normalization
    return normalized_ppg 

# Load the configuration file
with open('./config.json') as f:
    cfg = json.load(f)

device = 'cuda' if torch.cuda.is_available() else 'cpu'

# Initialize the model
model = Net1D(**cfg).to(device)

# Load model weights
model.load_state_dict(torch.load('./model.pth', map_location=device))

# Prepare raw PPG data (example) and normalize it
raw_ppg_data = np.random.randn(1, 1, 100)  # Example PPG data (batch size of 1, 1 channel, 100 time steps)
normalized_ppg_data = normalize_ppg(raw_ppg_data)  # Normalize the PPG data

# Convert normalized PPG data to tensor
ppg_data_tensor = torch.tensor(normalized_ppg_data, dtype=torch.float32).to(device)

# Perform inference
model.eval()
with torch.no_grad():
    vascular_age = model(ppg_data_tensor)

print(f"Estimated Vascular Age: {vascular_age.item()}")

API Inference

We also provide a remote API endpoint for vascular age prediction, enabling inference without local model deployment.

Endpoint

POST http://183.162.233.24:10021/VascularAge
Content-Type: application/json

Example Usage (Python)

import requests
import json
import numpy as np

def predict_vascular_age(ppg_beats):
    """
    Predicts vascular age using the provided PPG beats.

    Args:
        ppg_beats (list of lists): A list where each element is a list representing an averaged waveform of a single PPG beat.

    Returns:
        list: A list of predicted vascular ages corresponding to each input PPG beat.
    """
    # Convert the data to JSON format
    payload = json.dumps({'ppg_beats': ppg_beats})

    # Define the API endpoint
    url = "http://183.162.233.24:10021/VascularAge"

    # Set request headers (content type is JSON)
    headers = {'Content-Type': 'application/json'}

    try:
        # Send POST request to the API
        response = requests.post(url=url, headers=headers, data=payload)

        # Raise an exception if the request was unsuccessful
        response.raise_for_status()

        # Parse the response JSON
        response_data = response.json()

        # Check if the response code indicates success
        if response_data['code'] == 200:
            return response_data['data']
        else:
            print(f"Error: {response_data['msg']}")
            return None
    except requests.exceptions.HTTPError as http_err:
        print(f"HTTP error occurred: {http_err}")
    except Exception as err:
        print(f"An error occurred: {err}")

# Example usage
if __name__ == "__main__":
    # Prepare PPG beats (each beat is an averaged waveform of a single PPG beat)
    example_ppg_beats = [
        np.random.randn(120).tolist(),  # Example beat 1
        np.random.randn(80).tolist(),   # Example beat 2
        np.random.randn(100).tolist()   # Example beat 3
    ]

    # Predict vascular age
    predicted_ages = predict_vascular_age(example_ppg_beats)

    # Print the prediction result
    if predicted_ages is not None:
        print("Predicted Vascular Ages:", predicted_ages)

Response Format

The API returns a JSON object with the following structure:

{
  "code": 200,
  "msg": "SUCCESS",
  "data": [[38.0], [41.0], [40.0]]
}

• code: Status code (200 indicates success)
• msg: Status message
• data: A list of predicted vascular ages corresponding to each input PPG beat. Each vascular age is wrapped in a list (e.g., [[38.0]]) for consistency with batch processing.