Upload 11 files

Dockerfile

@@ -0,0 +1,13 @@
+
+FROM python:3.11
+
+WORKDIR /app
+
+COPY . /app
+
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+EXPOSE 8000
+
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -1,12 +1,12 @@
----
-title: Speech Defects
-emoji: 📈
-colorFrom: red
-colorTo: yellow
-sdk: docker
-pinned: false
-license: mit
-short_description: Model api for detecting speech defects for ai challange
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: Speech Model
+emoji: 🏢
+colorFrom: yellow
+colorTo: blue
+sdk: streamlit
+sdk_version: 1.38.0
+app_file: app.py
+pinned: false
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,126 @@
+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 librosa
+import numpy as np
+import keras
+from utils import (
+    create_cnn_model,
+    get_features,
+    extract_features,
+    pad_or_trim,
+    noise,
+    stretch,
+    pitch,
+)
+
+app = FastAPI(port=8000)
+
+# origins = [
+#     "http://localhost:3000",
+#     "http://127.0.0.1:3000",
+#     # Add more origins if needed
+# ]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],#origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    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)
+target_shape = (32, 200)
+
+
+@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)
+
+
+logging.basicConfig(
+    level=logging.INFO,
+    filename="server.log",
+    filemode="w",
+    format="%(asctime)s - %(levelname)s - %(message)s",
+)
+
+
+@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()  # Make sure data is written to disk
+        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(...)):
+    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()
+        logging.info(
+            f"Received audio bytes: {len(audio_bytes)} bytes"
+        )  # Log size of audio 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)
+            # Add interpretation of prediction here (e.g., class labels)
+            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}")
+    except Exception as e:
+        logging.exception(f"Error processing audio: {e}")  # Log the full traceback
+        raise HTTPException(status_code=500, detail="Internal server error")
+

forms.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel
+
+class UserRegistration(BaseModel):
+    login: str
+    password: str
+
+class UserLoginForm(BaseModel):
+    login: str
+    password: str

main.py

@@ -0,0 +1,85 @@
+from fastapi import FastAPI, File, UploadFile, HTTPException
+from models import User, Course, connection
+from forms import UserRegistration, UserLoginForm
+from fastapi.responses import JSONResponse
+from utils import create_cnn_model, get_features, extract_features, pad_or_trim, noise, stretch, pitch
+from peewee import *
+import numpy as np
+import tensorflow as tf
+import keras
+import requests
+import io
+import os
+
+from fastapi.middleware.cors import CORSMiddleware
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+UPLOAD_DIR = 'audio'
+os.makedirs(UPLOAD_DIR, exist_ok=True)
+
+MODEL_SERVER_URL = "http://model-server-url/predict"
+
+@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(UPLOAD_DIR, file.filename)
+    
+    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)
+    
+    
+model = tf.keras.models.load_model("cnn_1_v6_final_model.keras", compile=False)
+
+@app.post("/process-audio")
+async def process_audio(audio: UploadFile = File(...)):
+    if audio.content_type != "audio/mpeg":
+        raise HTTPException(status_code=400, detail="Invalid file type. Please upload an MP3 file.")
+
+    audio_bytes = await audio.read()
+
+    features = get_features(audio_bytes)
+
+    if features is None:
+        raise HTTPException(status_code=400, detail="Invalid audio file. Please upload a valid MP3 file.")
+
+    prediction = model.predict(np.expand_dims(features, axis=0))
+
+    return {"prediction": prediction}
+
+
+'''
+@router.post("/login")
+async def login(user_data: UserLoginForm):
+    user = User.get(User.login == user_data.login)
+    if not user or user_data.password != user.password:
+        return {"message": "Invalid login or password"}
+    token_content = {"user_id": user.user_id}
+    jwt_token = jwt.encode(token_content, SECRET_KEY, algorithm=ALGORITHM)
+    return {"token": jwt_token}
+
+
+@router.post("/registration")
+async def registration(user_data: UserRegistration):
+    try:
+        new_user = User.create(login=user_data.login, password=user_data.password)
+        new_user.save()
+        return {"message": "User registered successfully"}
+    except IntegrityError:
+        return {"message": "User with this login already exists"} 
+'''

models.py

@@ -0,0 +1,56 @@
+from peewee import *
+
+connection = SqliteDatabase('database.db')
+
+
+
+class BaseModel(Model):
+    class Meta:
+        database = connection
+
+class User(BaseModel):
+    user_id = AutoField()
+    login = CharField(unique=True)
+    password = CharField()
+
+    class Meta:
+        db_table = 'Users'
+        order_by = ('user_id',)
+
+
+class Course(BaseModel):
+    course_id = AutoField()
+    name = CharField()
+    progress = IntegerField()
+
+    class Meta:
+        db_table = 'Courses'
+        order_by = ('course_id',)
+from peewee import *
+
+connection = SqliteDatabase('database.db')
+
+
+
+class BaseModel(Model):
+    class Meta:
+        database = connection
+
+class User(BaseModel):
+    user_id = AutoField()
+    login = CharField(unique=True)
+    password = CharField()
+
+    class Meta:
+        db_table = 'Users'
+        order_by = ('user_id',)
+
+
+class Course(BaseModel):
+    course_id = AutoField()
+    name = CharField()
+    progress = IntegerField()
+
+    class Meta:
+        db_table = 'Courses'
+        order_by = ('course_id',)

