Blame for app.py

from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
from fastapi.responses import JSONResponse, Response
import requests
import uvicorn
import os
import logging
import torch 
from transformers import pipeline
import subprocess
import re
from piper.voice import PiperVoice
import wave
from pydub import AudioSegment
from gradio_client import Client
import time
import gc
import threading

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

app = FastAPI()
HF_READ_TKN = os.getenv("HF_READ_TKN") 
hope_client = None
Lock = threading.Lock()
TIMESTAMP_LAST_CALL_TO_HOPE = None

#Ongoing llama chat; Dict will tell if we are in chat with this person and what llama input earlier was
ONGOING_CONVO = {}

def is_space_sleeping():
    # Stage can be SLEEPING, PAUSED, RUNNING, or APP_STARTING
    headers = {"Authorization": f"Bearer {HF_READ_TKN}"}
    response = requests.get("https://huggingface.co/api/spaces/HopeLLM/HopeV1.0", headers=headers)
    if response.status_code == 200:
        space_info = response.json()
        status = space_info.get("runtime", {}).get("stage", "") 
        if status == "SLEEPING":
            return True
        elif status == "APP_STARTING": 
            return True 
        elif status == "RESTARTING": 
            return True 
        elif status == "BUILDING": 
            return True 
        elif status == "RUNNING": 
            return False 
        elif status == "PAUSED": 
            return "unknown" 
        else: 
            return False
    return "unknown"  # If API call fails, assume the space is sleeping

def call_hope_api(input_txt, sender_id):
    global hope_client
    global TIMESTAMP_LAST_CALL_TO_HOPE

    TIMESTAMP_LAST_CALL_TO_HOPE = time.time()
    with Lock:
        if hope_client == None:
            if(is_space_sleeping()==True):
                logger.info("Hope is sleeping, trying to wake it up")
                send_whatsapp_message(sender_id, None, "Hope is sleeping, waking it up ;)\nKindly wait for some time", text=True)
                try:
                    Client("HopeLLM/HopeV1.0", hf_token=HF_READ_TKN)
                except:
                    while(is_space_sleeping()):
                        #Sleep for 30sec
                        time.sleep(30)
            elif (is_space_sleeping() == "unknown"):
                logger.error("Something wrong with Hope Check")
            hope_client = Client("HopeLLM/HopeV1.0", hf_token=HF_READ_TKN)
        return hope_client.predict(inp = input_txt, api_name = "/answer_question") 

def reset_hope_client(exit_fn=None):
    #The HOPE GPU Space may sleep; Hence we need to remove this hope_client variable before that
    #If we dont do that, we get runtime error in our WhatsAppServer Space 
    global hope_client
    global TIMESTAMP_LAST_CALL_TO_HOPE
    #If More than an hour to last access to HOPE
    try:
        while True:
            time.sleep(3600)
            with Lock:
                if hope_client and TIMESTAMP_LAST_CALL_TO_HOPE and time.time() - TIMESTAMP_LAST_CALL_TO_HOPE > 3600: 
                    logger.info("Closing Hope Client Connection. An hour since the last iteraction.")
                    hope_client = None
                    gc.collect()

            ##Added cleanup as well here after 40 min sleep time
            logger.debug("Cleaning Ongoing convo")
            for sender_id in ONGOING_CONVO.keys():
                logger.debug(f"Trying to clean up sender {sender_id}")
                if( ONGOING_CONVO[sender_id] and (time.time() - ONGOING_CONVO[sender_id].get("lst_msg_time", 0) > 2400)):
                    send_whatsapp_message(sender_id, None, "Session Time-Out", text=True)
                    clean_up_audio_files(sender_id)
                    ONGOING_CONVO[sender_id] = None
    finally:
        if exit_fn:
            exit_fn()

def notify_parent_of_exit():
    logger.error("The cleaner thread has exited. Check!!")

def clean_up_audio_files(sender_id):
    command = f"rm -f /tmp/*{sender_id}*"
    subprocess.run(command, text = True, shell = True)

# Your WhatsApp verification token
VERIFY_TOKEN = os.getenv("WHATSAPP_VERIFY_TOKEN", "your_verification_token")

# WhatsApp API credentials
WHATSAPP_BASE_URL = "https://graph.facebook.com/v22.0"
WHATSAPP_API_URL = f"{WHATSAPP_BASE_URL}/599521673246967/messages"
WHATSAPP_MEDIA_URL = f"{WHATSAPP_BASE_URL}/599521673246967/media"
WHATSAPP_ACCESS_TOKEN = os.getenv("WHATSAPP_ACCESS_TOKEN", "your_access_token")


