auralis model & files

api/routes.py

@@ -0,0 +1,55 @@
+from fastapi import File, UploadFile, APIRouter
+from audio.preprocessing import load_audio, extract_features
+from model.ecapa import ECAPAENCODER
+# from model.scorer import fatigue_score_0_to_100, prosody_score  ## for prosody scoring
+from model.scorer import fatigue_score_0_to_100
+from fastapi.responses import JSONResponse
+import numpy as np
+from utils.logger import logger
+from utils.file_utils import save_temp_audio
+from core.config import LOW_PERCENTILE, HIGH_PERCENTILE, FATIGUE_AXIS, REF_C_H, MAX_DURATION_SEC
+# from audio.feature_extractor import get_prosody_stats
+from fastapi import HTTPException, status
+from audio.validators import validate_audio_duration, validate_audio_file, AudioValidationError
+
+
+C_h = np.load(REF_C_H)
+fatigue_axis = np.load(FATIGUE_AXIS)
+low = float(np.load(LOW_PERCENTILE)["arr_0"])
+high = float(np.load(HIGH_PERCENTILE)["arr_0"])
+
+router = APIRouter()
+
+encoder = ECAPAENCODER()
+
+@router.post("/score")
+async def score_voice(file: UploadFile = File(...)):
+    try:
+
+        path = save_temp_audio(file)
+        validate_audio_file(
+            file_path=path,
+            original_filename=file.filename
+        )
+        wav = load_audio(path)
+        # prosody_features = get_prosody_stats(wav)
+        # p_score, report = prosody_score(prosody_features)
+        features = extract_features(wav)
+        wav = wav.squeeze()
+        emb = encoder.encode(wav)
+        score = float(fatigue_score_0_to_100(emb, C_h, fatigue_axis, low, high))
+        # return {"fatigue_score": score, "prosody_score": p_score, "prosody_report": report}
+        return {"fatigue_score" : score}
+
+    except AudioValidationError as e:
+        logger.warning(str(e))
+        raise HTTPException(
+            status_code= status.HTTP_400_BAD_REQUEST,
+            detail = str(e)
+        )
+    except Exception as e:
+        logger.exception("Unexpected server error")
+        raise HTTPException(
+            status_code= status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail = "Unexpected server error."
+        )

audio/feature_extractor.py

@@ -0,0 +1,22 @@
+# import numpy as np
+# import parselmouth
+# from core.config import SAMPLE_RATE
+
+# def get_prosody_stats(waveforms, sr=SAMPLE_RATE):
+#     feats = {"pitch_mean" : [], "pitch_std" : [], "jitter" : [], "shimmer" : [], "hnr" : []}
+
+#     for wav in waveforms:
+#         snd = parselmouth.Sound(wav.numpy, sampling_frequency=sr)
+#         pitch = snd.to_pitch()
+
+#         feats["pitch_mean"].append(pitch.mean())
+#         feats["pitch_std"].append(pitch.stdev())
+#         feats["jitter"].append(snd.get_jitter_local())
+#         feats["shimmer"].append(snd.get_shimmer_local())
+#         feats["hnr"].append(snd.to_harmonicity().mean())
+
+#         thresholds = {}
+
+#         for k, v in feats.items():
+#             thresholds[k] = (np.percentile(v, 5), np.percentile(v, 95))
+#         return thresholds

audio/preprocessing.py

