added whisper operating

app.py

@@ -1,30 +1,29 @@
+from fastapi import FastAPI, File, UploadFile, Form, HTTPException
+import whisper
 import logging
 from contextlib import contextmanager
-from fastapi import FastAPI, File, UploadFile, HTTPException
-from fastapi.responses import JSONResponse
-from fastapi.middleware.cors import CORSMiddleware
 import tempfile
 import os
+import keras
 import librosa
 import numpy as np
-import keras
+import re
+import Levenshtein
+from fastapi.responses import JSONResponse
+
+from fastapi.middleware.cors import CORSMiddleware
 from utils import (
-    create_cnn_model,
-    get_features,
     extract_features,
     pad_or_trim,
-    noise,
-    stretch,
-    pitch,
 )
 
-app = FastAPI(port=8000)
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", 
+    handlers=[logging.StreamHandler()] 
+)
 
-# origins = [
-#     "http://localhost:3000",
-#     "http://127.0.0.1:3000",
-#     # Add more origins if needed
-# ]
+app = FastAPI(port=8000)
 
 app.add_middleware(
     CORSMiddleware,
@@ -34,6 +33,25 @@ app.add_middleware(
     allow_headers=["*"],
 )
 
+filepath = os.path.abspath("cnn_1_v6_final_model.h5")
+if not os.path.exists(filepath):
+    raise FileNotFoundError(f"Model file not found at {filepath}")
+
+model = keras.models.load_model(filepath, compile=False)
+whisper_model = whisper.load_model("tiny")
+
+@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)
+
 @app.get("/")
 async def read_root():
     return {"message": "Welcome to the Defects_model API"}
@@ -73,21 +91,12 @@ logging.basicConfig(
 )
 
 
-@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)
-
 
 @app.post("/process-audio")
-async def process_audio(audio: UploadFile = File(...)):
+async def process_audio(
+    audio: UploadFile = File(...), 
+    phrase: str = Form(...)
+):
     if audio.content_type != "audio/mpeg":
         raise HTTPException(
             status_code=400, detail="Invalid file type. Only MP3 files are supported."
@@ -95,36 +104,57 @@ async def process_audio(audio: UploadFile = File(...)):
 
     try:
         audio_bytes = await audio.read()
-        logging.info(
-            f"Received audio bytes: {len(audio_bytes)} bytes"
-        ) 
+
+        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 = extract_features(audio_data, sample_rate)
             logging.info(f"Features extracted: shape = {features.shape}")
+
             target_shape = (1, model.input_shape[1])
             features = pad_or_trim(features, target_shape[1])
             features = np.expand_dims(features, axis=0)
 
             prediction = model.predict(features)
-
             logging.info(f"Prediction: {prediction}")
-            return {"prediction": prediction.tolist()}
-
-    except librosa.util.exceptions.ParameterError as e:
-        logging.error(f"Librosa error: {e}")
-        raise HTTPException(status_code=400, detail=f"Invalid audio file: {e}")
-    except ValueError as e:
-        logging.error(f"Value error: {e}")
-        raise HTTPException(status_code=400, detail=f"Invalid audio data: {e}")
+
+            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))
+
+            # Допускаем различие в 40% длины исходной фразы
+            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": prediction.tolist(),
+                "match_phrase": match_phrase,
+                "lev_distance": lev_distance,
+                "transcribed_text": transcribed_text_clean
+            }
+
     except Exception as e:
-        logging.exception(f"Error processing audio: {e}") 
+        logging.exception(f"Error processing audio: {e}")
         raise HTTPException(status_code=500, detail="Internal server error")
-

test.py

@@ -3,6 +3,7 @@ import numpy as np
 import keras
 import httpx
 import librosa
+import whisper
 
 from utils import (
     extract_features,
@@ -48,4 +49,14 @@ model = keras.models.load_model(filepath, compile=False)
 prediction = model.predict(features)
 print(f"Prediction: {prediction.tolist()}")
 
-test_get_answer(audio_file_path)
+
+
+def transcribe_russian(audio_file, model_name="tiny"):
+    model = whisper.load_model(model_name)
+    result = model.transcribe(audio_file, language="russian")
+    return result["text"]
+
+# Example usage:
+audio_file = "audio.mp3"
+text = transcribe_russian(audio_file)
+print(text)