update backend functions

backend_functions.py

@@ -13,6 +13,8 @@ from google.cloud import storage
 from elevenlabs.client import ElevenLabs, AsyncElevenLabs
 from elevenlabs import play, save, Voice, stream
 from pymongo.mongo_client import MongoClient
+from utils import create_folders
+from gcp import download_credentials
 from dotenv import load_dotenv
 load_dotenv()
 
@@ -67,175 +69,95 @@ def _add_question_vectorstore(question: str, response: str):
 
 
 def _update_elements(question, chatbot, output, history_messages, url_audio, url_video, df_table_times):
-    
-    if tf.test.is_gpu_available():
-        chatbot.append([question, output])
-        new_comp_audio = gr.Audio(value=str(url_audio), autoplay=False, label="Audio")
-        new_comp_video = gr.Video(value=str(url_video), autoplay=True, height=400, label="Video")
-
-        history_messages.append({'role': 'user', 'content': question})
-        history_messages.append({'role': 'assistant', 'content': output})
-    else:
-        chatbot.append([question, output])
-        new_comp_audio = gr.Audio(value=str(url_audio), autoplay=False, label="Audio")
-        new_comp_video = gr.Video(value=str(url_video), autoplay=True, height=400, label="Video")
+    chatbot.append([question, output])
+    new_comp_audio = gr.Audio(value=str(url_audio), autoplay=False, label="Audio")
+    new_comp_video = gr.Video(value=str(url_video), autoplay=True, height=400, label="Video")
 
-        history_messages.append({'role': 'user', 'content': question})
-        history_messages.append({'role': 'assistant', 'content': output})
+    history_messages.append({'role': 'user', 'content': question})
+    history_messages.append({'role': 'assistant', 'content': output})
 
     return chatbot, new_comp_audio, new_comp_video, df_table_times
 
 
 def _query_pinecone(embedding):
-    
-    if tf.test.is_gpu_available():
-        results = index.query(
-                        vector=embedding,
-                        top_k=10,
-                        include_metadata=True,
-                    )
-
-        final_results = """"""
-        for result in results['matches']:
-            final_results += f"{result['metadata']['text']}\n"
-    else:
-        results = index.query(
-                        vector=embedding,
-                        top_k=10,
-                        include_metadata=True,
-                    )
+    results = index.query(
+                    vector=embedding,
+                    top_k=10,
+                    include_metadata=True,
+                )
 
-        final_results = """"""
-        for result in results['matches']:
-            final_results += f"{result['metadata']['text']}\n"
+    final_results = """"""
+    for result in results['matches']:
+        final_results += f"{result['metadata']['text']}\n"
 
     return final_results
 
 
 def _general_prompt(context):
-
-    if tf.test.is_gpu_available():
-        with open("prompt_general.txt", "r") as file:
-            file_prompt = file.read().replace("\n", "")
-        
-        context_prompt = file_prompt.replace('CONTEXT', context)
-        print(context_prompt)
-        print("--------------------")
-    else:
-        with open("prompt_general.txt", "r") as file:
-            file_prompt = file.read().replace("\n", "")
-        
-        context_prompt = file_prompt.replace('CONTEXT', context)
-        print(context_prompt)
-        print("--------------------")
+    with open("prompt_general.txt", "r") as file:
+        file_prompt = file.read().replace("\n", "")
+    
+    context_prompt = file_prompt.replace('CONTEXT', context)
+    print(context_prompt)
+    print("--------------------")
 
     return context_prompt
 
 
 def _call_embedding(text: str):
-
-    if tf.test.is_gpu_available():
-        response = openai_client.embeddings.create(
-            input=text,
-            model='text-embedding-ada-002'
-        )
-    else:
-        response = openai_client.embeddings.create(
-            input=text,
-            model='text-embedding-ada-002'
-        )
+    response = openai_client.embeddings.create(
+        input=text,
+        model='text-embedding-ada-002'
+    )
 
     return response.data[0].embedding
 
 
 def _call_gpt(prompt: str, message: str):
