Update process_interview.py

process_interview.py

@@ -113,13 +113,13 @@ def load_models():
     return speaker_model, nlp, tokenizer, llm_model
 
 speaker_model, nlp, tokenizer, llm_model = load_models()
-
 def convert_to_wav(audio_path: str, output_dir: str = OUTPUT_DIR) -> str:
     try:
         audio = AudioSegment.from_file(audio_path)
         if audio.channels > 1:
             audio = audio.set_channels(1)
         audio = audio.set_frame_rate(16000)
+
         wav_file = os.path.join(output_dir, f"{uuid.uuid4()}.wav")
         audio.export(wav_file, format="wav")
         return wav_file
@@ -127,15 +127,18 @@ def convert_to_wav(audio_path: str, output_dir: str = OUTPUT_DIR) -> str:
         logger.error(f"Audio conversion failed: {str(e)}")
         raise
 
+
 def extract_prosodic_features(audio_path: str, start_ms: int, end_ms: int) -> Dict:
     try:
         audio = AudioSegment.from_file(audio_path)
         segment = audio[start_ms:end_ms]
         temp_path = os.path.join(OUTPUT_DIR, f"temp_{uuid.uuid4()}.wav")
         segment.export(temp_path, format="wav")
+
         y, sr = librosa.load(temp_path, sr=16000)
         pitches = librosa.piptrack(y=y, sr=sr)[0]
         pitches = pitches[pitches > 0]
+
         features = {
             'duration': (end_ms - start_ms) / 1000,
             'mean_pitch': float(np.mean(pitches)) if len(pitches) > 0 else 0.0,
@@ -147,16 +150,24 @@ def extract_prosodic_features(audio_path: str, start_ms: int, end_ms: int) -> Di
             'intensityMax': float(np.max(librosa.feature.rms(y=y)[0])),
             'intensitySD': float(np.std(librosa.feature.rms(y=y)[0])),
         }
+
         os.remove(temp_path)
         return features
     except Exception as e:
         logger.error(f"Feature extraction failed: {str(e)}")
         return {
-            'duration': 0.0, 'mean_pitch': 0.0, 'min_pitch': 0.0, 'max_pitch': 0.0,
-            'pitch_sd': 0.0, 'intensityMean': 0.0, 'intensityMin': 0.0,
-            'intensityMax': 0.0, 'intensitySD': 0.0
+            'duration': (end_ms - start_ms) / 1000,
+            'mean_pitch': 0.0,
+            'min_pitch': 0.0,
+            'max_pitch': 0.0,
+            'pitch_sd': 0.0,
+            'intensityMean': 0.0,
+            'intensityMin': 0.0,
+            'intensityMax': 0.0,
+            'intensitySD': 0.0,
         }
 
+
 def transcribe(audio_path: str) -> Dict:
     try:
         with open(audio_path, 'rb') as f:
@@ -166,6 +177,7 @@ def transcribe(audio_path: str) -> Dict:
                 data=f
             )
         audio_url = upload_response.json()['upload_url']
+
         transcript_response = requests.post(
             "https://api.assemblyai.com/v2/transcript",
             headers={"authorization": ASSEMBLYAI_KEY},
@@ -176,20 +188,24 @@ def transcribe(audio_path: str) -> Dict:
             }
         )
         transcript_id = transcript_response.json()['id']
+
         while True:
             result = requests.get(
                 f"https://api.assemblyai.com/v2/transcript/{transcript_id}",
                 headers={"authorization": ASSEMBLYAI_KEY}
             ).json()
+
             if result['status'] == 'completed':
                 return result
             elif result['status'] == 'error':
                 raise Exception(result['error'])
+
             time.sleep(5)
     except Exception as e:
         logger.error(f"Transcription failed: {str(e)}")
         raise
 
+
 def process_utterance(utterance, full_audio, wav_file):
     try:
         start = utterance['start']
@@ -197,198 +213,254 @@ def process_utterance(utterance, full_audio, wav_file):
         segment = full_audio[start:end]
         temp_path = os.path.join(OUTPUT_DIR, f"temp_{uuid.uuid4()}.wav")
         segment.export(temp_path, format="wav")
