Update app.py

app.py

@@ -8,9 +8,9 @@ import requests
 from datetime import datetime
 from transformers import pipeline
 
-# Load detection model
+# 🎙️ Load detection model
 try:
-    print("[INFO] Loading Hugging Face audio classification model...")
+    print("[INFO] Loading Hugging Face model...")
     classifier = pipeline(
         "audio-classification",
         model="padmalcom/wav2vec2-large-nonverbalvocalization-classification"
@@ -19,7 +19,7 @@ except Exception as e:
     print(f"[ERROR] Failed to load model: {e}")
     classifier = None
 
-# Convert input audio to 16kHz mono WAV
+# === Audio Conversion ===
 def convert_audio(input_path, output_path="input.wav"):
     try:
         cmd = [
@@ -28,32 +28,35 @@ def convert_audio(input_path, output_path="input.wav"):
             output_path, "-y"
         ]
         subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        print(f"[INFO] Audio converted successfully: {output_path}")
+        print(f"[DEBUG] Audio converted to WAV: {output_path}")
         return output_path
     except subprocess.CalledProcessError as e:
-        print(f"[ERROR] Audio conversion failed: {e.stderr.decode()}")
+        print(f"[ERROR] ffmpeg conversion failed: {e.stderr.decode()}")
         raise RuntimeError("Audio conversion failed.")
 
-# Detect scream using classifier
+# === Scream Detection ===
 def detect_scream(audio_path):
-    if not classifier:
-        print("[ERROR] Classifier not loaded.")
-        return {"label": None, "score": 0.0}
-    audio, sr = librosa.load(audio_path, sr=16000)
-    results = classifier(audio)
-    print(f"[DEBUG] Model results: {results}")
-    if not results:
-        return {"label": None, "score": 0.0}
-    top = results[0]
-    return {"label": top['label'].lower(), "score": float(top['score']) * 100}
-
-# Send alert to Salesforce
+    try:
+        audio, sr = librosa.load(audio_path, sr=16000)
+        print(f"[DEBUG] Loaded audio: {len(audio)} samples at {sr} Hz")
+        if len(audio) == 0:
+            return {"label": "none", "score": 0.0}
+        results = classifier(audio)
+        print(f"[DEBUG] Model output: {results}")
+        if not results:
+            return {"label": "none", "score": 0.0}
+        top = results[0]
+        return {"label": top["label"].lower(), "score": float(top["score"]) * 100}
+    except Exception as e:
+        print(f"[ERROR] Detection failed: {e}")
+        return {"label": "error", "score": 0.0}
+
+# === Send Alert to Salesforce ===
 def send_salesforce_alert(audio_meta, detection):
     SF_URL = os.getenv("SF_ALERT_URL")
     SF_TOKEN = os.getenv("SF_API_TOKEN")
     if not SF_URL or not SF_TOKEN:
-        print("[ERROR] Salesforce credentials missing.")
-        raise RuntimeError("Missing Salesforce URL or token.")
+        raise RuntimeError("Salesforce config missing.")
 
     headers = {
         "Authorization": f"Bearer {SF_TOKEN}",
@@ -64,20 +67,16 @@ def send_salesforce_alert(audio_meta, detection):
         "DetectedLabel": detection["label"],
         "Score": detection["score"],
         "AlertLevel": audio_meta["alert_level"],
-        "Timestamp": audio_meta["timestamp"]
+        "Timestamp": audio_meta["timestamp"],
     }
 
-    try:
-        resp = requests.post(SF_URL, json=payload, headers=headers, timeout=5)
-        resp.raise_for_status()
-        print(f"[INFO] Salesforce alert sent successfully: {resp.json()}")
-        return resp.json()
-    except requests.RequestException as e:
-        print(f"[ERROR] Salesforce request failed: {e}")
-        raise RuntimeError("Salesforce alert failed.")
-
-# Main Gradio handler
-def process_uploaded(audio_file, start_stop, high_thresh, med_thresh, test_mode=False):
+    print(f"[DEBUG] Sending payload to Salesforce: {payload}")
+    resp = requests.post(SF_URL, json=payload, headers=headers, timeout=5)
+    resp.raise_for_status()
+    return resp.json()
+
+# === Main Gradio Function ===
+def process_uploaded(audio_file, start_stop, high_thresh, med_thresh):
     if start_stop != "Start":
         return "🛑 System is stopped."
 
@@ -86,16 +85,11 @@ def process_uploaded(audio_file, start_stop, high_thresh, med_thresh, test_mode=
     except Exception as e:
         return f"❌ Audio conversion error: {e}"
 
-    if test_mode:
-        detection = {"label": "screaming", "score": 85.0}
-        print("[TEST MODE] Simulating scream detection with 85.0%")
-    else:
-        detection = detect_scream(wav_path)
-
+    detection = detect_scream(wav_path)
     label = detection["label"]
     score = detection["score"]
 
-    # Determine alert level
+    # Determine risk level
     if label and "scream" in label and score >= high_thresh:
         level = "High-Risk"
     elif label and "scream" in label and score >= med_thresh:
@@ -109,36 +103,35 @@ def process_uploaded(audio_file, start_stop, high_thresh, med_thresh, test_mode=
         "alert_level": level
     }
 
+    # Send to Salesforce if needed
     if level in ("High-Risk", "Medium-Risk"):
         try:
             sf_resp = send_salesforce_alert(audio_meta, detection)
-            return f"✅ Detection: {label} ({score:.1f}%) — {level} — Alert logged (ID: {sf_resp.get('id', 'N/A')})"
+            return f"✅ Detection: {label} ({score:.1f}%) — {level} — Alert sent (ID: {sf_resp.get('id', 'N/A')})"
         except Exception as e:
             return f"⚠️ Detection: {label} ({score:.1f}%) — {level} — ERROR: {e}"
 
     return f"🟢 Detection: {label} ({score:.1f}%) — Alert Level: {level}"
 
-# Gradio interface
+# === Gradio UI ===
 iface = gr.Interface(
     fn=process_uploaded,
     inputs=[
         gr.Audio(type="filepath", label="Upload Audio"),
         gr.Radio(["Start", "Stop"], label="System State", value="Start"),
         gr.Slider(0, 100, value=80, step=1, label="High-Risk Threshold (%)"),
-        gr.Slider(0, 100, value=50, step=1, label="Medium-Risk Threshold (%)"),
-        gr.Checkbox(label="Test Mode (Simulate Detection)", value=False)
+        gr.Slider(0, 100, value=50, step=1, label="Medium-Risk Threshold (%)")
     ],
     outputs="text",
     title="📢 Scream Detection & Salesforce Alerts",
     description="""
-🎙 Upload or record audio to detect panic screams.  
-⚙️ Adjust thresholds and toggle test mode for simulation.  
-📡 Alerts are sent to Salesforce when scream intensity is high.
+🎧 Upload or record audio. System classifies screams and triggers alerts.  
+⚠️ Alerts are sent to Salesforce for High/Medium-Risk detections.
 """,
     allow_flagging="never"
 )
 
-# Optional Raspberry Pi listener
+# === Optional Real-Time Listener ===
 def pi_listener(high_thresh=80, med_thresh=50, interval=1.0):
     import sounddevice as sd
     import numpy as np
@@ -161,12 +154,14 @@ def pi_listener(high_thresh=80, med_thresh=50, interval=1.0):
             print(f"[{timestamp}] {level} scream detected ({sc:.1f}%) – alert sent.")
 
     with sd.InputStream(channels=1, samplerate=16000, callback=callback):
-        print("🔊 Raspberry Pi real-time listener running...")
+        print("🔊 Real-time detection started...")
         while True:
             time.sleep(interval)
 
+# === App Entry ===
 if __name__ == "__main__":
-    # Uncomment to enable real-time listener
+    # Optional: enable real-time listener
     # pi_thread = threading.Thread(target=pi_listener, daemon=True)
     # pi_thread.start()
+
     iface.launch(server_name="0.0.0.0", server_port=int(os.getenv("PORT", 7860)))