@@ -0,0 +1,109 @@
+import torch
+import torchaudio
+import torch.nn.functional as F
+from core.config import SAMPLE_RATE, DEVICE, N_MELS, TARGET_LEN
+from pydub import AudioSegment
+import numpy as np
+
+mel_transform = torchaudio.transforms.MelSpectrogram(
+    sample_rate = SAMPLE_RATE,
+    n_fft = 400,
+    hop_length = 256,
+    n_mels = N_MELS    
+).to(DEVICE)
+
+amp_to_db = torchaudio.transforms.AmplitudeToDB().to(DEVICE)
+
+# def load_audio(path: str) -> torch.Tensor:
+#     wav, sr = torchaudio.load(path)
+#     if sr != SAMPLE_RATE:
+#         wav = torchaudio.transforms.Resample(wav, sr, SAMPLE_RATE)
+#     if wav.shape[0] > 1:
+#         wav = wav.mean(dim = 0)
+#     return wav.to(DEVICE)  
+
+import torch
+import torchaudio
+import torch.nn.functional as F
+import numpy as np
+from pydub import AudioSegment
+
+class AudioLoadError(Exception):
+    pass
+
+def load_audio(path: str) -> torch.Tensor:
+    waveform = None
+    sr = None
+
+    # --- primary loader ---
+    try:
+        waveform, sr = torchaudio.load(path)
+    except Exception as e1:
+        # --- fallback loader ---
+        try:
+            audio = AudioSegment.from_file(path)
+            audio = audio.set_channels(1).set_frame_rate(SAMPLE_RATE)
+            samples = np.array(audio.get_array_of_samples(), dtype=np.float32)
+            if samples.size == 0:
+                raise AudioLoadError("Empty audio file")
+
+            waveform = torch.from_numpy(samples)
+            sr = SAMPLE_RATE
+
+        except Exception as e2:
+            raise AudioLoadError(
+                f"Failed to decode audio file: {str(e2)}"
+            ) from e2
+
+    # ---- sanity checks ----
+    if waveform is None or waveform.numel() == 0:
+        raise AudioLoadError("Loaded audio is empty")
+
+    # mono
+    if waveform.dim() > 1:
+        waveform = waveform.mean(dim=0)
+
+    # resample
+    if sr != SAMPLE_RATE:
+        waveform = torchaudio.transforms.Resample(sr, SAMPLE_RATE)(waveform)
+
+    # duration control
+    if waveform.numel() < TARGET_LEN:
+        raise AudioLoadError("Audio too short for analysis")
+
+    if waveform.numel() > TARGET_LEN:
+        waveform = waveform[:TARGET_LEN]
+    else:
+        waveform = F.pad(waveform, (0, TARGET_LEN - waveform.numel()))
+
+    return waveform.float()
+
+
+
+def waveform_to_mel(waveform: torch.Tensor):
+    """
+    waveform: [T]
+    returns: [1, T, N_MELS]
+    """
+    mel = mel_transform(waveform.unsqueeze(0))   # [1, n_mels, frames]
+    mel = amp_to_db(mel)
+    mel = mel.transpose(1, 2)                     # [1, frames, n_mels]
+    return mel
+
+def pad_time_dim(mel):
+    T = mel.shape[1]
+    pad_len = (8 - (T % 8)) % 8
+    if pad_len > 0:
+        mel = F.pad(mel, (0, 0, 0, pad_len))
+    return mel
+
+
+def extract_features(wav: torch.Tensor) -> torch.Tensor:
+    mel = mel_transform(wav.unsqueeze(0))
+    mel = amp_to_db(mel)
+    if mel.dim == 4:
+        mel = mel.squeeze(1)
+
+    mel.transpose(1, 2)  # [B, T, N_MELS]
+    mel = pad_time_dim(mel)
+    return mel

audio/validators.py

@@ -0,0 +1,36 @@
+import soundfile as sf
+from core.config import MAX_DURATION_SEC, MIN_DURATION_SEC, ALLOWED_EXTENSIONS
+from utils.logger import logger
+
+class AudioValidationError(ValueError):
+    pass
+
+
+def validate_audio_duration(filepath: str, max_duration: float = MAX_DURATION_SEC):
+    try:
+        info = sf.info(filepath)
+    except RuntimeError:
+        raise AudioValidationError("Invalid or corrupted audio file.")
+
+    duration = info.frames / float(info.samplerate)
+
+    if duration > max_duration or duration < MIN_DURATION_SEC:
+        logger.warning(f"Unsupported file length received: {original_filename}")
+        raise AudioValidationError(
+            f"Audio duration {duration:.2f}s invalid. "
+            f"Allowed range: {MIN_DURATION_SEC:.2f}s – {max_duration:.2f}s."
+        )
+
+    return duration
+
+def validate_audio_file(file_path: str, original_filename: str):
+    ext = original_filename.lower().rsplit(".", 1)[-1]
+    ext = "." + ext
+
+    if ext not in ALLOWED_EXTENSIONS:
+        logger.warning(f"Unsupported file format received: {original_filename}")
+        raise AudioValidationError(
+            f"Unsupported file type {ext}. Allowed formats are: " + ", ".join(ALLOWED_EXTENSIONS)
+            )
+
+    validate_audio_duration(file_path)

core/config.py

@@ -0,0 +1,29 @@
+from pathlib import Path
+import torch
+import yaml
+
+BASE_DIR = Path(__file__).resolve().parents[1]
+
+MODEL_DIR = f"{BASE_DIR}/models/ecapa_supcon_model.pth"
+REF_EMB = f"{BASE_DIR}/data/reference_embeddings_192-d.npy"
+REF_C_H = f"{BASE_DIR}/data/centroid_healthy.npy"
+FATIGUE_AXIS = f"{BASE_DIR}/data/fatigue_axis.npy"
+LOW_PERCENTILE = f"{BASE_DIR}/data/low_percentile.npz"
+HIGH_PERCENTILE = f"{BASE_DIR}/data/high_percentile.npz"
+CONFIG_PATH = f"{BASE_DIR}/model/config.yaml"
+
+SAMPLE_RATE = 16000
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+TARGET_SEC = 5
+N_MELS = 80
+TARGET_LEN = SAMPLE_RATE * TARGET_SEC
+MAX_DURATION_SEC = 10.0
+MIN_DURATION_SEC = 5.0
+ALLOWED_EXTENSIONS = {".wav", ".mp3", ".m4a"}
+
+with open(CONFIG_PATH, "r") as f:
+    CONFIG = yaml.safe_load(f)
+
+print(f"Model directory is set to: {MODEL_DIR}")
+print(f"base dir: {BASE_DIR}")
+print(f"ref emb path: {REF_EMB}")

