Pneumonia Risk Assessment API

import os import tempfile import warnings from flask import Flask, request, jsonify from inference_service import ( load_classifier, preprocess_audio, generate_embeddings, predict_pneumonia, aggregate_predictions, SAMPLE_RATE, CLIP_DURATION, CLIP_OVERLAP_PERCENT, CLIP_IGNORE_SILENT_CLIPS, SILENCE_RMS_THRESHOLD_DB ) warnings.filterwarnings("ignore", category=UserWarning, module="soundfile") warnings.filterwarnings("ignore", module="librosa") app = Flask(__name__) # Load classifier globally classifier_model = load_classifier() @app.route('/') def home(): """API documentation homepage""" html_content = """ Pneumonia Risk Assessment API

⚠️ Medical Disclaimer: This is an AI assessment tool, not a medical diagnostic device. Results should not be used as the sole basis for medical decisions. Always consult qualified healthcare professionals.

📡 API Endpoint

POST /predict_pneumonia

Description: Analyze respiratory audio file for pneumonia risk assessment

Content-Type: multipart/form-data

Parameter: audio_file - Audio file (.wav, .mp3, etc.)

📥 Response Format

{
  "filename": "recording.wav",
  "pneumonia_risk_score": 0.7234,
  "risk_level": "High",
  "note": "This is an AI assessment, not a medical diagnosis..."
}

Risk Levels:

Low: Risk score < 0.4
Moderate: Risk score 0.4 - 0.7
High: Risk score > 0.7

💻 Usage Example

Using cURL:

curl -X POST \\
  -F "audio_file=@recording.wav" \\
  https://root16285-pneumonia-space.hf.space/predict_pneumonia

Using Python:

import requests

url = "https://root16285-pneumonia-space.hf.space/predict_pneumonia"
files = {"audio_file": open("recording.wav", "rb")}
response = requests.post(url, files=files)
print(response.json())

🧪 Test the API

Upload an audio file to test the pneumonia risk assessment:

""" return html_content @app.route('/predict_pneumonia', methods=['POST']) def predict_pneumonia_endpoint(): if 'audio_file' not in request.files: return jsonify({"error": "No audio_file part in the request"}), 400 audio_file = request.files['audio_file'] if audio_file.filename == '': return jsonify({"error": "No selected file"}), 400 temp_dir = tempfile.mkdtemp() temp_audio_path = os.path.join(temp_dir, audio_file.filename) audio_file.save(temp_audio_path) try: if classifier_model is None: return jsonify({"error": "Classifier not loaded"}), 500 audio_clips = preprocess_audio( temp_audio_path, SAMPLE_RATE, CLIP_DURATION, CLIP_OVERLAP_PERCENT, CLIP_IGNORE_SILENT_CLIPS, SILENCE_RMS_THRESHOLD_DB ) if audio_clips.size == 0: return jsonify({"result": "No valid audio clips", "risk_score": None}), 200 # Generate embeddings using OpenL3 embeddings = generate_embeddings(audio_clips) if embeddings.size == 0: return jsonify({"result": "No embeddings generated", "risk_score": None}), 200 # Predict pneumonia risk clip_predictions, clip_probabilities = predict_pneumonia(embeddings, classifier_model) if clip_predictions is None or clip_probabilities is None: return jsonify({"result": "Prediction failed", "risk_score": None}), 200 final_prediction, risk_score = aggregate_predictions(clip_predictions, clip_probabilities) return jsonify({ "filename": audio_file.filename, "pneumonia_risk_score": round(float(risk_score), 4), "risk_level": "High" if risk_score > 0.7 else "Moderate" if risk_score > 0.4 else "Low", "note": "This is an AI assessment, not a medical diagnosis. Consult a healthcare professional." }), 200 finally: os.remove(temp_audio_path) os.rmdir(temp_dir) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000)