# Initialize the Whisper model pipeline
device = "cuda" if torch.cuda.is_available() else "cpu"
transcriber = pipeline(
    task="automatic-speech-recognition",
    model="openai/whisper-small",
    device=device,
    generate_kwargs={"task": "translate", "language":"hi"}
)
transcriber.model.config.forced_decoder_ids = transcriber.tokenizer.get_decoder_prompt_ids(language="hi", task="translate")

amy_voice = PiperVoice.load("./piper/en_US-amy-medium.onnx")

FIXED_PROMPT = "You are a helpful English Teaching Bot. You need to be friendly. Since you teach kids less than 7 years old, be motivating and don't point out any mistakes. Dont have long responses, keep your responses short. Assume I am your students and refer me as your student when replying"
llama_input_template = "<s> [INST] <<SYS>> {fixed_prompt} <</SYS>> Understand the story below and ask me one basic question regarding the story.  Then wait for me to reply, when I reply, appreciate my participation and same way keep asking more questions one by one. Keep the questions one-liners and do not give much context. Keep everything brief \n {story_to_use} [/INST]"
llama_input_gen_random_story = f"<s> [INST] <<SYS>> You are a helpful English Teacher. You teach kids less than 7 years old.<</SYS>> Give me an english story for today's english learning session. Don't make it very big. Just output the story nothing else, no chatting with me [/INST]"

@app.get("/webhook")
async def verify_webhook(request: Request):
    """
    Endpoint for WhatsApp to verify the webhook during setup.
    """
    params = request.query_params
    if (
        params.get("hub.mode") == "subscribe" and
        params.get("hub.verify_token") == VERIFY_TOKEN
    ):
        logger.info("Webhook verified successfully.")
        return Response(content=params.get("hub.challenge"), media_type="text/plain")
    else:
        logger.warning("Webhook verification failed.")
        raise HTTPException(status_code=403, detail="Verification failed")

@app.post("/webhook")
async def receive_message(request: Request, background_tasks: BackgroundTasks):
    """
    Endpoint to receive incoming WhatsApp messages.
    """
    data = await request.json()
    # Extract the message and sender information
    try:
        logger.info(f"Incoming message data: {data}")
        entry = data["entry"][0]
        changes = entry["changes"][0]
        value = changes["value"]
        message = value["messages"][0]
        background_tasks.add_task(backgnd_process_message, message)
        return JSONResponse(status_code=200, content={"message": "Webhook received successfully"})
        
    except (KeyError, IndexError) as e:
        # If the expected fields are not present, return a 400 response
        logger.error(f"Error parsing incoming message: {e}")
        return JSONResponse(status_code=200, content={"message": "Webhook received successfully"})

    return {"status": "success"}


