MIF_E31231708/ml/predict_api.py

from pathlib import Path

import joblib
import numpy as np

from audio_utils_api import convert_audio_to_wav
from features import (
    LABEL_PD,
    LABEL_TPD,
    analyze_audio,
    build_prediction_explanation,
)


BASE_DIR = Path(__file__).resolve().parent
PROJECT_DIR = BASE_DIR.parent
MODEL_DIR = BASE_DIR / "models"
MODEL_PATH = MODEL_DIR / "svm_voice_confidence_model.joblib"

CONFIDENCE_THRESHOLD = 0.60
MODEL_NOT_FOUND_MESSAGE = "Model tidak ditemukan. Pastikan file model berada di folder ml/models/."

LABEL_DESCRIPTION = {
    LABEL_PD: "Percaya Diri",
    LABEL_TPD: "Tidak Percaya Diri",
}


def find_model_path():
    if MODEL_PATH.exists():
        return MODEL_PATH

    available_models = sorted(MODEL_DIR.glob("*.joblib"))
    if available_models:
        return available_models[0]

    raise FileNotFoundError(MODEL_NOT_FOUND_MESSAGE)


def load_prediction_model(model_path=None):
    model_path = Path(model_path) if model_path else find_model_path()
    if not model_path.exists():
        raise FileNotFoundError(MODEL_NOT_FOUND_MESSAGE)
    return joblib.load(model_path)


def get_expected_feature_count(model):
    if hasattr(model, "named_steps") and "scaler" in model.named_steps:
        return getattr(model.named_steps["scaler"], "n_features_in_", None)
    return getattr(model, "n_features_in_", None)


def calculate_indicator_probability(model_probability_pd, indicators):
    indicator_pd = (
        0.15 * indicators["volume_score"]
        + 0.35 * indicators["intonation_score"]
        + 0.50 * indicators["pause_score"]
    )

    adjusted_pd = 0.80 * model_probability_pd + 0.20 * indicator_pd

    if indicators["pause_score"] < 0.35:
        adjusted_pd -= (0.35 - indicators["pause_score"]) * 0.25
    if indicators["volume_score"] < 0.18:
        adjusted_pd -= (0.18 - indicators["volume_score"]) * 0.10
    if indicators["intonation_score"] < 0.45:
        adjusted_pd -= (0.45 - indicators["intonation_score"]) * 0.15

    return float(np.clip(adjusted_pd, 0.01, 0.99)), float(np.clip(indicator_pd, 0.0, 1.0))


def predict_audio(audio_path, model_path=None):
    model = load_prediction_model(model_path)
    wav_path = convert_audio_to_wav(audio_path)

    try:
        try:
            analysis = analyze_audio(wav_path, validate_quality=True)
        except ValueError as error:
            fallback_analysis = analyze_audio(wav_path, validate_quality=False)
            return {
                "is_valid_audio": False,
                "error_message": str(error),
                "predicted_label": None,
                "description": "Audio tidak valid",
                "confidence": 0.0,
                "probability_pd": 0.0,
                "probability_tpd": 0.0,
                "margin": 0.0,
                "audio_quality": fallback_analysis["audio_quality"],
                "voice_indicators": fallback_analysis["voice_indicators"],
                "explanation": str(error),
            }

        features = analysis["features"]
        feature_matrix = np.asarray(features, dtype=np.float32).reshape(1, -1)

        expected_feature_count = get_expected_feature_count(model)
        if expected_feature_count and feature_matrix.shape[1] != expected_feature_count:
            raise ValueError(
                "Jumlah fitur audio tidak sesuai dengan model. "
                f"Audio menghasilkan {feature_matrix.shape[1]} fitur, "
                f"sedangkan model mengharapkan {expected_feature_count} fitur."
            )

        prediction = model.predict(feature_matrix)
        probabilities = model.predict_proba(feature_matrix)

        predicted_label = str(prediction[0])
        class_probabilities = {
            str(label): float(probability)
            for label, probability in zip(model.classes_, probabilities[0])
        }
        model_probability_pd = float(class_probabilities.get(LABEL_PD, 0.0))
        probability_pd, indicator_pd_score = calculate_indicator_probability(
            model_probability_pd,
            analysis["voice_indicators"],
        )
        probability_tpd = float(1.0 - probability_pd)
        predicted_label = LABEL_PD if probability_pd >= probability_tpd else LABEL_TPD
        confidence = float(max(probability_pd, probability_tpd))
        margin = float(abs(probability_pd - probability_tpd))
        explanation = build_prediction_explanation(
            predicted_label,
            confidence,
            analysis["voice_indicators"],
        )

        return {
            "is_valid_audio": True,
            "predicted_label": predicted_label,
            "label": predicted_label,
            "description": LABEL_DESCRIPTION.get(predicted_label, predicted_label),
            "confidence": confidence,
            "probability_pd": probability_pd,
            "probability_tpd": probability_tpd,
            "margin": margin,
            "probabilities": {
                LABEL_PD: probability_pd,
                LABEL_TPD: probability_tpd,
            },
            "audio_quality": analysis["audio_quality"],
            "voice_indicators": analysis["voice_indicators"],
            "indicator_pd_score": indicator_pd_score,
            "model_probability_pd": model_probability_pd,
            "model_probability_tpd": float(class_probabilities.get(LABEL_TPD, 0.0)),
            "explanation": explanation,
        }
    finally:
        Path(wav_path).unlink(missing_ok=True)


def get_model_info():
    model_path = find_model_path()
    model = load_prediction_model(model_path)

    return {
        "model_path": str(model_path.relative_to(PROJECT_DIR)),
        "model_type": type(model).__name__,
        "classes": [str(label) for label in getattr(model, "classes_", [])],
        "expected_features": get_expected_feature_count(model),
    }