Upload app.py

app.py

@@ -1,269 +1,72 @@
-# -*- coding: utf-8 -*-
-# Install required libraries if running outside Colab
-# !pip install gradio yt-dlp moviepy pillow speechrecognition llama-index lancedb google-generativeai
-
+_H='./mixed_data/'
+_G='text'
+_F='uploader'
+_E='title'
+_D='./video_data/'
+_C='value'
+_B='label'
+_A=True
 import gradio as gr
 from moviepy import VideoFileClip
 from pathlib import Path
 import speech_recognition as sr
 from PIL import Image
-import os
-import shutil
-import json
-import matplotlib.pyplot as plt
-import yt_dlp
-import requests
-import base64
+import os,shutil,json,matplotlib.pyplot as plt,yt_dlp,requests,base64
 from io import BytesIO
-
-# Add your existing methods here (download_video, video_to_images, video_to_audio, audio_to_text, prepare_video...)
-
 def plot_images(image_paths):
-    images_shown = 0
-    plt.figure(figsize=(16, 9))
-    img_files = []
-    for img_path in image_paths:
-        if os.path.isfile(img_path):
-            img_files.append(img_path)
-            images_shown += 1
-            if images_shown >= 7:
-                break
-    return img_files
-
-def download_video(video_url, output_video_path="./video_data/"):
-    ydl_opts = {
-        "format": "bestvideo+bestaudio/best",
-        "merge_output_format": "mp4",
-        "outtmpl": f"{output_video_path}/input_vid.mp4",
-        "noplaylist": True,
-        "quiet": False,
-        # Uncomment and set your cookie file path if required
-        # "cookiefile": "cookies.txt",
-    }
-    Path(output_video_path).mkdir(parents=True, exist_ok=True)
-    with yt_dlp.YoutubeDL(ydl_opts) as ydl:
-        info = ydl.extract_info(video_url, download=True)
-        info = ydl.sanitize_info(info)
-        return {
-            "title": info.get("title"),
-            "uploader": info.get("uploader"),
-            "views": info.get("view_count"),
-        }
-
-def video_to_images(video_path, output_folder):
-    Path(output_folder).mkdir(parents=True, exist_ok=True)
-    clip = VideoFileClip(video_path)
-    clip.write_images_sequence(
-        os.path.join(output_folder, "frame%04d.png"), fps=0.2
-    )
-
-def video_to_audio(video_path, output_audio_path):
-    clip = VideoFileClip(video_path)
-    audio = clip.audio
-    audio.write_audiofile(output_audio_path)
-
+	A=0;plt.figure(figsize=(16,9));B=[]
+	for C in image_paths:
+		if os.path.isfile(C):
+			B.append(C);A+=1
+			if A>=7:break
+	return B
+def download_video(video_url,output_video_path=_D):
+	B=output_video_path;D={'format':'bestvideo+bestaudio/best','merge_output_format':'mp4','outtmpl':f"{B}/input_vid.mp4",'noplaylist':_A,'quiet':False};Path(B).mkdir(parents=_A,exist_ok=_A)
+	with yt_dlp.YoutubeDL(D)as C:A=C.extract_info(video_url,download=_A);A=C.sanitize_info(A);return{_E:A.get(_E),_F:A.get(_F),'views':A.get('view_count')}
+def video_to_images(video_path,output_folder):A=output_folder;Path(A).mkdir(parents=_A,exist_ok=_A);B=VideoFileClip(video_path);B.write_images_sequence(os.path.join(A,'frame%04d.png'),fps=.2)
+def video_to_audio(video_path,output_audio_path):A=VideoFileClip(video_path);B=A.audio;B.write_audiofile(output_audio_path)
 def audio_to_text(audio_path):
