app.py

from fastapi import FastAPI, File, UploadFile
import librosa
import numpy as np
import shutil
import uvicorn
import os
from funasr import AutoModel
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    middleware=[
        Middleware(
            CORSMiddleware,
            allow_origins=["*"],  # Cho phép tất cả các origin
            allow_credentials=True,
            allow_methods=["*"],  # Cho phép tất cả các phương thức
            allow_headers=["*"],  # Cho phép tất cả các header
        )
    ]
)
# Tạo thư mục temp nếu chưa có
if not os.path.exists("temp"):
    os.makedirs("temp")

# Load mô hình SenseVoiceSmall từ Hugging Face
model_dir = "FunAudioLLM/SenseVoiceSmall"
model = AutoModel(
    model=model_dir,
    vad_model="fsmn-vad",
    vad_kwargs={"max_single_segment_time": 30000},
    device="cuda:0",
    hub="hf",
)

# Hàm tính RMS energy
def calculate_rms_energy(audio_path):
    y, sr = librosa.load(audio_path)
    rms = librosa.feature.rms(y=y)[0]
    return np.mean(rms)

# Hàm phát hiện tiếng ồn
def detect_noise(audio_path):
    rms_energy = calculate_rms_energy(audio_path)
    res = model.generate(input=audio_path, language="auto", audio_event_detection=True)
    audio_events = res[0].get("audio_event_detection", {})

    if rms_energy > 0.02:
        return "ồn ào"
    elif rms_energy > 0.01:
        for event_label, event_score in audio_events.items():
            if event_score > 0.7 and event_label in ["laughter", "applause", "crying", "coughing"]:
                return f"ồn ào ({event_label})"
    return "yên tĩnh"

@app.get("/")
def read_root():
    return {"message": "Hello, World!"}

print(app.routes)

# API nhận file âm thanh từ Flutter
@app.post("/detect-noise/")
async def detect_noise_api(file: UploadFile = File(...)):
    try:
        logger.info("Tên file: %s", file.filename)
        logger.info("Loại file: %s", file.content_type)
        file_size = len(await file.read())
        logger.info("Kích thước file: %s bytes", file_size)
        await file.seek(0)  # Reset lại vị trí đọc file

        file_path = f"temp/{file.filename}"
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer)

        result = detect_noise(file_path)
        return {"noise_level": result}
    except Exception as e:
        logger.exception("Lỗi trong API: %s", e)
        return {"error": str(e)}

# Chạy FastAPI trên Hugging Face Spaces
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=7860)


# from fastapi import FastAPI, UploadFile, File
# from starlette.middleware import Middleware
# from starlette.middleware.cors import CORSMiddleware
# import logging

# logging.basicConfig(level=logging.INFO)
# logger = logging.getLogger(__name__)

# app = FastAPI(
#     middleware=[
#         Middleware(
#             CORSMiddleware,
#             allow_origins=["*"],
#             allow_credentials=True,
#             allow_methods=["*"],
#             allow_headers=["*"],
#         )
#     ]
# )

# @app.post("/detect-noise/")
# async def detect_noise_api(file: UploadFile = File(...)):
#     logger.info("Đã nhận được yêu cầu!")
#     return {"message": "OK"}