def backgnd_process_message(message):

    sender_id = message["from"]
    message_type = message["type"]
    message_id = message["id"]

    if (ONGOING_CONVO.get(sender_id, None) == None):
        #Send the button to start story
        welcome_prompt = "Hi there!! I am Hope, your english teaching bot :)\nGive me a story or news article and we can chat around it.\n\n_Note: To end this teaching session, send the message_ END"
        ONGOING_CONVO[sender_id] = {'Story': None, 'Llama_conv' : None, 'lst_msg_time': time.time()}
        send_whatsapp_message(sender_id, message_id, welcome_prompt, text=True)
        send_story_option(sender_id)
    else:
        ONGOING_CONVO[sender_id]['lst_msg_time'] = time.time()
        if message_type == "text":
            message_text = message["text"]["body"]
            if(message_text.lower() == "end"):
                ONGOING_CONVO[sender_id] = None
                response_text = "I hope you had a wonderful learning session!! Keep up the good work and best of luck"
                clean_up_audio_files(sender_id)
            elif (ONGOING_CONVO[sender_id]["Story"] == "PENDING"):
                ONGOING_CONVO[sender_id]["Story"] = message_text
                response_text = message_text #The input story. We will repeat the story
                ## Generate question from llama 
                llama_input  = llama_input_template.format(story_to_use = message_text, fixed_prompt=FIXED_PROMPT)
                llama_output = call_hope_api(llama_input,sender_id)
                logger.info(f"LLama Ouput: {llama_output}")
                response_text = response_text + "\n \n" + llama_output
                ONGOING_CONVO[sender_id]['Llama_conv'] = llama_input + llama_output + "</s> <s> [INST] "
            else:
                llama_input = ONGOING_CONVO[sender_id]['Llama_conv'] + message_text + "[/INST]"
                llama_output = call_hope_api(llama_input,sender_id)
                logger.info(f"LLama Ouput: {llama_output}")
                response_text = llama_output
                ONGOING_CONVO[sender_id]['Llama_conv'] = llama_input + llama_output + "</s> <s> [INST] "
        elif message_type == "audio":
            media_id = message["audio"]["id"]
            audio_url = fetch_media_url(media_id)
            audio_path = download_audio(audio_url, sender_id)
            transcription = transcribe_audio(audio_path)
            if (ONGOING_CONVO[sender_id]["Story"] == "PENDING"):
                ONGOING_CONVO[sender_id]["Story"] = transcription
                response_text = transcription
                ## Generate question from llama 
                llama_input  = llama_input_template.format(story_to_use = transcription, fixed_prompt=FIXED_PROMPT)
                llama_output = call_hope_api(llama_input,sender_id)
                logger.info(f"LLama Ouput: {llama_output}")
                response_text = response_text + "\n \n" + llama_output
                ONGOING_CONVO[sender_id]['Llama_conv'] = llama_input + llama_output + "</s> <s> [INST] "
            else:
                llama_input = ONGOING_CONVO[sender_id]['Llama_conv'] + transcription + "[/INST]"
                llama_output = call_hope_api(llama_input,sender_id)
                logger.info(f"LLama Ouput: {llama_output}")
                response_text = llama_output
                ONGOING_CONVO[sender_id]['Llama_conv'] = llama_input + llama_output + "</s> <s> [INST] "
        elif message_type == "interactive":
            if message["interactive"]["button_reply"]["id"] == "random_story":
                ONGOING_CONVO[sender_id]['Story'] = "random_story"
                #send_whatsapp_message(sender_id, message_id, "Perfect, let me find a nice story for today.")
                story_for_chat = call_hope_api(llama_input_gen_random_story,sender_id)
                ONGOING_CONVO[sender_id]['Story'] = story_for_chat
                logger.info(f"Llama gave story {story_for_chat}")

                ## Generate question from llama 
                llama_input  = llama_input_template.format(story_to_use = story_for_chat, fixed_prompt=FIXED_PROMPT)
                llama_output = call_hope_api(llama_input,sender_id)
                logger.info(f"LLama Ouput: {llama_output}")
                response_text = story_for_chat + "\n \n" + llama_output
                ONGOING_CONVO[sender_id]['Llama_conv'] = llama_input + llama_output + "</s> <s> [INST] "
            elif message["interactive"]["button_reply"]["id"] == "own_story":
                response_text = "Great!! Send me your story as the next message, it can be text or audio!!" 
                ONGOING_CONVO[sender_id]['Story'] = "PENDING"
            else:
                button_type = message["interactive"]["button_reply"]["id"]
                logger.warning(f"Unsupported button reply: {button_type}")
                response_text = "Sorry, invalid story type, send a valid type"
        else:
            logger.warning(f"Unsupported message type: {message_type}")
            response_text = "Sorry, I can only process text and audio messages at the moment."

        send_whatsapp_message(sender_id, message_id, response_text) #TO-DO remove audio files every hr or something

def process_text_message(message_text: str, sender_id) -> str:
    """
    Process the incoming message and generate a response.
    Customize this function based on your requirements.
    """
    # For demonstration, echoing back the received message
    logger.info(f"Processing message: {message_text}")
    if ONGOING_CONVO[sender_id]["Story"] == None:
        ONGOING_CONVO[sender_id]["Story"] = message_text
    return f"Received your message: {message_text}"

def fetch_media_url(media_id: str) -> str:
    """
    Fetch the media URL using the media ID.
    """
    url = f"{WHATSAPP_BASE_URL}/{media_id}"
    headers = {
        "Authorization": f"Bearer {WHATSAPP_ACCESS_TOKEN}"
    }
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        media_url = response.json().get("url")
        logger.info(f"Fetched media URL: {media_url}")
        return media_url
    else:
        logger.error(f"Failed to fetch media URL: {response.status_code} - {response.text}")
        raise HTTPException(status_code=500, detail="Failed to fetch media URL")

def download_audio(audio_url: str, sender_id) -> str:
    """
    Download the audio file from the provided URL.
    """
    headers = {
        "Authorization": f"Bearer {WHATSAPP_ACCESS_TOKEN}"
    }
    response = requests.get(audio_url, headers=headers)
    if response.status_code == 200:
        audio_path = f"/tmp/voice_message{sender_id}.ogg"
        with open(audio_path, "wb") as audio_file:
            audio_file.write(response.content)
        logger.info(f"Audio file downloaded to: {audio_path}")
        return audio_path
    else:
        logger.error(f"Failed to download audio file: {response.status_code} - {response.text}")
        raise HTTPException(status_code=500, detail="Failed to download audio file")