-    recognizer = sr.Recognizer()
-    try:
-        with sr.AudioFile(audio_path) as source:
-            audio_data = recognizer.record(source)
-            text = recognizer.recognize_google(audio_data)
-            return text
-    except sr.UnknownValueError:
-        print("Google Speech Recognition could not understand the audio.")
-    except sr.RequestError as e:
-        print(f"Could not request results: {e}")
-    return None
-
-def prepare_all_videos(
-    video_folder="./video_data/",
-    output_folder="./mixed_data/"
-):
-    """
-    Processes all video files in video_folder, extracting images and text for each,
-    and stores them in unique subfolders under output_folder.
-    Returns a list of metadata dicts for all videos.
-    """
-    Path(output_folder).mkdir(parents=True, exist_ok=True)
-    video_files = [f for f in os.listdir(video_folder) if f.lower().endswith(('.mp4', '.mov', '.avi', '.mkv'))]
-    all_metadata = []
-    for video_file in video_files:
-        video_path = os.path.join(video_folder, video_file)
-        video_name = Path(video_file).stem
-        video_output_folder = os.path.join(output_folder, video_name)
-        Path(video_output_folder).mkdir(parents=True, exist_ok=True)
-        audio_path = os.path.join(video_output_folder, "output_audio.wav")
-        # Extract images and audio
-        video_to_images(video_path, video_output_folder)
-        video_to_audio(video_path, audio_path)
-        # Transcribe audio
-        text_data = audio_to_text(audio_path)
-        text_path = os.path.join(video_output_folder, "output_text.txt")
-        with open(text_path, "w") as file:
-            file.write(text_data if text_data else "")
-        os.remove(audio_path)
-        # Dummy metadata, you can enhance this as needed
-        meta = {
-            "title": video_name,
-            "uploader": "unknown",
-            "views": "unknown",
-            "file": video_file
-        }
-        all_metadata.append({"meta": meta, "text": text_data, "folder": video_output_folder})
-    return all_metadata
-
+	A=sr.Recognizer()
+	try:
+		with sr.AudioFile(audio_path)as B:C=A.record(B);D=A.recognize_google(C);return D
+	except sr.UnknownValueError:print('Google Speech Recognition could not understand the audio.')
+	except sr.RequestError as E:print(f"Could not request results: {E}")
+def prepare_all_videos(video_folder=_D,output_folder=_H):
+	'\n    Processes all video files in video_folder, extracting images and text for each,\n    and stores them in unique subfolders under output_folder.\n    Returns a list of metadata dicts for all videos.\n    ';J='unknown';F=output_folder;E=video_folder;Path(F).mkdir(parents=_A,exist_ok=_A);K=[A for A in os.listdir(E)if A.lower().endswith(('.mp4','.mov','.avi','.mkv'))];G=[]
+	for B in K:
+		H=os.path.join(E,B);I=Path(B).stem;A=os.path.join(F,I);Path(A).mkdir(parents=_A,exist_ok=_A);C=os.path.join(A,'output_audio.wav');video_to_images(H,A);video_to_audio(H,C);D=audio_to_text(C);L=os.path.join(A,'output_text.txt')
+		with open(L,'w')as M:M.write(D if D else'')
+		os.remove(C);N={_E:I,_F:J,'views':J,'file':B};G.append({'meta':N,_G:D,'folder':A})
+	return G
 from llama_index.core.indices import MultiModalVectorStoreIndex
-from llama_index.core import SimpleDirectoryReader, StorageContext
+from llama_index.core import SimpleDirectoryReader,StorageContext
 from llama_index.vector_stores.lancedb import LanceDBVectorStore
 from llama_index.embeddings.huggingface import HuggingFaceEmbedding
 from llama_index.core import Settings
-
-def create_vector_db_for_all(image_txt_root_folder: str):
-    """
-    Loads all subfolders in image_txt_root_folder as documents for the vector DB.
-    """
-    text_store = LanceDBVectorStore(uri="lancedb", table_name="text_collection")
-    image_store = LanceDBVectorStore(uri="lancedb", table_name="image_collection")
-    storage_context = StorageContext.from_defaults(
-        vector_store=text_store, image_store=image_store
-    )
-    Settings.embed_model = HuggingFaceEmbedding(
-        model_name="sentence-transformers/all-MiniLM-L6-v2"
-    )
-    # Load all subfolders as documents
-    documents = []
-    for subfolder in Path(image_txt_root_folder).iterdir():
-        if subfolder.is_dir():
-            documents.extend(SimpleDirectoryReader(str(subfolder)).load_data())
-    index = MultiModalVectorStoreIndex.from_documents(
-        documents,
-        storage_context=storage_context,
-    )
-    retriever_engine = index.as_retriever(
-        similarity_top_k=2, image_similarity_top_k=3
-    )
-    return retriever_engine
-
+def create_vector_db_for_all(image_txt_root_folder):
+	'\n    Loads all subfolders in image_txt_root_folder as documents for the vector DB.\n    ';C='lancedb';D=LanceDBVectorStore(uri=C,table_name='text_collection');E=LanceDBVectorStore(uri=C,table_name='image_collection');F=StorageContext.from_defaults(vector_store=D,image_store=E);Settings.embed_model=HuggingFaceEmbedding(model_name='sentence-transformers/all-MiniLM-L6-v2');A=[]
+	for B in Path(image_txt_root_folder).iterdir():
+		if B.is_dir():A.extend(SimpleDirectoryReader(str(B)).load_data())
+	G=MultiModalVectorStoreIndex.from_documents(A,storage_context=F);H=G.as_retriever(similarity_top_k=2,image_similarity_top_k=3);return H
 from llama_index.core.schema import ImageNode
