live.py

import warnings
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 functions.facs import *
# 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


def analyze_live_video(video_path: str, uid: str, user_id: str, count: int, final: bool, log: Callable[[str], None]):
	try:
		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)
		models_folder = 'models'
		print(models_folder)
		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) 
				
		speech_model=os.path.join(models_folder,'speech.keras')
		fer_model=os.path.join(models_folder,'22.6_AffectNet_10K_part2.pt')
		valence_arousal_model=os.path.join(models_folder,'emotion_model.pt')
		facs_model=os.path.join(models_folder,'incept_v3_10fps_full_dp0.4.keras')
		val_ar_feat_path=os.path.join(models_folder,'resnet_features.pt')
		valence_dict_path=os.path.join(models_folder,'valence-NRC-VAD-Lexicon.txt')
		arousal_dict_path=os.path.join(models_folder,'arousal-NRC-VAD-Lexicon.txt')
		dominance_dict_path=os.path.join(models_folder,'dominance-NRC-VAD-Lexicon.txt')
		dnn_net = cv2.dnn.readNetFromCaffe(os.path.join(models_folder,"deploy.prototxt"), os.path.join(models_folder,"res10_300x300_ssd_iter_140000.caffemodel"))
		predictor = dlib.shape_predictor(os.path.join(models_folder,"shape_predictor_68_face_landmarks.dat"))

		eye=[]
		fer=[]
		blinks=[]
		class_wise_frame_counts=[]
		ser_major_emotions=[]
		speech_data=[]
		speech_emotions=[]
		word_weights_list=[ ]
		# Load previous data
		if final:
			print("Gathering data from previous runs")
			log(f"Gathering data from previous runs")
			for i in range(count):
				folder_name = f"{i+1}"
				folder_path = os.path.join(output_dir, folder_name)
				if os.path.isdir(folder_path):
					for file in os.listdir(folder_path):
						file_path = os.path.join(folder_path, file)
						if file.endswith('.csv'):
							df = pd.read_csv(file_path)
							if 'eye' in file:
								eye.append(df)
							elif 'fer' in file:
								fer.append(df)
							elif 'speech' in file:
								speech_emotions.append(df)
						elif file == 'meta_data.json':
							with open(file_path, 'r') as json_file:
								data = json.load(json_file)
								
								blinks.append(data['eye_emotion_recognition']['blink_durations'])
								class_wise_frame_counts.append(data['facial_emotion_recognition']['class_wise_frame_count'])
								ser_major_emotions.append(data['speech_emotion_recognition']['major_emotion'])
								speech_data.append([data['speech_emotion_recognition']['pause_length'], data['speech_emotion_recognition']['articulation_rate'], data['speech_emotion_recognition']['speaking_rate']])
								word_weights_list.append(data['speech_emotion_recognition']['word_weights'])
				
		folder_name = f"{count}"
		folder_path = os.path.join(output_dir, folder_name)
		os.makedirs(folder_path, exist_ok=True)
		print(folder_path)
		#output paths
		fer_log_path = os.path.join(folder_path,"fer_log.csv")
		Speech_log_path = os.path.join(folder_path,"speech_log.csv")
		eye_log_path = os.path.join(folder_path,"eyetrack_log.csv")
		word_path = os.path.join(folder_path,"word_log.csv")
		json_path = os.path.join(folder_path, "meta_data.json")
		valence_plot=os.path.join(folder_path,"valence.png")
		arousal_plot=os.path.join(folder_path,"arousal.png")
		stress_plot=os.path.join(folder_path,"stress.png")
		facs_log=os.path.join(folder_path,'facs_log.csv')

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

		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.')



		#FACS PREDICT
		facs_df=facs_pred(faces,facs_model)
		facs_df.to_csv(facs_log,index=False)
		print('facs log saved to ',facs_log)

		##EYE TRACKING
		fc=Facetrack()
		column=['timestamp','total_blinks']
		log(f"Extracting eye features for question - {count}")
		preds,blink_durations,total_blinks=eye_track_predict(fc,faces,fps)
		print("total_blinks- ",total_blinks)
		eye_df=pd.DataFrame(preds,columns=column)
		eye_df.to_csv(eye_log_path,index=False)


		#FACIAL EXPRESSION RECOGNITION
		log(f"Extracting facial features for question - {count}")
		fer_df,class_wise_frame_count,em_tensors=fer_predict(faces,fps,fer_model)
		valence_list,arousal_list,stress_list=va_predict(valence_arousal_model,val_ar_feat_path,faces,list(em_tensors))
		fer_df['arousal']=arousal_list
		fer_df['valence']=valence_list
		fer_df['stress']=stress_list
		timestamps=list(fer_df['timestamp'])
		frame_index=[i+1 for i in range(len(timestamps))]
		fer_df.to_csv(fer_log_path, index=False)
		plot_graph(filter(frame_index,valence_list),'valence',valence_plot)
		plot_graph(filter(frame_index,arousal_list),'arousal',arousal_plot)
		plot_graph(filter(frame_index,stress_list),'Stress',stress_plot)

		#SPEECH EMOTION RECOGNITION
		log(f"Extracting speech features for question - {count}")
		emotions_df,major_emotion,word=speech_predict(audio_path,speech_model,valence_dict_path,arousal_dict_path,dominance_dict_path,word_path)
		emotions_df.to_csv(Speech_log_path, index=False)


		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
		# try:
		# 	avg_blink_duration=float(sum(blink_durations)/(len(blink_durations)))
		# 	meta_data['avg_blink_durations']=avg_blink_duration
		# except Exception as e:
		# 	print(f"An error occurred: {e}")

		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']
		}
		with open(json_path, 'w') as json_file:
			json.dump(meta_data, json_file)

		# Save CSV Logs
		eye.append(eye_df)
		fer.append(fer_df)
		blinks.append(blink_durations)
		class_wise_frame_counts.append(class_wise_frame_count)
		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)])
		ser_major_emotions.append(major_emotion)
		speech_emotions.append(emotions_df)
		word_weights_list.append(word['word_weights'])

		file_path=audio_path
		if os.path.exists(file_path):
				os.remove(file_path)
		file_path='segment.wav'
		if os.path.exists(file_path):
				os.remove(file_path)

		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")


		combined_json_path = os.path.join(output_dir, "combined_data.json")
		combined_valence_path = os.path.join(output_dir, "comb_valence.png")
		combined_arousal_path = os.path.join(output_dir, "comb_arousal.png")
		combined_stress_path=os.path.join(output_dir,'combined_stress.png')
		# Process each DataFrame and update timestamps sequentially
		current_max = 0
		for i, df in enumerate(fer):
				df['timestamp'] = df['timestamp'] + current_max
				current_max = df['timestamp'].max()
		combined_fer_df = pd.concat(fer).reset_index(drop=True)
		combined_fer_df.to_csv(os.path.join(output_dir,'fer_combined.csv'),index=False)

		current_max = 0
		for i, df in enumerate(speech_emotions):
			df['timestamp'] = df['timestamp'] + current_max
			current_max = df['timestamp'].max()
			combined_speech_df=pd.concat(speech_emotions).reset_index(drop=True)
			combined_speech_df.to_csv(os.path.join(output_dir,'speech_combined.csv'),index=False)

		combined_valence=list(combined_fer_df['valence'])
		combined_arousal=list(combined_fer_df['arousal'])
		combined_stress=list(combined_fer_df['stress'])
		timestamps=list(combined_fer_df['timestamp'])
		frame_index=[i+1 for i in range(len(timestamps))]
		plot_graph(filter(frame_index,combined_valence),'combined_valence',combined_valence_path)
		plot_graph(filter(frame_index,combined_arousal),'combined_arousal',combined_arousal_path)	
		plot_graph(filter(frame_index,combined_stress),'Stress',combined_stress_path)

		combined_meta_data={}
		current_max = 0
		c1=0
		combined_weights = Counter()
		for word_weight in word_weights_list:
			combined_weights.update(word_weight)
		combined_weights_dict = dict(combined_weights)

		# Process each DataFrame and update timestamps sequentially
		print(f"lenght of eye :{len(eye)}, length of blinks :{len(blinks)}")
		for i,df in enumerate(eye):
			df['timestamp'] = df['timestamp'] +current_max
			current_max = df['timestamp'].max()
			add_value=c1
			def add_integer(val):
				if isinstance(val,(int,float)):
					return val+add_value
				return val
			df['total_blinks'] = df['total_blinks'].apply(add_integer)
			c1=c1+len(blinks[i])

		# Combine the DataFrames into one
		combined_eye_df = pd.concat(eye).reset_index(drop=True)
		combined_eye_df.to_csv(os.path.join(output_dir,'eye_combined.csv'),index=False)
		flattened_list = [item for sublist in blinks for item in sublist]
		try:
			avg_blink_duration=float(sum(flattened_list)/len(flattened_list))
		except:
			avg_blink_duration=0
		numeric_values = pd.to_numeric(combined_eye_df['total_blinks'], errors='coerce')
		max_value = numeric_values.max()
		dict_list = class_wise_frame_counts

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

		combined_meta_data={}
		combined_meta_data['facial_emotion_recognition']={
			'class_wise_frame_count': result
		}
		combined_meta_data['eye_emotion_recognition'] = {
			'avg_blink_duration': avg_blink_duration,
			'total_blinks': int(max_value)
		}
		combined_meta_data['speech_emotion_recognition'] = {
			'major_emotion': str(statistics.mode(ser_major_emotions)),
			'pause_length': statistics.mean([row[0] for row in speech_data]),
			'articulation_rate': statistics.mean([row[1] for row in speech_data]),
			'speaking_rate': statistics.mean([row[2] for row in speech_data]),
			'word_weights':combined_weights_dict
		}

		with open(combined_json_path, 'w') as json_file:
			json.dump(combined_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_dir)
		print(f"Deleted output directory: {output_dir}")
		

# st=time.time()
# # analyze_live_video(video_path: video_path=, uid: str, user_id: str, count: int, final: bool, log: Callable[[str], None])
# analyze_live_video('videos/s2.webm', 1,1,1,True,print)
# analyze_live_video('videos/s2.webm', 1,1,2,True,print)
# print("time taken - ",time.time()-st)