def transcribe_audio(audio_path: str) -> str:
    """
    Transcribe the audio file to text.
    """
    try:
        logger.info(f"Transcribing audio file: {audio_path}")
        transcription = transcriber(audio_path, return_timestamps=True)["text"]
        logger.info(f"Transcription: {transcription}")
        return transcription
    except Exception as e:
        logger.error(f"Failed to transcribe audio: {e}")
        return "I'm sorry, I couldn't understand the audio message."

def convert_text_to_mp3(text, mp3_name = '/tmp/amy_said.mp3'):
    text = re.sub(r':', ". ", text) #To add pause need to add period and not space
    text = re.sub(r'!', ". ", text) #To add pause need to add period and not space
    text = re.sub(r'\n', ". ", text) #To add pause need to add period and not space
    text = re.sub(r"\*[^*]+\*", " ", text)
    text = re.sub(r'[^A-Za-z0-9.,$ \'?]+', " ", text)
    text = text.encode('cp1252', 'ignore').decode('cp1252')
    logger.info(f"Input to amy: {text}")
    if text.isspace() or text == "":
        return None

    with wave.open(mp3_name+'.wav', 'w') as wav_file:
        amy_voice.synthesize(text, wav_file, length_scale=1.3)

    audio = AudioSegment.from_wav(mp3_name+'.wav')
    audio.export(mp3_name, format='mp3')

def get_meta_id(audio_file_path, file_name):
    # Open the audio file in binary mode
    with open(audio_file_path, 'rb') as audio_file:
        # Prepare the data for the request
        files = {
            'file': (file_name, audio_file, 'audio/mpeg')
        }
        payload = {
            'messaging_product': 'whatsapp',
            'type': 'audio/mpeg'
        }
        headers = {
            'Authorization': f'Bearer {WHATSAPP_ACCESS_TOKEN}'
        }

        # Send the POST request to upload media
        response = requests.post(WHATSAPP_MEDIA_URL, headers=headers, data=payload, files=files)

    # Check for successful upload
    if response.status_code == 200:
        media_id = response.json().get('id')
        logger.info(f"Media uploaded successfully. Media ID: {media_id}")
    else:
        media_id = -1
        logger.info(f"Failed to upload media: {response.status_code} {response.text}")

    return media_id

def send_whatsapp_message(recipient_id: str,message_id:str, message_text: str, text=False):
    """
    Send a message back to the WhatsApp user.
    """
    headers = {
        "Authorization": f"Bearer {WHATSAPP_ACCESS_TOKEN}",
        "Content-Type": "application/json"
    }
    if text:
        payload = {
            "messaging_product": "whatsapp",
            "to": recipient_id,
            "type": "text",
            "text": {"body": message_text}
        }
    else:
        payload = {
        'messaging_product': 'whatsapp',
        'to': recipient_id,
        'type': 'audio',
        'audio': {
            'id': None
            }
        }   
        audio_file = convert_text_to_mp3(message_text, "/tmp/" + recipient_id + message_id +".mp3")
        payload['audio']['id'] = get_meta_id("/tmp/" + recipient_id + message_id +".mp3", recipient_id + message_id +".mp3")

    response = requests.post(WHATSAPP_API_URL, json=payload, headers=headers)
    if response.status_code == 200:
        logger.info(f"Message sent successfully to {recipient_id}.")
    else:
        logger.error(f"Failed to send message: {response.status_code} - {response.text}")

def send_story_option(recipient_id):
    headers = {
        "Authorization": f"Bearer {WHATSAPP_ACCESS_TOKEN}",
        "Content-Type": "application/json"
    }
    payload = {
        "messaging_product": "whatsapp",
        "to": recipient_id,
        "type": "interactive",
        "interactive": {
          "type": "button",
          "body": {
            "text": "Story option:"
          },
          "action": {
            "buttons": [
              {
                "type": "reply",
                "reply": {
                  "id": "random_story",
                  "title": "Pick a story"
                }
              },
              {
                "type": "reply",
                "reply": {
                  "id": "own_story",
                  "title": "I have my own story"
                }
              }
            ]
          }
        }
    }
    response = requests.post(WHATSAPP_API_URL, json=payload, headers=headers)
    if response.status_code == 200:
        logger.info(f"Message sent successfully to {recipient_id}.")
    else:
        logger.error(f"Failed to send message: {response.status_code} - {response.text}")



if __name__ == "__main__":
    TIMESTAMP_LAST_CALL_TO_HOPE = time.time()
    threading.Thread(target=reset_hope_client, kwargs={'exit_fn': notify_parent_of_exit},daemon=True).start()
    uvicorn.run(app, host="0.0.0.0", port=7860)