[Python] Why is my CNN+BiLSTM model's accuracy and loss fluctuating despite high F1-score...

# -*- coding: utf-8 -*-

# Uninstalling all relevant libraries to ensure a clean state
!pip uninstall -y numpy mediapipe opencv-python tensorflow pandas

# Installing all necessary libraries together
# This helps pip resolve compatible versions
!pip install numpy mediapipe opencv-python tensorflow pandas

import kagglehub

# Download latest version
path = kagglehub.dataset_download("nandwalritik/yoga-pose-videos-dataset")

print("Path to dataset files:", path)

import os
import shutil
import re

def group_videos_by_pose(source_folder, destination_folder):
"""
Groups video files in the source folder into subfolders by yoga pose name
extracted from the filename, and copies them into the destination folder.

Args:
source_folder (str): Path to the folder containing original yoga videos.
destination_folder (str): Path where grouped videos will be copied.
"""
if not os.path.exists(source_folder):
print(f"Error: Source folder '{source_folder}' not found.")
return

# Create destination folder if it doesn't exist
if not os.path.exists(destination_folder):
os.makedirs(destination_folder)
print(f"Destination folder created: {destination_folder}")

print(f"Grouping videos in '{source_folder}' by pose...")

# List all files in the source folder
for item_name in os.listdir(source_folder):
item_path = os.path.join(source_folder, item_name)

# Process only video files
if os.path.isfile(item_path) and item_name.lower().endswith(".mp4"):
# Extract the pose name from the file name
# Assumes format: personname_posename.mp4 (separated by last underscore)
match = re.search(r'_([^_]+)\.mp4$', item_name, re.IGNORECASE)
if match:
pose_name = match.group(1)

# Create destination subfolder for the pose
destination_pose_folder = os.path.join(destination_folder, pose_name)

if not os.path.exists(destination_pose_folder):
os.makedirs(destination_pose_folder)
print(f"Created pose folder: {pose_name}/")

# Copy video file to the corresponding pose folder
destination_file_path = os.path.join(destination_pose_folder, item_name)
try:
shutil.copy2(item_path, destination_file_path) # Preserve metadata
print(f" '{item_name}' → '{pose_name}/'")
except Exception as e:
print(f" Error copying '{item_name}': {e}")
else:
print(f" Could not extract pose name from '{item_name}', skipped.")
elif os.path.isdir(item_path):
print(f" '{item_name}/' is a folder, skipped.")
else:
print(f" '{item_name}' is not a valid video file, skipped.")

print("✅ Video grouping by pose completed.")

# Example usage:
# The source folder should be the extracted 'Yoga_Vid_Collected' folder from the ZIP file.
# Adjust the path depending on where you extracted the dataset.
# Your notebook variable 'extracted_video_dataset' may point to this structure.
# Example path: /content/extracted_dataset/Yoga_Vid_Collected
source_folder_path = os.path.join(path, "Yoga_Vid_Collected")
destination_folder_path = "/content/grouped_yoga_videos"

group_videos_by_pose(source_folder_path, destination_folder_path)

# You can print the folder structure afterward to verify:

# === 0. Setup and Imports ===
import os
import shutil
import zipfile
import csv
import re
import cv2
import numpy as np
# import mediapipe as mp # Uncomment when using MediaPipe
from google.colab import drive


pose_name = "Yoga_Vid_Collected"
video_folder_path = f"/content/drive/MyDrive/dataset2/{pose_name}.zip"
extracted_video_dataset = "/content/grouped_yoga_videos"
extracted_csv_dataset = "/content/extracted_csv_dataset"
drive_hedef_yolu = "MyDrive/Csv_Dataset_Mediapipe"
zip_adı = pose_name

# === 3. Keypoint Extraction with MediaPipe ===
mp_pose = mp.solutions.pose
pose_model = mp_pose.Pose(static_image_mode=False, min_detection_confidence=0.5)

def extract_mediapipe_keypoints(frame):
image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
results = pose_model.process(image_rgb)