+
         with torch.no_grad():
-            embedding = speaker_model.get_embedding(temp_path).cpu().numpy()
-        embedding_list = embedding.flatten().tolist()
+            embedding = speaker_model.get_embedding(temp_path).to(device)
+
         query_result = index.query(
-            vector=embedding_list,
+            vector=embedding.cpu().numpy().tolist(),
             top_k=1,
             include_metadata=True
         )
+
         if query_result['matches'] and query_result['matches'][0]['score'] > 0.7:
             speaker_id = query_result['matches'][0]['id']
             speaker_name = query_result['matches'][0]['metadata']['speaker_name']
         else:
             speaker_id = f"unknown_{uuid.uuid4().hex[:6]}"
             speaker_name = f"Speaker_{speaker_id[-4:]}"
-            index.upsert([(speaker_id, embedding_list, {"speaker_name": speaker_id})])
+            index.upsert([(speaker_id, embedding.tolist(), {"speaker_name": speaker_name})])
+
         os.remove(temp_path)
+
         return {
-            ...
-            **speech, 'speaker': speaker_name,
+            **utterance,
+            'speaker': speaker_name,
             'speaker_id': speaker_id,
-            'embedding': embedding_list
+            'embedding': embedding.cpu().numpy().tolist()
         }
     except Exception as e:
-        logger.error(f"Utterance processing failed: {str(e)}", exc_info=True)
+        logger.error(f"Utterance processing failed: {str(e)}")
         return {
-            ...
-            speech, 'speech': 'Unknown',
-            'speaker_id': speaker_id,
-            'embedding_id': None
+            **utterance,
+            'speaker': 'Unknown',
+            'speaker_id': 'unknown',
+            'embedding': None
         }
 
-def identify_speakers(audio: Dict, text: str) -> List[Dict]:
+
+def identify_speakers(transcript: Dict, wav_file: str) -> List[Dict]:
     try:
-        audio = AudioSegment.from_wav(text)
-        speakers = audio['speech']
-        with ThreadPoolExecutor(max_workers=5) as executor:
+        full_audio = AudioSegment.from_wav(wav_file)
+        utterances = transcript['utterances']
+
+        with ThreadPoolExecutor(max_workers=5) as executor:  # Changed to 5 workers
             futures = [
-                executor.submit(process_speech, speech, speakers, text)
-                for speech in speakers
+                executor.submit(process_utterance, utterance, full_audio, wav_file)
+                for utterance in utterances
             ]
             results = [f.result() for f in futures]
+
         return results
     except Exception as e:
         logger.error(f"Speaker identification failed: {str(e)}")
         raise
 
-def train_role_classifier(speakers: List[Dict]):
+
+def train_role_classifier(utterances: List[Dict]):
     try:
-        speech = [u['speech'].split()]
-        vectorizer = TfidfVectorizer(max_features=500, ngram_range=(1,2))
-        X_text = vectorizer.fit_transform(speech)
+        texts = [u['text'] for u in utterances]
+        vectorizer = TfidfVectorizer(max_features=500, ngram_range=(1, 2))
+        X_text = vectorizer.fit_transform(texts)
+
         features = []
         labels = []
-        for i, speaker in enumerate(speakers):
-            utterance = speaker['speech_features']
+
+        for i, utterance in enumerate(utterances):
+            prosodic = utterance['prosodic_features']
             feat = [
-                utterance['duration'], utterance['speech_rate'], utterance['duration'], utterance['mean_pitch'], 
-                utterance['min_pitch'], utterance['max_pitch'], 
-                utterance['speech_sd'], utterance['intensityLevel'], 
-                utterance['intensity_level'], 
-                utterance['speechMax']], utterance['speechSD'],
+                prosodic['duration'],
+                prosodic['mean_pitch'],
+                prosodic['min_pitch'],
+                prosodic['max_pitch'],
+                prosodic['pitch_sd'],
+                prosodic['intensityMean'],
+                prosodic['intensityMin'],
+                prosodic['intensityMax'],
+                prosodic['intensitySD'],
             ]
