Par-ity_Project / app /models /llm_analyzer.py
chenemii's picture
Fix front-facing golf metrics: improve accuracy and reliability
33b443b
"""
LLM-based Golf Swing Analysis
This module handles LLM analysis and prompt generation for golf swing feedback.
"""
import os
import json
import requests
import re
from typing import Dict, Any, Optional
from .metrics_calculator import (
calculate_back_tilt_degree, calculate_knee_bend_degree,
calculate_shoulder_tilt_swing_plane_at_top,
compute_dtl_three
)
from .front_facing_metrics import compute_front_facing_metrics
# Import grading functions to avoid duplication
try:
from ..streamlit_app import (
get_shoulder_tilt_swing_plane_grading,
get_back_tilt_grading,
get_knee_flexion_grading,
get_head_drop_grading,
get_shoulder_tilt_impact_grading,
get_hip_sway_grading,
get_wrist_hinge_grading,
get_hip_shoulder_separation_impact_grading
)
except ImportError:
# Fallback if imports fail - define minimal grading functions
def get_shoulder_tilt_swing_plane_grading(value, confidence):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_back_tilt_grading(value, confidence):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_knee_flexion_grading(value, confidence):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_hip_depth_grading(value, confidence):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_shoulder_tilt_impact_grading(value, confidence):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_hip_sway_grading(value, confidence, position):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_wrist_hinge_grading(value, confidence, position):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_hip_shoulder_separation_impact_grading(value, confidence):
return {'value': value, 'status': 'n/a', 'badge': '⚪'}
def get_head_drop_grading(value, confidence):
"""Grade head drop metric based on percentage of torso length
Args:
value: percentage value (positive = head moved down)
confidence: confidence score (not used in current implementation)
"""
if value is None:
return {'value': None, 'status': 'error', 'badge': '❌'}
# Address → Top: ~2–6% typical; >8% = too much "sit"
if 2 <= value <= 6:
return {'value': f"{value:.1f}%", 'status': 'good', 'badge': '✅'}
elif 6 < value <= 8:
return {'value': f"{value:.1f}%", 'status': 'caution', 'badge': '⚠️'}
else:
return {'value': f"{value:.1f}%", 'status': 'poor', 'badge': '❌'}
def safe_fmt_deg(v):
"""Format angle value safely, returning 'n/a' for None or invalid values"""
if v is None or (isinstance(v, str) and v.lower() in ['none', 'n/a', '']):
return 'n/a'
try:
return f"{float(v):.1f}°"
except (ValueError, TypeError):
return 'n/a'
def safe_fmt_percent(v):
"""Format percentage value safely, returning 'n/a' for None or invalid values"""
if v is None or (isinstance(v, str) and v.lower() in ['none', 'n/a', '']):
return 'n/a'
try:
return f"{float(v):.1f}%"
except (ValueError, TypeError):
return 'n/a'
def check_llm_services():
"""Check availability of LLM services"""
services = {
'openai': {'available': bool(os.getenv('OPENAI_API_KEY'))},
'ollama': {'available': False} # Will check if running
}
# Check if Ollama is running
try:
response = requests.get('http://localhost:11434/api/version', timeout=2)
services['ollama']['available'] = response.status_code == 200
except:
pass
return services
def call_openai_service(prompt, config):
"""Call OpenAI API for swing analysis"""
api_key = os.getenv('OPENAI_API_KEY')
if not api_key:
return None
headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
payload = {
'model': config.get('model', 'gpt-4'),
'messages': [{'role': 'user', 'content': prompt}],
'max_tokens': config.get('max_tokens', 2000),
'temperature': config.get('temperature', 0.7)
}
try:
response = requests.post(
'https://api.openai.com/v1/chat/completions',
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
return response.json()['choices'][0]['message']['content']
except Exception as e:
print(f"OpenAI API error: {e}")
return None
def call_ollama_service(prompt, config):
"""Call local Ollama service for swing analysis"""
try:
response = requests.post(
'http://localhost:11434/api/generate',
json={
'model': config.get('model', 'llama2'),
'prompt': prompt,
'stream': False
},
timeout=60
)
response.raise_for_status()
return response.json().get('response')
except Exception as e:
print(f"Ollama service error: {e}")
return None
def compute_core_metrics(pose_data, swing_phases, frame_timestamps_ms=None, total_ms=None, player_handedness='right', is_front_facing=False, frames=None, world_landmarks=None, club="iron"):
"""Compute core golf swing metrics
Args:
pose_data (dict): Dictionary mapping frame indices to pose keypoints
swing_phases (dict): Dictionary mapping phase names to lists of frame indices
frame_timestamps_ms (list, optional): List of frame timestamps in milliseconds
total_ms (float, optional): Total video duration in milliseconds
player_handedness (str): 'right' or 'left' handed player
is_front_facing (bool): True for front-facing camera view
club (str): Club type for grading ("driver", "iron", "wedge")
Returns:
dict: Core metrics with confidence scores and status
"""
if is_front_facing:
# Use front-facing metrics
return compute_front_facing_core_metrics(pose_data, swing_phases, frames, world_landmarks, club)
else:
# Use DTL metrics (existing implementation)
return compute_dtl_core_metrics(pose_data, swing_phases, frame_timestamps_ms, total_ms, player_handedness, frames)
def compute_dtl_core_metrics(pose_data, swing_phases, frame_timestamps_ms=None, total_ms=None, player_handedness='right', frames=None):
"""Compute DTL (Down-the-Line) golf swing metrics using new unified function"""
# Extract frame dimensions from frames if available
frame_w, frame_h = 1920, 1080 # Default values
if frames and len(frames) > 0:
if hasattr(frames[0], 'shape'):
frame_h, frame_w = frames[0].shape[:2]
# Use the new unified DTL metrics function
try:
dtl_metrics = compute_dtl_three(pose_data, swing_phases, frame_w, frame_h, frames)
if dtl_metrics is None:
# Fallback to individual calculations if unified function fails
return compute_dtl_core_metrics_fallback(pose_data, swing_phases, frame_timestamps_ms, total_ms, player_handedness, frames)
# Convert to the expected format with grading
core_metrics = {}
# Map the new metric names to the expected format (hip depth and hip turn removed)
metric_mapping = {
'shoulder_plane_top_deg': ('shoulder_tilt_swing_plane_top_deg', get_shoulder_tilt_swing_plane_grading),
'back_tilt_setup_deg': ('back_tilt_deg', get_back_tilt_grading),
'knee_flexion_deg': ('knee_flexion_deg', get_knee_flexion_grading),
'head_drop_top_pct': ('head_drop_top_pct', get_head_drop_grading),
# Hip turn mapping removed per user request
}
for new_key, (old_key, grading_func) in metric_mapping.items():
value = dtl_metrics.get(new_key)
if value is not None:
try:
grading = grading_func(value, 0.8)
if grading:
# Hip turn debug info removed per user request
core_metrics[old_key] = grading
except Exception:
# Fallback format if grading fails
core_metrics[old_key] = {
'value': value,
'status': 'Calculated',
'badge': '✅'
}
# Add validation info if available
validation = dtl_metrics.get('_validation')
if validation:
core_metrics['_validation'] = validation
# Add debug info if available
debug_info = dtl_metrics.get('_debug')
if debug_info:
core_metrics['_debug'] = debug_info
return core_metrics
except Exception as e:
print(f"Error in compute_dtl_three: {e}")
# Fallback to individual calculations
return compute_dtl_core_metrics_fallback(pose_data, swing_phases, frame_timestamps_ms, total_ms, player_handedness, frames)
def compute_dtl_core_metrics_fallback(pose_data, swing_phases, frame_timestamps_ms=None, total_ms=None, player_handedness='right', frames=None):
"""Fallback function using individual metric calculations"""
# Extract frame dimensions from frames if available
frame_w, frame_h = 1920, 1080 # Default values
if frames and len(frames) > 0:
if hasattr(frames[0], 'shape'):
frame_h, frame_w = frames[0].shape[:2]
# Get phase frame indices
setup_frames = swing_phases.get("setup", [])
backswing_frames = swing_phases.get("backswing", [])
downswing_frames = swing_phases.get("downswing", [])
impact_frames = swing_phases.get("impact", [])
# Get key frame indices
address_idx = setup_frames[0] if setup_frames else 0
top_idx = backswing_frames[-1] if backswing_frames else address_idx
impact_idx = impact_frames[0] if impact_frames else top_idx
# Initialize core metrics with new DTL metrics - don't set default status, let grading functions handle it
core_metrics = {}
# Calculate Shoulder Tilt / Swing Plane Angle at Top - professional = 36°, 30 handicap = 29°
try:
# Ensure we have valid indices and data
if backswing_frames and top_idx in pose_data and pose_data[top_idx] is not None:
shoulder_tilt_swing_plane = calculate_shoulder_tilt_swing_plane_at_top(pose_data, swing_phases, top_idx, frames)
if shoulder_tilt_swing_plane is not None and isinstance(shoulder_tilt_swing_plane, (int, float)):
grading = get_shoulder_tilt_swing_plane_grading(shoulder_tilt_swing_plane, 0.8)
if grading and isinstance(grading, dict):
core_metrics["shoulder_tilt_swing_plane_top_deg"] = grading
except Exception as e:
# Silently continue if calculation fails
pass
# Calculate Back Tilt @ Setup
try:
# Ensure we have valid indices and data
if setup_frames and address_idx in pose_data and pose_data[address_idx] is not None:
back_tilt = calculate_back_tilt_degree(pose_data, swing_phases, address_idx, frames)
if back_tilt is not None and isinstance(back_tilt, (int, float)):
grading = get_back_tilt_grading(back_tilt, 0.8)
if grading and isinstance(grading, dict):
core_metrics["back_tilt_deg"] = grading
except Exception as e:
# Silently continue if calculation fails
pass
# Calculate Knee Flexion @ Setup
knee_bend_data = calculate_knee_bend_degree(pose_data, address_idx)
if knee_bend_data is not None:
primary_value = knee_bend_data.get('primary_value') # Average or single value
if primary_value is not None:
grading = get_knee_flexion_grading(primary_value, 0.8)
# Add separate lead/trail values if available
lead_flexion = knee_bend_data.get('lead_knee_flexion')
trail_flexion = knee_bend_data.get('trail_knee_flexion')
if lead_flexion is not None:
grading['lead_knee_flexion'] = round(lead_flexion, 1)
if trail_flexion is not None:
grading['trail_knee_flexion'] = round(trail_flexion, 1)
core_metrics["knee_flexion_deg"] = grading
# Calculate Hip Depth / Early Extension
try:
hip_depth_data = calculate_hip_depth_early_extension(pose_data, swing_phases)
if hip_depth_data:
# Check if it's an error message
if 'error' in hip_depth_data:
# Store error but still try to display it
core_metrics["hip_depth_early_extension"] = {
'value': None,
'detailed_data': hip_depth_data,
'status': f'Error: {hip_depth_data["error"]}',
'badge': '🔴',
'error': True
}
elif isinstance(hip_depth_data, dict) and 'depth_loss_pct' in hip_depth_data:
# Valid data structure - proceed with grading
grading = get_hip_depth_grading(hip_depth_data, 0.8)
if grading:
core_metrics["hip_depth_early_extension"] = grading
else:
# Unexpected data structure - create a fallback entry
core_metrics["hip_depth_early_extension"] = {
'value': None,
'status': 'Data format error',
'badge': '🔴'
}
except Exception as e:
# Create a fallback entry for display
core_metrics["hip_depth_early_extension"] = {
'value': None,
'status': 'Calculation failed',
'badge': '🔴'
}
# Hip turn calculation removed per user request
return core_metrics
def compute_front_facing_core_metrics(pose_data, swing_phases, frames=None, world_landmarks=None, club="iron"):
"""Compute front-facing golf swing metrics"""
# Get front-facing metrics from the new module
front_metrics = compute_front_facing_metrics(pose_data, swing_phases, world_landmarks=world_landmarks, frames=frames, club=club, handedness="right")
# Convert to the expected format with status and badges
core_metrics = {}
# Process each front-facing metric with new calibration values
for metric_name, metric_data in front_metrics.items():
# Skip hip turn metric entirely
if 'hip_turn_impact' in metric_name:
continue
# Handle other metrics with traditional processing
value = metric_data.get('value') if isinstance(metric_data, dict) else metric_data
if 'shoulder_tilt_impact' in metric_name:
core_metrics[metric_name] = get_shoulder_tilt_impact_grading(value, 0.9)
elif 'hip_shoulder_separation_impact' in metric_name:
confidence = metric_data.get('confidence', 0.8) if isinstance(metric_data, dict) else 0.8
core_metrics[metric_name] = get_hip_shoulder_separation_impact_grading(value, confidence)
elif 'hip_sway_top' in metric_name:
core_metrics[metric_name] = get_hip_sway_grading(value, 0.8, "top")
elif 'wrist_hinge_top' in metric_name:
core_metrics[metric_name] = get_wrist_hinge_grading(value, 0.8, "top")
else:
# Default case
core_metrics[metric_name] = {'value': value, 'status': 'n/a'}
return core_metrics
def prepare_data_for_llm(pose_data, swing_phases, trajectory_data=None, fps=30.0, frame_shape=None, frame_timestamps_ms=None, total_ms=None, is_front_facing=False, frames=None):
"""Prepare swing data for LLM analysis
Args:
pose_data (dict): Dictionary mapping frame indices to pose keypoints
swing_phases (dict): Dictionary mapping phase names to lists of frame indices
trajectory_data (dict, optional): Ball trajectory data
fps (float): Video frame rate for timing calculations
frame_shape (tuple, optional): Frame shape (H, W)
frame_timestamps_ms (list, optional): List of frame timestamps in milliseconds
total_ms (float, optional): Total video duration in milliseconds
is_front_facing (bool): True for front-facing camera view
Returns:
dict: Formatted swing data for LLM
"""
# Compute core metrics
core_metrics = compute_core_metrics(
pose_data, swing_phases, frame_timestamps_ms, total_ms,
player_handedness='right', is_front_facing=is_front_facing, frames=frames
)
# Calculate timing data
setup_frames = swing_phases.get("setup", [])
backswing_frames = swing_phases.get("backswing", [])
downswing_frames = swing_phases.get("downswing", [])
impact_frames = swing_phases.get("impact", [])
follow_through_frames = swing_phases.get("follow_through", [])
total_frames = len(setup_frames) + len(backswing_frames) + len(downswing_frames) + len(impact_frames) + len(follow_through_frames)
if total_ms is None:
total_ms = total_frames * (1000.0 / fps)
dt = total_ms / max(total_frames, 1) / 1000.0 # seconds per frame
swing_data = {
"swing_phases": {
"setup": {
"frame_count": len(setup_frames),
"duration_ms": len(setup_frames) * dt * 1000.0
},
"backswing": {
"frame_count": len(backswing_frames),
"duration_ms": len(backswing_frames) * dt * 1000.0
},
"downswing": {
"frame_count": len(downswing_frames),
"duration_ms": len(downswing_frames) * dt * 1000.0
},
"impact": {
"frame_count": len(impact_frames),
"duration_ms": len(impact_frames) * dt * 1000.0
},
"follow_through": {
"frame_count": len(follow_through_frames),
"duration_ms": len(follow_through_frames) * dt * 1000.0
}
},
"timing_metrics": {
"total_swing_frames": total_frames,
"total_swing_time_ms": total_frames * dt * 1000.0,
"actual_fps": round(1.0 / dt if dt > 0 else fps, 1)
},
"core_metrics": core_metrics
}
return swing_data
def create_llm_prompt(analysis_data):
"""Create LLM prompt from swing analysis data"""
prompt_template = """# Golf Swing Analysis
## NEW METRICS CALIBRATION
Use these new professional/amateur benchmarks for scoring. These represent updated golf swing mechanics based on the latest analysis:
### **NEW FRONT-FACING METRICS:**
**Shoulder Tilt @ Impact:**
- Professional = 39°
- 30 Handicap = 27°
**Hip Sway @ Top:**
- Professional = 3.9" towards target
- 30 Handicap = 2.5" towards target
**Wrist Hinge @ Top:**
- To be measured and calibrated
### **NEW DTL METRICS:**
**Shoulder Tilt/Swing Plane Angle @ Top:**
- Professional = 36° (Iron avg: 34.3°, Driver avg: 30.5°)
- 30 Handicap = 29°
**Back Tilt (°):**
- Professional = 31° (Iron avg: 30.8°, Driver avg: 32.3°)
**Knee Flexion (°):**
- Professional = 22° (Iron avg: 21.6°, Driver avg: 27.2°)
**Head Drop/Rise @ Top (%):**
- Professional = 0-5% drop (Iron: 0-7.6% range, Driver: 4.6-5.5% drop)
### **ANALYSIS INSTRUCTIONS**
**GOLF SWING ANALYSIS FORMAT**
Use the benchmarks above to guide your evaluation. Follow this exact format:
**OVERALL_SUMMARY:** [1-2 sentences maximum providing a concise evaluation of the swing's overall quality and main strengths/areas for improvement]
**PERFORMANCE_CLASSIFICATION:** [XX%]
(XX = number from 10% to 100%)
**Metric Evaluations**
For each metric below, write exactly 3 sentences evaluating the metric:
1. First sentence: State if it's good, bad, or needs improvement compared to professional standards
2. Second sentence: Compare the specific value to professional/amateur ranges
3. Third sentence: Brief explanation of impact on swing performance
**Classification Bands:**
- **90–100%**: Tour-level
- **80–89%**: Advanced amateur
- **70–79%**: Skilled
- **60–69%**: Intermediate
- **50–59%**: Developing
- **40–49%**: Beginner
- **10–39%**: Novice
**STYLE & FORMATTING RULES:**
- Use these headers: OVERALL_SUMMARY, PERFORMANCE_CLASSIFICATION, Metric Evaluations
- No emojis anywhere in the response
- Write 1-2 sentences maximum for the overall summary
- Write exactly 3 sentences for each metric evaluation
- Use a positive, coaching tone throughout
- Focus on biomechanics and compare actual values to the professional ranges provided
"""
# Format metrics for prompt
core_metrics = analysis_data.get('core_metrics', {})
metrics_text = "SWING METRICS:\n"
for metric_name, metric_data in core_metrics.items():
if metric_data.get('value') is not None:
value = metric_data['value']
confidence = metric_data.get('confidence', 0.0)
status = metric_data.get('status', 'unknown')
if isinstance(value, dict):
# Handle complex metrics like hip depth
if 'depth_loss_pct' in value:
metrics_text += f"- Hip Depth/Early Extension: {value['depth_loss_pct']}% loss (confidence: {confidence:.0%}) - {status}\n"
elif 'displacement_pct' in value:
direction = value.get('displacement_direction', 'unknown')
metrics_text += f"- Head Displacement: {value['displacement_pct']}% {direction} (confidence: {confidence:.0%}) - {status}\n"
else:
metrics_text += f"- {metric_name.replace('_', ' ').title()}: {value}° (confidence: {confidence:.0%}) - {status}\n"
# Combine template with data
full_prompt = f"{prompt_template}\n\n{metrics_text}"
return full_prompt
def generate_swing_analysis(pose_data, swing_phases, trajectory_data):
"""Generate swing analysis using LLM"""
# Prepare data for LLM
analysis_data = prepare_data_for_llm(pose_data, swing_phases, trajectory_data)
# Create prompt
prompt = create_llm_prompt(analysis_data)
# Check available services
services = check_llm_services()
# Configuration
config = {
'model': 'gpt-4',
'max_tokens': 2000,
'temperature': 0.7
}
# Try OpenAI first, then Ollama
analysis = None
if services.get('openai', {}).get('available'):
analysis = call_openai_service(prompt, config)
elif services.get('ollama', {}).get('available'):
analysis = call_ollama_service(prompt, config)
if analysis:
return analysis
else:
return "LLM analysis unavailable. Please check your API configuration."
def parse_and_format_analysis(raw_analysis):
"""Parse and format the raw LLM analysis"""
if not raw_analysis:
return None
# Simple parsing - just clean up the text
cleaned_analysis = raw_analysis.strip()
# Remove any markdown formatting for cleaner display
cleaned_analysis = re.sub(r'\*\*(.*?)\*\*', r'\1', cleaned_analysis)
cleaned_analysis = re.sub(r'\*(.*?)\*', r'\1', cleaned_analysis)
return {
'formatted_analysis': cleaned_analysis,
'raw_analysis': raw_analysis
}
def display_formatted_analysis(analysis_data):
"""Display formatted analysis (for compatibility)"""
if not analysis_data:
return "No analysis data available"
return analysis_data.get('formatted_analysis', analysis_data.get('raw_analysis', 'No analysis available'))
def test_dtl_five_metrics_fixes(pose_data, swing_phases, frames=None):
"""Test function to validate the DTL metric fixes
This function tests the fixes for:
1. Shoulder tilt no longer returning ~90° due to yaw correction
2. Hip turn no longer returning ~90° due to width-ratio method
3. Hip depth properly comparing posterior hip positions
4. All 5 metrics being computed and returned
Args:
pose_data (dict): Dictionary mapping frame indices to pose keypoints
swing_phases (dict): Dictionary mapping phase names to lists of frame indices
frames (list, optional): Video frames for analysis
Returns:
dict: Test results showing before/after values and validation
"""
print("\n=== Testing DTL Five Metrics Fixes ===")
# Test the new aggregator
metrics = compute_dtl_five_metrics(pose_data, swing_phases, frames)
if metrics is None:
print("❌ Failed: compute_dtl_five_metrics returned None")
return None
print(f"✅ Success: Got {len(metrics)} metrics")
# Validate each metric
expected_metrics = [
"shoulder_plane_top_deg",
"back_tilt_setup_deg",
"knee_flexion_deg",
"hip_depth_pct",
# "hip_turn_impact_deg" # Removed hip turn metric
]
results = {
"total_metrics": len(metrics),
"expected_metrics": len(expected_metrics),
"metric_values": {},
"validation_results": {}
}
for metric_name in expected_metrics:
value = metrics.get(metric_name)
results["metric_values"][metric_name] = value
if value is None:
print(f"⚠️ {metric_name}: None (calculation failed or outside sanity bounds)")
results["validation_results"][metric_name] = "failed_or_clamped"
elif metric_name.endswith("_deg") and value >= 80:
print(f"❌ {metric_name}: {value}° (unrealistic high value - check calculation)")
results["validation_results"][metric_name] = "unrealistic_high"
else:
print(f"✅ {metric_name}: {value}{'%' if 'pct' in metric_name else '°'} (reasonable value)")
results["validation_results"][metric_name] = "success"
# Summary
successful_metrics = sum(1 for v in results["validation_results"].values() if v == "success")
print(f"\n📊 Summary: {successful_metrics}/{len(expected_metrics)} metrics returning reasonable values")
if successful_metrics >= 4:
print("🎉 Overall: GOOD - Most metrics working properly")
elif successful_metrics >= 2:
print("⚠️ Overall: FAIR - Some metrics still need work")
else:
print("❌ Overall: POOR - Major issues remain")
results["overall_status"] = "good" if successful_metrics >= 4 else "fair" if successful_metrics >= 2 else "poor"
return results
def get_hip_depth_grading_from_value(value, confidence):
"""Helper function to create hip depth grading from just the percentage value"""
if value is None:
return None
# Create a mock hip_depth_data structure for the existing grading function
mock_data = {
'depth_loss_pct': value,
'confidence': confidence
}
# Try to import the grading function from streamlit_app
try:
from ..streamlit_app import get_hip_depth_grading
return get_hip_depth_grading(mock_data, confidence)
except ImportError:
# Fallback grading logic
if value <= 5:
status = "Excellent - minimal early extension"
badge = "🟢"
elif value <= 15:
status = "Good - slight early extension"
badge = "🟡"
else:
status = "Needs work - significant early extension"
badge = "🔴"
return {
'value': value,
'status': status,
'badge': badge,
'detailed_data': mock_data
}