| """ |
| example_keypoint_adder.py |
| Manages example videos and their precomputed keypoint JSON files and rendered keypoint videos. |
| Ensures that each example entry has: |
| 1. video_path: Path to the input ASL video. |
| 2. keypoints_path: Path to the extracted keypoints JSON. |
| 3. keypoints_video_path: Path to the video with keypoints rendered on top. |
| |
| Can be run as a standalone script to batch-process all examples or add new ones. |
| """ |
|
|
| import os |
| import sys |
| import json |
| import argparse |
| import numpy as np |
|
|
| |
| if hasattr(sys.stdout, 'reconfigure'): |
| try: |
| sys.stdout.reconfigure(encoding='utf-8', errors='replace') |
| except Exception: |
| pass |
|
|
| from predict_pose import create_mediapipe_models, predict_pose, load_video_cv |
| from visualize_pose import render_keypoints_video |
|
|
|
|
| def to_json_serializable(obj): |
| """Recursively converts NumPy arrays, numbers, and tensors to JSON-serializable Python types.""" |
| if isinstance(obj, np.ndarray): |
| return obj.tolist() |
| if isinstance(obj, (np.float32, np.float64, np.float16)): |
| return float(obj) |
| if isinstance(obj, (np.int32, np.int64, np.int16, np.int8, np.uint8)): |
| return int(obj) |
| if isinstance(obj, dict): |
| return {str(k): to_json_serializable(v) for k, v in obj.items()} |
| if isinstance(obj, (list, tuple)): |
| return [to_json_serializable(v) for v in obj] |
| return obj |
|
|
|
|
| def load_examples_json(json_path="examples.json"): |
| """Loads examples configuration from a JSON file.""" |
| if not os.path.exists(json_path): |
| return [] |
| try: |
| with open(json_path, "r", encoding="utf-8") as f: |
| data = json.load(f) |
| if isinstance(data, list): |
| return data |
| elif isinstance(data, dict) and "examples" in data: |
| return data["examples"] |
| return [] |
| except Exception as e: |
| print(f"Error loading {json_path}: {e}") |
| return [] |
|
|
|
|
| def save_examples_json(examples, json_path="examples.json"): |
| """Saves examples configuration to a JSON file.""" |
| try: |
| with open(json_path, "w", encoding="utf-8") as f: |
| json.dump(examples, f, indent=2, ensure_ascii=False) |
| print(f"Saved {len(examples)} examples to {json_path}") |
| except Exception as e: |
| print(f"Error saving {json_path}: {e}") |
|
|
|
|
| def process_example( |
| video_path, |
| title=None, |
| keypoints_path=None, |
| keypoints_video_path=None, |
| pose_models=None, |
| force=False, |
| checkpoint_folder="checkpoints/pose/" |
| ): |
| """ |
| Checks if keypoints JSON and rendered keypoints video exist for a video. |
| If not (or if force=True), extracts keypoints and renders the visualization video. |
| Returns the updated example dictionary. |
| """ |
| video_path = os.path.normpath(video_path) |
| if not os.path.exists(video_path): |
| print(f"[Warning] Video file not found: {video_path}") |
| return None |
|
|
| video_dir = os.path.dirname(video_path) |
| video_stem = os.path.splitext(os.path.basename(video_path))[0] |
| |
| if not title: |
| title = video_stem.replace("_", " ").capitalize() |
|
|
| if not keypoints_path: |
| keypoints_path = os.path.join(video_dir, f"{video_stem}_keypoints.json") |
| keypoints_path = os.path.normpath(keypoints_path) |
|
|
| if not keypoints_video_path: |
| keypoints_video_path = os.path.join(video_dir, f"{video_stem}_keypoints.mp4") |
| keypoints_video_path = os.path.normpath(keypoints_video_path) |
|
|
| kp_exists = os.path.exists(keypoints_path) and os.path.getsize(keypoints_path) > 0 |
| vid_exists = os.path.exists(keypoints_video_path) and os.path.getsize(keypoints_video_path) > 0 |
|
|
| if kp_exists and vid_exists and not force: |
| print(f"[OK] Keypoints and video already exist for '{title}' ({video_path})") |
| return { |
| "title": title, |
| "video_path": video_path.replace("\\", "/"), |
| "keypoints_path": keypoints_path.replace("\\", "/"), |
| "keypoints_video_path": keypoints_video_path.replace("\\", "/"), |
| } |
|
|
| print(f"\n[Processing] Generating keypoints for '{title}' ({video_path})...") |
|
|
| |
| if pose_models is None: |
| print(" Initializing MediaPipe & YOLO models...") |
| pose_models = create_mediapipe_models(checkpoint_folder) |
|
|
| |
| video_frames, fps = load_video_cv(video_path) |
| if fps is None or fps <= 0 or np.isnan(fps): |
| fps = 25.0 |
| print(f" Loaded {len(video_frames)} frames @ {fps:.1f} FPS") |
|
|
| |
| pose_results = predict_pose(video_frames, pose_models) |
|
|
| |
| kps_list = pose_results.get("keypoints", []) |
| bbox_face_list = pose_results.get("bbox_face", []) |
| bbox_lh_list = pose_results.get("bbox_left_hand", []) |
| bbox_rh_list = pose_results.get("bbox_right_hand", []) |
|
|
| for i, kp in enumerate(kps_list): |
| if i < len(bbox_face_list): |
| kp['bbox_face'] = bbox_face_list[i] |
| if i < len(bbox_lh_list): |
| kp['bbox_left_hand'] = bbox_lh_list[i] |
| if i < len(bbox_rh_list): |
| kp['bbox_right_hand'] = bbox_rh_list[i] |
|
|
| |
| heavy_image_keys = {'images', 'cropped_images', 'cropped_left_hand', 'cropped_right_hand', 'cropped_face'} |
| clean_pose_results = {k: v for k, v in pose_results.items() if k not in heavy_image_keys} |
|
|
| os.makedirs(os.path.dirname(os.path.abspath(keypoints_path)), exist_ok=True) |
| serializable_results = to_json_serializable(clean_pose_results) |
| with open(keypoints_path, "w", encoding="utf-8") as f: |
| json.dump(serializable_results, f) |
| print(f" [Saved] Lightweight Keypoints JSON -> {keypoints_path}") |
|
|
| |
| rendered_path = render_keypoints_video( |
| video_frames=video_frames, |
| keypoints_list=kps_list, |
| fps=fps, |
| output_path=keypoints_video_path, |
| ) |
| print(f" [Saved] Rendered Keypoints Video -> {rendered_path}") |
|
|
| return { |
| "title": title, |
| "video_path": video_path.replace("\\", "/"), |
| "keypoints_path": keypoints_path.replace("\\", "/"), |
| "keypoints_video_path": keypoints_video_path.replace("\\", "/"), |
| } |
|
|
|
|
| def ensure_all_examples( |
| json_path="examples.json", |
| default_dir="examples/displayed", |
| force=False, |
| checkpoint_folder="checkpoints/pose/" |
| ): |
| """ |
| Ensures all examples in `examples.json` (or found in default_dir) |
| have precomputed keypoints JSON and rendered MP4 video. |
| """ |
| existing_examples = load_examples_json(json_path) |
|
|
| |
| if not existing_examples and os.path.exists(default_dir): |
| print(f"Discovering example videos in {default_dir}...") |
| for fname in sorted(os.listdir(default_dir)): |
| if fname.lower().endswith(".mp4") and not fname.endswith("_keypoints.mp4"): |
| vpath = os.path.join(default_dir, fname) |
| stem = os.path.splitext(fname)[0] |
| existing_examples.append({ |
| "title": stem.capitalize(), |
| "video_path": vpath.replace("\\", "/"), |
| "keypoints_path": os.path.join(default_dir, f"{stem}_keypoints.json").replace("\\", "/"), |
| "keypoints_video_path": os.path.join(default_dir, f"{stem}_keypoints.mp4").replace("\\", "/"), |
| }) |
|
|
| if not existing_examples: |
| print("No examples found to process.") |
| return [] |
|
|
| pose_models = None |
| |
| needs_processing = force or any( |
| not ( |
| os.path.exists(ex.get("keypoints_path", "")) and |
| os.path.exists(ex.get("keypoints_video_path", "")) and |
| os.path.getsize(ex.get("keypoints_path", "")) > 0 and |
| os.path.getsize(ex.get("keypoints_video_path", "")) > 0 |
| ) |
| for ex in existing_examples |
| ) |
|
|
| if needs_processing: |
| print("Initializing pose estimation models for processing...") |
| pose_models = create_mediapipe_models(checkpoint_folder) |
|
|
| updated_examples = [] |
| for ex in existing_examples: |
| vpath = ex.get("video_path", "") |
| title = ex.get("title", "") |
| kpath = ex.get("keypoints_path", "") |
| kvpath = ex.get("keypoints_video_path", "") |
|
|
| res = process_example( |
| video_path=vpath, |
| title=title, |
| keypoints_path=kpath, |
| keypoints_video_path=kvpath, |
| pose_models=pose_models, |
| force=force, |
| checkpoint_folder=checkpoint_folder, |
| ) |
| if res: |
| updated_examples.append(res) |
| else: |
| updated_examples.append(ex) |
|
|
| save_examples_json(updated_examples, json_path) |
| return updated_examples |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser( |
| description="Ensure all examples have precomputed keypoints JSON and rendered pose videos." |
| ) |
| parser.add_argument( |
| "--json", |
| default="examples.json", |
| help="Path to examples.json (default: examples.json)" |
| ) |
| parser.add_argument( |
| "--dir", |
| default="examples/displayed", |
| help="Directory to scan for example videos if JSON is empty (default: examples/displayed)" |
| ) |
| parser.add_argument( |
| "--add", |
| default=None, |
| help="Add a specific video path to examples.json and generate its keypoints" |
| ) |
| parser.add_argument( |
| "--title", |
| default=None, |
| help="Optional title/label when adding a single video" |
| ) |
| parser.add_argument( |
| "--force", |
| action="store_true", |
| help="Force re-generation of keypoints and videos even if they exist" |
| ) |
| args = parser.parse_args() |
|
|
| if args.add: |
| print(f"Adding single video: {args.add}") |
| examples = load_examples_json(args.json) |
| pose_models = create_mediapipe_models("checkpoints/pose/") |
| res = process_example( |
| video_path=args.add, |
| title=args.title, |
| pose_models=pose_models, |
| force=args.force |
| ) |
| if res: |
| |
| norm_new = os.path.normpath(res["video_path"]) |
| exists_idx = -1 |
| for idx, ex in enumerate(examples): |
| if os.path.normpath(ex.get("video_path", "")) == norm_new: |
| exists_idx = idx |
| break |
| if exists_idx >= 0: |
| examples[exists_idx] = res |
| else: |
| examples.append(res) |
| save_examples_json(examples, args.json) |
| else: |
| ensure_all_examples( |
| json_path=args.json, |
| default_dir=args.dir, |
| force=args.force |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