if not results.pose_landmarks:
return None

keypoints = []
for lm in results.pose_landmarks.landmark:
keypoints.extend([lm.x, lm.y, lm.visibility])
return keypoints # 33 keypoints * 3 values (x, y, visibility) = 99 values


# === 4. Convert a Single Video to CSV ===
def video_to_csv_mediapipe(video_path, csv_path):
cap = cv2.VideoCapture(video_path)
frame_index = 0
fps = cap.get(cv2.CAP_PROP_FPS)

with open(csv_path, mode="w", newline="", encoding="utf-8") as csv_file:
fieldnames = ["timestamp_sec"] + [f"{coord}{i}" for i in range(1, 34) for coord in ['x', 'y', 'v']]
writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
writer.writeheader()

while True:
ret, frame = cap.read()
if not ret:
break
timestamp = frame_index / fps
keypoints = extract_mediapipe_keypoints(frame)
if keypoints is None:
frame_index += 1
continue
row = {"timestamp_sec": round(timestamp, 2)}
for i, val in enumerate(keypoints):
row[fieldnames[i + 1]] = val
writer.writerow(row)
frame_index += 1

cap.release()
print(f"✅ CSV file created: {csv_path}")

# === 5. Process All Videos ===
def is_video_file(filename):
return filename.lower().endswith(".mp4")

def process_all_videos_mediapipe(root_path):
for folder in os.listdir(root_path):
folder_path = os.path.join(root_path, folder)
if not os.path.isdir(folder_path):
continue
for video_file in os.listdir(folder_path):
if is_video_file(video_file):
video_path = os.path.join(folder_path, video_file)
video_id = os.path.splitext(video_file)[0]
csv_path = os.path.join(folder_path, f"{video_id}.csv")
print(f" İşleniyor: {video_file}")
video_to_csv_mediapipe(video_path, csv_path)

process_all_videos_mediapipe(extracted_video_dataset)


# === 6. Move CSV Files into csv/ Subfolder ===
def move_csv_files_to_subfolder(root_folder):
for yoga_folder in os.listdir(root_folder):
folder_path = os.path.join(root_folder, yoga_folder)
if not os.path.isdir(folder_path):
continue
csv_subfolder_path = os.path.join(folder_path, "csv")
os.makedirs(csv_subfolder_path, exist_ok=True)
for file in os.listdir(folder_path):
if file.endswith(".csv"):
shutil.move(os.path.join(folder_path, file), os.path.join(csv_subfolder_path, file))

move_csv_files_to_subfolder(extracted_video_dataset)


# === 7. Copy CSV Files to Upper-Level Folder ===
def copy_csv_files_to_upper_folder(source_root_folder, target_root_folder):
if not os.path.exists(target_root_folder):
os.makedirs(target_root_folder)
for pose_folder_name in os.listdir(source_root_folder):
pose_folder_path = os.path.join(source_root_folder, pose_folder_name)
csv_subfolder = os.path.join(pose_folder_path, "csv")
if os.path.isdir(csv_subfolder):
target_pose_folder = os.path.join(target_root_folder, pose_folder_name)
os.makedirs(target_pose_folder, exist_ok=True)
for file in os.listdir(csv_subfolder):
if file.endswith(".csv"):
shutil.copy2(os.path.join(csv_subfolder, file), os.path.join(target_pose_folder, file))

copy_csv_files_to_upper_folder(extracted_video_dataset, extracted_csv_dataset)





# -*- coding: utf-8 -*-
"""TezWoA02_TrainModel_MediaPipe.ipynb"""

# === 1. Libraries and Helper Functions ===
import os
import zipfile
from shutil import rmtree
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split

# Path to the folder containing extracted CSV files
extract_csv_dataset = "/content/dataset"

# === 3. Dataset Preparation ===
def load_sequences_mediapipe(base_path, test_size=0.40, window_size=50, step=10):
person_data = {}
label_map = {}
current_label = 0

for pose_folder in sorted(os.listdir(base_path)):
pose_path = os.path.join(base_path, pose_folder)
if not os.path.isdir(pose_path):
continue

