live.py

import warnings
from functions.models import models_dict
warnings.filterwarnings('ignore', category=UserWarning, module='tensorflow')
import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
os.environ['TF_ENABLE_ONEDNN_OPTS'] = '0'
import logging
logging.getLogger('absl').setLevel(logging.ERROR)
from moviepy.editor import VideoFileClip
import pandas as pd
from tqdm import tqdm
import time
import json
import cv2
import dlib
from collections import Counter
import statistics
import shutil
import asyncio
import traceback

from functions.valence_arousal import va_predict
from functions.speech import speech_predict
from functions.eye_track import Facetrack, eye_track_predict
from functions.fer import extract_face,fer_predict,plot_graph,filter
# from app.utils.session import send_analytics, send_individual_analytics_files, send_combined_analytics_files, send_error
# from app.utils.socket import ConnectionManager
from typing import Callable
session_data={}
dnn_net=models_dict['face'][0]
predictor=models_dict['face'][1]
speech_model=models_dict['speech']
valence_dict_path=models_dict['vad'][0]
arousal_dict_path=models_dict['vad'][1]
dominance_dict_path=models_dict['vad'][2]
valence_arousal_model=models_dict['valence_fer'][1]
val_ar_feat_model=models_dict['valence_fer'][0]
fer_model=models_dict['fer']

