Spaces:
Runtime error
Runtime error
| ######### | |
| # path_to_key = 'credentials/plenary-stacker-457109-k5-6b664f4fa9ba.json' | |
| path_to_key = None | |
| ############ | |
| from fastapi import FastAPI, Request,UploadFile, File | |
| from fastapi.responses import HTMLResponse, JSONResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.templating import Jinja2Templates | |
| import asyncio | |
| from google.cloud import dialogflowcx_v3 as dialogflow | |
| # | |
| import os | |
| import json | |
| from google.oauth2 import service_account | |
| from google.api_core.client_options import ClientOptions | |
| # | |
| import soundfile as sf | |
| import sounddevice as sd | |
| import io | |
| import speech_recognition as sr | |
| from pydub import AudioSegment | |
| import base64 | |
| ########################### === Setup server === ########################### | |
| app = FastAPI() | |
| app.mount("/static", StaticFiles(directory="static"), name="static") | |
| templates = Jinja2Templates(directory="templates") | |
| ################angle detection and pose detection################## | |
| #### Audio recording and transcription ######### | |
| async def transcribe_audio(file: UploadFile = File(...)): | |
| status_message = '' | |
| input_text = '' | |
| try: | |
| # Step 1: Read uploaded file | |
| uploaded_bytes = await file.read() | |
| # Step 2: Save the audio file in buffer | |
| audio = AudioSegment.from_file(io.BytesIO(uploaded_bytes)) | |
| wav_io = io.BytesIO() | |
| audio.export(wav_io, format="wav") | |
| wav_io.seek(0) | |
| # Step 4: Transcribe using SpeechRecognition | |
| recognizer = sr.Recognizer() | |
| with sr.AudioFile(wav_io) as source: | |
| audio = recognizer.record(source) | |
| try: | |
| input_text = recognizer.recognize_google(audio) | |
| status_message = "Recording Done" | |
| except sr.UnknownValueError: | |
| status_message = "Could not understand the audio" | |
| except sr.RequestError as e: | |
| status_message = f"Request failed: {e}" | |
| except Exception as e: | |
| input_text = "Error " | |
| status_message = f"Unexpected error: {str(e)}" | |
| return JSONResponse(content={ | |
| "message": status_message, | |
| "input_text": input_text, | |
| }) | |
| ########## Chatbot setup ############## | |
| clt1 = None | |
| s_path = None | |
| def set_session(): | |
| api_endpoint_ = 'europe-west2-dialogflow.googleapis.com' | |
| project_id_ = 'plenary-stacker-457109-k5' | |
| session_id_ = 'unique-session-id' | |
| location_ = 'europe-west2' | |
| agent_id_ = 'c827abce-ac66-40e5-a7e5-aba7e90b74ae' | |
| if path_to_key is None: | |
| credentials_info = json.loads(os.getenv("GOOGLE_CREDENTIALS_JSON")) | |
| credentials = service_account.Credentials.from_service_account_info(credentials_info) | |
| client1 = dialogflow.SessionsClient( | |
| credentials=credentials, | |
| client_options=ClientOptions(api_endpoint=api_endpoint_) | |
| ) | |
| else: | |
| client1 = dialogflow.SessionsClient( | |
| credentials=service_account.Credentials.from_service_account_file(path_to_key), | |
| client_options=ClientOptions(api_endpoint=api_endpoint_) | |
| ) | |
| session_path1 = client1.session_path(project_id_, location_, agent_id_, session_id_) | |
| return client1, session_path1 | |
| async def get_response(message_): | |
| global clt1, s_path | |
| if clt1 is None or s_path is None: | |
| clt1, s_path = set_session() | |
| try: | |
| query_input = dialogflow.QueryInput( | |
| text=dialogflow.TextInput(text=message_), | |
| language_code='en' | |
| ) | |
| loop = asyncio.get_event_loop() | |
| response = await loop.run_in_executor( | |
| None, | |
| lambda: clt1.detect_intent(request={"session": s_path, "query_input": query_input}) | |
| ) | |
| return extract_message(response) | |
| except Exception as e: | |
| return f'Error: {e}' | |
| def extract_message(response): | |
| response_messages = response.query_result.response_messages | |
| message_text = '' | |
| for message in response_messages: | |
| if message.text and hasattr(message.text.text, '__iter__'): | |
| message_text += ''.join(message.text.text) | |
| if not message_text: | |
| message_text = response.query_result.fulfillment_text | |
| return message_text | |
| ####### Chatbot API ########### | |
| async def chat_post(request: Request): | |
| data = await request.json() | |
| user_input = data.get("message", "") | |
| # user_input = 'hey man' | |
| response_text = await get_response(user_input) | |
| return {"reply": response_text} | |
| ###########basic API for testing ###########` | |
| # ` | |
| async def index(request: Request): | |
| return templates.TemplateResponse("index.html", {"request": request}) | |