-            feat.extend(X_text[i].toarray()[0])
-            doc = nlp(speaker['speech'])
-            speech.extend([
-                int(speaker['speech'].endswith('?'))),
-                len(re.findall(r'\b(why|how|what|when|where|who|which)\b', speaker['speech'].lower())),
-                len(speaker['speech'].split())),
-                sum(frequency for token in speech if token.pos_ == 'VERB'),
-                sum(frequency for token in speech if token.pos == 'NOUN')
+
+            feat.extend(X_text[i].toarray()[0].tolist())
+
+            doc = nlp(utterance['text'])
+            feat.extend([
+                int(utterance['text'].endswith('?')),
+                len(re.findall(r'\b(why|how|what|when|where|who|which)\b', utterance['text'].lower())),
+                len(utterance['text'].split()),
+                sum(1 for token in doc if token.pos_ == 'VERB'),
+                sum(1 for token in doc if token.pos_ == 'NOUN')
             ])
+
             features.append(feat)
-            labels.append((0 if i % 2 == 0 else 1))
+            labels.append(0 if i % 2 == 0 else 1)
+
         scaler = StandardScaler()
         X = scaler.fit_transform(features)
+
         clf = RandomForestClassifier(
-            n_estimators=150, max_depth=10, random_state=42, class_weight='balanced'
+            n_estimators=150,
+            max_depth=10,
+            random_state=42,
+            class_weight='balanced'
         )
         clf.fit(X, labels)
+
         joblib.dump(clf, os.path.join(OUTPUT_DIR, 'role_classifier.pkl'))
         joblib.dump(vectorizer, os.path.join(OUTPUT_DIR, 'text_vectorizer.pkl'))
         joblib.dump(scaler, os.path.join(OUTPUT_DIR, 'feature_scaler.pkl'))
+
         return clf, vectorizer, scaler
     except Exception as e:
         logger.error(f"Classifier training failed: {str(e)}")
         raise
 
-def classify_roles(speakers: List[Dict], clf, vectorizer, scaler):
+
+def classify_roles(utterances: List[Dict], clf, vectorizer, scaler):
     try:
-        speech = [u['speech'] for u in speakers]
-        X_text = vectorizer.transform(speech)
+        texts = [u['text'] for u in utterances]
+        X_text = vectorizer.transform(texts)
+
         results = []
-        for i, speaker in enumerate(speakers):
-            prosodic = speaker['speech_features']
+        for i, utterance in enumerate(utterances):
+            prosodic = utterance['prosodic_features']
             feat = [
-                prosodic['duration'], prosodic['mean_pitch'], prosodic['min_pitch'],
-                prosodic['max_pitch'], prosodic['pitch_sd'], prosodic['intensityMean'],
-                prosodic['intensityMin'], prosodic['intensityMax'], prosodic['intensitySD'],
+                prosodic['duration'],
+                prosodic['mean_pitch'],
+                prosodic['min_pitch'],
+                prosodic['max_pitch'],
+                prosodic['pitch_sd'],
+                prosodic['intensityMean'],
+                prosodic['intensityMin'],
+                prosodic['intensityMax'],
+                prosodic['intensitySD'],
             ]
+
             feat.extend(X_text[i].toarray()[0].tolist())
-            doc = nlp(speaker['speech'])
+
+            doc = nlp(utterance['text'])
             feat.extend([
-                int(speaker['speech'].endswith('?')),
-                len(re.findall(r'\b(why|how|what|when|where|who|which)\b', speaker['speech'].lower())),
-                len(speaker['speech'].split()),
+                int(utterance['text'].endswith('?')),
+                len(re.findall(r'\b(why|how|what|when|where|who|which)\b', utterance['text'].lower())),
+                len(utterance['text'].split()),
                 sum(1 for token in doc if token.pos_ == 'VERB'),
                 sum(1 for token in doc if token.pos_ == 'NOUN')
             ])
+
             X = scaler.transform([feat])
             role = 'Interviewer' if clf.predict(X)[0] == 0 else 'Interviewee'
-            results.append({**speaker, 'role': role})
+
+            results.append({**utterance, 'role': role})
+
         return results
     except Exception as e:
         logger.error(f"Role classification failed: {str(e)}")
         raise
 
