Delete app.py

app.py

@@ -1,213 +0,0 @@
-# -*- coding: utf-8 -*-
-# Install required libraries if running outside Colab
-# !pip install gradio yt-dlp moviepy pillow speechrecognition llama-index lancedb google-generativeai
-
-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
-
-# 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)
-
-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_video(video_url,
-                 output_video_path="./video_data/",
-                 output_folder="./mixed_data/",
-                 output_audio_path="./mixed_data/output_audio.wav"):
-    filepath = os.path.join(output_video_path, "input_vid.mp4")
-    #meta = download_video(video_url, output_video_path)
-    video_to_images(filepath, output_folder)
-    video_to_audio(filepath, output_audio_path)
-    text_data = audio_to_text(output_audio_path)
-    text_path = os.path.join(output_folder, "output_text.txt")
-    with open(text_path, "w") as file:
-        file.write(text_data if text_data else "")
-    os.remove(output_audio_path)
-    meta = {
-            "title": "test",
-            "uploader": "uploader",
-            "views": "view_count",
-        }
-    return meta, text_data
-
-from llama_index.core.indices import MultiModalVectorStoreIndex
-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(image_txt_folder_path: str):
-    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"
-    )
-    documents = SimpleDirectoryReader(image_txt_folder_path).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
-
-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: "
-)
-
-import google.generativeai as genai
-
-def get_response(retriever_engine, query_str, metadata_str, output_folder):
-    img, txt = retrieve(retriever_engine=retriever_engine, query_str=query_str)
-    context_str = "".join(txt)
-    prompt = qa_tmpl_str.format(
-        context_str=context_str, query_str=query_str, metadata_str=metadata_str
-    )
-
-    GOOGLE_API_KEY = "AIzaSyD0sn-z1CmYcyhzSyE_4t2_nSQFGmnKFWc"
-    genai.configure(api_key=GOOGLE_API_KEY)
-    gemini_model = genai.GenerativeModel('gemini-1.5-flash-latest')
-
-
-    content_parts = [prompt]
-    image_paths = []
-    for img_path in img:
-        try:
-            image = Image.open(img_path)
-            content_parts.append(image)
-            image_paths.append(img_path)
-        except Exception as e:
-            print(f"Error loading image {img_path}: {e}")
-
-    response_1 = gemini_model.generate_content(content_parts)
-    result_text = response_1.text if hasattr(response_1, 'text') else str(response_1)
-    return result_text, image_paths
-
-# Gradio interface function
-def gradio_chat(query):
-    output_video_path = "./video_data/"
-    output_folder = "./mixed_data/"
-    output_audio_path = "./mixed_data/output_audio.wav"
-    video_url=""
-
-    try:
-        metadata_vid, text_data = prepare_video(
-            video_url, output_video_path, output_folder, output_audio_path
-        )
-        metadata_str = json.dumps(metadata_vid)
-        retriever_engine = create_vector_db(output_folder)
-        result_text, image_paths = get_response(
-            retriever_engine, query, metadata_str, output_folder
-        )
-        # Cleanup
-        #if os.path.exists(output_video_path):
-        #   shutil.rmtree(output_video_path)
-        #if os.path.exists(output_folder):
-        #    shutil.rmtree(output_folder)
-        # Gradio can return text plus images (as list of file paths)
-        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="Try:Beaches in Kerala"),
-    ],
-    outputs=[
-        gr.Textbox(label="Vega Response"),
-        gr.Gallery(label="Relevant Images", allow_preview=True),
-    ],
-    title="",
-    description=""
-)
-
-if __name__ == "__main__":
-    gradio_ui.launch(share=True)