if pose_folder not in label_map:
label_map[pose_folder] = current_label
current_label += 1

for file in os.listdir(pose_path):
if not file.endswith(".csv") or file.endswith("_genelEvre.csv"):
continue
person_name = file.split("_")[0]
csv_path = os.path.join(pose_path, file)
person_data.setdefault(person_name, []).append((pose_folder, csv_path))

train_people, test_people = train_test_split(sorted(person_data), test_size=test_size, random_state=42)

def augment_sequence(window, noise_std=0.005, time_shift_max=3):
# Add Gaussian noise
noise = np.random.normal(0, noise_std, window.shape)
window_noisy = window + noise

# Apply time shifting
shift = np.random.randint(-time_shift_max, time_shift_max + 1)
if shift > 0:
window_shifted = np.pad(window_noisy, ((shift, 0), (0, 0)), mode='edge')[:-shift]
elif shift < 0:
window_shifted = np.pad(window_noisy, ((0, -shift), (0, 0)), mode='edge')[-shift:]
else:
window_shifted = window_noisy

return window_shifted.astype(np.float32)

def spatial_scaling(window, scale_range=(0.9, 1.1)):
scale = np.random.uniform(*scale_range)
return (window * scale).astype(np.float32)

def mirror_pose(window):
# Mirror pose by flipping x-coordinates: new_x = 1 - x
flipped = window.copy()
flipped[:, ::2] = 1.0 - flipped[:, ::2]
return flipped.astype(np.float32)


def create_dataset(people, augment=False):
X, y = []
coord_columns = [f"{axis}{i}" for i in range(1, 34) for axis in ['x', 'y']]
for person in people:
for pose_label, csv_file in person_data[person]:
df = pd.read_csv(csv_file)
if not set(coord_columns).issubset(df.columns):
continue
features = df[coord_columns].to_numpy(dtype=np.float32)
if np.isnan(features).any() or np.isinf(features).any():
continue

for start in range(0, features.shape[0] - window_size + 1, step):
window = features[start:start + window_size]
if window.shape[0] == window_size:
X.append(window)
y.append(label_map[pose_label])

# Optional: add scaled version of the window
# X.append(spatial_scaling(window))
# y.append(label_map[pose_label])

# Add mirrored version of the window
X.append(mirror_pose(window))
y.append(label_map[pose_label])

# Add augmented version if enabled
if augment:
aug_window = augment_sequence(window)
X.append(aug_window)
y.append(label_map[pose_label])
return np.array(X), np.array(y)

X_train, y_train = create_dataset(train_people, augment=True)
X_test, y_test = create_dataset(test_people, augment=False)

return X_train, X_test, y_train, y_test, label_map

# Call the dataset loading function
X_train, X_test, y_train, y_test, label_map = load_sequences_mediapipe(extract_csv_dataset)
print("Training data:", X_train.shape)
print("Test data:", X_test.shape)

# === 4. PyTorch Dataset and DataLoader ===
import torch
from torch.utils.data import Dataset, DataLoader

class PoseSequenceDataset(torch.utils.data.Dataset):
def __init__(self, X, y):
self.X = torch.tensor(X, dtype=torch.float32)
self.y = torch.tensor(y, dtype=torch.long)

def __len__(self):
return len(self.X)

def __getitem__(self, idx):
return self.X[idx], self.y[idx]

# Create PyTorch datasets and loaders
train_dataset = PoseSequenceDataset(X_train, y_train)
test_dataset = PoseSequenceDataset(X_test, y_test)

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

# === 5. LSTM Model ===
import torch
import torch.nn as nn