-def analyze_interviewee_voice(audio_path: str, speakers: List[Dict]) -> Dict:
+
+
+
+def analyze_interviewee_voice(audio_path: str, utterances: List[Dict]) -> Dict:
     try:
         y, sr = librosa.load(audio_path, sr=16000)
-        interviewee_speakers = [u for u in speakers if u['role'] == 'Interviewee']
-        if not interviewee_speakers:
-            return {'error': 'No interviewee speeches found'}
+
+        interviewee_utterances = [u for u in utterances if u['role'] == 'Interviewee']
+        if not interviewee_utterances:
+            return {'error': 'No interviewee utterances found'}
+
         segments = []
-        for u in interviewee_speakers:
+        for u in interviewee_utterances:
             start = int(u['start'] * sr / 1000)
             end = int(u['end'] * sr / 1000)
             segments.append(y[start:end])
-        total_duration = sum(u['speech_features']['duration'] for u in interviewee_speakers)
-        total_words = sum(len(u['speech'].split()) for u in interviewee_speakers)
+
+        combined_audio = np.concatenate(segments)
+
+        total_duration = sum(u['prosodic_features']['duration'] for u in interviewee_utterances)
+        total_words = sum(len(u['text'].split()) for u in interviewee_utterances)
         speaking_rate = total_words / total_duration if total_duration > 0 else 0
+
         filler_words = ['um', 'uh', 'like', 'you know', 'so', 'i mean']
-        filler_count = sum(sum(u['speech'].lower().count(fw) for fw in filler_words) for u in interviewee_speakers)
+        filler_count = sum(
+            sum(u['text'].lower().count(fw) for fw in filler_words)
+            for u in interviewee_utterances
+        )
         filler_ratio = filler_count / total_words if total_words > 0 else 0
-        all_words = ' '.join(u['speech'].lower() for u in interviewee_speakers).split()
+
+        all_words = ' '.join(u['text'].lower() for u in interviewee_utterances).split()
         word_counts = {}
         for i in range(len(all_words) - 1):
             bigram = (all_words[i], all_words[i + 1])
             word_counts[bigram] = word_counts.get(bigram, 0) + 1
-        repetition_score = sum(1 for count in word_counts.values() if count > 1) / len(word_counts) if word_counts else 0
+        repetition_score = sum(1 for count in word_counts.values() if count > 1) / len(
+            word_counts) if word_counts else 0
+
         pitches = []
         for segment in segments:
             f0, voiced_flag, _ = librosa.pyin(segment, fmin=80, fmax=300, sr=sr)
             pitches.extend(f0[voiced_flag])
+
         pitch_mean = np.mean(pitches) if len(pitches) > 0 else 0
         pitch_std = np.std(pitches) if len(pitches) > 0 else 0
         jitter = np.mean(np.abs(np.diff(pitches))) / pitch_mean if len(pitches) > 1 and pitch_mean > 0 else 0
+
         intensities = []
         for segment in segments:
             rms = librosa.feature.rms(y=segment)[0]
             intensities.extend(rms)
+
         intensity_mean = np.mean(intensities) if intensities else 0
         intensity_std = np.std(intensities) if intensities else 0
-        shimmer = np.mean(np.abs(np.diff(intensities))) / intensity_mean if len(intensities) > 1 and intensity_mean > 0 else 0
+        shimmer = np.mean(np.abs(np.diff(intensities))) / intensity_mean if len(
+            intensities) > 1 and intensity_mean > 0 else 0
+
         anxiety_score = 0.6 * (pitch_std / pitch_mean) + 0.4 * (jitter + shimmer) if pitch_mean > 0 else 0
         confidence_score = 0.7 * (1 / (1 + intensity_std)) + 0.3 * (1 / (1 + filler_ratio))
         hesitation_score = filler_ratio + repetition_score
