sirius/gesture_rec
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

started dymanic gect with lstm

Browse Source
This commit is contained in:
moscovskayaliza
2026-03-31 17:01:42 +03:00
parent f40e126b6d
commit 0d61d47cfa
7 changed files with 324 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+8 -1
View File
@@ -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
```
ml_gestures_dynamic/evaluate.py
+59
View File
@@ -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)
ml_gestures_dynamic/feature_extractor.py
+15
View File
@@ -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)
ml_gestures_dynamic/predict.py
+31
View File
@@ -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()
ml_gestures_dynamic/sequence_utils.py
+47
View File
@@ -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
ml_gestures_dynamic/train.py
+78
View File
@@ -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)
utils/record_dynamic.py
+86
View File
@@ -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()