Create app.py

app.py

@@ -0,0 +1,178 @@
+import streamlit as st
+import torch
+import torchaudio
+from speechbrain.pretrained import EncoderClassifier, SpectralMaskEnhancement
+from speechbrain.pretrained import KWS
+import os
+from sklearn.metrics.pairwise import cosine_similarity
+import numpy as np
+
+# --- Konfigurasi dan Pemuatan Model (Dijalankan sekali) ---
+
+@st.cache_resource
+def load_models():
+    """Memuat model verifikasi speaker dan KWS."""
+    # Model untuk Verifikasi Speaker (Tahap 1)
+    spk_model = EncoderClassifier.from_hparams(
+        source="speechbrain/spkrec-xvect-voxceleb",
+        savedir="pretrained_models/spkrec-xvect-voxceleb"
+    )
+    
+    # Model untuk Deteksi Perintah (Tahap 2)
+    kws_model = KWS.from_hparams(
+        source="speechbrain/google_speech_command_xvector",
+        savedir="pretrained_models/google_speech_command_xvector"
+    )
+    
+    # Model untuk membersihkan audio (Opsional tapi bagus)
+    enhancer = SpectralMaskEnhancement.from_hparams(
+        source="speechbrain/metricgan-plus-voicebank",
+        savedir="pretrained_models/metricgan-plus-voicebank"
+    )
+    return spk_model, kws_model, enhancer
+
+# Memuat model
+spk_model, kws_model, enhancer = load_models()
+
+# Direktori pendaftaran
+ENROLL_DIR = "enroll/"
+THRESHOLD = 0.85 # Ambang batas kemiripan
+
+# --- Fungsi Helper ---
+
+def preprocess_audio(wav_file):
+    """Memuat, membersihkan, dan mengubah sample rate audio."""
+    try:
+        # Muat audio dari file yang di-upload
+        sig, fs = torchaudio.load(wav_file)
+
+        # Bersihkan noise (jika model enhancer ada)
+        if enhancer:
+            enhanced_sig = enhancer.enhance_batch(sig, lengths=torch.tensor([sig.shape[1]]))
+            sig = enhanced_sig.squeeze(0)
+
+        # Resample ke 16kHz (wajib untuk model)
+        if fs != 16000:
+            resampler = torchaudio.transforms.Resample(orig_freq=fs, new_freq=16000)
+            sig = resampler(sig)
+        
+        return sig
+    except Exception as e:
+        st.error(f"Error memproses audio: {e}")
+        return None
+
+@st.cache_data
+def get_enrollment_embeddings():
+    """
+    Membuat embedding (sidik jari suara) rata-rata 
+    untuk setiap pengguna di folder /enroll.
+    """
+    enrollment_data = {}
+    if not os.path.exists(ENROLL_DIR):
+        st.warning(f"Folder '{ENROLL_DIR}' tidak ditemukan.")
+        return {}
+
+    for speaker_name in os.listdir(ENROLL_DIR):
+        speaker_dir = os.path.join(ENROLL_DIR, speaker_name)
+        if os.path.isdir(speaker_dir):
+            embeddings = []
+            for wav_file in os.listdir(speaker_dir):
+                if wav_file.endswith(".wav"):
+                    wav_path = os.path.join(speaker_dir, wav_file)
+                    try:
+                        sig, fs = torchaudio.load(wav_path)
+                        if fs != 16000:
+                            resampler = torchaudio.transforms.Resample(orig_freq=fs, new_freq=16000)
+                            sig = resampler(sig)
+                        
+                        # Buat embedding
+                        with torch.no_grad():
+                            emb = spk_model.encode_batch(sig)
+                            emb = emb.squeeze()
+                            embeddings.append(emb.numpy())
+                    except Exception as e:
+                        st.error(f"Gagal memproses {wav_path}: {e}")
+            
+            if embeddings:
+                # Ambil rata-rata embedding untuk speaker ini
+                enrollment_data[speaker_name] = np.mean(embeddings, axis=0)
+    
+    return enrollment_data
+
+# --- Antarmuka Streamlit ---
+st.title("Sistem Verifikasi Perintah Suara 🔐")
+st.write("Unggah file .wav untuk verifikasi.")
+
+# Muat data pendaftaran
+enrollment_embeddings = get_enrollment_embeddings()
+
+if not enrollment_embeddings:
+    st.error("Tidak ada data pendaftaran yang ditemukan. Pastikan folder 'enroll' ada dan berisi file .wav.")
+else:
+    st.success(f"Berhasil memuat data pendaftaran untuk: {list(enrollment_embeddings.keys())}")
+
+uploaded_file = st.file_uploader("Pilih file audio...", type=["wav"])
+
+if uploaded_file is not None:
+    st.audio(uploaded_file, format="audio/wav")
+    
+    if st.button("Verifikasi Sekarang"):
+        with st.spinner("Memproses audio..."):
+            signal = preprocess_audio(uploaded_file)
+        
+        if signal is not None:
+            # --- TAHAP 1: VERIFIKASI SPEAKER (SIAPA?) ---
+            st.subheader("Tahap 1: Verifikasi Speaker")
+            
+            with torch.no_grad():
+                upload_embedding = spk_model.encode_batch(signal).squeeze().numpy()
+            
+            best_score = 0
+            best_match = "Tidak Dikenali"
+            
+            # Bandingkan dengan setiap speaker yang terdaftar
+            for speaker_name, enrolled_emb in enrollment_embeddings.items():
+                score = cosine_similarity(
+                    upload_embedding.reshape(1, -1),
+                    enrolled_emb.reshape(1, -1)
+                )[0][0]
+                
+                st.write(f"Skor kemiripan dengan {speaker_name}: **{score:.2f}**")
+                
+                if score > best_score:
+                    best_score = score
+                    best_match = speaker_name
+
+            # --- KEPUTUSAN TAHAP 1 ---
+            if best_score > THRESHOLD:
+                st.success(f"✅ **Akses Diberikan**: Dikenali sebagai **{best_match}** (Skor: {best_score:.2f})")
+                
+                # --- TAHAP 2: DETEKSI PERINTAH (APA?) ---
+                st.subheader("Tahap 2: Deteksi Perintah")
+                with st.spinner("Mendeteksi perintah..."):
+                    with torch.no_grad():
+                        # Model KWS memprediksi probabilitas
+                        prediction = kws_model.classify_batch(signal)
+                        
+                        # Ambil label dengan probabilitas tertinggi
+                        # 'prediction[0]' adalah tensor probabilitas
+                        # 'prediction[3]' adalah labelnya
+                        top_prob = torch.max(prediction[0]).item()
+                        top_label = prediction[3][0]
+
+                        # Logika untuk perintah "Buka" (Up) atau "Tutup" (Down)
+                        # Catatan: Sesuaikan label ini ("Up", "Down") dengan output model KWS Anda
+                        # Model Google Speech Command menggunakan "Up", "Down", "Left", "Right", "Yes", "No", dll.
+                        
+                        st.write(f"Perintah terdeteksi: **{top_label}** (Keyakinan: {top_prob:.2f})")
+
+                        if top_label.lower() == "up": # Asumsikan 'Up' = 'Buka'
+                            st.balloons()
+                            st.success(f"🔓 **Perintah Diterima**: `{best_match}` berkata 'BUKA'.")
+                        elif top_label.lower() == "down": # Asumsikan 'Down' = 'Tutup'
+                            st.success(f"🔒 **Perintah Diterima**: `{best_match}` berkata 'TUTUP'.")
+                        else:
+                            st.warning(f"Perintah '{top_label}' tidak dikenali sebagai 'Buka' atau 'Tutup'.")
+
+            else:
+                st.error(f"❌ **Akses Ditolak**: Suara tidak dikenali (Skor tertinggi: {best_score:.2f})")