data/centroid_healthy.npy

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:7290aa86b3a3c2d6dc739fc1a305f969d0ab442a81fd65e1bd0157032d0a4bac
+size 896

data/fatigue_axis.npy

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:4b68e5a5ded25edd95efc6ee0f4a45804c46023a994cc5fa616a3a89100a698f
+size 896

data/high_percentile.npz

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:e24a1e11b2ea4e347d9452a96be376d8886ec59a2f3cc8b8de3d3102dd1115d1
+size 211

data/low_percentile.npz

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:5a2e8eb7c090098d787b2a72e9300c77c811815307b808ab05ab4d3bef6ad639
+size 211

data/reference_embeddings_192-d.npy

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:60ecf6c5194ccf55d9729786aec24c02308a519097cf7fd408d6d6f0bccd9783
+size 57448

main.py

@@ -0,0 +1,34 @@
+from fastapi import FastAPI, APIRouter, Request
+from api.routes import router
+import time
+import logging
+
+logging.basicConfig(
+    level = logging.INFO,
+    format = "%(asctime)s - %(levelname)s - %(message)s"
+)
+
+app = FastAPI(title = "Vocal fatigue scoring API")
+
+@app.middleware("HTTP")   
+async def log_requests(request: Request, call_next):
+    start = time.time()
+    response = await call_next(request)
+    duration = time.time() - start
+
+    logging.info(
+        f"{request.method} {request.url.path} "
+        f"status_code = {response.status_code} "
+        f"time = {duration:.3f}"
+    )
+    return response
+
+api_v1 = APIRouter(prefix="/api/v1")
+api_v1.include_router(router, prefix="/voice")
+
+app.include_router(api_v1)
+
+@app.get("/health")
+def health():
+    return {"status" : "ok"}  
+    

model/config.yaml

@@ -0,0 +1,26 @@
+model:
+  name: ecapa_fatigue
+  version: 1.0.0
+  encoder: ecapa_tdnn
+  embedding_dim: 192
+  sampling_rate: 16000
+  N_mels: 80
+scoring:
+  method: calibrated_sigmoid
+  raw_low_percentile: 5
+  raw_high_percentile: 95
+preprocessing:
+  mono: True
+  target_len: 48000
+prosody_thresholds:
+  pitch_mean: [110, 180]       # Hz
+  pitch_std: [5, 40]           # Hz
+  jitter: [0.0, 0.5]           # %
+  shimmer: [0.0, 1.0]          # %
+  hnr: [10, 30]                # dB
+audio:
+  min_duration: 5.0            # seconds
+  max_duration: 10.0           # seconds
+  allow_trim: False
+  allowed_formats: ['.wav', '.m4a', '.mp3']
+  required_sampling_rate: 16000

model/ecapa.py

@@ -0,0 +1,50 @@
+import torch
+import torchaudio
+from speechbrain.lobes.models.ECAPA_TDNN import ECAPA_TDNN
+from core.config import MODEL_DIR, DEVICE, N_MELS
+from audio.preprocessing import waveform_to_mel
+import numpy as np
+
+
+class ECAPAENCODER:
+    def __init__(self):
+        self.ecapa = ECAPA_TDNN(
+            input_size = N_MELS,
+            lin_neurons = 192,
+            channels = [512, 512, 512],
+            kernel_sizes = [5, 3, 3],
+            dilations = [1, 2 , 3]
+        ).to(DEVICE)
+
+        checkpoint = torch.load(MODEL_DIR, map_location = DEVICE)
+        self.ecapa.load_state_dict(checkpoint['ecapa_state_dict'])
+        self.ecapa.eval()
+
+    @torch.no_grad()
+    def encode(self, waveform):
+        """
+        waveform: Tensor [T]
+        returns: np.ndarray [192]
+        """
+
+        # ---- safety checks ----
+        if waveform.dim() == 2 and waveform.shape[0] == 1:
+            waveform = waveform.squeeze(0)
+        if waveform.dim() != 1:
+            raise ValueError(f"Expected waveform [T], got {waveform.shape}")
+
+        waveform = waveform.float().to(DEVICE)
+        # waveform = waveform.unsqueeze(0)          # [1, T]
+
+        mel = waveform_to_mel(waveform)              # [1, n_mels, frames]
+
+        # if mel.dim() == 4:
+        #     mel = mel.squeeze(1)
+
+        # mel = mel.transpose(1, 2).contiguous()     # [1, T, n_mels]
+
+        # ---- critical debug line (keep this while testing) ----
+        print("ECAPA INPUT SHAPE:", mel.shape)
+
+        emb = self.ecapa(mel)                     # [1, 192]
+        return emb.squeeze(0).cpu().numpy()