-        anxiety_level = 'High' if anxiety_score > 0.15 else 'Moderate' if anxiety_score > 0.07 else 'Low'
-        confidence_level = 'High' if confidence_score > 0.7 else 'Moderate' if confidence_score > 0.5 else 'Low'
-        fluency_level = 'Fluent' if (filler_ratio < 0.05 and repetition_score < 0.1) else 'Moderate' if (filler_ratio < 0.1 and repetition_score < 0.2) else 'Disfluent'
+
+        anxiety_level = 'high' if anxiety_score > 0.15 else 'moderate' if anxiety_score > 0.07 else 'low'
+        confidence_level = 'high' if confidence_score > 0.7 else 'moderate' if confidence_score > 0.5 else 'low'
+        fluency_level = 'fluent' if (filler_ratio < 0.05 and repetition_score < 0.1) else 'moderate' if (
+                    filler_ratio < 0.1 and repetition_score < 0.2) else 'disfluent'
+
         return {
             'speaking_rate': float(round(speaking_rate, 2)),
             'filler_ratio': float(round(filler_ratio, 4)),
             'repetition_score': float(round(repetition_score, 4)),
-            'pitch_analysis': {'mean': float(round(pitch_mean, 2)), 'std_dev': float(round(pitch_std, 2)), 'jitter': float(round(jitter, 4))},
-            'intensity_analysis': {'mean': float(round(intensity_mean, 2)), 'std_dev': float(round(intensity_std, 2)), 'shimmer': float(round(shimmer, 4))},
-            'composite_scores': {'anxiety': float(round(anxiety_score, 4)), 'confidence': float(round(confidence_score, 4)), 'hesitation': float(round(hesitation_score, 4))},
-            'interpretation': {'anxiety_level': anxiety_level, 'confidence_level': confidence_level, 'fluency_level': fluency_level}
+            'pitch_analysis': {
+                'mean': float(round(pitch_mean, 2)),
+                'std_dev': float(round(pitch_std, 2)),
+                'jitter': float(round(jitter, 4))
+            },
+            'intensity_analysis': {
+                'mean': float(round(intensity_mean, 2)),
+                'std_dev': float(round(intensity_std, 2)),
+                'shimmer': float(round(shimmer, 4))
+            },
+            'composite_scores': {
+                'anxiety': float(round(anxiety_score, 4)),
+                'confidence': float(round(confidence_score, 4)),
+                'hesitation': float(round(hesitation_score, 4))
+            },
+            'interpretation': {
+                'anxiety_level': anxiety_level,
+                'confidence_level': confidence_level,
+                'fluency_level': fluency_level
+            }
         }
     except Exception as e:
         logger.error(f"Voice analysis failed: {str(e)}")
         return {'error': str(e)}
 
-def generate_voice_interpretation(analysis: Dict) -> str:
-    if 'error' in analysis:
-        return "Voice analysis unavailable due to processing limitations."
-    interpretation_lines = [
-        "Vocal Performance Profile:",
-        f"- Speaking Rate: {analysis['speaking_rate']} words/sec - Benchmark: 2.0-3.0 wps for clear delivery",
-        f"- Filler Word Frequency: {analysis['filler_ratio'] * 100:.1f}% - Measures non-content words",
-        f"- Repetition Index: {analysis['repetition_score']:.3f} - Frequency of repeated phrases",
-        f"- Anxiety Indicator: {analysis['interpretation']['anxiety_level']} (Score: {analysis['composite_scores']['anxiety']:.3f}) - Pitch and vocal stability",
-        f"- Confidence Indicator: {analysis['interpretation']['confidence_level']} (Score: {analysis['composite_scores']['confidence']:.3f}) - Vocal strength",
-        f"- Fluency Rating: {analysis['interpretation']['fluency_level']} - Speech flow and coherence",
-        "",
-        "HR Insights:",
-        "- Rapid speech (>3.0 wps) may signal enthusiasm but risks clarity.",
-        "- High filler word use reduces perceived professionalism.",
-        "- Elevated anxiety suggests pressure; training can build resilience.",
-        "- Strong confidence aligns with leadership presence.",
-        "- Fluent speech enhances engagement, critical for team roles."
-    ]
-    return "\n".join(interpretation_lines)
 
 def generate_anxiety_confidence_chart(composite_scores: Dict, chart_path_or_buffer):
     try: