Reorganisation

.gitignore

@@ -2,6 +2,7 @@
 __pycache__/
 *.py[cod]
 *$py.class
+.idea/
 
 # C extensions
 *.so
@@ -182,3 +183,4 @@ data/*
 
 # Mac
 .DS_Store
+*.pth

src/config.py

@@ -0,0 +1,20 @@
+import os
+import torch
+from dotenv import load_dotenv
+
+# Charger les variables d'environnement
+load_dotenv()
+HF_API_KEY = os.getenv("HF_API_KEY")
+
+if not HF_API_KEY:
+    raise ValueError("Le token Hugging Face n'a pas été trouvé dans .env")
+
+# Labels d'émotions
+LABELS = {"colere": 0, "neutre": 1, "joie": 2}
+NUM_LABELS = len(LABELS)
+
+# Choisir le device
+DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
+
+# Modèle Wav2Vec2
+MODEL_NAME = "facebook/wav2vec2-large-xlsr-53-french"

src/model/__init__.py

@@ -0,0 +1 @@
+

src/model/emotion_classifier.py

@@ -1,17 +1,15 @@
-import torch
 import torch.nn as nn
-from transformers import Wav2Vec2Model
 
-class Wav2Vec2EmotionClassifier(nn.Module):
+class EmotionClassifier(nn.Module):
+    def __init__(self, feature_dim, num_labels):
+        super(EmotionClassifier, self).__init__()
+        self.fc1 = nn.Linear(feature_dim, 256)
+        self.relu = nn.ReLU()
+        self.dropout = nn.Dropout(0.3)
+        self.fc2 = nn.Linear(256, num_labels)
 
-    def __init__(self, model_name="facebook/wav2vec2-large-xlsr-53-french", num_labels=3):
-        super(Wav2Vec2EmotionClassifier, self).__init__()
-        self.wav2vec2 = Wav2Vec2Model.from_pretrained(model_name)
-        self.fc = nn.Linear(self.wav2vec2.config.hidden_size, num_labels)
-        self.softmax = nn.Softmax(dim=1)
-
-    def forward(self, input_values):
-        outputs = self.wav2vec2(input_values).last_hidden_state
-        pooled_output = torch.mean(outputs, dim=1)  
-        logits = self.fc(pooled_output)
-        return self.softmax(logits)
+    def forward(self, x):
+        x = self.fc1(x)
+        x = self.relu(x)
+        x = self.dropout(x)
+        return self.fc2(x)

src/model/emotion_dataset.py

@@ -1,29 +0,0 @@
-import librosa
-import torch
-import pandas as pd
-from torch.utils.data import Dataset
-import os
-
-class EmotionDataset(Dataset):
-    def __init__(self, csv_file, processor):
-        self.data = pd.read_csv(csv_file, sep=",", header=0)
-        # print(self.data.info())  # Pour voir les premières lignes du dataset
-        self.processor = processor
-        self.emotion_labels = {"joie": 0, "colere": 1, "neutre": 2}
-        # print(self.data["emotion"].unique())  # Pour voir les valeurs exactes
-
-
-    def __len__(self):
-        return len(self.data)
-
-    def __getitem__(self, idx):
-        base_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data"))
-        audio_file = self.data.iloc[idx, 0]
-        label = self.emotion_labels[self.data.iloc[idx, 1].strip()]
-
-        audio_path = os.path.join(base_path, audio_file)
-        waveform, _ = librosa.load(audio_path, sr=16000)  # Chargement audio
-        input_values = self.processor(waveform, return_tensors="pt", sampling_rate=16000).input_values
-
-        return input_values.squeeze(0), torch.tensor(label, dtype=torch.long)
-

src/model/feature_extrator.py

@@ -0,0 +1,6 @@
+import torch
+from transformers import Wav2Vec2Model, Wav2Vec2Processor
+from config import MODEL_NAME, DEVICE
+
+processor = Wav2Vec2Processor.from_pretrained(MODEL_NAME)
+feature_extractor = Wav2Vec2Model.from_pretrained(MODEL_NAME).to(DEVICE)

src/model/test_wav2vec.py

@@ -1,62 +0,0 @@
-from transformers import Wav2Vec2Processor, Wav2Vec2ForCTC
-import torch
-import librosa
-import numpy as np
-import matplotlib.pyplot as plt
-
-# Charger le modèle et le processeur Wav2Vec 2.0
-model_name = "facebook/wav2vec2-large-xlsr-53-french"
-processor = Wav2Vec2Processor.from_pretrained(model_name)
-model = Wav2Vec2ForCTC.from_pretrained(model_name)
-
-# Charger l'audio
-audio_file = "C:\\Users\\fkpamegan\\Downloads\\datasets_oreau2_m_sessp_07a01Pa.wav"
-y, sr = librosa.load(audio_file, sr=16000) 
-
-# Prétraiter l'audio avec le processeur Wav2Vec 2.0
-input_values = processor(y, return_tensors="pt").input_values
-
-# Obtenir la prédiction (logits)
-with torch.no_grad():
-    logits = model(input_values).logitsa
-
-# Obtenir les IDs des tokens prédits (transcription)
-predicted_ids = torch.argmax(logits, dim=-1)
-
-# Décoder les IDs pour obtenir le texte transcrit
-transcription = processor.decode(predicted_ids[0])
-
-print("Transcription:", transcription)
-
-
-# Extraire le pitch (hauteur tonale) et l'intensité
-pitch, magnitudes = librosa.core.piptrack(y=y, sr=sr)
-intensity = librosa.feature.rms(y=y)  # Intensité (volume)
-
-# Calculer le tempo (vitesse de parole)
-tempo, _ = librosa.beat.beat_track(y=y, sr=sr)
-
-# Affichage du pitch
-plt.figure(figsize=(10, 6))
-librosa.display.specshow(pitch, x_axis='time', y_axis='log')
-plt.colorbar()
-plt.title("Pitch (Hauteur Tonale)")
-plt.show()
-
-# Affichage de l'intensité
-plt.figure(figsize=(10, 6))
-librosa.display.specshow(intensity, x_axis='time')
-plt.colorbar()
-plt.title("Intensité")
-plt.show()
-
-# Fusionner la transcription avec les caractéristiques prosodiques (pitch, intensité, tempo)
-features = np.hstack([
-    np.mean(intensity, axis=1),  # Moyenne de l'intensité
-    np.mean(pitch, axis=1),  # Moyenne du pitch
-    tempo  # Tempo
-])
-
-# Afficher les caractéristiques extraites
-print("Caractéristiques combinées :")
-print(features)

src/model/train.py

@@ -1,51 +0,0 @@
-import torch
-import torch.optim as optim
-import torch.nn as nn
-from torch.utils.data import DataLoader
-from transformers import Wav2Vec2Processor
-from emotion_dataset import EmotionDataset
-from emotion_classifier import Wav2Vec2EmotionClassifier
-import os
-from utils import collate_fn
-
-
-# Charger le processeur et le dataset
-processor = Wav2Vec2Processor.from_pretrained("facebook/wav2vec2-large-xlsr-53-french")
-data_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "data", "dataset.csv"))
-if not os.path.exists(data_path):
-    raise FileNotFoundError(f"Le fichier {data_path} est introuvable.")
-
-dataset = EmotionDataset(data_path, processor)
-dataloader = DataLoader(dataset, batch_size=4, shuffle=True, collate_fn=collate_fn)  # collate_fn ajouté
-
-
-# Initialiser le modèle
-device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-model = Wav2Vec2EmotionClassifier().to(device)
-
-# Définir la fonction de perte et l'optimiseur
-criterion = nn.CrossEntropyLoss()
-optimizer = optim.AdamW(model.parameters(), lr=5e-5)
-
-# Entraînement du modèle
-num_epochs = 10
-for epoch in range(num_epochs):
-    model.train()
-    total_loss = 0
-
-    for inputs, labels in dataloader:
-        inputs, labels = inputs.to(device), labels.to(device)
-
-        optimizer.zero_grad()
-        outputs = model(inputs)
-        loss = criterion(outputs, labels)
-        loss.backward()
-        optimizer.step()
-
-        total_loss += loss.item()
-
-    print(f"Epoch {epoch+1}, Loss: {total_loss:.4f}")
-
-# Sauvegarde du modèle
-torch.save(model.state_dict(), "wav2vec2_emotion.pth")
-print("Modèle sauvegardé !")

src/model/utils.py

@@ -1,8 +0,0 @@
-import torch
-from torch.nn.utils.rnn import pad_sequence
-
-def collate_fn(batch):
-    inputs, labels = zip(*batch)  # Séparer les features et les labels
-    inputs = pad_sequence(inputs, batch_first=True, padding_value=0)  # Padding des audios
-    labels = torch.tensor(labels, dtype=torch.long)  # Conversion en tensor
-    return inputs, labels

src/speech2.py

@@ -1,201 +0,0 @@
-import os
-import torch
-import torch.nn as nn
-import torch.optim as optim
-import soundfile as sf  
-import torchaudio
-import numpy as np
-from datasets import Dataset
-from transformers import (
-    Wav2Vec2Model,
-    Wav2Vec2Processor
-)
-from dotenv import load_dotenv
-from sklearn.metrics import accuracy_score
-
-# Charger .env pour Hugging Face API Key
-load_dotenv()
-HF_API_KEY = os.getenv("HF_API_KEY")
-
-if not HF_API_KEY:
-    raise ValueError("Le token Hugging Face n'a pas été trouvé dans .env")
-
-# Définition des labels pour la classification des émotions
-LABELS = {"colere": 0, "neutre": 1, "joie": 2}
-NUM_LABELS = len(LABELS)
-
-# Charger le processeur et le modèle pour l'extraction de features
-model_name = "facebook/wav2vec2-large-xlsr-53-french"
-device = "cuda" if torch.cuda.is_available() else "cpu"
-
-processor = Wav2Vec2Processor.from_pretrained(model_name)
-feature_extractor = Wav2Vec2Model.from_pretrained(model_name).to(device)
-
-# Resampleur pour convertir en 16 kHz
-resampler = torchaudio.transforms.Resample(orig_freq=48_000, new_freq=16_000)
-
-# Définition du classifieur amélioré
-class EmotionClassifier(nn.Module):
-    def __init__(self, feature_dim, num_labels):
-        super(EmotionClassifier, self).__init__()
-        self.fc1 = nn.Linear(feature_dim, 256)
-        self.relu = nn.ReLU()
-        self.dropout = nn.Dropout(0.3)
-        self.fc2 = nn.Linear(256, num_labels)
-
-    def forward(self, x):
-        x = self.fc1(x)
-        x = self.relu(x)
-        x = self.dropout(x)
-        return self.fc2(x)
-
-# Instancier le classifieur
-classifier = EmotionClassifier(feature_extractor.config.hidden_size, NUM_LABELS).to(device)
-
-# Charger les fichiers audio et leurs labels
-def load_audio_data(data_dir):
-    data = []
-    for label_name, label_id in LABELS.items():
-        label_dir = os.path.join(data_dir, label_name)
-        for file in os.listdir(label_dir):
-            if file.endswith(".wav"):
-                file_path = os.path.join(label_dir, file)
-                data.append({"path": file_path, "label": label_id})
-    return Dataset.from_list(data)
-
-# Chargement du dataset
-data_dir = "./dataset"
-ds = load_audio_data(data_dir)
-
-# Charger les fichiers audio avec SoundFile et rééchantillonner à 16 kHz
-def preprocess_audio(batch):
-    speech, sample_rate = sf.read(batch["path"], dtype="float32")
-
-    if sample_rate != 16000:
-        speech = torch.tensor(speech).unsqueeze(0)
-        speech = resampler(speech).squeeze(0).numpy()
-    
-    batch["speech"] = speech.tolist()  # Convertir en liste pour éviter les erreurs de PyArrow
-    batch["sampling_rate"] = 16000
-    return batch
-
-ds = ds.map(preprocess_audio)
-
-# Vérifier la distribution des longueurs des fichiers audio
-lengths = [len(sample["speech"]) for sample in ds]
-max_length = int(np.percentile(lengths, 95))
-
-# Transformer l'audio en features utilisables par le modèle
-def prepare_features(batch):
-    features = processor(
-        batch["speech"], 
-        sampling_rate=16000,  
-        padding=True,
-        truncation=True,
-        max_length=max_length,  
-        return_tensors="pt"
-    )
-    batch["input_values"] = features.input_values.squeeze(0)
-    batch["label"] = torch.tensor(batch["label"], dtype=torch.long)
-    return batch
-
-ds = ds.map(prepare_features)
-
-# Diviser les données en train et test
-ds = ds.train_test_split(test_size=0.2)
-train_ds = ds["train"]
-test_ds = ds["test"]
-
-# Fonction d'évaluation sur les données de test
-def evaluate(classifier, feature_extractor, test_ds):
-    classifier.eval()
-    correct = 0
-    total = 0
-
-    with torch.no_grad():
-        for batch in test_ds:
-            input_values = processor(
-                batch["speech"], 
-                sampling_rate=16000,  
-                return_tensors="pt", 
-                padding=True, 
-                truncation=True, 
-                max_length=max_length  
-            ).input_values.to(device)
-
-            features = feature_extractor(input_values).last_hidden_state.mean(dim=1)
-            logits = classifier(features)
-            predictions = logits.argmax(dim=-1)
-            labels = torch.tensor(batch["label"], dtype=torch.long, device=device)
-
-            correct += (predictions == labels).sum().item()
-            total += 1
-
-    return correct / total
-
-# Fonction d'entraînement
-def train_classifier(feature_extractor, classifier, train_ds, test_ds, epochs=10, batch_size=16):
-    optimizer = optim.Adam(classifier.parameters(), lr=1e-4)
-    scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=3, gamma=0.7)
-    loss_fn = nn.CrossEntropyLoss()
-
-    best_accuracy = 0.0  # Variable pour stocker la meilleure accuracy
-
-    for epoch in range(epochs):
-        classifier.train()
-        total_loss, correct = 0, 0
-        batch_count = 0
-
-        for i in range(0, len(train_ds), batch_size):
-            batch = train_ds[i: i + batch_size]
-            optimizer.zero_grad()
-
-            input_values = processor(
-                batch["speech"], 
-                sampling_rate=16000,  
-                return_tensors="pt", 
-                padding=True, 
-                truncation=True, 
-                max_length=max_length  
-            ).input_values.to(device)
-
-            with torch.no_grad():
-                features = feature_extractor(input_values).last_hidden_state.mean(dim=1)
-                features = (features - features.mean()) / features.std()  # Normalisation
-
-            logits = classifier(features)
-            labels = torch.tensor(batch["label"], dtype=torch.long, device=device)
-
-            if labels.numel() == 0:
-                continue
-
-            loss = loss_fn(logits, labels)
-            loss.backward()
-            optimizer.step()
-
-            total_loss += loss.item()
-            correct += (logits.argmax(dim=-1) == labels).sum().item()
-            batch_count += 1
-
-        train_acc = correct / len(train_ds)
-        test_acc = evaluate(classifier, feature_extractor, test_ds)
-        scheduler.step()
-
-        # Sauvegarde uniquement si l'accuracy sur test est la meilleure obtenue
-        if test_acc > best_accuracy:
-            best_accuracy = test_acc
-            torch.save({
-                "classifier_state_dict": classifier.state_dict(),
-                "feature_extractor_state_dict": feature_extractor.state_dict(),
-                "processor": processor
-            }, "best_emotion_model.pth")
-            print(f"✅ Nouveau meilleur modèle sauvegardé ! Accuracy Test: {best_accuracy:.4f}")
-
-        print(f"Epoch {epoch+1}/{epochs} - Loss: {total_loss/batch_count:.4f} - Train Accuracy: {train_acc:.4f} - Test Accuracy: {test_acc:.4f}")
-
-    return classifier
-
-# Entraînement
-trained_classifier = train_classifier(feature_extractor, classifier, train_ds, test_ds, epochs=10, batch_size=16)
-
-print("✅ Entraînement terminé, le meilleur modèle a été sauvegardé !")

src/train.py

@@ -0,0 +1,93 @@
+import torch
+import torch.optim as optim
+import torch.nn as nn
+import numpy as np
+from sklearn.metrics import accuracy_score
+from utils.dataset import load_audio_data
+from utils.preprocessing import preprocess_audio, prepare_features
+from model.emotion_classifier import EmotionClassifier
+from model.feature_extrator import feature_extractor, processor
+from config import DEVICE, NUM_LABELS
+import os
+
+# Charger les données
+data_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "data"))
+print(f"data dir  {data_dir}")
+ds = load_audio_data(data_dir)
+
+# Prétraitement
+ds = ds.map(preprocess_audio)
+
+# Ajustement de la longueur maximale
+lengths = [len(sample["speech"]) for sample in ds]
+max_length = int(np.percentile(lengths, 95))
+
+ds = ds.map(lambda batch: prepare_features(batch, max_length))
+
+# Séparation en train et test
+ds = ds.train_test_split(test_size=0.2)
+train_ds, test_ds = ds["train"], ds["test"]
+
+# Instancier le modèle
+classifier = EmotionClassifier(feature_extractor.config.hidden_size, NUM_LABELS).to(DEVICE)
+
+# Fonction d'entraînement
+def train_classifier(classifier, train_ds, test_ds, epochs=20, batch_size=8):
+    optimizer = optim.AdamW(classifier.parameters(), lr=2e-5, weight_decay=0.01)
+    loss_fn = nn.CrossEntropyLoss()
+    best_accuracy = 0.0
+
+    for epoch in range(epochs):
+        classifier.train()
+        total_loss, correct = 0, 0
+        batch_count = 0
+
+        for i in range(0, len(train_ds), batch_size):
+            batch = train_ds[i: i + batch_size]
+            optimizer.zero_grad()
+
+            input_values = processor(
+                batch["speech"], 
+                sampling_rate=16000,  
+                return_tensors="pt", 
+                padding=True, 
+                truncation=True, 
+                max_length=max_length  
+            ).input_values.to(DEVICE)
+
+            with torch.no_grad():
+                features = feature_extractor(input_values).last_hidden_state.mean(dim=1)
+
+            logits = classifier(features)
+            labels = torch.tensor(batch["label"], dtype=torch.long, device=DEVICE)
+
+            if labels.numel() == 0:
+                continue
+
+            loss = loss_fn(logits, labels)
+            loss.backward()
+            optimizer.step()
+
+            total_loss += loss.item()
+            correct += (logits.argmax(dim=-1) == labels).sum().item()
+            batch_count += 1
+
+        train_acc = correct / len(train_ds)
+
+        if train_acc > best_accuracy:
+            best_accuracy = train_acc
+            torch.save({
+                "classifier_state_dict": classifier.state_dict(),
+                "feature_extractor_state_dict": feature_extractor.state_dict(),
+                "processor": processor
+            }, "acc_model.pth")
+            print(f"Nouveau meilleur modèle sauvegardé ! Accuracy: {best_accuracy:.4f}")
+
+        print(f"Epoch {epoch+1}/{epochs} - Loss: {total_loss/batch_count:.4f} - Accuracy: {train_acc:.4f}")
+
+    return classifier
+
+# Lancer l'entraînement
+trained_classifier = train_classifier(classifier, train_ds, test_ds, epochs=20, batch_size=8)
+
+print("✅ Entraînement terminé, le meilleur modèle a été sauvegardé !")

src/utils/__init__.py

@@ -0,0 +1 @@
+

src/utils/dataset.py

@@ -0,0 +1,13 @@
+import os
+from datasets import Dataset
+from config import LABELS
+
+def load_audio_data(data_dir):
+    data = []
+    for label_name, label_id in LABELS.items():
+        label_dir = os.path.join(data_dir, label_name)
+        for file in os.listdir(label_dir):
+            if file.endswith(".wav"):
+                file_path = os.path.join(label_dir, file)
+                data.append({"path": file_path, "label": label_id})
+    return Dataset.from_list(data)

src/utils/preprocessing.py

@@ -0,0 +1,33 @@
+import soundfile as sf
+import torch
+import torchaudio
+import numpy as np
+from model.feature_extrator import processor
+from config import DEVICE
+
+# Resampler
+resampler = torchaudio.transforms.Resample(orig_freq=48_000, new_freq=16_000)
+
+def preprocess_audio(batch):
+    speech, sample_rate = sf.read(batch["path"], dtype="float32")
+
+    if sample_rate != 16000:
+        speech = torch.tensor(speech).unsqueeze(0)
+        speech = resampler(speech).squeeze(0).numpy()
+    
+    batch["speech"] = speech.tolist()
+    batch["sampling_rate"] = 16000
+    return batch
+
+def prepare_features(batch, max_length):
+    features = processor(
+        batch["speech"], 
+        sampling_rate=16000,  
+        padding=True,
+        truncation=True,
+        max_length=max_length,  
+        return_tensors="pt"
+    )
+    batch["input_values"] = features.input_values.squeeze(0)
+    batch["label"] = torch.tensor(batch["label"], dtype=torch.long)
+    return batch