Create app.py

app.py

@@ -0,0 +1,310 @@
+import gradio as gr
+import requests
+import os
+from huggingface_hub import InferenceClient,HfApi
+import random
+import json
+import datetime
+import uuid
+import yt_dlp
+import cv2
+import whisper
+
+from agent import (
+    PREFIX,
+    COMPRESS_DATA_PROMPT,
+    COMPRESS_DATA_PROMPT_SMALL,
+    LOG_PROMPT,
+    LOG_RESPONSE,
+)
+client = InferenceClient(
+    "mistralai/Mixtral-8x7B-Instruct-v0.1"
+)
+
+#save_data=f'https://huggingface.co/datasets/{reponame}/raw/main/'
+#token_self = os.environ['HF_TOKEN']
+#api=HfApi(token=token_self)
+
+sizes = list(whisper._MODELS.keys())
+langs = ["none"] + sorted(list(whisper.tokenizer.LANGUAGES.values()))
+current_size = "base"
+loaded_model = whisper.load_model(current_size)
+
+VERBOSE = True
+MAX_HISTORY = 100
+MAX_DATA = 20000
+
+def dl(inp,img):
+    uid=uuid.uuid4()
+    fps="Error"
+    out = None
+    out_file=[]
+    if img == None and inp !="":
+        try:
+            inp_out=inp.replace("https://","")
+            inp_out=inp_out.replace("/","_").replace(".","_").replace("=","_").replace("?","_")
+            if "twitter" in inp:
+                os.system(f'yt-dlp "{inp}" --extractor-arg "twitter:api=syndication" --trim-filenames 160 -o "{uid}/{inp_out}.mp4" -S res,mp4 --recode mp4')  
+            else:
+                os.system(f'yt-dlp "{inp}" --trim-filenames 160 -o "{uid}/{inp_out}.mp4" -S res,mp4 --recode mp4')  
+                
+            out = f"{uid}/{inp_out}.mp4"
+            capture = cv2.VideoCapture(out)
+            fps = capture.get(cv2.CAP_PROP_FPS)
+            capture.release()
+        except Exception as e:
+            print(e)
+            out = None
+    elif img !=None and inp == "":
+        capture = cv2.VideoCapture(img)
+        fps = capture.get(cv2.CAP_PROP_FPS)
+        capture.release()
+        out = f"{img}"
+    return out
+  
+def csv(segments):
+    output = ""
+    for segment in segments:
+      output += f"{segment['start']},{segment['end']},{segment['text']}\n"
+    return output
+def transcribe(path,lang,size):
+    yield (None,[("","Transcribing Video...")])
+    #if size != current_size:
+    loaded_model = whisper.load_model(size)
+    current_size = size
+    results = loaded_model.transcribe(path, language=lang)
+    subs = ".csv"
+    if subs == "None":
+      yield results["text"],[("","Transcription Complete...")]
+    elif subs == ".csv":
+      yield csv(results["segments"]),[("","Transcription Complete...")]
+
+
+def format_prompt(message, history):
+  prompt = "<s>"
+  for user_prompt, bot_response in history:
+    prompt += f"[INST] {user_prompt} [/INST]"
+    prompt += f" {bot_response}</s> "
+  prompt += f"[INST] {message} [/INST]"
+  return prompt
+
+
+
+def run_gpt(
+    prompt_template,
+    stop_tokens,
+    max_tokens,
+    seed,
+    **prompt_kwargs,
+):
+    print(seed)
+    timestamp=datetime.datetime.now()
+    
+    generate_kwargs = dict(
+        temperature=0.9,
+        max_new_tokens=max_tokens,
+        top_p=0.95,
+        repetition_penalty=1.0,
+        do_sample=True,
+        seed=seed,
+    )
+    
+    content = PREFIX.format(
+        timestamp=timestamp,
+        purpose="Compile the provided data and complete the users task"
+    ) + prompt_template.format(**prompt_kwargs)
+    if VERBOSE:
+        print(LOG_PROMPT.format(content))
+    
+    
+    #formatted_prompt = format_prompt(f"{system_prompt}, {prompt}", history)
+    #formatted_prompt = format_prompt(f'{content}', history)
+
+    stream = client.text_generation(content, **generate_kwargs, stream=True, details=True, return_full_text=False)
+    resp = ""
+    for response in stream:
+        resp += response.token.text
+        #yield resp
+
+    if VERBOSE:
+        print(LOG_RESPONSE.format(resp))
+    return resp
+
+    
+def compress_data(c, instruct, history):
+    seed=random.randint(1,1000000000)
+    
+    print (f'c:: {c}')
+    #tot=len(purpose)
+    #print(tot)
+    divr=int(c)/MAX_DATA
+    divi=int(divr)+1 if divr != int(divr) else int(divr)
+    chunk = int(int(c)/divr)
+    print(f'chunk:: {chunk}')
+    print(f'divr:: {divr}')
+    print (f'divi:: {divi}')
+    out = []
+    #out=""
+    s=0
+    e=chunk
+    print(f'e:: {e}')
+    new_history=""
+    #task = f'Compile this data to fulfill the task: {task}, and complete the purpose: {purpose}\n'
+    for z in range(divi):
+        print(f's:e :: {s}:{e}')
+        
+        hist = history[s:e]
+        
+        resp = run_gpt(
+            COMPRESS_DATA_PROMPT_SMALL,
+            stop_tokens=["observation:", "task:", "action:", "thought:"],
+            max_tokens=8192,
+            seed=seed,
+            direction=instruct,
+            knowledge="",
+            history=hist,
+        )
+        out.append(resp)
+        #new_history = resp
+        #print (resp)
+        #out+=resp
+        e=e+chunk
+        s=s+chunk
+    return out
+
+    
+def compress_data_og(c, instruct, history):
+    seed=random.randint(1,1000000000)
+    
+    print (c)
+    #tot=len(purpose)
+    #print(tot)
+    divr=int(c)/MAX_DATA
+    divi=int(divr)+1 if divr != int(divr) else int(divr)
+    chunk = int(int(c)/divr)
+    print(f'chunk:: {chunk}')
+    print(f'divr:: {divr}')
+    print (f'divi:: {divi}')
+    out = []
+    #out=""
+    s=0
+    e=chunk
+    print(f'e:: {e}')
+    new_history=""
+    #task = f'Compile this data to fulfill the task: {task}, and complete the purpose: {purpose}\n'
+    for z in range(divi):
+        print(f's:e :: {s}:{e}')
+        
+        hist = history[s:e]
+        
+        resp = run_gpt(
+            COMPRESS_DATA_PROMPT,
+            stop_tokens=["observation:", "task:", "action:", "thought:"],
+            max_tokens=8192,
+            seed=seed,
+            direction=instruct,
+            knowledge=new_history,
+            history=hist,
+        )
+        
+        new_history = resp
+        print (resp)
+        out+=resp
+        e=e+chunk
+        s=s+chunk
+
+    print ("final" + resp)
+    #history = "observation: {}\n".format(resp)
+    return resp
+
+
+
+def summarize(inp,history,mem_check,data=None):
+    
+    json_box=[]
+    error_box=""
+    json_out={}
+    rawp="Error"
+    if inp == "":
+        inp = "Process this data"
+    history.clear()
+    history = [(inp,"Summarizing Transcription...")] 
+    yield "",history,error_box,json_box
+
+    if data != "Error" and data != "" and data != None:
+        print(inp)
+        out = str(data)
+        rl = len(out)
+        print(f'rl:: {rl}')
+        c=1
+        for i in str(out):
+            print(f'i:: {i}')
+            if i == " " or i=="," or i=="\n":
+                c +=1
+        print (f'c:: {c}')
+        json_out = compress_data(c,inp,out)  
+        history = [(inp,"Generating Report...")] 
+        yield "", history,error_box,json_out
+
+        out = str(json_out)
+        print (out)
+        rl = len(out)
+        print(f'rl:: {rl}')
+        c=1
+        for i in str(out):
+            if i == " " or i=="," or i=="\n":
+                c +=1
+        print (f'c2:: {c}')
+        rawp = compress_data_og(c,inp,out)
+        history.clear()
+        history.append((inp,rawp))                
+
+        yield "", history,error_box,json_out
+    else:
+        rawp = "Provide a valid data source"
+        history.clear()
+        history.append((inp,rawp))
+        yield "", history,error_box,json_out
+
+    
+#################################
+def clear_fn():
+    return "",[(None,None)]
+
+with gr.Blocks() as app:
+    gr.HTML("""<center><h1>Video Summarizer</h1>""")
+    with gr.Row():
+        with gr.Column():
+            with gr.Row():
+                inp_url = gr.Textbox(label="Video URL")
+                url_btn = gr.Button("Load Video")
+            vid = gr.Video()
+        #trans_btn=gr.Button("Transcribe")
+        trans = gr.Textbox(interactive=True)
+    chatbot = gr.Chatbot(label="Mixtral 8x7B Chatbot",show_copy_button=True)
+    
+    with gr.Row():
+        with gr.Column(scale=3):
+            prompt=gr.Textbox(label = "Instructions (optional)")
+        with gr.Column(scale=1):
+            mem_check=gr.Checkbox(label="Memory", value=False)
+            button=gr.Button()
+        
+        #models_dd=gr.Dropdown(choices=[m for m in return_list],interactive=True)
+    with gr.Row():
+        stop_button=gr.Button("Stop")
+        clear_btn = gr.Button("Clear")
+    with gr.Row():
+        sz = gr.Dropdown(label="Model Size", choices=sizes, value='base')
+        lang = gr.Dropdown(label="Language (Optional)", choices=langs, value="English")
+    json_out=gr.JSON()
+    e_box=gr.Textbox()
+    #text=gr.JSON()
+    #inp_query.change(search_models,inp_query,models_dd)
+
+    url_btn.click(dl,[inp_url,vid],vid)
+    #trans_btn.click(transcribe,[vid,lang,sz],trans)
+    clear_btn.click(clear_fn,None,[prompt,chatbot])
+    go=button.click(transcribe,[vid,lang,sz],[trans,chatbot]).then(summarize,[prompt,chatbot,mem_check,trans],[prompt,chatbot,e_box,json_out])
+    stop_button.click(None,None,None,cancels=[go])
+app.queue(default_concurrency_limit=20).launch(show_api=False)