def analyze_live_video(video_path: str, uid: str, user_id: str, count: int, final: bool, log: Callable[[str], None]):
	try:
		#initilalizing lists 
		global session_data
		if uid not in session_data:
			session_data[uid] = {
					"vcount":[],
					"duration":[],

					"eye": [],

					"fer": [],
					"valence":[],
					"arousal":[],
					"stress":[],

					"blinks": [],
					"class_wise_frame_counts": [],

					"speech_emotions": [],
					"speech_data":[],
					"word_weights_list": []
				}
		print(f"UID: {uid}, User ID: {user_id}, Count: {count}, Final: {final}, Video: {video_path}")
		log(f"Analyzing video for question - {count}")

		output_dir = os.path.join('output',str(uid))
		print(output_dir)
		if not os.path.exists(output_dir):
				os.makedirs(output_dir)
		# Wait for previous files to be written if final
		if final and count > 1:
			for i in range(1, count):
				previous_file_name = os.path.join(output_dir, f"{i}.json")
				print(previous_file_name)
				while not os.path.exists(previous_file_name):
					time.sleep(1) 

		video_clip = VideoFileClip(video_path)
		video_clip = video_clip.set_fps(30)
		print("Duration: ", video_clip.duration)
		session_data[uid]['vcount'].append(count)
		session_data[uid]['duration'].append(video_clip.duration)
		fps = video_clip.fps
		audio = video_clip.audio
		audio_path = os.path.join(output_dir,'extracted_audio.wav')
		audio.write_audiofile(audio_path)
		video_frames = [frame for frame in video_clip.iter_frames()]

		#Face extraction
		print("extracting faces")
		faces=[extract_face(frame,dnn_net,predictor) for frame in tqdm(video_frames)]
		print(f'{len([face for face in faces if face is not None])} faces found.')


		##EYE TRACKING
		fc=Facetrack()
		log(f"Extracting eye features for question - {count}")
		eye_preds,blink_durations,total_blinks=eye_track_predict(fc,faces,fps)
		print(len(eye_preds))
		print("total_blinks- ",total_blinks)	
		session_data[uid]['eye'].append(eye_preds)
		session_data[uid]['blinks'].append(blink_durations)
		

		#FACIAL EXPRESSION RECOGNITION
		log(f"Extracting facial features for question - {count}")
		fer_emotions,class_wise_frame_count,em_tensors=fer_predict(faces,fps,fer_model)
		print("face emotions",len(fer_emotions))
		session_data[uid]['fer'].append(fer_emotions)
		session_data[uid]['class_wise_frame_counts'].append(class_wise_frame_count)
		
		#VALENCE AROUSAL STRESS
		valence_list,arousal_list,stress_list=va_predict(valence_arousal_model,val_ar_feat_model,faces,list(em_tensors))
		session_data[uid]['valence'].append(valence_list)
		session_data[uid]['arousal'].append(arousal_list)
		session_data[uid]['stress'].append(stress_list)
		log(f"Extracting speech features for question - {count}")
		emotions,major_emotion,word=speech_predict(audio_path,speech_model,valence_dict_path,arousal_dict_path,dominance_dict_path)
		session_data[uid]['speech_emotions'].append(emotions)
		session_data[uid]['word_weights_list'].append(word['word_weights'])
		session_data[uid]['speech_data'].append([float(word['average_pause_length'] if word and word['average_pause_length'] else 0),float(word['articulation_rate'] if word and word['articulation_rate'] else 0),float(word['speaking_rate'] if word and word['speaking_rate'] else 0)])
		log(f"Generating the metadata for question - {count}")

		
		# Create Meta Data
		meta_data={}
		try:
			avg_blink_duration= float(sum(blink_durations)/(len(blink_durations)))
		except:
			avg_blink_duration=0
		meta_data['vcount']=count
		meta_data['eye_emotion_recognition'] = {
			"blink_durations": blink_durations,
			"avg_blink_duration":avg_blink_duration,
			"total_blinks": total_blinks,
			"duration":video_clip.duration
		}
    
		meta_data['facial_emotion_recognition'] = {
			"class_wise_frame_count": class_wise_frame_count,
		}
		meta_data['speech_emotion_recognition'] = {
		'major_emotion':str(major_emotion),
		'pause_length':float(word['average_pause_length']),
		'articulation_rate':float(word['articulation_rate']),
		'speaking_rate':float(word['speaking_rate']),
		'word_weights':word['word_weights']
		}


		file_path=audio_path
		if os.path.exists(file_path):
				os.remove(file_path)
		print(f"{file_path} deleted")
		file_path='segment.wav'
		if os.path.exists(file_path):
				os.remove(file_path)
		print(f"{file_path} deleted")
		

		print("Individual: ", meta_data)
		if not final:
			print("Not final Executing")
			log(f"Saving analytics for question - {count}")
		# 	send_analytics(valence_plot, arousal_plot,{
		# 		"uid": uid,
		# 		"user_id": user_id, 
		# 		"individual": meta_data,
		# 		"count": count
		# 	})
			print("Sent analytics")
		# 	send_individual_analytics_files(uid, output_dir, count)
			dummy_file_path = os.path.join(output_dir, f'{count}.json')
			print("Writing dummy file: ", dummy_file_path)
			with open(dummy_file_path, 'w') as dummy_file:
				json.dump({"status": "completed"}, dummy_file)
			return
	
		# Process combined
		log(f"Processing gathered data for final output")

		vcount=session_data[uid]['vcount']
		sorted_indices = sorted(range(len(vcount)), key=lambda i: vcount[i])
		for key in session_data[uid]:
			# Only sort lists that are the same length as vcount
			if len(session_data[uid][key]) == len(vcount):
				session_data[uid][key] = [session_data[uid][key][i] for i in sorted_indices]

		videos=len(session_data[uid]['vcount'])
		#INDIV PLOT SAVING
		combined_speech=[]
		combined_valence=[]
		combined_arousal=[]
		combined_stress=[]
		combined_fer=[]
		combined_eye=[]
		vid_index=[]
		combined_speech=[]
		combined_blinks=[]
		for i in range(videos):
			for j in range(len(session_data[uid]['speech_emotions'][i])):
				vid_index.append(i+1)
			combined_speech+=session_data[uid]['speech_emotions'][i]
		timestamps=[i*3 for i in range(len(combined_speech))]
		df = pd.DataFrame({
			'timestamps':timestamps,
			'video_index':vid_index,
			'speech_emotion':combined_speech
		})
		df.to_csv(os.path.join(output_dir,'combined_speech.csv'), index=False)



		vid_index=[]
		for i in range(videos):
			timestamps=[j/30 for j in range(len(session_data[uid]['valence'][i]))]			
			for j in range(len(timestamps)):
				vid_index.append(i+1)
			folder_path=os.path.join(output_dir,f"{session_data[uid]['vcount'][i]}")
			os.makedirs(folder_path, exist_ok=True)
			plot_graph(timestamps,session_data[uid]['valence'][i],'valence',os.path.join(folder_path,'valence.png'))
			plot_graph(timestamps,session_data[uid]['arousal'][i],'arousal',os.path.join(folder_path,'arousal.png'))
			plot_graph(timestamps,session_data[uid]['stress'][i],'stress',os.path.join(folder_path,'stress.png'))
			combined_arousal+=session_data[uid]['arousal'][i]
			combined_valence+=session_data[uid]['valence'][i]
			combined_stress+=session_data[uid]['stress'][i]
			combined_fer+=session_data[uid]['fer'][i]
			combined_blinks+=session_data[uid]['blinks'][i]
			# combined_class_wise_frame_count+=session_data[uid]['class_wise_frame_counts'][i]
			try:
				max_value=max([x for x in combined_eye if isinstance(x, (int, float))])
			except:
				max_value=0
			session_data[uid]['eye'][i]=[x + max_value if isinstance(x, (int, float)) else x for x in session_data[uid]['eye'][i]]
			combined_eye+=session_data[uid]['eye'][i]
		
		timestamps=[i/fps for i in range(len(combined_arousal))]
		plot_graph(timestamps,combined_valence,'valence',os.path.join(output_dir,'valence.png'))
		plot_graph(timestamps,combined_arousal,'arousal',os.path.join(output_dir,'arousal.png'))
		plot_graph(timestamps,combined_stress,'stress',os.path.join(output_dir,'stress.png'))
		print(len(timestamps),len(vid_index),len(combined_fer),len(combined_valence),len(combined_arousal),len(combined_stress),len(combined_eye))
		df = pd.DataFrame({
			'timestamps':timestamps,
			'video_index': vid_index,  # Add a column for video index
			'fer': combined_fer,
			'valence': combined_valence,
			'arousal': combined_arousal,
			'stress': combined_stress,
			'eye': combined_eye,
		})
		df.to_csv(os.path.join(output_dir,'combined_data.csv'), index=False)

		#generate metadata for Combined
		comb_meta_data={}
		try:
			avg_blink_duration= float(sum(combined_blinks)/(len(combined_blinks)))
		except:
			avg_blink_duration=0

		total_blinks=max([x for x in combined_eye if isinstance(x, (int, float))])

		comb_meta_data['eye_emotion_recognition'] = {
			"avg_blink_duration":avg_blink_duration,
			"total_blinks": total_blinks,
		}

		dict_list = session_data[uid]['class_wise_frame_counts']

		result = {}
		for d in dict_list:
			for key,value in d.items():
				result[key]=result.get(key,0)+value
		comb_meta_data['facial_emotion_recognition'] = {
			"class_wise_frame_count": result,
		}

		combined_weights = Counter()
		for word_weight in session_data[uid]['word_weights_list']:
			combined_weights.update(word_weight)
		combined_weights_dict = dict(combined_weights)
		print(combined_weights_dict)
		comb_meta_data['speech_emotion_recognition'] = {
		'major_emotion':str(major_emotion),
		'pause_length':statistics.mean([row[0] for row in session_data[uid]['speech_data']]),
		'articulation_rate':statistics.mean([row[1] for row in session_data[uid]['speech_data']]),
		'speaking_rate':statistics.mean([row[2] for row in session_data[uid]['speech_data']]),
		'word_weights':combined_weights_dict
		}
		with open(os.path.join(output_dir,'combined.json'), 'w') as json_file:
			json.dump(comb_meta_data, json_file)
		log(f"Saving analytics for final output")
		# send_analytics(valence_plot, arousal_plot,{
		# 	"uid": uid,
		# 	"user_id": user_id, 
		# 	"individual": meta_data,
		# 	"combined": combined_meta_data,
		# 	"count": count
		# })
		# send_individual_analytics_files(uid, output_dir, count)
		# send_combined_analytics_files(uid, output_dir)

		# shutil.rmtree(output_dir)
		# print(f"Deleted output directory: {output_dir}")
	except Exception as e:
		print("Error analyzing video...: ", e)
		error_trace = traceback.format_exc()
		print("Error Trace: ", error_trace)
		log(f"Error analyzing video for question - {count}")
		# send_error(uid, {
		# 	"message": str(e),
		# 	"trace": error_trace
		# })
		shutil.rmtree('output')
		print(f"Deleted output directory: {output_dir}")
		

# st=time.time()
# # analyze_live_video(video_path, uid, user_id, count, final, log)
# analyze_live_video('videos/s2.webm', 1,1,1,False,print)
# analyze_live_video('videos/a4.webm', 1,1,2,True,print)

# analyze_live_video('videos/s2.webm', 1,1,2,True,print)
# print("time taken - ",time.time()-st)