from main import *
from tts_api import *
from stt_api import *
from sentiment_api import *
from imagegen_api import *
from musicgen_api import *
from translation_api import *
from codegen_api import *
from text_to_video_api import *
from summarization_api import *
from image_to_3d_api import *
from flask import Flask, request, jsonify, Response, send_file, stream_with_context
from flask_cors import CORS
import torch
import torch.nn.functional as F
import torchaudio
import numpy as np
from PIL import Image
import io
import tempfile
import queue
import json
import base64
from markupsafe import Markup
from markupsafe import escape
app = Flask(__name__)
CORS(app)
html_code = """
AI Conversational Avatar
AI POWERED
Generate Avatar Response
Start Recording
Stop Recording
Transcription:
Response:
"""
feedback_queue = queue.Queue()
@app.route("/")
def index():
return html_code
@app.route("/api/v1/generate_stream", methods=["GET"])
def generate_stream():
text = request.args.get("text", "")
temp = float(request.args.get("temp", 0.7))
top_k = int(request.args.get("top_k", 40))
top_p = float(request.args.get("top_p", 0.0))
reppenalty = float(request.args.get("reppenalty", 1.2))
response_queue = queue.Queue()
reasoning_queue.put({
'text_input': text,
'temperature': temp,
'top_k': top_k,
'top_p': top_p,
'repetition_penalty': reppenalty,
'response_queue': response_queue
})
@stream_with_context
def event_stream():
while True:
output = response_queue.get()
if "error" in output:
yield "data: \n\n"
break
text_chunk = output.get("text")
if text_chunk:
for word in text_chunk.split(' '):
clean_word = word.strip()
if clean_word:
yield "data: " + clean_word + "\n\n"
yield "data: \n\n"
break
return Response(event_stream(), mimetype="text/event-stream")
@app.route("/api/v1/generate", methods=["POST"])
def generate():
data = request.get_json()
text = data.get("text", "")
temp = float(data.get("temp", 0.7))
top_k = int(data.get("top_k", 40))
top_p = float(data.get("top_p", 0.0))
reppenalty = float(data.get("reppenalty", 1.2))
response_queue = queue.Queue()
reasoning_queue.put({
'text_input': text,
'temperature': temp,
'top_k': top_k,
'top_p': top_p,
'repetition_penalty': reppenalty,
'response_queue': response_queue
})
output = response_queue.get()
if "error" in output:
return jsonify({"error": output["error"]}), 500
result_text = output.get("text", "").strip()
return jsonify({"response": result_text})
@app.route("/api/v1/feedback", methods=["POST"])
def feedback():
data = request.get_json()
feedback_text = data.get("feedback_text")
correct_category = data.get("correct_category")
if feedback_text and correct_category:
feedback_queue.put((feedback_text, correct_category))
return jsonify({"status": "feedback received"})
return jsonify({"status": "feedback failed"}), 400
@app.route("/api/v1/tts", methods=["POST"])
def tts_api():
return tts_route()
@app.route("/api/v1/stt", methods=["POST"])
def stt_api():
return stt_route()
@app.route("/api/v1/sentiment", methods=["POST"])
def sentiment_api():
return sentiment_route()
@app.route("/api/v1/imagegen", methods=["POST"])
def imagegen_api():
return imagegen_route()
@app.route("/api/v1/musicgen", methods=["POST"])
def musicgen_api():
return musicgen_route()
@app.route("/api/v1/translation", methods=["POST"])
def translation_api():
return translation_route()
@app.route("/api/v1/codegen", methods=["POST"])
def codegen_api():
return codegen_route()
@app.route("/api/v1/text_to_video", methods=["POST"])
def text_to_video_api():
return text_to_video_route()
@app.route("/api/v1/summarization", methods=["POST"])
def summarization_api():
return summarization_route()
@app.route("/api/v1/image_to_3d", methods=["POST"])
def image_to_3d_api():
return image_to_3d_route()
@app.route("/api/v1/sadtalker", methods=["POST"])
def sadtalker():
from sadtalker_api import router as sadtalker_router
return sadtalker_router.create_video()
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860)