class CNNPerFrame(nn.Module):
"""
A CNN block that processes a single frame of keypoints (33 keypoints × 2D coordinates).
"""
def __init__(self, input_channels=66, cnn_out_channels=32, kernel_size=3, dropout=0.2):
super(CNNPerFrame, self).__init__()
self.conv1 = nn.Conv1d(in_channels=2, out_channels=cnn_out_channels,
kernel_size=kernel_size, padding=kernel_size // 2)
self.relu = nn.ReLU()
self.bn = nn.BatchNorm1d(cnn_out_channels)
self.dropout = nn.Dropout(dropout)

def forward(self, x):
# Input shape: (B*T, 66) → reshape to (B*T, 2, 33)
x = x.view(-1, 33, 2).permute(0, 2, 1) # Shape: (B*T, 2, 33)
x = self.conv1(x) # Shape: (B*T, cnn_out_channels, 33)
x = self.relu(x)
x = self.bn(x)
x = self.dropout(x)
return x.view(x.size(0), -1) # Flatten to shape: (B*T, cnn_out_channels * 33)

class CNNLSTM_YogaModel(nn.Module):
"""
A hybrid model that applies CNN to each frame and LSTM across time steps for yoga pose classification.
"""
def __init__(self, frame_cnn_out=32*33, lstm_hidden_size=128, num_classes=6):
super().__init__()
self.cnn_per_frame = CNNPerFrame()
self.lstm = nn.LSTM(input_size=frame_cnn_out,
hidden_size=lstm_hidden_size,
num_layers=1,
batch_first=True,
bidirectional=True)
self.fc = nn.Linear(lstm_hidden_size * 2, num_classes) # Bidirectional → 2×hidden_size

def forward(self, x):
B, T, C = x.size() # Input shape: (batch_size, time_steps, features)
x = x.view(B * T, C) # Flatten to (B*T, features) for CNN
cnn_features = self.cnn_per_frame(x) # Apply CNN → (B*T, cnn_feature_size)
cnn_features = cnn_features.view(B, T, -1) # Reshape back to (B, T, cnn_feature_size)
lstm_out, _ = self.lstm(cnn_features) # LSTM output shape: (B, T, 2*hidden_size)
out = self.fc(lstm_out[:, -1, :]) # Use the last time step: (B, num_classes)
return out
# === 6. Training Preparation ===
from sklearn.utils.class_weight import compute_class_weight
from sklearn.metrics import accuracy_score, f1_score
import torch.nn.functional as F

# Set device to GPU if available
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

# Compute class weights to handle class imbalance
class_weights = compute_class_weight(class_weight='balanced', classes=np.unique(y_train), y=y_train)
class_weights_tensor = torch.tensor(class_weights, dtype=torch.float32).to(device)

# Initialize the model
model = CNNLSTM_YogaModel(
frame_cnn_out=32 * 33, # Output size of CNN per frame (cnn_out_channels × number of keypoints)
lstm_hidden_size=128, # Hidden size of the LSTM
num_classes=6 # Number of yoga pose classes
).to(device)

# Define loss function with class weights and optimizer
criterion = nn.CrossEntropyLoss(weight=class_weights_tensor)
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)


# === 7. Training Loop ===
num_epochs = 75
patience = 10
best_val_loss = float('inf')
epochs_no_improve = 0
early_stop = False

train_losses, val_losses = [], []
train_accuracies, val_accuracies, val_f1_scores = [], [], []

for epoch in range(num_epochs):
if early_stop:
break

model.train()
train_loss = 0
train_preds, train_labels = [], []

for inputs, labels in train_loader:
inputs, labels = inputs.to(device), labels.to(device)
outputs = model(inputs)
loss = criterion(outputs, labels)

optimizer.zero_grad()
loss.backward()
optimizer.step()

train_loss += loss.item()
preds = torch.argmax(outputs, dim=1)
train_preds.extend(preds.cpu().numpy())
train_labels.extend(labels.cpu().numpy())

train_acc = accuracy_score(train_labels, train_preds)

# Validation
model.eval()
val_loss = 0
val_preds, val_labels = [], []

with torch.no_grad():
for inputs, labels in test_loader:
inputs, labels = inputs.to(device), labels.to(device)
outputs = model(inputs)
loss = criterion(outputs, labels)
val_loss += loss.item()
preds = torch.argmax(outputs, dim=1)
val_preds.extend(preds.cpu().numpy())
val_labels.extend(labels.cpu().numpy())

