From 0d61d47cfa084cc210f384fb2493763f0c0bc660 Mon Sep 17 00:00:00 2001 From: moscovskayaliza Date: Tue, 31 Mar 2026 17:01:42 +0300 Subject: [PATCH] started dymanic gect with lstm --- README.md | 9 ++- ml_gestures_dynamic/evaluate.py | 59 ++++++++++++++++ ml_gestures_dynamic/feature_extractor.py | 15 +++++ ml_gestures_dynamic/predict.py | 31 +++++++++ ml_gestures_dynamic/sequence_utils.py | 47 +++++++++++++ ml_gestures_dynamic/train.py | 78 +++++++++++++++++++++ utils/record_dynamic.py | 86 ++++++++++++++++++++++++ 7 files changed, 324 insertions(+), 1 deletion(-) create mode 100644 ml_gestures_dynamic/evaluate.py create mode 100644 ml_gestures_dynamic/feature_extractor.py create mode 100644 ml_gestures_dynamic/predict.py create mode 100644 ml_gestures_dynamic/sequence_utils.py create mode 100644 ml_gestures_dynamic/train.py create mode 100644 utils/record_dynamic.py diff --git a/README.md b/README.md index 1cb575b..ec7b3eb 100644 --- a/README.md +++ b/README.md @@ -16,12 +16,19 @@ gesture_robot/ │ ├── feature_extractor.py │ ├── predict.py │ └── train.py +├── ml_gestures_dynamic/ # ML для динамических жестов +│ ├── feature_extractor.py +│ ├── sequence_utils.py +│ ├── predict.py +│ ├── evaluate.py +│ └── train.py ├── robot/ # Робот (симулятор или заглушка) │ ├── map_simulator.py │ └── dummy.py ├── utils/ # Вспомогательные скрипты │ ├── annotate.py # Разметка изображений -│ └── capture_photo.py # Съёмка фото с камеры +│ ├── capture_photo.py # Съёмка фото с камеры +│ └── record_dynamic.py # Разметка видеопоследовательности └── requirements.txt ``` diff --git a/ml_gestures_dynamic/evaluate.py b/ml_gestures_dynamic/evaluate.py new file mode 100644 index 0000000..6f287e0 --- /dev/null +++ b/ml_gestures_dynamic/evaluate.py @@ -0,0 +1,59 @@ +import numpy as np +import joblib +import json +import argparse +import tensorflow as tf +from sklearn.metrics import classification_report, confusion_matrix, accuracy_score +from .sequence_utils import load_sequences_from_csv + +def evaluate(data_path, model_path, max_len=30, test_size=0.2): + # Загружаем данные + X_train, X_test, y_train, y_test, le = load_sequences_from_csv(data_path, max_len, test_size) + # Загружаем модель + model = tf.keras.models.load_model(model_path) + # Загружаем классы (или используем le из данных) + try: + classes_path = model_path.replace('.h5', '_classes.pkl') + with open(classes_path, 'rb') as f: + saved_classes = joblib.load(f) + # Проверяем соответствие классов + if list(saved_classes) != list(le.classes_): + print("Warning: classes in model and data differ. Using data classes.") + classes = le.classes_ + else: + classes = saved_classes + except: + classes = le.classes_ + + # Предсказание + y_pred = np.argmax(model.predict(X_test), axis=1) + acc = accuracy_score(y_test, y_pred) + report = classification_report(y_test, y_pred, target_names=classes, output_dict=True) + cm = confusion_matrix(y_test, y_pred).tolist() + + print(f"Test accuracy: {acc:.4f}") + print("\nClassification Report:") + for cls in classes: + print(f"{cls}: precision={report[cls]['precision']:.3f}, recall={report[cls]['recall']:.3f}, f1={report[cls]['f1-score']:.3f}") + print("\nConfusion Matrix:") + for row in cm: + print(row) + + # Сохраняем отчёт + report_path = model_path.replace('.h5', '_evaluation_report.json') + with open(report_path, 'w') as f: + json.dump({ + 'accuracy': acc, + 'classification_report': report, + 'confusion_matrix': cm + }, f, indent=2) + print(f"\nEvaluation report saved to {report_path}") + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description='Оценка LSTM для динамических жестов') + parser.add_argument('--data', required=True, help='Путь к CSV-файлу или папке с CSV-файлами') + parser.add_argument('--model', required=True, help='Путь к обученной модели (.h5)') + parser.add_argument('--max_len', type=int, default=30, help='Длина последовательности (должна совпадать с обучением)') + parser.add_argument('--test_size', type=float, default=0.2, help='Доля тестовой выборки') + args = parser.parse_args() + evaluate(args.data, args.model, args.max_len, args.test_size) diff --git a/ml_gestures_dynamic/feature_extractor.py b/ml_gestures_dynamic/feature_extractor.py new file mode 100644 index 0000000..8d2850d --- /dev/null +++ b/ml_gestures_dynamic/feature_extractor.py @@ -0,0 +1,15 @@ +import numpy as np +from ml_gestures.feature_extractor import normalize_landmarks + +def extract_sequence(landmarks_seq): + """ + landmarks_seq: список массивов (каждый (33,4) или (33,3)) + Возвращает np.array формы (seq_len, 99) + """ + seq = [] + for lm in landmarks_seq: + if lm is None: + seq.append(np.zeros(99)) # если пропущен кадр + else: + seq.append(normalize_landmarks(lm)) + return np.array(seq) diff --git a/ml_gestures_dynamic/predict.py b/ml_gestures_dynamic/predict.py new file mode 100644 index 0000000..a9c59ca --- /dev/null +++ b/ml_gestures_dynamic/predict.py @@ -0,0 +1,31 @@ +import numpy as np +import tensorflow as tf +import joblib +from collections import deque +from .feature_extractor import extract_sequence + +class DynamicGesturePredictor: + def __init__(self, model_path, classes_path, window_size=30, threshold=0.7): + self.model = tf.keras.models.load_model(model_path) + with open(classes_path, 'rb') as f: + self.classes = joblib.load(f) + self.window_size = window_size + self.buffer = deque(maxlen=window_size) + self.threshold = threshold + + def add_frame(self, landmarks): + self.buffer.append(landmarks) + + def predict(self): + if len(self.buffer) < self.window_size: + return None + seq = extract_sequence(list(self.buffer)) + seq = np.expand_dims(seq, axis=0) # (1, window, 99) + probs = self.model.predict(seq, verbose=0)[0] + idx = np.argmax(probs) + if probs[idx] > self.threshold and self.classes[idx] != 'none': + return self.classes[idx] + return None + + def reset(self): + self.buffer.clear() diff --git a/ml_gestures_dynamic/sequence_utils.py b/ml_gestures_dynamic/sequence_utils.py new file mode 100644 index 0000000..d239c28 --- /dev/null +++ b/ml_gestures_dynamic/sequence_utils.py @@ -0,0 +1,47 @@ +import pandas as pd +import numpy as np +import os +from sklearn.model_selection import train_test_split + +def load_sequences_from_csv(csv_path, max_len=30, test_size=0.2, random_state=42): + """ + Загружает последовательности из одного или нескольких CSV-файлов. + Возвращает X_train, X_test, y_train, y_test, le (LabelEncoder). + """ + if os.path.isdir(csv_path): + dfs = [] + for f in os.listdir(csv_path): + if f.endswith('.csv'): + dfs.append(pd.read_csv(os.path.join(csv_path, f))) + df = pd.concat(dfs, ignore_index=True) + else: + df = pd.read_csv(csv_path) + + sequences = {} + for seq_id, group in df.groupby('sequence_id'): + group = group.sort_values('frame') + label = group['label'].iloc[0] + features = group[[f'f{i}' for i in range(99)]].values + # Обрезаем или падинг + if len(features) > max_len: + features = features[:max_len] + elif len(features) < max_len: + pad = np.zeros((max_len - len(features), 99)) + features = np.vstack([features, pad]) + sequences[seq_id] = (label, features) + + labels = [] + X = [] + for label, feats in sequences.values(): + labels.append(label) + X.append(feats) + + X = np.array(X) + from sklearn.preprocessing import LabelEncoder + le = LabelEncoder() + y = le.fit_transform(labels) + + X_train, X_test, y_train, y_test = train_test_split( + X, y, test_size=test_size, stratify=y, random_state=random_state + ) + return X_train, X_test, y_train, y_test, le diff --git a/ml_gestures_dynamic/train.py b/ml_gestures_dynamic/train.py new file mode 100644 index 0000000..d60623d --- /dev/null +++ b/ml_gestures_dynamic/train.py @@ -0,0 +1,78 @@ +import numpy as np +import joblib +import json +import argparse +import tensorflow as tf +from tensorflow.keras import layers, models +from sklearn.metrics import classification_report, confusion_matrix, accuracy_score +from .sequence_utils import load_sequences_from_csv + +def train(data_path, model_path, max_len=30, lstm_units=64, epochs=50, batch_size=16, test_size=0.2): + # Загрузка данных + X_train, X_test, y_train, y_test, le = load_sequences_from_csv(data_path, max_len, test_size) + num_classes = len(le.classes_) + print(f"Classes: {le.classes_}") + print(f"Train samples: {len(X_train)}, Test samples: {len(X_test)}") + + # Построение модели + model = models.Sequential([ + layers.LSTM(lstm_units, input_shape=(max_len, 99), return_sequences=True), + layers.Dropout(0.3), + layers.LSTM(lstm_units), + layers.Dropout(0.3), + layers.Dense(num_classes, activation='softmax') + ]) + model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy']) + model.fit(X_train, y_train, epochs=epochs, batch_size=batch_size, validation_data=(X_test, y_test)) + + # Оценка + y_pred = np.argmax(model.predict(X_test), axis=1) + acc = accuracy_score(y_test, y_pred) + report = classification_report(y_test, y_pred, target_names=le.classes_, output_dict=True) + cm = confusion_matrix(y_test, y_pred).tolist() + + print(f"\nTest accuracy: {acc:.4f}") + print("\nClassification Report:") + for cls in le.classes_: + print(f"{cls}: precision={report[cls]['precision']:.3f}, recall={report[cls]['recall']:.3f}, f1={report[cls]['f1-score']:.3f}") + print("\nConfusion Matrix:") + print(cm) + + # Сохранение модели и метаданных + model.save(model_path) + classes_path = model_path.replace('.h5', '_classes.pkl') + joblib.dump(le.classes_, classes_path) + + report_data = { + 'model_type': 'LSTM', + 'max_len': max_len, + 'lstm_units': lstm_units, + 'epochs': epochs, + 'batch_size': batch_size, + 'test_size': test_size, + 'accuracy': acc, + 'classification_report': report, + 'confusion_matrix': cm, + 'classes': le.classes_.tolist(), + 'train_samples': len(X_train), + 'test_samples': len(X_test) + } + report_path = model_path.replace('.h5', '_report.json') + with open(report_path, 'w') as f: + json.dump(report_data, f, indent=2) + + print(f"\nModel saved to {model_path}") + print(f"Classes saved to {classes_path}") + print(f"Report saved to {report_path}") + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description='Обучение LSTM для динамических жестов') + parser.add_argument('--data', required=True, help='Путь к CSV-файлу или папке с CSV-файлами') + parser.add_argument('--model', required=True, help='Путь для сохранения модели (.h5)') + parser.add_argument('--max_len', type=int, default=30, help='Длина последовательности') + parser.add_argument('--lstm_units', type=int, default=64, help='Количество нейронов в LSTM') + parser.add_argument('--epochs', type=int, default=50, help='Количество эпох') + parser.add_argument('--batch_size', type=int, default=16, help='Размер батча') + parser.add_argument('--test_size', type=float, default=0.2, help='Доля тестовой выборки') + args = parser.parse_args() + train(args.data, args.model, args.max_len, args.lstm_units, args.epochs, args.batch_size, args.test_size) diff --git a/utils/record_dynamic.py b/utils/record_dynamic.py new file mode 100644 index 0000000..902ff5f --- /dev/null +++ b/utils/record_dynamic.py @@ -0,0 +1,86 @@ +import cv2 +import numpy as np +import csv +import sys +import time +from pathlib import Path +import argparse + +sys.path.append(str(Path(__file__).parent.parent)) +from skeleton.mediapipe_detector import MediaPipeDetector +from ml_gestures.dynamic.feature_extractor import extract_sequence + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument('--label', required=True, help='Название жеста (например, wave_left)') + parser.add_argument('--output', default='dynamic_data.csv', help='CSV файл для сохранения') + parser.add_argument('--camera', type=int, default=0) + parser.add_argument('--duration', type=float, default=3.0, help='Длительность записи (сек)') + args = parser.parse_args() + + detector = MediaPipeDetector() + cap = cv2.VideoCapture(args.camera) + if not cap.isOpened(): + print("Камера не найдена") + sys.exit(1) + + # Определяем следующий ID последовательности + try: + import pandas as pd + df = pd.read_csv(args.output) + next_id = df['sequence_id'].max() + 1 if not df.empty else 0 + except: + next_id = 0 + + print(f"Запись жеста: {args.label}. Нажмите SPACE для начала, q для выхода.") + recording = False + start_time = 0 + sequence = [] + + while True: + ret, frame = cap.read() + if not ret: + break + frame = cv2.flip(frame, 1) # зеркало для удобства + result = detector.detect(frame) + if result['success']: + landmarks = result['landmarks'] + vis = detector.draw_landmarks(frame, result['pose_landmarks']) + if recording: + sequence.append(landmarks) + elapsed = time.time() - start_time + if elapsed >= args.duration: + recording = False + # Сохраняем последовательность + seq_features = extract_sequence(sequence) + with open(args.output, 'a', newline='') as f: + writer = csv.writer(f) + for i, feat in enumerate(seq_features): + writer.writerow([args.label, next_id, i] + feat.tolist()) + print(f"Сохранено {len(sequence)} кадров для жеста {args.label}, ID={next_id}") + sequence = [] + next_id += 1 + else: + vis = frame + + if recording: + cv2.putText(vis, f"RECORDING... {elapsed:.1f}/{args.duration}", (10, 30), + cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,0,255), 2) + else: + cv2.putText(vis, f"Press SPACE to record '{args.label}'", (10, 30), + cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0,255,0), 2) + + cv2.imshow('Record dynamic gesture', vis) + key = cv2.waitKey(1) & 0xFF + if key == ord('q'): + break + if key == ord(' ') and not recording: + recording = True + start_time = time.time() + sequence = [] + + cap.release() + cv2.destroyAllWindows() + +if __name__ == '__main__': + main()