model/scorer.py

@@ -0,0 +1,69 @@
+from core.config import LOW_PERCENTILE, HIGH_PERCENTILE, FATIGUE_AXIS, REF_C_H
+import numpy as np
+from core.config import CONFIG
+
+
+C_h = np.load(REF_C_H)
+fatigue_axis = np.load(FATIGUE_AXIS)
+low = float(np.load(LOW_PERCENTILE)["arr_0"])
+high = float(np.load(HIGH_PERCENTILE)["arr_0"])
+
+# def fatigue_score_0_to_100(emb: np.ndarray) -> float:
+#     raw = np.dot(emb - C_h, fatigue_axis)
+#     raw = np.clip(raw, low , high)
+#     return 100 * (raw - low) / (high - low)
+
+def fatigue_score_0_to_100(embedding, C_h, fatigue_axis, raw_low, raw_high, method='sigmoid'):
+    """
+    Compute a continuous fatigue score (0-100) from an embedding.
+
+    embedding: 192-d numpy array
+    C_h: healthy centroid (192-d)
+    fatigue_axis: unit vector from healthy -> fatigued (192-d)
+    raw_low, raw_high: training percentile values along the fatigue axis
+    method: 'linear', 'sigmoid', or 'smooth_linear'
+
+    Returns: float [0, 100]
+    """
+    # Project embedding along fatigue axis
+    raw = np.dot(embedding - C_h, fatigue_axis)
+
+    # Normalize raw value to [0, 1] within training range
+    normalized = (raw - raw_low) / (raw_high - raw_low)
+
+    # Clamp slightly beyond training range to avoid extreme scores
+    normalized = np.clip(normalized, -0.05, 1.05)
+
+    if method == 'linear':
+        score = normalized * 100  # simple linear scaling
+    elif method == 'sigmoid':
+        # Smooth sigmoid, less steep
+        midpoint = 0.5
+        scale = 0.25  # tune this for slope; bigger = smoother
+        score = 1 / (1 + np.exp(-(normalized - midpoint) / scale)) * 100
+    elif method == 'smooth_linear':
+        # Combine linear scaling with mild sigmoid smoothing at ends
+        # This gives a natural 0-100 spread but saturates near extremes
+        scale = 10  # controls smoothness near 0 and 100
+        score = normalized * 100
+        score = 100 / (1 + np.exp(- (score - 50) / scale))
+    else:
+        raise ValueError("method must be 'linear', 'sigmoid', or 'smooth_linear'")
+
+    # Ensure the output is float and bounded
+    return float(np.clip(score, 0, 100))
+
+
+# def prosody_score(prosody_feats):
+#     report = []
+#     thresholds = CONFIG['prosody_thresholds']
+
+#     for feat, val in prosody_feats.items():
+#         low, high = thresholds[feat]
+#         if val < low:
+#             report.append(f"{feat} is low → potential fatigue")
+#         elif val > high:
+#             report.append(f"{feat} is high → potential fatigue")
+    
+#     score = len(report)  # simple count, or map to 0-100 if needed
+#     return score, report

models/ecapa_supcon_model.pth

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:145325909e3e53c13bbb351537117727f4caf34828aea9c2e55b1d0f7262bfc6
+size 9208363

requirements.txt

@@ -0,0 +1,11 @@
+torch==2.1.1
+pytest
+fastapi
+torchaudio==2.1.1
+speechbrain
+numpy==1.26.4
+pathlib
+pydub
+uvicorn
+soundfile
+python-multipart

utils/file_utils.py

@@ -0,0 +1,8 @@
+import tempfile
+from fastapi import UploadFile
+
+def save_temp_audio(file: UploadFile) -> str:
+    suffix = file.filename.split(".")[-1]
+    with tempfile.NamedTemporaryFile(delete=False, suffix=f".{suffix}") as tmp_file:
+        tmp_file.write(file.file.read())
+        return tmp_file.name

utils/logger.py

@@ -0,0 +1,8 @@
+import logging
+
+logging.basicConfig(
+    level= logging.INFO,
+    format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+
+logger = logging.getLogger("vocal-fatigue-api")