val_acc = accuracy_score(val_labels, val_preds)
val_f1 = f1_score(val_labels, val_preds, average='weighted')

# Logging
train_losses.append(train_loss / len(train_loader))
val_losses.append(val_loss / len(test_loader))
train_accuracies.append(train_acc)
val_accuracies.append(val_acc)
val_f1_scores.append(val_f1)

if val_loss < best_val_loss:
best_val_loss = val_loss
epochs_no_improve = 0
torch.save(model.state_dict(), "best_model.pt")
else:
epochs_no_improve += 1
if epochs_no_improve >= patience:
early_stop = True

print(f"Epoch {epoch+1}/{num_epochs} - loss: {train_losses[-1]:.4f} - acc: {train_acc:.4f} - "
f"val_loss: {val_losses[-1]:.4f} - val_acc: {val_acc:.4f} - val_f1: {val_f1:.4f}")

# === 8. Evaluation and Model Saving ===
from sklearn.metrics import classification_report

model.eval()
all_preds, all_labels = [], []

with torch.no_grad():
for inputs, labels in test_loader:
inputs = inputs.to(device)
outputs = model(inputs)
preds = torch.argmax(outputs, dim=1).cpu().numpy()
all_preds.extend(preds)
all_labels.extend(labels.numpy())

print("Test Accuracy:", accuracy_score(all_labels, all_preds))
print(classification_report(all_labels, all_preds, target_names=label_map.keys()))

# Save the trained model
torch.save(model.state_dict(), "pose_lstm_model.pt")
print("✅ Model saved successfully.")

# === Loss and Accuracy Plots ===
import matplotlib.pyplot as plt

epochs = range(1, len(train_losses) + 1)

plt.figure(figsize=(14, 5))

# --- LOSS PLOT ---
plt.subplot(1, 2, 1)
plt.plot(epochs, train_losses, label="Train Loss")
plt.plot(epochs, val_losses, label="Validation Loss")
plt.xlabel("Epoch")
plt.ylabel("Loss")
plt.title("Loss Curve")
plt.legend()
plt.grid(True)

# --- ACCURACY PLOT ---
plt.subplot(1, 2, 2)
plt.plot(epochs, train_accuracies, label="Train Accuracy")
plt.plot(epochs, val_accuracies, label="Validation Accuracy")
plt.xlabel("Epoch")
plt.ylabel("Accuracy")
plt.title("Accuracy Curve")
plt.legend()
plt.grid(True)

plt.tight_layout()
plt.show()



# === Confusion Matrix ===
from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay

# Compute confusion matrix
cm = confusion_matrix(all_labels, all_preds)

# Class label names
class_names = list(label_map.keys())

# Display
disp = ConfusionMatrixDisplay(confusion_matrix=cm, display_labels=class_names)
fig, ax = plt.subplots(figsize=(10, 10))
disp.plot(ax=ax, cmap=plt.cm.Blues, xticks_rotation=45)
plt.title("Confusion Matrix")
plt.grid(False)
plt.show()




I'm training a CNN+BiLSTM model in PyTorch to classify yoga poses using 2D (x, y) coordinates from 33 keypoints extracted with MediaPipe. Each frame is passed through a 1D CNN to extract spatial features, which are then fed into a BiLSTM layer for temporal modeling. The model achieves an excellent test F1-score of 0.99, but the training and validation accuracy and loss curves fluctuate noticeably over epochs. I'm using CrossEntropyLoss with class weights, Adam optimizer (lr=0.001), batch size 32, and early stopping (patience = 10). The dataset is split by subject (60% train / 40% test), and data augmentation (Gaussian noise, mirroring, time shifting) is applied only on the training set. Despite close train/validation loss values and no clear signs of overfitting, the instability in the loss/accuracy graphs makes me question whether this behavior is normal. Could this be due to the model structure, augmentation strategy, or something else in my training loop?

Continue reading...