visiontest / evaluate_from_url.py

Upload folder using huggingface_hub

e4189f9 verified 11 months ago

9.95 kB

import argparse
import json
import tempfile
from pathlib import Path
from typing import List, Tuple, Dict
import urllib.request
import urllib.parse
import urllib.error

import cv2
import numpy as np

from miner1 import TVFrameResult, BoundingBox
from keypoint_evaluation import (
    load_template_from_file,
)
from test_predict_batch import (
    evaluate_keypoints_batch,
    visualize_keypoint_evaluation,
)


def fetch_json_data(url: str) -> dict:
    """Fetch JSON data from URL."""
    print(f"Fetching data from {url}...")
    
    # Create a request with headers to avoid 403 errors
    req = urllib.request.Request(url)
    req.add_header('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36')
    req.add_header('Accept', 'application/json, text/plain, */*')
    req.add_header('Accept-Language', 'en-US,en;q=0.9')
    
    try:
        with urllib.request.urlopen(req) as response:
            data = json.loads(response.read().decode('utf-8'))
        predictions = data.get('predictions', {})
        frames_list = predictions.get('frames', [])
        print(f"Successfully fetched data with {len(frames_list)} frames")
        return data
    except urllib.error.HTTPError as e:
        print(f"HTTP Error {e.code}: {e.reason}")
        if e.code == 403:
            print("403 Forbidden: The server is blocking the request. This might require authentication or different headers.")
        raise
    except urllib.error.URLError as e:
        print(f"URL Error: {e.reason}")
        raise


def download_video(video_url: str, output_path: Path) -> Path:
    """Download video from URL to local file."""
    print(f"Downloading video from {video_url}...")
    output_path.parent.mkdir(parents=True, exist_ok=True)
    urllib.request.urlretrieve(video_url, str(output_path))
    print(f"Video downloaded to {output_path}")
    return output_path


def extract_frames_from_video(video_path: Path, frame_ids: List[int] = None) -> Dict[int, np.ndarray]:
    """Extract frames from video, optionally only specific frame IDs."""
    print(f"Extracting frames from {video_path}...")
    cap = cv2.VideoCapture(str(video_path))
    if not cap.isOpened():
        raise RuntimeError(f"Unable to open video: {video_path}")
    
    frames = {}
    frame_count = 0
    
    while True:
        ret, frame = cap.read()
        if not ret:
            break
        
        if frame_ids is None or frame_count in frame_ids:
            frames[frame_count] = frame
        
        frame_count += 1
    
    cap.release()
    print(f"Extracted {len(frames)} frames from video")
    return frames


def convert_keypoints_format(json_keypoints: List[List[int]]) -> List[Tuple[int, int]]:
    """Convert keypoints from JSON format [[x,y], [x,y], ...] to List[Tuple[int, int]]."""
    return [(int(kp[0]), int(kp[1])) for kp in json_keypoints]


def convert_json_to_tvframe_results(
    json_data: dict,
    frames: Dict[int, np.ndarray],
) -> List[TVFrameResult]:
    """
    Convert JSON data to TVFrameResult objects.
    
    Args:
        json_data: JSON data containing predictions with frames, boxes, and keypoints
        frames: Dictionary mapping frame_id to frame image
    
    Returns:
        List of TVFrameResult objects
    """
    predictions = json_data.get('predictions', {})
    frames_data = predictions.get('frames', [])
    
    results = []
    for frame_data in frames_data:
        frame_id = frame_data.get('frame_id')
        if frame_id not in frames:
            print(f"Warning: Frame {frame_id} not found in extracted frames, skipping")
            continue
        
        # Convert boxes
        json_boxes = frame_data.get('boxes', [])
        boxes = []
        for box_data in json_boxes:
            box = BoundingBox(
                x1=int(box_data.get('x1', 0)),
                y1=int(box_data.get('y1', 0)),
                x2=int(box_data.get('x2', 0)),
                y2=int(box_data.get('y2', 0)),
                cls_id=int(box_data.get('cls_id', 0)),
                conf=float(box_data.get('conf', 0.0)),
            )
            boxes.append(box)
        
        # Convert keypoints
        json_keypoints = frame_data.get('keypoints', [])
        keypoints = convert_keypoints_format(json_keypoints)
        
        result = TVFrameResult(
            frame_id=frame_id,
            boxes=boxes,
            keypoints=keypoints,
        )
        results.append(result)
    
    return results


