from fastapi import FastAPI, File, UploadFile, Form, HTTPException import whisper import logging from contextlib import contextmanager import tempfile import os import keras import librosa import numpy as np import re import Levenshtein import tensorflow as tf from fastapi.responses import JSONResponse from fastapi.middleware.cors import CORSMiddleware from utils_api import get_features #вывод в консоль для просмотри на hugging face logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()] ) # Установка временной директории для кэша Numba os.environ['NUMBA_CACHE_DIR'] = '/tmp' # Инициализация FastAPI приложения app = FastAPI(port=8000) # Настройка CORS (Cross-Origin Resource Sharing) для обработки запросов с разных доменов app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["GET, POST"], allow_headers=["*"], ) # Инициализация и загрузка модели Whisper для распознавания речи cache_dir = "/tmp/whisper_cache" os.makedirs(cache_dir, exist_ok=True) whisper_model = whisper.load_model("tiny", download_root=cache_dir) # загрузка параметров модели filepath = "best_model.keras" if not os.path.exists(filepath): raise FileNotFoundError(f"Model file not found at {filepath}")\ model = tf.keras.models.load_model(filepath, compile=False) logging.info(model.summary()) # Контекстный менеджер для временных аудио файлов @contextmanager def temporary_audio_file(audio_bytes): """ Создает временный файл для хранения аудио данных и автоматически удаляет его после использования """ with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file: tmp_file.write(audio_bytes) tmp_file.flush() tmp_filename = tmp_file.name try: yield tmp_filename finally: if os.path.exists(tmp_filename): os.remove(tmp_filename) # Корневой endpoint @app.get("/") async def read_root(): return {"message": "Welcome to the Defects_model API"} # Endpoint для сохранения аудио файлов @app.post("/save-audio") async def save_audio(file: UploadFile = File(...)): """ Обработчик для сохранения загруженных аудио файлов """ if not file.content_type.startswith("audio/"): raise HTTPException(status_code=400, detail="Invalid file type") file_path = os.path.join("audio", file.filename) os.makedirs("audio", exist_ok=True) try: with open(file_path, "wb") as f: content = await file.read() f.write(content) return JSONResponse( content={"message": "File saved successfully", "filePath": file_path}, status_code=200, ) except Exception as e: return JSONResponse(content={"error": str(e)}, status_code=500) # Настройка пути для файла логов log_file_path = os.path.join("/tmp", "server.log") # Настройка логирования для отслеживания работы сервера logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.StreamHandler()] ) # Основной endpoint для обработки аудио @app.post("/process-audio") async def process_audio( audio: UploadFile = File(...), phrase: str = Form(...) ): """ Главный обработчик для анализа аудио файлов: - Делает предсказание моделью - Прогоняет аудио через openai-whisper для проверки фразы - Сравнивает полученный текст с ожидаемой фразой """ # Проверка формата файла if audio.content_type != "audio/mpeg": raise HTTPException( status_code=400, detail="Invalid file type. Only MP3 files are supported." ) try: # Чтение аудио файла audio_bytes = await audio.read() if not audio_bytes: raise HTTPException(status_code=400, detail="Received empty file") logging.info(f"Received audio bytes: {len(audio_bytes)} bytes") # Обработка аудио во временном файле with temporary_audio_file(audio_bytes) as tmp_filename: logging.info(f"Temporary file created: {tmp_filename}") # Загрузка аудио данных audio_data, sample_rate = librosa.load(tmp_filename, sr=None) logging.info( f"Audio loaded: sample rate = {sample_rate}, data shape = {audio_data.shape}" ) if not audio_data.any() or sample_rate == 0: raise ValueError("Empty or invalid audio data.") # Извлечение признаков из аудио features = get_features(tmp_filename) # features = np.expand_dims(features, axis=0) # Add batch dimension logging.info(f"Features extracted: shape = {features.shape}") # Получение предсказания от модели class_weights = {0: 0.5460790960451978, 1: 1.0068333333333332, 2: 1000.696369636963697} prediction = model.predict(features) logging.info(f"Prediction shape: {prediction.shape}") #умножаем предикт на веса классов for j in range(prediction.shape[1]): prediction[0, j] *= class_weights.get(j, 1.0) prediction[0, j] *= 10 logging.info(f"Prediction: {prediction}") response_answer = np.argmax(prediction) if (response_answer == 0): response_answer = 1 else: response_answer = 0 logging.info(f"Right or with defects: 1 or 0: {response_answer}") # Транскрибация аудио с помощью Whisper transcription_result = whisper_model.transcribe(tmp_filename, language="russian") transcribed_text = transcription_result["text"].lower().strip() # Очистка транскрибированного текста transcribed_text_clean = re.sub(r'[^\w\s]', '', transcribed_text) logging.info(f"Transcribed text (cleaned): {transcribed_text_clean}") # Сравнение с ожидаемой фразой lev_distance = Levenshtein.distance(transcribed_text_clean, phrase.lower().strip()) phrase_length = max(len(transcribed_text_clean), len(phrase)) # Определение допустимого расстояния Левенштейна max_acceptable_distance = 0.5 * phrase_length match_phrase = lev_distance <= max_acceptable_distance logging.info(f"Expected phrase: {phrase}, Is correct: {match_phrase}, Transcribed text: {transcribed_text_clean}, Levenshtein distance: {lev_distance}") # Возврат результатов return { "prediction": response_answer, "match_phrase": match_phrase } except Exception as e: logging.exception(f"Error processing audio: {e}") raise HTTPException(status_code=500, detail="Internal server error")