-
-    if tf.test.is_gpu_available():
-        response = openai_client.chat.completions.create(
-            model=MODEL_OPENAI,
-            temperature=0.2,
-            messages=[
-                {'role': 'system', 'content': prompt},
-                {'role': 'user', 'content': message}
-            ]
-        )
-    else:
-        response = openai_client.chat.completions.create(
-            model=MODEL_OPENAI,
-            temperature=0.2,
-            messages=[
-                {'role': 'system', 'content': prompt},
-                {'role': 'user', 'content': message}
-            ]
-        )
+    response = openai_client.chat.completions.create(
+        model=MODEL_OPENAI,
+        temperature=0.2,
+        messages=[
+            {'role': 'system', 'content': prompt},
+            {'role': 'user', 'content': message}
+        ]
+    )
         
     return response.choices[0].message.content
 
 
 def _call_gpt_standalone(prompt: str):
-
-    if tf.test.is_gpu_available():
-        response = openai_client.chat.completions.create(
-            model=MODEL_OPENAI,
-            temperature=0.2,
-            messages=[
-                {'role': 'system', 'content': prompt},
-            ]
-        )
-    else:
-        response = openai_client.chat.completions.create(
-            model=MODEL_OPENAI,
-            temperature=0.2,
-            messages=[
-                {'role': 'system', 'content': prompt},
-            ]
-        )
+    response = openai_client.chat.completions.create(
+        model=MODEL_OPENAI,
+        temperature=0.2,
+        messages=[
+            {'role': 'system', 'content': prompt},
+        ]
+    )
 
     return response.choices[0].message.content
 
 
 def _get_standalone_question(question, history_messages):
-        
-    if tf.test.is_gpu_available():
-        with open("prompt_standalone_message.txt", "r") as file:
-            file_prompt_standalone = file.read().replace("\n", "")
-
-        history = ''
-        for i, msg in enumerate(history_messages):
-            try:
-                if i == 0:
-                    continue  # Omit the prompt
-                if i % 2 == 0:
-                    history += f'user: {msg["content"]}\n'
-                else:
-                    history += f'assistant: {msg["content"]}\n'
-            except Exception as e:
-                print(e)
-        
-        prompt_standalone = file_prompt_standalone.replace('HISTORY', history).replace('QUESTION', question)
-        standalone_msg_q = _call_gpt_standalone(prompt_standalone)
-        print(standalone_msg_q)
-        print("------------------")
-    else:
-        with open("prompt_standalone_message.txt", "r") as file:
-            file_prompt_standalone = file.read().replace("\n", "")
-
-        history = ''
-        for i, msg in enumerate(history_messages):
-            try:
-                if i == 0:
-                    continue  # Omit the prompt
-                if i % 2 == 0:
-                    history += f'user: {msg["content"]}\n'
-                else:
-                    history += f'assistant: {msg["content"]}\n'
-            except Exception as e:
-                print(e)
-        
-        prompt_standalone = file_prompt_standalone.replace('HISTORY', history).replace('QUESTION', question)
-        standalone_msg_q = _call_gpt_standalone(prompt_standalone)
-        print(standalone_msg_q)
-        print("------------------")
+    with open("prompt_standalone_message.txt", "r") as file:
+        file_prompt_standalone = file.read().replace("\n", "")
+
+    history = ''
+    for i, msg in enumerate(history_messages):
+        try:
+            if i == 0:
+                continue  # Omit the prompt
+            if i % 2 == 0:
+                history += f'user: {msg["content"]}\n'
+            else:
+                history += f'assistant: {msg["content"]}\n'
+        except Exception as e:
+            print(e)
+    
+    prompt_standalone = file_prompt_standalone.replace('HISTORY', history).replace('QUESTION', question)
+    standalone_msg_q = _call_gpt_standalone(prompt_standalone)
+    print(standalone_msg_q)
+    print("------------------")
 
     return standalone_msg_q
 
@@ -246,243 +168,130 @@ def _create_clean_message(text: str):
 
 
 def _create_audio(clean_text: str):