requirements.txt

@@ -0,0 +1,13 @@
+fastapi
+uvicorn
+torch
+librosa
+requests
+keras
+requests
+io
+os
+logging
+tempfile
+tensorflow
+keras

test.py

@@ -0,0 +1,51 @@
+import os
+import numpy as np
+import keras
+import httpx
+import librosa
+
+from utils import (
+    extract_features,
+    pad_or_trim,
+)
+
+def test_get_answer(audio_file_path: str):
+    url = "http://127.0.0.1:8000/process-audio"
+    headers = {
+        "accept": "application/json",
+    }
+
+    with open(audio_file_path, "rb") as audio_file:
+        files = {
+            "audio": ("test.mp3", audio_file, "audio/mp3")
+        }
+        response = httpx.post(url, headers=headers, files=files)
+    print("Status Code:", response.status_code)
+    print("Response JSON:", response.json())
+
+
+audio_file_path = "test_audio.mp3"
+if not os.path.exists(audio_file_path):
+    raise FileNotFoundError(f"Audio file not found at {audio_file_path}")
+
+audio_data, sample_rate = librosa.load(audio_file_path)
+
+features = extract_features(audio_data, sample_rate) 
+
+target_shape = (32, 200) 
+features = pad_or_trim(features, target_shape[1])
+
+
+features = np.expand_dims(features, axis=0)
+
+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)
+
+
+prediction = model.predict(features)
+print(f"Prediction: {prediction.tolist()}")
+
+test_get_answer(audio_file_path)

utils.py