def evaluate_keypoints_from_json(
    json_data: dict,
    frames: Dict[int, np.ndarray],
    template_image: np.ndarray,
    template_keypoints: List[Tuple[int, int]],
    visualization_output_dir: Path = None,
) -> Dict[str, float]:
    """
    Evaluate keypoint accuracy from JSON data using the same function as test_predict_batch.py.
    
    Args:
        json_data: JSON data containing predictions with frames and keypoints
        frames: Dictionary mapping frame_id to frame image
        template_image: Template image for evaluation
        template_keypoints: Template keypoints
        visualization_output_dir: Optional directory to save visualization images
    
    Returns:
        Dictionary with keypoint evaluation statistics
    """
    # Convert JSON data to TVFrameResult objects
    results = convert_json_to_tvframe_results(json_data, frames)
    
    if len(results) == 0:
        print("No valid frames found in JSON data")
        return {
            "keypoint_avg_score": 0.0,
            "keypoint_valid_frames": 0,
            "keypoint_total_frames": 0,
        }
    
    print(f"Evaluating {len(results)} frames using evaluate_keypoints_batch...")
    
    # Use the same evaluation function as test_predict_batch.py
    stats = evaluate_keypoints_batch(
        results=results,
        original_frames=frames,
        template_image=template_image,
        template_keypoints=template_keypoints,
        visualization_output_dir=visualization_output_dir,
    )
    
    print("\n=== Keypoint Evaluation Results ===")
    print(f"Total frames: {stats['keypoint_total_frames']}")
    print(f"Valid frames: {stats['keypoint_valid_frames']}")
    print(f"Average score: {stats['keypoint_avg_score']:.3f}")
    print(f"Max score: {stats['keypoint_max_score']:.3f}")
    print(f"Min score: {stats['keypoint_min_score']:.3f}")
    print(f"Frames with score > 0.5: {stats['keypoint_frames_above_0.5']}")
    print(f"Frames with score > 0.7: {stats['keypoint_frames_above_0.7']}")
    
    return stats


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Fetch video and keypoint data from URL, evaluate keypoints, and visualize results."
    )
    parser.add_argument(
        "--url",
        type=str,
        default="https://pub-7b4130b6af75472f800371248bca15b6.r2.dev/scorevision/results_soccer/5Fnhz5fDihvno4DfssfRogL84VFvdDRRsgu19grbqEDPbJGv/responses/007115302-f9bd4226d1f4248c782a3179764e3203ce2fc520642eed4f7b02c40e61db55eb.json",
        help="URL to fetch JSON data containing video_url and predictions.",
    )
    parser.add_argument(
        "--template-image",
        type=Path,
        default='football_pitch_template.png',
        help="Path to football pitch template image.",
    )
    parser.add_argument(
        "--output-dir",
        type=Path,
        default='outputs/url_evaluation',
        help="Directory to save visualizations and downloaded video.",
    )
    parser.add_argument(
        "--delete-video",
        action="store_true",
        help="Delete downloaded video file after processing (default: keep video).",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    
    # Create output directory
    args.output_dir.mkdir(parents=True, exist_ok=True)
    
    # Fetch JSON data
    json_data = fetch_json_data(args.url)
    
    # Get video URL
    video_url = json_data.get('video_url')
    if not video_url:
        raise ValueError("No video_url found in JSON data")
    
    # Download video
    video_filename = Path(urllib.parse.urlparse(video_url).path).name
    if not video_filename:
        video_filename = "video.mp4"
    video_path = args.output_dir / video_filename
    
    download_video(video_url, video_path)
    
    # Get video filename without extension for folder naming
    video_name_without_ext = Path(video_filename).stem
    
    # Get frame IDs from JSON
    predictions = json_data.get('predictions', {})
    frames_data = predictions.get('frames', [])
    frame_ids = [frame_data.get('frame_id') for frame_data in frames_data]
    
    # Extract frames from video
    frames = extract_frames_from_video(video_path, frame_ids=frame_ids if frame_ids else None)
    
    # Load template
    template_image, template_keypoints = load_template_from_file(str(args.template_image))
    
    # Create visualization directory with video filename
    visualization_dir = args.output_dir / f"visualizations_{video_name_without_ext}"
    
    # Evaluate keypoints
    stats = evaluate_keypoints_from_json(
        json_data=json_data,
        frames=frames,
        template_image=template_image,
        template_keypoints=template_keypoints,
        visualization_output_dir=visualization_dir,
    )
    
    # Clean up video if requested
    if args.delete_video:
        video_path.unlink()
        print(f"Deleted video file: {video_path}")
    else:
        print(f"Video saved at: {video_path}")
    
    print(f"\nResults saved to: {args.output_dir}")
    print(f"Visualizations saved to: {visualization_dir}")


if __name__ == "__main__":
    main()

	import argparse
	import json
	import tempfile
	from pathlib import Path
	from typing import List, Tuple, Dict
	import urllib.request
	import urllib.parse
	import urllib.error

	import cv2
	import numpy as np

	from miner1 import TVFrameResult, BoundingBox
	from keypoint_evaluation import (
	load_template_from_file,
	)
	from test_predict_batch import (
	evaluate_keypoints_batch,
	visualize_keypoint_evaluation,
	)


	def fetch_json_data(url: str) -> dict:
	"""Fetch JSON data from URL."""
	print(f"Fetching data from {url}...")

	# Create a request with headers to avoid 403 errors
	req = urllib.request.Request(url)
	req.add_header('User-Agent', 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36')
	req.add_header('Accept', 'application/json, text/plain, /')
	req.add_header('Accept-Language', 'en-US,en;q=0.9')

	try:
	with urllib.request.urlopen(req) as response:
	data = json.loads(response.read().decode('utf-8'))
	predictions = data.get('predictions', {})
	frames_list = predictions.get('frames', [])
	print(f"Successfully fetched data with {len(frames_list)} frames")
	return data
	except urllib.error.HTTPError as e:
	print(f"HTTP Error {e.code}: {e.reason}")
	if e.code == 403:
	print("403 Forbidden: The server is blocking the request. This might require authentication or different headers.")
	raise
	except urllib.error.URLError as e:
	print(f"URL Error: {e.reason}")
	raise


	def download_video(video_url: str, output_path: Path) -> Path:
	"""Download video from URL to local file."""
	print(f"Downloading video from {video_url}...")
	output_path.parent.mkdir(parents=True, exist_ok=True)
	urllib.request.urlretrieve(video_url, str(output_path))
	print(f"Video downloaded to {output_path}")
	return output_path


	def extract_frames_from_video(video_path: Path, frame_ids: List[int] = None) -> Dict[int, np.ndarray]:
	"""Extract frames from video, optionally only specific frame IDs."""
	print(f"Extracting frames from {video_path}...")
	cap = cv2.VideoCapture(str(video_path))
	if not cap.isOpened():
	raise RuntimeError(f"Unable to open video: {video_path}")

	frames = {}
	frame_count = 0

	while True:
	ret, frame = cap.read()
	if not ret:
	break

	if frame_ids is None or frame_count in frame_ids:
	frames[frame_count] = frame

	frame_count += 1

	cap.release()
	print(f"Extracted {len(frames)} frames from video")
	return frames


	def convert_keypoints_format(json_keypoints: List[List[int]]) -> List[Tuple[int, int]]:
	"""Convert keypoints from JSON format [[x,y], [x,y], ...] to List[Tuple[int, int]]."""
	return [(int(kp[0]), int(kp[1])) for kp in json_keypoints]


	def convert_json_to_tvframe_results(
	json_data: dict,
	frames: Dict[int, np.ndarray],
	) -> List[TVFrameResult]:
	"""
	Convert JSON data to TVFrameResult objects.

	Args:
	json_data: JSON data containing predictions with frames, boxes, and keypoints
	frames: Dictionary mapping frame_id to frame image

	Returns:
	List of TVFrameResult objects
	"""
	predictions = json_data.get('predictions', {})
	frames_data = predictions.get('frames', [])

	results = []
	for frame_data in frames_data:
	frame_id = frame_data.get('frame_id')
	if frame_id not in frames:
	print(f"Warning: Frame {frame_id} not found in extracted frames, skipping")
	continue

	# Convert boxes
	json_boxes = frame_data.get('boxes', [])
	boxes = []
	for box_data in json_boxes:
	box = BoundingBox(
	x1=int(box_data.get('x1', 0)),
	y1=int(box_data.get('y1', 0)),
	x2=int(box_data.get('x2', 0)),
	y2=int(box_data.get('y2', 0)),
	cls_id=int(box_data.get('cls_id', 0)),
	conf=float(box_data.get('conf', 0.0)),
	)
	boxes.append(box)

	# Convert keypoints
	json_keypoints = frame_data.get('keypoints', [])
	keypoints = convert_keypoints_format(json_keypoints)

	result = TVFrameResult(
	frame_id=frame_id,
	boxes=boxes,
	keypoints=keypoints,
	)
	results.append(result)

	return results


	def evaluate_keypoints_from_json(
	json_data: dict,
	frames: Dict[int, np.ndarray],
	template_image: np.ndarray,
	template_keypoints: List[Tuple[int, int]],
	visualization_output_dir: Path = None,
	) -> Dict[str, float]:
	"""
	Evaluate keypoint accuracy from JSON data using the same function as test_predict_batch.py.

	Args:
	json_data: JSON data containing predictions with frames and keypoints
	frames: Dictionary mapping frame_id to frame image
	template_image: Template image for evaluation
	template_keypoints: Template keypoints
	visualization_output_dir: Optional directory to save visualization images

	Returns:
	Dictionary with keypoint evaluation statistics
	"""
	# Convert JSON data to TVFrameResult objects
	results = convert_json_to_tvframe_results(json_data, frames)

	if len(results) == 0:
	print("No valid frames found in JSON data")
	return {
	"keypoint_avg_score": 0.0,
	"keypoint_valid_frames": 0,
	"keypoint_total_frames": 0,
	}

	print(f"Evaluating {len(results)} frames using evaluate_keypoints_batch...")

	# Use the same evaluation function as test_predict_batch.py
	stats = evaluate_keypoints_batch(
	results=results,
	original_frames=frames,
	template_image=template_image,
	template_keypoints=template_keypoints,
	visualization_output_dir=visualization_output_dir,
	)

	print("\n=== Keypoint Evaluation Results ===")
	print(f"Total frames: {stats['keypoint_total_frames']}")
	print(f"Valid frames: {stats['keypoint_valid_frames']}")
	print(f"Average score: {stats['keypoint_avg_score']:.3f}")
	print(f"Max score: {stats['keypoint_max_score']:.3f}")
	print(f"Min score: {stats['keypoint_min_score']:.3f}")
	print(f"Frames with score > 0.5: {stats['keypoint_frames_above_0.5']}")
	print(f"Frames with score > 0.7: {stats['keypoint_frames_above_0.7']}")

	return stats


	def parse_args() -> argparse.Namespace:
	parser = argparse.ArgumentParser(
	description="Fetch video and keypoint data from URL, evaluate keypoints, and visualize results."
	)
	parser.add_argument(
	"--url",
	type=str,
	default="https://pub-7b4130b6af75472f800371248bca15b6.r2.dev/scorevision/results_soccer/5Fnhz5fDihvno4DfssfRogL84VFvdDRRsgu19grbqEDPbJGv/responses/007115302-f9bd4226d1f4248c782a3179764e3203ce2fc520642eed4f7b02c40e61db55eb.json",
	help="URL to fetch JSON data containing video_url and predictions.",
	)
	parser.add_argument(
	"--template-image",
	type=Path,
	default='football_pitch_template.png',
	help="Path to football pitch template image.",
	)
	parser.add_argument(
	"--output-dir",
	type=Path,
	default='outputs/url_evaluation',
	help="Directory to save visualizations and downloaded video.",
	)
	parser.add_argument(
	"--delete-video",
	action="store_true",
	help="Delete downloaded video file after processing (default: keep video).",
	)
	return parser.parse_args()


	def main() -> None:
	args = parse_args()

	# Create output directory
	args.output_dir.mkdir(parents=True, exist_ok=True)

	# Fetch JSON data
	json_data = fetch_json_data(args.url)

	# Get video URL
	video_url = json_data.get('video_url')
	if not video_url:
	raise ValueError("No video_url found in JSON data")

	# Download video
	video_filename = Path(urllib.parse.urlparse(video_url).path).name
	if not video_filename:
	video_filename = "video.mp4"
	video_path = args.output_dir / video_filename

	download_video(video_url, video_path)

	# Get video filename without extension for folder naming
	video_name_without_ext = Path(video_filename).stem

	# Get frame IDs from JSON
	predictions = json_data.get('predictions', {})
	frames_data = predictions.get('frames', [])
	frame_ids = [frame_data.get('frame_id') for frame_data in frames_data]

	# Extract frames from video
	frames = extract_frames_from_video(video_path, frame_ids=frame_ids if frame_ids else None)

	# Load template
	template_image, template_keypoints = load_template_from_file(str(args.template_image))

	# Create visualization directory with video filename
	visualization_dir = args.output_dir / f"visualizations_{video_name_without_ext}"

	# Evaluate keypoints
	stats = evaluate_keypoints_from_json(
	json_data=json_data,
	frames=frames,
	template_image=template_image,
	template_keypoints=template_keypoints,
	visualization_output_dir=visualization_dir,
	)

	# Clean up video if requested
	if args.delete_video:
	video_path.unlink()
	print(f"Deleted video file: {video_path}")
	else:
	print(f"Video saved at: {video_path}")

	print(f"\nResults saved to: {args.output_dir}")
	print(f"Visualizations saved to: {visualization_dir}")


	if __name__ == "__main__":
	main()