Compare commits
No commits in common. "master" and "aioptimize" have entirely different histories.
master
...
aioptimize
File diff suppressed because it is too large
Load Diff
@ -1,558 +0,0 @@
|
|||||||
import pickle
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
import cv2
|
|
||||||
import mediapipe as mp
|
|
||||||
import numpy as np
|
|
||||||
from mediapipe.tasks.python import vision
|
|
||||||
from sklearn.ensemble import RandomForestClassifier
|
|
||||||
from sklearn.linear_model import LogisticRegression
|
|
||||||
from sklearn.neighbors import KNeighborsClassifier
|
|
||||||
from sklearn.preprocessing import LabelEncoder, StandardScaler
|
|
||||||
from sklearn.svm import SVC
|
|
||||||
|
|
||||||
|
|
||||||
WRIST = 0
|
|
||||||
|
|
||||||
THUMB_CMC = 1
|
|
||||||
THUMB_MCP = 2
|
|
||||||
THUMB_IP = 3
|
|
||||||
THUMB_TIP = 4
|
|
||||||
|
|
||||||
INDEX_MCP = 5
|
|
||||||
INDEX_PIP = 6
|
|
||||||
INDEX_DIP = 7
|
|
||||||
INDEX_TIP = 8
|
|
||||||
|
|
||||||
MIDDLE_MCP = 9
|
|
||||||
MIDDLE_PIP = 10
|
|
||||||
MIDDLE_DIP = 11
|
|
||||||
MIDDLE_TIP = 12
|
|
||||||
|
|
||||||
RING_MCP = 13
|
|
||||||
RING_PIP = 14
|
|
||||||
RING_DIP = 15
|
|
||||||
RING_TIP = 16
|
|
||||||
|
|
||||||
PINKY_MCP = 17
|
|
||||||
PINKY_PIP = 18
|
|
||||||
PINKY_DIP = 19
|
|
||||||
PINKY_TIP = 20
|
|
||||||
|
|
||||||
HAND_CONNECTIONS = [
|
|
||||||
(0, 1), (1, 2), (2, 3), (3, 4),
|
|
||||||
(0, 5), (5, 6), (6, 7), (7, 8),
|
|
||||||
(0, 9), (9, 10), (10, 11), (11, 12),
|
|
||||||
(0, 13), (13, 14), (14, 15), (15, 16),
|
|
||||||
(0, 17), (17, 18), (18, 19), (19, 20),
|
|
||||||
(5, 9), (9, 13), (13, 17),
|
|
||||||
]
|
|
||||||
|
|
||||||
DEFAULT_UNKNOWN_THRESHOLD = 0.7
|
|
||||||
DEFAULT_KNN_K = 3
|
|
||||||
DEFAULT_RF_TREES = 200
|
|
||||||
DEFAULT_NUM_HANDS = 2
|
|
||||||
|
|
||||||
|
|
||||||
def landmark_to_pixel(lm, width, height):
|
|
||||||
return int(lm.x * width), int(lm.y * height)
|
|
||||||
|
|
||||||
|
|
||||||
def vec2_from_landmarks(hand_landmarks):
|
|
||||||
return np.array([[lm.x, lm.y] for lm in hand_landmarks], dtype=np.float32)
|
|
||||||
|
|
||||||
|
|
||||||
def vec3_from_world_landmarks(world_landmarks):
|
|
||||||
return np.array([[lm.x, lm.y, lm.z] for lm in world_landmarks], dtype=np.float32)
|
|
||||||
|
|
||||||
|
|
||||||
def hand_scale_2d(points2d):
|
|
||||||
s = np.linalg.norm(points2d[MIDDLE_MCP] - points2d[WRIST])
|
|
||||||
return max(float(s), 1e-6)
|
|
||||||
|
|
||||||
|
|
||||||
def hand_scale_3d(points3d):
|
|
||||||
s = np.linalg.norm(points3d[MIDDLE_MCP] - points3d[WRIST])
|
|
||||||
return max(float(s), 1e-6)
|
|
||||||
|
|
||||||
|
|
||||||
def normalize_2d_features(hand_landmarks):
|
|
||||||
pts = vec2_from_landmarks(hand_landmarks)
|
|
||||||
origin = pts[WRIST].copy()
|
|
||||||
pts = pts - origin
|
|
||||||
scale = hand_scale_2d(pts + origin)
|
|
||||||
pts = pts / scale
|
|
||||||
return pts.flatten().astype(np.float32)
|
|
||||||
|
|
||||||
|
|
||||||
def normalize_3d_features(world_landmarks):
|
|
||||||
pts = vec3_from_world_landmarks(world_landmarks)
|
|
||||||
origin = pts[WRIST].copy()
|
|
||||||
pts = pts - origin
|
|
||||||
scale = hand_scale_3d(pts + origin)
|
|
||||||
pts = pts / scale
|
|
||||||
return pts.flatten().astype(np.float32)
|
|
||||||
|
|
||||||
|
|
||||||
def angle_between(v1, v2):
|
|
||||||
n1 = np.linalg.norm(v1)
|
|
||||||
n2 = np.linalg.norm(v2)
|
|
||||||
if n1 < 1e-6 or n2 < 1e-6:
|
|
||||||
return 0.0
|
|
||||||
c = np.dot(v1, v2) / (n1 * n2)
|
|
||||||
c = np.clip(c, -1.0, 1.0)
|
|
||||||
return float(np.arccos(c))
|
|
||||||
|
|
||||||
|
|
||||||
def joint_angle(points, a, b, c):
|
|
||||||
ba = points[a] - points[b]
|
|
||||||
bc = points[c] - points[b]
|
|
||||||
return angle_between(ba, bc)
|
|
||||||
|
|
||||||
|
|
||||||
def extract_angle_features(world_landmarks):
|
|
||||||
pts = vec3_from_world_landmarks(world_landmarks)
|
|
||||||
|
|
||||||
thumb_angle_1 = joint_angle(pts, THUMB_CMC, THUMB_MCP, THUMB_IP)
|
|
||||||
thumb_angle_2 = joint_angle(pts, THUMB_MCP, THUMB_IP, THUMB_TIP)
|
|
||||||
|
|
||||||
index_angle_1 = joint_angle(pts, INDEX_MCP, INDEX_PIP, INDEX_DIP)
|
|
||||||
index_angle_2 = joint_angle(pts, INDEX_PIP, INDEX_DIP, INDEX_TIP)
|
|
||||||
|
|
||||||
middle_angle_1 = joint_angle(pts, MIDDLE_MCP, MIDDLE_PIP, MIDDLE_DIP)
|
|
||||||
middle_angle_2 = joint_angle(pts, MIDDLE_PIP, MIDDLE_DIP, MIDDLE_TIP)
|
|
||||||
|
|
||||||
ring_angle_1 = joint_angle(pts, RING_MCP, RING_PIP, RING_DIP)
|
|
||||||
ring_angle_2 = joint_angle(pts, RING_PIP, RING_DIP, RING_TIP)
|
|
||||||
|
|
||||||
pinky_angle_1 = joint_angle(pts, PINKY_MCP, PINKY_PIP, PINKY_DIP)
|
|
||||||
pinky_angle_2 = joint_angle(pts, PINKY_PIP, PINKY_DIP, PINKY_TIP)
|
|
||||||
|
|
||||||
wrist_to_thumb = pts[THUMB_TIP] - pts[WRIST]
|
|
||||||
wrist_to_index = pts[INDEX_TIP] - pts[WRIST]
|
|
||||||
wrist_to_middle = pts[MIDDLE_TIP] - pts[WRIST]
|
|
||||||
wrist_to_ring = pts[RING_TIP] - pts[WRIST]
|
|
||||||
wrist_to_pinky = pts[PINKY_TIP] - pts[WRIST]
|
|
||||||
|
|
||||||
spread_thumb_index = angle_between(wrist_to_thumb, wrist_to_index)
|
|
||||||
spread_index_middle = angle_between(wrist_to_index, wrist_to_middle)
|
|
||||||
spread_middle_ring = angle_between(wrist_to_middle, wrist_to_ring)
|
|
||||||
spread_ring_pinky = angle_between(wrist_to_ring, wrist_to_pinky)
|
|
||||||
|
|
||||||
feats = np.array([
|
|
||||||
thumb_angle_1, thumb_angle_2,
|
|
||||||
index_angle_1, index_angle_2,
|
|
||||||
middle_angle_1, middle_angle_2,
|
|
||||||
ring_angle_1, ring_angle_2,
|
|
||||||
pinky_angle_1, pinky_angle_2,
|
|
||||||
spread_thumb_index,
|
|
||||||
spread_index_middle,
|
|
||||||
spread_middle_ring,
|
|
||||||
spread_ring_pinky,
|
|
||||||
], dtype=np.float32)
|
|
||||||
|
|
||||||
return feats
|
|
||||||
|
|
||||||
|
|
||||||
def extract_feature_family(hand_landmarks, world_landmarks, family_name):
|
|
||||||
family_name = family_name.upper()
|
|
||||||
|
|
||||||
if family_name == "2D":
|
|
||||||
return normalize_2d_features(hand_landmarks)
|
|
||||||
if family_name == "3D":
|
|
||||||
return normalize_3d_features(world_landmarks)
|
|
||||||
if family_name == "ANGLES":
|
|
||||||
return extract_angle_features(world_landmarks)
|
|
||||||
|
|
||||||
raise ValueError(f"Unknown feature family: {family_name}")
|
|
||||||
|
|
||||||
|
|
||||||
def load_pickle(path):
|
|
||||||
with open(path, "rb") as f:
|
|
||||||
return pickle.load(f)
|
|
||||||
|
|
||||||
|
|
||||||
def family_key_from_name(name):
|
|
||||||
name = name.upper()
|
|
||||||
if name == "2D":
|
|
||||||
return "features_2d"
|
|
||||||
if name == "3D":
|
|
||||||
return "features_3d"
|
|
||||||
if name == "ANGLES":
|
|
||||||
return "features_angles"
|
|
||||||
raise ValueError(f"Unknown family name: {name}")
|
|
||||||
|
|
||||||
|
|
||||||
def ranking_key_from_method(method):
|
|
||||||
method = method.lower()
|
|
||||||
if method == "f":
|
|
||||||
return "rank_f"
|
|
||||||
if method == "mi":
|
|
||||||
return "rank_mi"
|
|
||||||
if method == "fisher":
|
|
||||||
return "rank_fisher"
|
|
||||||
raise ValueError(f"Unknown ranking method: {method}")
|
|
||||||
|
|
||||||
|
|
||||||
def flatten_training_data(data, family_key):
|
|
||||||
X = []
|
|
||||||
y = []
|
|
||||||
|
|
||||||
for label in sorted(data.keys()):
|
|
||||||
for sample in data[label]:
|
|
||||||
X.append(np.asarray(sample[family_key], dtype=np.float32))
|
|
||||||
y.append(label)
|
|
||||||
|
|
||||||
X = np.vstack([x.reshape(1, -1) for x in X])
|
|
||||||
y = np.asarray(y)
|
|
||||||
return X, y
|
|
||||||
|
|
||||||
|
|
||||||
def build_classifier(classifier_type):
|
|
||||||
classifier_type = classifier_type.lower()
|
|
||||||
|
|
||||||
if classifier_type == "knn":
|
|
||||||
return KNeighborsClassifier(n_neighbors=DEFAULT_KNN_K)
|
|
||||||
|
|
||||||
if classifier_type == "svm":
|
|
||||||
return SVC(kernel="rbf", probability=True)
|
|
||||||
|
|
||||||
if classifier_type == "rf":
|
|
||||||
return RandomForestClassifier(
|
|
||||||
n_estimators=DEFAULT_RF_TREES,
|
|
||||||
random_state=42,
|
|
||||||
class_weight=None,
|
|
||||||
)
|
|
||||||
|
|
||||||
if classifier_type == "logreg":
|
|
||||||
return LogisticRegression(
|
|
||||||
max_iter=2000,
|
|
||||||
solver="lbfgs",
|
|
||||||
)
|
|
||||||
|
|
||||||
raise ValueError(f"Unknown classifier type: {classifier_type}")
|
|
||||||
|
|
||||||
|
|
||||||
def get_selected_feature_indices(analysis, family_key, ranking_method, num_features):
|
|
||||||
rank_key = ranking_key_from_method(ranking_method)
|
|
||||||
rank = analysis[family_key][rank_key]
|
|
||||||
return np.asarray(rank[:num_features], dtype=np.int32)
|
|
||||||
|
|
||||||
|
|
||||||
def train_model(data, analysis, feature_family, classifier_type, ranking_method, num_features):
|
|
||||||
family_key = family_key_from_name(feature_family)
|
|
||||||
|
|
||||||
X, y = flatten_training_data(data, family_key)
|
|
||||||
selected_idx = get_selected_feature_indices(analysis, family_key, ranking_method, num_features)
|
|
||||||
|
|
||||||
X_sel = X[:, selected_idx]
|
|
||||||
|
|
||||||
scaler = StandardScaler()
|
|
||||||
X_scaled = scaler.fit_transform(X_sel)
|
|
||||||
|
|
||||||
label_encoder = LabelEncoder()
|
|
||||||
y_enc = label_encoder.fit_transform(y)
|
|
||||||
|
|
||||||
clf = build_classifier(classifier_type)
|
|
||||||
clf.fit(X_scaled, y_enc)
|
|
||||||
|
|
||||||
return {
|
|
||||||
"classifier": clf,
|
|
||||||
"scaler": scaler,
|
|
||||||
"label_encoder": label_encoder,
|
|
||||||
"selected_idx": selected_idx,
|
|
||||||
"family_key": family_key,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def predict_with_confidence(model_bundle, feature_vector, unknown_threshold=DEFAULT_UNKNOWN_THRESHOLD):
|
|
||||||
selected_idx = model_bundle["selected_idx"]
|
|
||||||
clf = model_bundle["classifier"]
|
|
||||||
scaler = model_bundle["scaler"]
|
|
||||||
label_encoder = model_bundle["label_encoder"]
|
|
||||||
|
|
||||||
x = feature_vector[selected_idx].reshape(1, -1)
|
|
||||||
x = scaler.transform(x)
|
|
||||||
|
|
||||||
pred_id = clf.predict(x)[0]
|
|
||||||
pred_label = label_encoder.inverse_transform([pred_id])[0]
|
|
||||||
|
|
||||||
if hasattr(clf, "predict_proba"):
|
|
||||||
probs = clf.predict_proba(x)[0]
|
|
||||||
conf = float(np.max(probs))
|
|
||||||
else:
|
|
||||||
conf = 1.0
|
|
||||||
|
|
||||||
if conf < unknown_threshold:
|
|
||||||
return "UNKNOWN", conf
|
|
||||||
|
|
||||||
return pred_label, conf
|
|
||||||
|
|
||||||
|
|
||||||
def create_hand_landmarker(
|
|
||||||
model_path="hand_landmarker.task",
|
|
||||||
num_hands=DEFAULT_NUM_HANDS,
|
|
||||||
min_hand_detection_confidence=0.5,
|
|
||||||
min_hand_presence_confidence=0.5,
|
|
||||||
min_tracking_confidence=0.5,
|
|
||||||
):
|
|
||||||
BaseOptions = mp.tasks.BaseOptions
|
|
||||||
HandLandmarkerOptions = vision.HandLandmarkerOptions
|
|
||||||
VisionRunningMode = vision.RunningMode
|
|
||||||
|
|
||||||
options = HandLandmarkerOptions(
|
|
||||||
base_options=BaseOptions(model_asset_path=model_path),
|
|
||||||
running_mode=VisionRunningMode.IMAGE,
|
|
||||||
num_hands=num_hands,
|
|
||||||
min_hand_detection_confidence=min_hand_detection_confidence,
|
|
||||||
min_hand_presence_confidence=min_hand_presence_confidence,
|
|
||||||
min_tracking_confidence=min_tracking_confidence,
|
|
||||||
)
|
|
||||||
|
|
||||||
return vision.HandLandmarker.create_from_options(options)
|
|
||||||
|
|
||||||
|
|
||||||
def load_gesture_model_bundle(
|
|
||||||
data_pickle="gesture_samples.pkl",
|
|
||||||
analysis_pickle="gesture_feature_analysis.pkl",
|
|
||||||
feature_family="ANGLES",
|
|
||||||
classifier_type="knn",
|
|
||||||
ranking_method="fisher",
|
|
||||||
num_features=10,
|
|
||||||
unknown_threshold=DEFAULT_UNKNOWN_THRESHOLD,
|
|
||||||
hand_model_path="hand_landmarker.task",
|
|
||||||
):
|
|
||||||
if not Path(data_pickle).exists():
|
|
||||||
raise FileNotFoundError(f"Missing {data_pickle}")
|
|
||||||
if not Path(analysis_pickle).exists():
|
|
||||||
raise FileNotFoundError(f"Missing {analysis_pickle}")
|
|
||||||
|
|
||||||
data = load_pickle(data_pickle)
|
|
||||||
analysis = load_pickle(analysis_pickle)
|
|
||||||
|
|
||||||
model_bundle = train_model(
|
|
||||||
data=data,
|
|
||||||
analysis=analysis,
|
|
||||||
feature_family=feature_family,
|
|
||||||
classifier_type=classifier_type,
|
|
||||||
ranking_method=ranking_method,
|
|
||||||
num_features=num_features,
|
|
||||||
)
|
|
||||||
|
|
||||||
hand_landmarker = create_hand_landmarker(model_path=hand_model_path)
|
|
||||||
|
|
||||||
return {
|
|
||||||
"model_bundle": model_bundle,
|
|
||||||
"hand_landmarker": hand_landmarker,
|
|
||||||
"feature_family": feature_family,
|
|
||||||
"classifier_type": classifier_type,
|
|
||||||
"ranking_method": ranking_method,
|
|
||||||
"num_features": num_features,
|
|
||||||
"unknown_threshold": unknown_threshold,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def _put_text_with_background(frame, text, origin, font_scale=0.6, text_color=(255, 255, 255), bg_color=(0, 0, 0)):
|
|
||||||
x, y = origin
|
|
||||||
font = cv2.FONT_HERSHEY_SIMPLEX
|
|
||||||
thickness = 2
|
|
||||||
(text_w, text_h), baseline = cv2.getTextSize(text, font, font_scale, thickness)
|
|
||||||
top_left = (x, max(0, y - text_h - baseline - 6))
|
|
||||||
bottom_right = (x + text_w + 6, max(0, y + 4))
|
|
||||||
cv2.rectangle(frame, top_left, bottom_right, bg_color, -1)
|
|
||||||
cv2.putText(
|
|
||||||
frame,
|
|
||||||
text,
|
|
||||||
(x + 3, y - 3),
|
|
||||||
font,
|
|
||||||
font_scale,
|
|
||||||
text_color,
|
|
||||||
thickness,
|
|
||||||
cv2.LINE_AA,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _ray_to_frame_bounds(start_point, direction_vector, frame_w, frame_h):
|
|
||||||
x0, y0 = start_point
|
|
||||||
dx, dy = direction_vector
|
|
||||||
|
|
||||||
norm = float(np.hypot(dx, dy))
|
|
||||||
if norm < 1e-6:
|
|
||||||
return None
|
|
||||||
|
|
||||||
dx /= norm
|
|
||||||
dy /= norm
|
|
||||||
|
|
||||||
candidates = []
|
|
||||||
|
|
||||||
if abs(dx) > 1e-6:
|
|
||||||
for x_bound in (0, frame_w - 1):
|
|
||||||
t = (x_bound - x0) / dx
|
|
||||||
if t <= 0:
|
|
||||||
continue
|
|
||||||
y = y0 + t * dy
|
|
||||||
if 0 <= y <= frame_h - 1:
|
|
||||||
candidates.append((t, (int(x_bound), int(round(y)))))
|
|
||||||
|
|
||||||
if abs(dy) > 1e-6:
|
|
||||||
for y_bound in (0, frame_h - 1):
|
|
||||||
t = (y_bound - y0) / dy
|
|
||||||
if t <= 0:
|
|
||||||
continue
|
|
||||||
x = x0 + t * dx
|
|
||||||
if 0 <= x <= frame_w - 1:
|
|
||||||
candidates.append((t, (int(round(x)), int(y_bound))))
|
|
||||||
|
|
||||||
if not candidates:
|
|
||||||
return None
|
|
||||||
|
|
||||||
candidates.sort(key=lambda item: item[0])
|
|
||||||
return candidates[0][1]
|
|
||||||
|
|
||||||
|
|
||||||
def draw_pointing_ray(frame, hand_landmarks, crop_x, crop_y, crop_w, crop_h, color=(255, 255, 0), thickness=3):
|
|
||||||
index_tip_x, index_tip_y = landmark_to_pixel(hand_landmarks[INDEX_TIP], crop_w, crop_h)
|
|
||||||
index_mcp_x, index_mcp_y = landmark_to_pixel(hand_landmarks[INDEX_MCP], crop_w, crop_h)
|
|
||||||
|
|
||||||
start_point = (index_tip_x + crop_x, index_tip_y + crop_y)
|
|
||||||
direction = (
|
|
||||||
(index_tip_x - index_mcp_x),
|
|
||||||
(index_tip_y - index_mcp_y),
|
|
||||||
)
|
|
||||||
|
|
||||||
end_point = _ray_to_frame_bounds(start_point, direction, frame.shape[1], frame.shape[0])
|
|
||||||
if end_point is None:
|
|
||||||
return
|
|
||||||
|
|
||||||
cv2.line(frame, start_point, end_point, color, thickness)
|
|
||||||
|
|
||||||
|
|
||||||
def draw_hand_on_frame(frame, hand_landmarks, crop_x, crop_y, crop_w, crop_h, label_text=""):
|
|
||||||
points = []
|
|
||||||
|
|
||||||
for a, b in HAND_CONNECTIONS:
|
|
||||||
x1, y1 = landmark_to_pixel(hand_landmarks[a], crop_w, crop_h)
|
|
||||||
x2, y2 = landmark_to_pixel(hand_landmarks[b], crop_w, crop_h)
|
|
||||||
x1 += crop_x
|
|
||||||
y1 += crop_y
|
|
||||||
x2 += crop_x
|
|
||||||
y2 += crop_y
|
|
||||||
cv2.line(frame, (x1, y1), (x2, y2), (0, 255, 0), 2)
|
|
||||||
|
|
||||||
for i, lm in enumerate(hand_landmarks):
|
|
||||||
x, y = landmark_to_pixel(lm, crop_w, crop_h)
|
|
||||||
x += crop_x
|
|
||||||
y += crop_y
|
|
||||||
points.append((x, y))
|
|
||||||
color = (0, 0, 255) if i in [THUMB_TIP, INDEX_TIP, MIDDLE_TIP, RING_TIP, PINKY_TIP] else (255, 255, 0)
|
|
||||||
radius = 5 if i in [THUMB_TIP, INDEX_TIP, MIDDLE_TIP, RING_TIP, PINKY_TIP] else 3
|
|
||||||
cv2.circle(frame, (x, y), radius, color, -1)
|
|
||||||
|
|
||||||
if label_text and points:
|
|
||||||
label_x = max(0, min(x for x, _ in points))
|
|
||||||
label_y = max(20, min(y for _, y in points) - 10)
|
|
||||||
_put_text_with_background(frame, label_text, (label_x, label_y), font_scale=0.55)
|
|
||||||
|
|
||||||
|
|
||||||
class GestureRecognizer:
|
|
||||||
def __init__(
|
|
||||||
self,
|
|
||||||
data_pickle="gesture_samples.pkl",
|
|
||||||
analysis_pickle="gesture_feature_analysis.pkl",
|
|
||||||
feature_family="ANGLES",
|
|
||||||
classifier_type="knn",
|
|
||||||
ranking_method="fisher",
|
|
||||||
num_features=10,
|
|
||||||
unknown_threshold=DEFAULT_UNKNOWN_THRESHOLD,
|
|
||||||
hand_model_path="hand_landmarker.task",
|
|
||||||
):
|
|
||||||
bundle = load_gesture_model_bundle(
|
|
||||||
data_pickle=data_pickle,
|
|
||||||
analysis_pickle=analysis_pickle,
|
|
||||||
feature_family=feature_family,
|
|
||||||
classifier_type=classifier_type,
|
|
||||||
ranking_method=ranking_method,
|
|
||||||
num_features=num_features,
|
|
||||||
unknown_threshold=unknown_threshold,
|
|
||||||
hand_model_path=hand_model_path,
|
|
||||||
)
|
|
||||||
self.model_bundle = bundle["model_bundle"]
|
|
||||||
self.hand_landmarker = bundle["hand_landmarker"]
|
|
||||||
self.feature_family = bundle["feature_family"]
|
|
||||||
self.classifier_type = bundle["classifier_type"]
|
|
||||||
self.ranking_method = bundle["ranking_method"]
|
|
||||||
self.num_features = bundle["num_features"]
|
|
||||||
self.unknown_threshold = bundle["unknown_threshold"]
|
|
||||||
|
|
||||||
def detect_gestures(self, crop_bgr):
|
|
||||||
crop_rgb = cv2.cvtColor(crop_bgr, cv2.COLOR_BGR2RGB)
|
|
||||||
mp_image = mp.Image(image_format=mp.ImageFormat.SRGB, data=crop_rgb)
|
|
||||||
result = self.hand_landmarker.detect(mp_image)
|
|
||||||
|
|
||||||
detections = []
|
|
||||||
if not result.hand_landmarks or not result.hand_world_landmarks:
|
|
||||||
return detections
|
|
||||||
|
|
||||||
hand_count = min(len(result.hand_landmarks), len(result.hand_world_landmarks))
|
|
||||||
for hand_index in range(hand_count):
|
|
||||||
hand_landmarks = result.hand_landmarks[hand_index]
|
|
||||||
world_landmarks = result.hand_world_landmarks[hand_index]
|
|
||||||
|
|
||||||
feature_vector = extract_feature_family(
|
|
||||||
hand_landmarks=hand_landmarks,
|
|
||||||
world_landmarks=world_landmarks,
|
|
||||||
family_name=self.feature_family,
|
|
||||||
)
|
|
||||||
pred_label, conf = predict_with_confidence(
|
|
||||||
self.model_bundle,
|
|
||||||
feature_vector,
|
|
||||||
unknown_threshold=self.unknown_threshold,
|
|
||||||
)
|
|
||||||
|
|
||||||
handedness_text = ""
|
|
||||||
if result.handedness and len(result.handedness) > hand_index and len(result.handedness[hand_index]) > 0:
|
|
||||||
handedness_text = result.handedness[hand_index][0].category_name
|
|
||||||
|
|
||||||
detections.append(
|
|
||||||
{
|
|
||||||
"hand_landmarks": hand_landmarks,
|
|
||||||
"label": pred_label,
|
|
||||||
"confidence": conf,
|
|
||||||
"handedness": handedness_text,
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
return detections
|
|
||||||
|
|
||||||
def draw_gesture(self, frame, detection, crop_x, crop_y, crop_w, crop_h):
|
|
||||||
label = detection["label"]
|
|
||||||
confidence = detection["confidence"]
|
|
||||||
handedness_text = detection["handedness"]
|
|
||||||
|
|
||||||
if handedness_text:
|
|
||||||
label_text = f"{handedness_text}: {label} {confidence:.2f}"
|
|
||||||
else:
|
|
||||||
label_text = f"{label} {confidence:.2f}"
|
|
||||||
|
|
||||||
draw_hand_on_frame(
|
|
||||||
frame,
|
|
||||||
detection["hand_landmarks"],
|
|
||||||
crop_x,
|
|
||||||
crop_y,
|
|
||||||
crop_w,
|
|
||||||
crop_h,
|
|
||||||
label_text=label_text,
|
|
||||||
)
|
|
||||||
|
|
||||||
if label.upper() == "A":
|
|
||||||
draw_pointing_ray(
|
|
||||||
frame,
|
|
||||||
detection["hand_landmarks"],
|
|
||||||
crop_x,
|
|
||||||
crop_y,
|
|
||||||
crop_w,
|
|
||||||
crop_h,
|
|
||||||
)
|
|
||||||
|
|
||||||
def close(self):
|
|
||||||
self.hand_landmarker.close()
|
|
||||||
@ -1,60 +0,0 @@
|
|||||||
#!/usr/bin/env python3
|
|
||||||
"""
|
|
||||||
06_osc_receiver.py
|
|
||||||
|
|
||||||
Very simple OSC receiver demo.
|
|
||||||
|
|
||||||
It shows that different OSC addresses are mapped
|
|
||||||
to different handler functions.
|
|
||||||
"""
|
|
||||||
|
|
||||||
from pathlib import Path
|
|
||||||
|
|
||||||
from pythonosc.dispatcher import Dispatcher
|
|
||||||
from pythonosc.osc_server import BlockingOSCUDPServer
|
|
||||||
from piper_module import Narrator, Language, Quality, LLM
|
|
||||||
|
|
||||||
|
|
||||||
IP = "100.83.246.218"
|
|
||||||
PORT = 9001
|
|
||||||
|
|
||||||
def on_read(address, *args):
|
|
||||||
print(f"on_read() address={address} args={args}")
|
|
||||||
narrator.read(str(args[0]))
|
|
||||||
|
|
||||||
def on_write_story(address, *args):
|
|
||||||
print(f"on_write_story() address={address} args={args}")
|
|
||||||
story = narrator.write_story(str(args[0]), str(args[1]))
|
|
||||||
print(story)
|
|
||||||
|
|
||||||
def on_read_story(address, *args):
|
|
||||||
print(f"on_read_story() address={address} args={args}")
|
|
||||||
story = narrator.write_story(str(args[0]), str(args[1]))
|
|
||||||
narrator.read(story)
|
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
|
||||||
dispatcher = Dispatcher()
|
|
||||||
|
|
||||||
dispatcher.map("/read", on_read)
|
|
||||||
dispatcher.map("/write_story", on_write_story)
|
|
||||||
dispatcher.map("/read_story", on_read_story)
|
|
||||||
|
|
||||||
server = BlockingOSCUDPServer((IP, PORT), dispatcher)
|
|
||||||
|
|
||||||
global narrator
|
|
||||||
llm = LLM("qwen2.5-7b-instruct-1m","http://100.83.153.234:1234/api/v1/chat" )
|
|
||||||
narrator = Narrator(Language.de_DE, "thorsten", Path("C:\Interaktion\WerWolf\piper-voices"), llm)
|
|
||||||
|
|
||||||
print(f"Listening for OSC on {IP}:{PORT}")
|
|
||||||
print("Try sending:")
|
|
||||||
print(" /read")
|
|
||||||
print(" /write_story")
|
|
||||||
print(" /read_story")
|
|
||||||
|
|
||||||
|
|
||||||
server.serve_forever()
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
main()
|
|
||||||
Binary file not shown.
@ -1,126 +0,0 @@
|
|||||||
import wave
|
|
||||||
from piper import PiperVoice
|
|
||||||
import pygame
|
|
||||||
import time
|
|
||||||
import os
|
|
||||||
import sys
|
|
||||||
import pyaudio
|
|
||||||
import requests
|
|
||||||
from pathlib import Path
|
|
||||||
from enum import Enum
|
|
||||||
|
|
||||||
class Language(Enum):
|
|
||||||
en_US = "en_US"
|
|
||||||
de_DE = "de_DE"
|
|
||||||
en_GB = "en_GB"
|
|
||||||
|
|
||||||
|
|
||||||
class Quality(Enum):
|
|
||||||
high = 3
|
|
||||||
medium = 2
|
|
||||||
low = 1
|
|
||||||
x_low = 0
|
|
||||||
|
|
||||||
class LLM:
|
|
||||||
def __init__(self, model: str, url: str):
|
|
||||||
self.url = url
|
|
||||||
self.model = model
|
|
||||||
|
|
||||||
|
|
||||||
class Narrator:
|
|
||||||
def __init__(self, language: Language, name: str, repo: Path, llm: LLM):
|
|
||||||
|
|
||||||
for quality in Quality:
|
|
||||||
filename = f"{language.name}-{name}-{quality.name}.onnx"
|
|
||||||
self.path = repo / language.name / name / quality.name / filename
|
|
||||||
|
|
||||||
print("Teste:", self.path)
|
|
||||||
|
|
||||||
if self.path.exists():
|
|
||||||
print("Gefunden:", self.path)
|
|
||||||
break
|
|
||||||
else:
|
|
||||||
raise FileNotFoundError("Keine passende Voice-Datei gefunden")
|
|
||||||
|
|
||||||
print("load voice")
|
|
||||||
|
|
||||||
self.voice = PiperVoice.load(self.path)
|
|
||||||
|
|
||||||
# for narrator texts -> connect to LLM
|
|
||||||
print(" create llm")
|
|
||||||
self.llm = llm
|
|
||||||
|
|
||||||
|
|
||||||
# Stream audio chunks and play them immediately
|
|
||||||
def read(self, text: str):
|
|
||||||
p = pyaudio.PyAudio()
|
|
||||||
|
|
||||||
stream = None
|
|
||||||
|
|
||||||
for audio_chunk in self.voice.synthesize(text):
|
|
||||||
|
|
||||||
# Stream erst öffnen, wenn erster Chunk da ist
|
|
||||||
if stream is None:
|
|
||||||
stream = p.open(
|
|
||||||
format=p.get_format_from_width(audio_chunk.sample_width),
|
|
||||||
channels=audio_chunk.sample_channels,
|
|
||||||
rate=audio_chunk.sample_rate,
|
|
||||||
output=True
|
|
||||||
)
|
|
||||||
|
|
||||||
# Audio sofort abspielen
|
|
||||||
stream.write(audio_chunk.audio_int16_bytes)
|
|
||||||
|
|
||||||
if stream:
|
|
||||||
stream.stop_stream()
|
|
||||||
stream.close()
|
|
||||||
|
|
||||||
p.terminate()
|
|
||||||
|
|
||||||
|
|
||||||
# play selected audiofile and delete it after playing
|
|
||||||
def read_audio(self, number: int):
|
|
||||||
#setup
|
|
||||||
filename = f"output{number}.wav"
|
|
||||||
pygame.mixer.init()
|
|
||||||
|
|
||||||
#play next audio
|
|
||||||
sound = pygame.mixer.Sound(filename)
|
|
||||||
channel = sound.play()
|
|
||||||
|
|
||||||
# wait until the sound has finished playing
|
|
||||||
while channel.get_busy():
|
|
||||||
time.sleep(0.1)
|
|
||||||
|
|
||||||
# delete now obsolete file
|
|
||||||
os.remove(filename)
|
|
||||||
|
|
||||||
pygame.mixer.quit()
|
|
||||||
|
|
||||||
|
|
||||||
# create wavefile from text
|
|
||||||
def create_audio(self, texts: list[str]):
|
|
||||||
|
|
||||||
for i, text in enumerate(texts):
|
|
||||||
filename = f"output{i}.wav"
|
|
||||||
|
|
||||||
with wave.open(f"output{i}.wav", "wb") as wav_file:
|
|
||||||
self.voice.synthesize_wav(text, wav_file)
|
|
||||||
|
|
||||||
# create a story from a base text
|
|
||||||
def write_story(self, text, count):
|
|
||||||
|
|
||||||
session = requests.Session()
|
|
||||||
session.trust_env = False
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
"model" : self.llm.model,
|
|
||||||
"input": "Formuliere:" + text +" zu einem erzählerischen Sprachtext um. Nicht mehr als " + count + " Sätze. Handlungsaufforderungen sollen klar heraushörbar sein."
|
|
||||||
}
|
|
||||||
|
|
||||||
response = requests.post(self.llm.url, json=payload, timeout=30)
|
|
||||||
response.raise_for_status()
|
|
||||||
data = response.json()
|
|
||||||
story = data["output"][0]["content"]
|
|
||||||
|
|
||||||
return story
|
|
||||||
@ -11,8 +11,4 @@
|
|||||||
<PackageReference Include="CoreOSC" Version="1.0.0" />
|
<PackageReference Include="CoreOSC" Version="1.0.0" />
|
||||||
</ItemGroup>
|
</ItemGroup>
|
||||||
|
|
||||||
<ItemGroup>
|
|
||||||
<Folder Include="Sprachausgabe\" />
|
|
||||||
</ItemGroup>
|
|
||||||
|
|
||||||
</Project>
|
</Project>
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user