@@ -0,0 +1,237 @@
+<<<<<<< HEAD
+import librosa
+import numpy as np
+from keras import layers, models
+
+def create_cnn_model(input_shape):
+    model = models.Sequential()
+
+    # First Convolutional Layer
+    model.add(layers.Conv1D(32, 3, activation='relu', input_shape=input_shape))
+    model.add(layers.MaxPooling1D(pool_size=2))
+
+    # Second Convolutional Layer
+    model.add(layers.Conv1D(64, 3, activation='relu'))
+    model.add(layers.MaxPooling1D(pool_size=2))
+
+    # Flatten layer
+    model.add(layers.Flatten())
+
+    # Dense layers
+    model.add(layers.Dense(128, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(256, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(512, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(512, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(256, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(128, activation='relu', input_shape=input_shape))
+
+    # Output layer
+    model.add(layers.Dense(1, activation='sigmoid'))
+
+    return model
+
+
+def get_features(path, duration=6):
+    try:
+        # Load audio file with specific duration and offset to handle silent parts
+        data, sample_rate = librosa.load(path, duration=2.5, offset=0.6)
+    except Exception as e:
+        print(f"Error loading {path}: {e}")
+        return None  # Skip the file if there's an error
+
+    # Without augmentation
+    res1 = extract_features(data, sample_rate)
+    result = np.array(res1)
+
+    # With noise
+    noise_data = noise(data)
+    res2 = extract_features(noise_data, sample_rate)
+    result = np.vstack((result, res2))
+
+    # Stretching and pitching
+    new_data = stretch(data)
+    data_stretch_pitch = pitch(new_data, sample_rate)
+    res3 = extract_features(data_stretch_pitch, sample_rate)
+    result = np.vstack((result, res3))
+
+    return result
+
+
+def extract_features(data, sample_rate, target_shape=40):
+    result = np.array([])
+
+    # ZCR
+    zcr = librosa.feature.zero_crossing_rate(y=data)
+    zcr = np.mean(zcr.T, axis=0)
+    zcr = pad_or_trim(zcr, target_shape)
+    result = np.hstack((result, zcr))
+
+    # Chroma_stft
+    stft = np.abs(librosa.stft(data))
+    chroma_stft = librosa.feature.chroma_stft(S=stft, sr=sample_rate)
+    chroma_stft = np.mean(chroma_stft.T, axis=0)
+    chroma_stft = pad_or_trim(chroma_stft, target_shape)
+    result = np.hstack((result, chroma_stft))
+
+    # MFCC
+    mfcc = librosa.feature.mfcc(y=data, sr=sample_rate, n_mfcc=13)
+    mfcc = np.mean(mfcc.T, axis=0)
+    mfcc = pad_or_trim(mfcc, target_shape)
+    result = np.hstack((result, mfcc))
+
+    # Root Mean Square Value
+    rms = librosa.feature.rms(y=data)
+    rms = np.mean(rms.T, axis=0)
+    rms = pad_or_trim(rms, target_shape)
+    result = np.hstack((result, rms))
+
+    # MelSpectrogram
+    mel = librosa.feature.melspectrogram(y=data, sr=sample_rate)
+    mel = np.mean(mel.T, axis=0)
+    mel = pad_or_trim(mel, target_shape)
+    result = np.hstack((result, mel))
+
+    return result
+
+
+def pad_or_trim(feature, target_shape):
+    """Pad or trim feature array to ensure a consistent shape."""
+    if len(feature) > target_shape:
+        feature = feature[:target_shape]
+    elif len(feature) < target_shape:
+        feature = np.pad(feature, (0, target_shape - len(feature)), mode='constant')
+    return feature
+
+
+def noise(data, noise_factor=0.005):
+    noise_amp = noise_factor * np.random.uniform() * np.amax(data)
+    data = data + noise_amp * np.random.normal(size=data.shape[0])
+    return data
+
+def stretch(data, rate=0.8):
+    return librosa.effects.time_stretch(data, rate=rate)
+
+def pitch(data, sample_rate, pitch_factor=0.7):
+    return librosa.effects.pitch_shift(data, sr=sample_rate, n_steps=pitch_factor)
+
+
+
+=======
+import librosa
+import numpy as np
+from keras import layers, models
+
+def create_cnn_model(input_shape):
+    model = models.Sequential()
+
+    # First Convolutional Layer
+    model.add(layers.Conv1D(32, 3, activation='relu', input_shape=input_shape))
+    model.add(layers.MaxPooling1D(pool_size=2))
+
+    # Second Convolutional Layer
+    model.add(layers.Conv1D(64, 3, activation='relu'))
+    model.add(layers.MaxPooling1D(pool_size=2))
+
+    # Flatten layer
+    model.add(layers.Flatten())
+
+    # Dense layers
+    model.add(layers.Dense(128, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(256, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(512, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(512, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(256, activation='relu', input_shape=input_shape))
+    model.add(layers.Dense(128, activation='relu', input_shape=input_shape))
+
+    # Output layer
+    model.add(layers.Dense(1, activation='sigmoid'))
+
+    return model
+
+
+def get_features(path, duration=6):
+    try:
+        # Load audio file with specific duration and offset to handle silent parts
+        data, sample_rate = librosa.load(path, duration=2.5, offset=0.6)
+    except Exception as e:
+        print(f"Error loading {path}: {e}")
+        return None  # Skip the file if there's an error
+
+    # Without augmentation
+    res1 = extract_features(data, sample_rate)
+    result = np.array(res1)
+
+    # With noise
+    noise_data = noise(data)
+    res2 = extract_features(noise_data, sample_rate)
+    result = np.vstack((result, res2))
+
+    # Stretching and pitching
+    new_data = stretch(data)
+    data_stretch_pitch = pitch(new_data, sample_rate)
+    res3 = extract_features(data_stretch_pitch, sample_rate)
+    result = np.vstack((result, res3))
+
+    return result
+
+
+def extract_features(data, sample_rate, target_shape=40):
+    result = np.array([])
+
+    # ZCR
+    zcr = librosa.feature.zero_crossing_rate(y=data)
+    zcr = np.mean(zcr.T, axis=0)
+    zcr = pad_or_trim(zcr, target_shape)
+    result = np.hstack((result, zcr))
+
+    # Chroma_stft
+    stft = np.abs(librosa.stft(data))
+    chroma_stft = librosa.feature.chroma_stft(S=stft, sr=sample_rate)
+    chroma_stft = np.mean(chroma_stft.T, axis=0)
+    chroma_stft = pad_or_trim(chroma_stft, target_shape)
+    result = np.hstack((result, chroma_stft))
+
+    # MFCC
+    mfcc = librosa.feature.mfcc(y=data, sr=sample_rate, n_mfcc=13)
+    mfcc = np.mean(mfcc.T, axis=0)
+    mfcc = pad_or_trim(mfcc, target_shape)
+    result = np.hstack((result, mfcc))
+
+    # Root Mean Square Value
+    rms = librosa.feature.rms(y=data)
+    rms = np.mean(rms.T, axis=0)
+    rms = pad_or_trim(rms, target_shape)
+    result = np.hstack((result, rms))
+
+    # MelSpectrogram
+    mel = librosa.feature.melspectrogram(y=data, sr=sample_rate)
+    mel = np.mean(mel.T, axis=0)
+    mel = pad_or_trim(mel, target_shape)
+    result = np.hstack((result, mel))
+
+    return result
+
+
+def pad_or_trim(feature, target_shape):
+    """Pad or trim feature array to ensure a consistent shape."""
+    if len(feature) > target_shape:
+        feature = feature[:target_shape]
+    elif len(feature) < target_shape:
+        feature = np.pad(feature, (0, target_shape - len(feature)), mode='constant')
+    return feature
+
+
+def noise(data, noise_factor=0.005):
+    noise_amp = noise_factor * np.random.uniform() * np.amax(data)
+    data = data + noise_amp * np.random.normal(size=data.shape[0])
+    return data
+
+def stretch(data, rate=0.8):
+    return librosa.effects.time_stretch(data, rate=rate)
+
+def pitch(data, sample_rate, pitch_factor=0.7):
+    return librosa.effects.pitch_shift(data, sr=sample_rate, n_steps=pitch_factor)
+
+
+
+>>>>>>> f3090616676ed6b7fcf9d16589c788e1843b194c