Add pneumonia detection Flask API with inference service and Docker support

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.9-slim-buster
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y \
+    ffmpeg libsndfile1 \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY app.py .
+COPY inference_service.py .
+COPY pneumonia_classifier.joblib .
+
+EXPOSE 5000
+
+CMD ["python", "app.py"]

app.py

@@ -1,7 +1,69 @@
-import gradio as gr
+import os
+import tempfile
+import warnings
+from flask import Flask, request, jsonify
+from inference_service import (
+    load_hear_model,
+    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
+)
 
-def greet(name):
-    return "Hello " + name + "!!"
+warnings.filterwarnings("ignore", category=UserWarning, module="soundfile")
+warnings.filterwarnings("ignore", module="librosa")
 
-demo = gr.Interface(fn=greet, inputs="text", outputs="text")
-demo.launch()
+app = Flask(__name__)
+
+# Load models globally
+hear_infer_fn = load_hear_model()
+classifier_model = load_classifier()
+
+@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 hear_infer_fn is None or classifier_model is None:
+            return jsonify({"error": "Models 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", "pneumonia_status": "Undetermined"}), 200
+
+        embeddings = generate_embeddings(audio_clips, hear_infer_fn)
+
+        if embeddings.size == 0:
+            return jsonify({"result": "No embeddings generated", "pneumonia_status": "Undetermined"}), 200
+
+        clip_predictions = predict_pneumonia(embeddings, classifier_model)
+        final_prediction = aggregate_predictions(clip_predictions)
+
+        result_label = 'Pneumonia' if final_prediction == 1 else 'No Pneumonia'
+        return jsonify({"filename": audio_file.filename, "pneumonia_status": result_label}), 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)

inference_service.py

@@ -0,0 +1,116 @@
+import os
+import numpy as np
+import librosa
+import joblib
+import soundfile as sf
+from huggingface_hub import from_pretrained_keras
+import warnings
+
+# Suppress warnings
+warnings.filterwarnings("ignore", category=UserWarning, module="soundfile")
+warnings.filterwarnings("ignore", module="librosa")
+
+# HeAR Parameters
+SAMPLE_RATE = 16000
+CLIP_DURATION = 2
+CLIP_LENGTH = SAMPLE_RATE * CLIP_DURATION
+CLIP_OVERLAP_PERCENT = 10
+CLIP_IGNORE_SILENT_CLIPS = True
+SILENCE_RMS_THRESHOLD_DB = -50
+
+def create_dummy_audio(filename='dummy_audio.wav', duration=5, sr=SAMPLE_RATE):
+    t = np.linspace(0, duration, int(sr * duration), endpoint=False)
+    dummy_audio = 0.5 * np.sin(2 * np.pi * 440 * t) + 0.3 * np.sin(2 * np.pi * 880 * t)
+    dummy_audio = dummy_audio.astype(np.float32)
+    sf.write(filename, dummy_audio, sr)
+    print(f"Dummy audio file '{filename}' created with duration {duration}s.")
+    return filename
+
+def load_hear_model():
+    print("Loading HeAR model from Hugging Face...")
+    try:
+        loaded_model = from_pretrained_keras("google/hear")
+        infer_fn = loaded_model.signatures["serving_default"]
+        print("HeAR model loaded successfully.")
+        return infer_fn
+    except Exception as e:
+        print(f"Error loading HeAR model: {e}")
+        return None
+
+def load_classifier(model_path='pneumonia_classifier.joblib'):
+    print(f"Loading classifier from '{model_path}'...")
+    try:
+        classifier_model = joblib.load(model_path)
+        print("Classifier loaded successfully.")
+        return classifier_model
+    except Exception as e:
+        print(f"Error loading classifier: {e}")
+        return None
+
+def preprocess_audio(audio_path, sample_rate, clip_duration, clip_overlap_percent, ignore_silent_clips, silence_rms_threshold_db):
+    print(f"Preprocessing audio file: {audio_path}")
+    try:
+        audio, sr = librosa.load(audio_path, sr=sample_rate, mono=True)
+    except Exception as e:
+        print(f"Error loading audio file {audio_path}: {e}")
+        return np.array([])
+
+    clip_batch = []
+    clip_length = sr * clip_duration
+    overlap_samples = int(clip_length * (clip_overlap_percent / 100))
+    step_size = clip_length - overlap_samples
+    num_clips = max(1, (len(audio) - overlap_samples) // step_size)
+
+    for i in range(num_clips):
+        start_sample = i * step_size
+        end_sample = start_sample + clip_length
+        clip = audio[start_sample:end_sample]
+
+        if end_sample > len(audio):
+            clip = np.pad(clip, (0, clip_length - len(clip)), 'constant')
+
+        rms_loudness = round(20 * np.log10(np.sqrt(np.mean(clip**2)) + 1e-10))
+        if ignore_silent_clips and rms_loudness < silence_rms_threshold_db:
+            continue
+
+        clip_batch.append(clip)
+
+    if not clip_batch:
+        print("No valid (non-silent) audio clips generated.")
+        return np.array([])
+
+    return np.asarray(clip_batch)
+
+def generate_embeddings(audio_clips, infer_fn):
+    if audio_clips.size == 0:
+        print("No audio clips to generate embeddings from.")
+        return np.array([])
+
+    print(f"Generating embeddings for {len(audio_clips)} clips...")
+    try:
+        embedding_batch = infer_fn(x=audio_clips)['output_0'].numpy()
+        print(f"Generated {embedding_batch.shape[0]} embeddings of dimension {embedding_batch.shape[1]}.")
+        return embedding_batch
+    except Exception as e:
+        print(f"Error generating embeddings: {e}")
+        return np.array([])
+
+def predict_pneumonia(embeddings, classifier_model):
+    if embeddings.size == 0:
+        print("No embeddings for prediction.")
+        return np.array([])
+
+    print(f"Making predictions for {len(embeddings)} embeddings...")
+    predictions = classifier_model.predict(embeddings)
+    return predictions
+
+def aggregate_predictions(predictions):
+    if predictions.size == 0:
+        print("No predictions to aggregate.")
+        return None
+
+    unique, counts = np.unique(predictions, return_counts=True)
+    final_prediction = unique[np.argmax(counts)]
+    result_label = 'Pneumonia' if final_prediction == 1 else 'No Pneumonia'
+    print(f"Aggregated prediction: {result_label}")
+    return final_prediction

requirements.txt

@@ -1,2 +1,7 @@
-# Add any additional dependencies here
-# gradio is already pre-installed by Hugging Face Spaces
+flask
+numpy
+librosa
+joblib
+soundfile
+huggingface_hub
+scikit-learn