import math import re import struct from typing import Dict, Any, List, Tuple HIGH_PAIN_PATTERNS = [ r"(?:10/10|9/10|8/10|10\s*out\s*of\s*10|9\s*out\s*of\s*10|8\s*out\s*of\s*10)", r"(?:unbearable|excruciating|worst\s*pain|extreme\s*pain|severe\s*chest\s*pain|cannot\s*breathe|gasping|dying)", r"(?:கடுமையான\s*வலி|தாங்க\s*முடியாத\s*வலி|நெஞ்சு\s*வலி|மூச்சு\s*திணறல்)", r"(?:असहनीय\s*दर्द|बहुत\s*तेज़\s*दर्द|सीने\s*में\s*दर्द|साँस\s*नहीं\s*आ\s*रही)", r"(?:చాలా\s*ఎక్కువ\s*నెప్పి|భరించలేని\s*నెప్పి|గుండె\s*నెప్పి)", r"(?:ಅಸಹನೀಯ\s*ನೋವು|ತೀವ್ರ\s*ನೋವು)" ] MODERATE_PAIN_PATTERNS = [ r"(?:7/10|6/10|5/10|moderate\s*pain|bad\s*pain|hurts\s*a\s*lot|வலி\s*அதிகமா|अत्यधिक\s*दर्द)", ] EMERGENCY_RED_FLAGS = [ r"(?:chest\s*pain|heart\s*attack|stroke|paralysis|unconscious|fainted|heavy\s*bleeding)", r"(?:நெஞ்சு\s*வலி|பக்கவாதம்|மயக்கம்|அதிக\s*ரத்தப்போக்கு)", r"(?:सीने\s*में\s*दर्द|हार्ट\s*अटैक|बेहोश|रक्तस्राव)" ] def calculate_audio_rms_energy(audio_bytes: bytes) -> float: """ Computes RMS (Root Mean Square) audio amplitude energy from raw mic bytes. Returns normalized energy float between 0.0 and 1.0. """ if not audio_bytes or len(audio_bytes) < 100: return 0.0 try: # Unpack 16-bit PCM audio samples sample_count = len(audio_bytes) // 2 if sample_count == 0: return 0.0 fmt = f"<{sample_count}h" samples = struct.unpack(fmt, audio_bytes[:sample_count * 2]) sum_squares = sum(s ** 2 for s in samples) rms = math.sqrt(sum_squares / sample_count) # Normalize 16-bit max amplitude (32768) norm_energy = min(1.0, rms / 15000.0) return round(norm_energy, 3) except Exception: return 0.0 def analyze_distress_and_pain(audio_bytes: bytes = b"", text: str = "") -> Dict[str, Any]: """ Acoustic & Semantic Distress Classifier: Analyzes spoken text and audio RMS energy to compute Pain Score (1-10) and Distress Score (0.0-1.0). """ txt_lower = (text or "").lower() pain_score = 3 distress_score = 0.2 is_emergency = False reasons: List[str] = [] # 1. Semantic High Pain Detection for pat in HIGH_PAIN_PATTERNS: if re.search(pat, txt_lower, re.IGNORECASE): pain_score = max(pain_score, 9) distress_score = max(distress_score, 0.85) reasons.append("Severe high pain expression detected") break # 2. Semantic Moderate Pain Detection if pain_score < 8: for pat in MODERATE_PAIN_PATTERNS: if re.search(pat, txt_lower, re.IGNORECASE): pain_score = max(pain_score, 6) distress_score = max(distress_score, 0.55) reasons.append("Moderate pain expression detected") break # 3. Emergency Red Flag Detection for pat in EMERGENCY_RED_FLAGS: if re.search(pat, txt_lower, re.IGNORECASE): is_emergency = True pain_score = max(pain_score, 9) distress_score = max(distress_score, 0.90) reasons.append("Critical emergency red flag detected") break # 4. Acoustic Audio RMS Energy Boost acoustic_energy = calculate_audio_rms_energy(audio_bytes) if acoustic_energy > 0.6: distress_score = min(1.0, distress_score + 0.15) reasons.append(f"High vocal acoustic energy detected (RMS: {acoustic_energy})") should_escalate = pain_score >= 8 or distress_score >= 0.75 or is_emergency return { "pain_score": pain_score, "distress_score": round(distress_score, 2), "acoustic_rms_energy": acoustic_energy, "is_emergency": is_emergency, "should_escalate": should_escalate, "reasons": reasons }