+    download_credentials()
+    create_folders()
+    
+    STORAGE_CLIENT = storage.Client.from_service_account_json(CREDENTIALS_GCP)
 
-    if tf.test.is_gpu_available():
-        STORAGE_CLIENT = storage.Client.from_service_account_json(CREDENTIALS_GCP)
-
-        unique_id = str(uuid.uuid4())
-
-        # Create audio file
-        client_elevenlabs = ElevenLabs(api_key=API_KEY_ELEVENLABS)
-        voice_custom = Voice(voice_id = "ZQe5CZNOzWyzPSCn5a3c")
-
-        audio = client_elevenlabs.generate(
-            text=clean_text,
-            voice=voice_custom,
-            model="eleven_multilingual_v2"
-        )
-
-        source_audio_file_name = f'./audios/file_audio_{unique_id}.wav'
-
-        try:
-            save(audio, source_audio_file_name)
-        except Exception as e:
-            print(e)
-
-        # Save audio and get url of gcp
-        destination_blob_name_audio = unique_id + '.wav'
-        
-        bucket = STORAGE_CLIENT.bucket(NAME_BUCKET)
-        blob = bucket.blob(destination_blob_name_audio)
-        try:
-            blob.upload_from_filename(source_audio_file_name)
-        except Exception as e:
-            print(e)
-
-        signed_url_audio = "None"
-        try:
-            url_expiration = timedelta(minutes=15)
-            signed_url_audio = blob.generate_signed_url(expiration=url_expiration)
-        except Exception as e:
-            print(e)
-    else:
-        STORAGE_CLIENT = storage.Client.from_service_account_json(CREDENTIALS_GCP)
-
-        unique_id = str(uuid.uuid4())
-
-        # Create audio file
-        client_elevenlabs = ElevenLabs(api_key=API_KEY_ELEVENLABS)
-        voice_custom = Voice(voice_id = "ZQe5CZNOzWyzPSCn5a3c")
+    unique_id = str(uuid.uuid4())
 
-        audio = client_elevenlabs.generate(
-            text=clean_text,
-            voice=voice_custom,
-            model="eleven_multilingual_v2"
-        )
+    # Create audio file
+    client_elevenlabs = ElevenLabs(api_key=API_KEY_ELEVENLABS)
+    voice_custom = Voice(voice_id = "ZQe5CZNOzWyzPSCn5a3c")
 
-        source_audio_file_name = f'./audios/file_audio_{unique_id}.wav'
+    audio = client_elevenlabs.generate(
+        text=clean_text,
+        voice=voice_custom,
+        model="eleven_multilingual_v2"
+    )
 
-        try:
-            save(audio, source_audio_file_name)
-        except Exception as e:
-            print(e)
+    source_audio_file_name = f'./audios/file_audio_{unique_id}.wav'
 
-        # Save audio and get url of gcp
-        destination_blob_name_audio = unique_id + '.wav'
-        
-        bucket = STORAGE_CLIENT.bucket(NAME_BUCKET)
-        blob = bucket.blob(destination_blob_name_audio)
-        try:
-            blob.upload_from_filename(source_audio_file_name)
-        except Exception as e:
-            print(e)
+    try:
+        save(audio, source_audio_file_name)
+    except Exception as e:
+        print(e)
 
-        signed_url_audio = "None"
-        try:
-            url_expiration = timedelta(minutes=15)
-            signed_url_audio = blob.generate_signed_url(expiration=url_expiration)
-        except Exception as e:
-            print(e)
+    # Save audio and get url of gcp
+    destination_blob_name_audio = unique_id + '.wav'
+    
+    bucket = STORAGE_CLIENT.bucket(NAME_BUCKET)
+    blob = bucket.blob(destination_blob_name_audio)
+    try:
+        blob.upload_from_filename(source_audio_file_name)
+    except Exception as e:
+        print(e)
+
+    signed_url_audio = "None"
+    try:
+        url_expiration = timedelta(minutes=15)
+        signed_url_audio = blob.generate_signed_url(expiration=url_expiration)
+    except Exception as e:
+        print(e)
 
     return signed_url_audio, unique_id
 
 
 def _create_video(link_audio: str, unique_id: str):