-
-def retrieve(retriever_engine, query_str):
-    retrieval_results = retriever_engine.retrieve(query_str)
-    retrieved_image = []
-    retrieved_text = []
-    for res_node in retrieval_results:
-        if isinstance(res_node.node, ImageNode):
-            retrieved_image.append(res_node.node.metadata["file_path"])
-        else:
-            retrieved_text.append(res_node.text)
-    return retrieved_image, retrieved_text
-
-qa_tmpl_str = (
-    "Given the provided information, including relevant images and retrieved context from the video, \
- accurately and precisely answer the query without any additional prior knowledge.\n"
-    "Please ensure honesty and responsibility, refraining from any racist or sexist remarks.\n"
-    "---------------------\n"
-    "Context: {context_str}\n"
-    "Metadata for video: {metadata_str} \n"
-    "---------------------\n"
-    "Query: {query_str}\n"
-    "Answer: "
-)
-
-# Define model values and their corresponding labels
-available_models = [
-    {"value": "meta-llama/llama-4-maverick:free", "label": "Llama"},
-    {"value": "qwen/qwen2.5-vl-72b-instruct:free", "label": "Qwen"},
-    {"value": "google/gemma-3-27b-it:free", "label": "Gemma"},
-    {"value": "moonshotai/kimi-vl-a3b-thinking:free", "label": "Kimi"},
-    {"value": "google/gemini-2.0-flash-exp:free", "label": "Gemini"},
-    # Add more models here if needed
-]
-
-# Helper to get value from label or vice versa
-model_value_to_label = {item["value"]: item["label"] for item in available_models}
-model_label_to_value = {item["label"]: item["value"] for item in available_models}
-
-# Gradio interface function
-def gradio_chat(query, model_label):
-    output_video_path = "./video_data/"
-    output_folder = "./mixed_data/"
-
-    try:
-        # Process all videos
-        all_metadata = prepare_all_videos(output_video_path, output_folder)
-        # Combine metadata for all videos
-        metadata_str = json.dumps([item["meta"] for item in all_metadata])
-        retriever_engine = create_vector_db_for_all(output_folder)
-
-        img, txt = retrieve(retriever_engine=retriever_engine, query_str=query)
-        context_str = "".join(txt)
-        prompt = qa_tmpl_str.format(
-            context_str=context_str, query_str=query, metadata_str=metadata_str
-        )
-
-        OPENROUTER_API_KEY = os.environ['OPENROUTER_API_KEY']
-        headers = {
-            "Authorization": f"Bearer {OPENROUTER_API_KEY}",
-            "Content-Type": "application/json",
-            "HTTP-Referer": "<YOUR_SITE_URL>",
-            "X-Title": "<YOUR_SITE_NAME>",
-        }
-
-        model_name = model_label_to_value.get(model_label, available_models[0]["value"])
-
-        messages = [{"role": "user", "content": [{"type": "text", "text": prompt}]}]
-        image_paths = []
-        for img_path in img:
-            try:
-                image = Image.open(img_path)
-                buffered = BytesIO()
-                image.save(buffered, format="JPEG")
-                img_base64 = base64.b64encode(buffered.getvalue()).decode('utf-8')
-                messages[0]["content"].append({
-                    "type": "image_url",
-                    "image_url": {"url": f"data:image/jpeg;base64,{img_base64}"}
-                })
-                image_paths.append(img_path)
-            except Exception as e:
-                print(f"Error loading image {img_path}: {e}")
-
-        data = {
-            "model": model_name,
-            "messages": messages,
-        }
-
-        response = requests.post(
-            url="https://openrouter.ai/api/v1/chat/completions",
-            headers=headers,
-            data=json.dumps(data)
-        )
-        response.raise_for_status()
-        result_text = response.json()['choices'][0]['message']['content']
-
-        return result_text, image_paths
-    except Exception as e:
-        return f"Error: {str(e)}", []
-
-# Gradio UI
-
-gradio_ui = gr.Interface(
-    fn=gradio_chat,
-    inputs=[
-        gr.Textbox(label="",placeholder="Try: Best island in Maldives"),
-        gr.Dropdown(
-            choices=[item["label"] for item in available_models],
-            value=available_models[0]["label"],
-            label="Select Model:"
-        )
-    ],
-    outputs=[
-        gr.Textbox(label="Vega Response:"),
-        gr.Gallery(label="Relevant Images", allow_preview=True),
-    ],
-    title="",
-    description="",
-    theme = gr.themes.Default(primary_hue="sky")
-)
-
-if __name__ == "__main__":
-    gradio_ui.launch(share=True)
+def retrieve(retriever_engine,query_str):
+	D=retriever_engine.retrieve(query_str);B=[];C=[]
+	for A in D:
+		if isinstance(A.node,ImageNode):B.append(A.node.metadata['file_path'])
+		else:C.append(A.text)
+	return B,C
+qa_tmpl_str='Given the provided information, including relevant images and retrieved context from the video,  accurately and precisely answer the query without any additional prior knowledge.\nPlease ensure honesty and responsibility, refraining from any racist or sexist remarks.\n---------------------\nContext: {context_str}\nMetadata for video: {metadata_str} \n---------------------\nQuery: {query_str}\nAnswer: '
+available_models=[{_C:'meta-llama/llama-4-maverick:free',_B:'Llama'},{_C:'qwen/qwen2.5-vl-72b-instruct:free',_B:'Qwen'},{_C:'google/gemma-3-27b-it:free',_B:'Gemma'},{_C:'moonshotai/kimi-vl-a3b-thinking:free',_B:'Kimi'},{_C:'google/gemini-2.0-flash-exp:free',_B:'Gemini'}]
+model_value_to_label={A[_C]:A[_B]for A in available_models}
+model_label_to_value={A[_B]:A[_C]for A in available_models}
+def gradio_chat(query,model_label):
+	K='image_url';J='type';D=query;C='content';L=_D;E=_H
+	try:
+		M=prepare_all_videos(L,E);N=json.dumps([A['meta']for A in M]);O=create_vector_db_for_all(E);P,Q=retrieve(retriever_engine=O,query_str=D);R=''.join(Q);S=qa_tmpl_str.format(context_str=R,query_str=D,metadata_str=N);T=os.environ['OPENROUTER_API_KEY'];U={'Authorization':f"Bearer {T}",'Content-Type':'application/json','HTTP-Referer':'<YOUR_SITE_URL>','X-Title':'<YOUR_SITE_NAME>'};V=model_label_to_value.get(model_label,available_models[0][_C]);F=[{'role':'user',C:[{J:_G,_G:S}]}];G=[]
+		for A in P:
+			try:W=Image.open(A);H=BytesIO();W.save(H,format='JPEG');X=base64.b64encode(H.getvalue()).decode('utf-8');F[0][C].append({J:K,K:{'url':f"data:image/jpeg;base64,{X}"}});G.append(A)
+			except Exception as B:print(f"Error loading image {A}: {B}")
+		Y={'model':V,'messages':F};I=requests.post(url='https://openrouter.ai/api/v1/chat/completions',headers=U,data=json.dumps(Y));I.raise_for_status();Z=I.json()['choices'][0]['message'][C];return Z,G
+	except Exception as B:return f"Error: {str(B)}",[]
+gradio_ui=gr.Interface(fn=gradio_chat,inputs=[gr.Textbox(label='',placeholder='Try: Best island in Maldives'),gr.Dropdown(choices=[A[_B]for A in available_models],value=available_models[0][_B],label='Select Model:')],outputs=[gr.Textbox(label='Vega Response:'),gr.Gallery(label='Relevant Images',allow_preview=_A)],title='',description='',theme=gr.themes.Default(primary_hue='sky'))
+if __name__=='__main__':gradio_ui.launch(share=_A)