-
-    if tf.test.is_gpu_available():
-        STORAGE_CLIENT = storage.Client.from_service_account_json(CREDENTIALS_GCP)
-        bucket = STORAGE_CLIENT.bucket(NAME_BUCKET)
-
-        # Create video talk with file audio created by elevenlabs api
-        url_did = "https://api.d-id.com/talks"
-
-        payload = {
-            "script": {
-                "type": "audio",
-                "provider": {
-                    "type": "microsoft",
-                    "voice_id": "en-US-JennyNeural"
-                },
-                "ssml": "false",
-                "audio_url": link_audio
-            },
-            "config": {
-                "fluent": "false",
-                "pad_audio": "0.0",
-                "stitch": True
-            },
-            "source_url": IMG_XAVY
-        }
-        headers = {
-            "accept": "application/json",
-            "content-type": "application/json",
-            "authorization": f"Basic {D_ID_KEY}"
-        }
-
-        request_create_talk = requests.post(url_did, json=payload, headers=headers)
-        resp_create_talk = request_create_talk.json()
-
-        talk_id = "None"
-        try:
-            talk_id = resp_create_talk['id']
-        except Exception as e:
-            print(e)
-
-        # Get url of video file
-        url_get_talk_id = f"https://api.d-id.com/talks/{talk_id}"
-
-        
-        while True:
-            request_video_url = requests.get(url_get_talk_id, headers=headers)
-            resp_video_url = request_video_url.json()
-
-            if resp_video_url['status'] == 'done':
-                break
-            # Sleep until the video is ready
-            time.sleep(0.5)
-
-        result_url_video = resp_video_url['result_url']
-
-        # Saves the video into a file to later upload it to the GCP
-        source_video_file_name = f'./videos/video_final_{unique_id}.mp4'
-        request_video = requests.get(result_url_video)
-        if request_video.status_code == 200:
-            with open(source_video_file_name, 'wb') as outfile:
-                outfile.write(request_video.content)
-
-        # Save video file to the GCP
-        destination_blob_name_video = unique_id + '.mp4'
-
-        # Configure bucket
-        blob = bucket.blob(destination_blob_name_video)
-        try:
-            blob.upload_from_filename(source_video_file_name)
-        except Exception as e:
-            print(e)
-
-        signed_url_video = "None"
-        try:
-            url_expiration_video = timedelta(minutes=15)
-            signed_url_video = blob.generate_signed_url(expiration=url_expiration_video)
-        except Exception as e:
-            print(e)
-    else:
-        STORAGE_CLIENT = storage.Client.from_service_account_json(CREDENTIALS_GCP)
-        bucket = STORAGE_CLIENT.bucket(NAME_BUCKET)
-
-        # Create video talk with file audio created by elevenlabs api
-        url_did = "https://api.d-id.com/talks"
-
-        payload = {
-            "script": {
-                "type": "audio",
-                "provider": {
-                    "type": "microsoft",
-                    "voice_id": "en-US-JennyNeural"
-                },
-                "ssml": "false",
-                "audio_url": link_audio
-            },
-            "config": {
-                "fluent": "false",
-                "pad_audio": "0.0",
-                "stitch": True
+    download_credentials()
+    create_folders()
+    
+    STORAGE_CLIENT = storage.Client.from_service_account_json(CREDENTIALS_GCP)
+    bucket = STORAGE_CLIENT.bucket(NAME_BUCKET)
+
+    # Create video talk with file audio created by elevenlabs api
+    url_did = "https://api.d-id.com/talks"
+
+    payload = {
+        "script": {
+            "type": "audio",
+            "provider": {
+                "type": "microsoft",
+                "voice_id": "en-US-JennyNeural"
             },
-            "source_url": IMG_XAVY
-        }
-        headers = {
-            "accept": "application/json",
-            "content-type": "application/json",
-            "authorization": f"Basic {D_ID_KEY}"
-        }
-
-        request_create_talk = requests.post(url_did, json=payload, headers=headers)
-        resp_create_talk = request_create_talk.json()
-
-        talk_id = "None"
-        try:
-            talk_id = resp_create_talk['id']
-        except Exception as e:
-            print(e)
-
-        # Get url of video file
-        url_get_talk_id = f"https://api.d-id.com/talks/{talk_id}"
-
-        
-        while True:
-            request_video_url = requests.get(url_get_talk_id, headers=headers)
-            resp_video_url = request_video_url.json()
-
-            if resp_video_url['status'] == 'done':
-                break
-            # Sleep until the video is ready
-            time.sleep(0.5)
-
-        result_url_video = resp_video_url['result_url']
+            "ssml": "false",
+            "audio_url": link_audio
+        },
+        "config": {
+            "fluent": "false",
+            "pad_audio": "0.0",
+            "stitch": True
+        },
+        "source_url": IMG_XAVY
+    }
+    headers = {
+        "accept": "application/json",
+        "content-type": "application/json",
+        "authorization": f"Basic {D_ID_KEY}"
+    }
 
-        # Saves the video into a file to later upload it to the GCP
-        source_video_file_name = f'./videos/video_final_{unique_id}.mp4'
-        request_video = requests.get(result_url_video)
-        if request_video.status_code == 200:
-            with open(source_video_file_name, 'wb') as outfile:
-                outfile.write(request_video.content)
+    request_create_talk = requests.post(url_did, json=payload, headers=headers)
+    resp_create_talk = request_create_talk.json()
 
-        # Save video file to the GCP
-        destination_blob_name_video = unique_id + '.mp4'
+    talk_id = "None"
+    try:
+        talk_id = resp_create_talk['id']
+    except Exception as e:
+        print(e)
 
-        # Configure bucket
-        blob = bucket.blob(destination_blob_name_video)
-        try:
-            blob.upload_from_filename(source_video_file_name)
-        except Exception as e:
-            print(e)
+    # Get url of video file
+    url_get_talk_id = f"https://api.d-id.com/talks/{talk_id}"
 
-        signed_url_video = "None"
-        try:
-            url_expiration_video = timedelta(minutes=15)
-            signed_url_video = blob.generate_signed_url(expiration=url_expiration_video)
-        except Exception as e:
-            print(e)
+    
+    while True:
+        request_video_url = requests.get(url_get_talk_id, headers=headers)
+        resp_video_url = request_video_url.json()
+
+        if resp_video_url['status'] == 'done':
+            break
+        # Sleep until the video is ready
+        time.sleep(0.5)
+
+    result_url_video = resp_video_url['result_url']
+
+    # Saves the video into a file to later upload it to the GCP
+    source_video_file_name = f'./videos/video_final_{unique_id}.mp4'
+    request_video = requests.get(result_url_video)
+    if request_video.status_code == 200:
+        with open(source_video_file_name, 'wb') as outfile:
+            outfile.write(request_video.content)
+
+    # Save video file to the GCP
+    destination_blob_name_video = unique_id + '.mp4'
+
+    # Configure bucket
+    blob = bucket.blob(destination_blob_name_video)
+    try:
+        blob.upload_from_filename(source_video_file_name)
+    except Exception as e:
+        print(e)
+
+    signed_url_video = "None"
+    try:
+        url_expiration_video = timedelta(minutes=15)
+        signed_url_video = blob.generate_signed_url(expiration=url_expiration_video)
+    except Exception as e:
+        print(e)
 
     return signed_url_video
 

gcp.py

@@ -15,7 +15,7 @@ def download_credentials() -> None:
     """
     Downloads the GCP credentials from Hugging Face Hub
     """
-    assets_dir = './assets'
+    assets_dir = 'assets'
     credentials_file = os.path.join(assets_dir, "credentials.json")
     print(credentials_file)