Spaces:
Sleeping
Sleeping
File size: 11,416 Bytes
23a9367 7b6b271 23a9367 7b6b271 23a9367 7b6b271 23a9367 7b6b271 23a9367 7b6b271 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 |
"""
Surf Evaluation Tool - Multi-Factor Scoring Algorithm.
This module implements a sophisticated surf condition evaluation system
that scores surf spots based on multiple environmental factors and user
preferences. The algorithm considers wave conditions, wind patterns,
swell direction, and skill compatibility.
Scoring Components (weighted):
- Wave Conditions (35%): Height matching and safety
- Wind Analysis (25%): Speed and direction optimization
- Swell Direction (25%): Angular matching to optimal window
- Skill Compatibility (15%): Break type vs experience level
The scoring system uses normalized 0-100 scales with progressive
curves that reward optimal conditions while penalizing dangerous
or suboptimal scenarios.
Example:
>>> evaluator = SurfEvaluatorTool()
>>> result = await evaluator.run({
... "spot": surf_spot_data,
... "conditions": current_wave_data,
... "prefs": {"skill_level": "intermediate"}
... })
>>> print(f"Score: {result['score']}/100")
Author: Surf Spot Finder Team
License: MIT
"""
from pydantic import BaseModel
from typing import Dict, Any, List, Tuple
import math
class SurfEvaluatorTool:
"""
Advanced surf evaluation tool with multi-factor scoring algorithm.
This tool implements a comprehensive evaluation system that analyzes
surf conditions across multiple dimensions to produce a single score
representing the quality and suitability of surfing conditions.
The algorithm weighs different factors based on their importance:
- Wave conditions: Safety and optimal height ranges
- Wind patterns: Offshore vs onshore preferences
- Swell direction: Alignment with spot's optimal angles
- User skill level: Appropriate difficulty matching
All scores are normalized to 0-100 scale for consistency.
Attributes:
name: Tool identifier for MCP registration.
description: Human-readable tool description.
Example:
>>> evaluator = SurfEvaluatorTool()
>>> input_data = SurfEvaluatorTool.SurfEvalInput(
... spot=spot_data,
... conditions=wave_conditions,
... prefs={"skill_level": "beginner"}
... )
>>> result = await evaluator.run(input_data.dict())
"""
class SurfEvalInput(BaseModel):
"""Input schema for surf evaluation.
Attributes:
spot: Surf spot data including location and characteristics.
conditions: Current wave/wind conditions from marine APIs.
prefs: User preferences including skill level and board type.
"""
spot: Dict[str, Any]
conditions: Dict[str, Any]
prefs: Dict[str, Any] = {}
def is_direction_in_range(self, direction: float, direction_range: List[float]) -> bool:
"""Check if direction falls within optimal range for the surf spot.
Handles both normal ranges and those crossing 0° (e.g., 315-45°).
Args:
direction: Current direction in degrees (0-360).
direction_range: [start, end] optimal range in degrees.
Returns:
True if direction is within the optimal range.
"""
if len(direction_range) != 2:
return False
start, end = direction_range
# Handle ranges that cross 0° (e.g., [315, 45])
if start > end:
return direction >= start or direction <= end
else:
return start <= direction <= end
def calculate_direction_score(self, direction: float, optimal_range: List[float]) -> float:
"""Calculate score based on direction proximity to optimal range.
Uses progressive scoring where perfect alignment = 1.0,
and score decreases with angular distance from optimal range.
Args:
direction: Current direction in degrees.
optimal_range: [start, end] optimal range in degrees.
Returns:
Score between 0.0-1.0 based on direction quality.
"""
if not optimal_range or len(optimal_range) != 2:
return 0.5 # Neutral score if no preference
if self.is_direction_in_range(direction, optimal_range):
return 1.0 # Perfect score if in range
start, end = optimal_range
# Calculate distance to nearest edge of range
if start > end: # Range crosses 0°
dist_to_start = min(abs(direction - start), 360 - abs(direction - start))
dist_to_end = min(abs(direction - end), 360 - abs(direction - end))
else:
dist_to_start = abs(direction - start)
dist_to_end = abs(direction - end)
min_distance = min(dist_to_start, dist_to_end)
# Score decreases with distance (max penalty at 90° off)
return max(0, 1 - (min_distance / 90))
def evaluate_wave_height(self, wave_height: float, spot_prefs: Dict[str, Any], user_prefs: Dict[str, Any]) -> Tuple[float, str]:
"""Evaluate wave height against spot and user preferences."""
min_height = spot_prefs.get("min_wave_height", 0.3)
max_height = spot_prefs.get("max_wave_height", 10.0)
# User skill level adjustments
skill_level = user_prefs.get("skill_level", "intermediate")
if skill_level == "beginner":
max_height = min(max_height, 2.0)
elif skill_level == "expert":
min_height = max(min_height, 1.5)
if wave_height < min_height:
score = wave_height / min_height * 0.5 # Partial score for smaller waves
explanation = f"Wave height {wave_height}m below minimum {min_height}m"
elif wave_height > max_height:
score = max(0, 1 - (wave_height - max_height) / max_height)
explanation = f"Wave height {wave_height}m above comfortable maximum {max_height}m"
else:
# Optimal range - score based on how close to ideal
ideal_height = (min_height + max_height) / 2
distance_from_ideal = abs(wave_height - ideal_height)
range_size = max_height - min_height
score = 1 - (distance_from_ideal / (range_size / 2)) * 0.3 # Max 30% penalty
explanation = f"Wave height {wave_height}m in optimal range {min_height}-{max_height}m"
return max(0, score), explanation
def evaluate_wind(self, wind_speed: float, wind_direction: float, spot_prefs: Dict[str, Any]) -> Tuple[float, str]:
"""Evaluate wind conditions for surfing."""
optimal_wind_dir = spot_prefs.get("wind_direction", [])
# Wind speed scoring (offshore/light winds preferred)
if wind_speed <= 5:
wind_speed_score = 1.0
elif wind_speed <= 15:
wind_speed_score = 1 - ((wind_speed - 5) / 10) * 0.5
else:
wind_speed_score = max(0, 0.5 - ((wind_speed - 15) / 20) * 0.5)
# Wind direction scoring
wind_dir_score = self.calculate_direction_score(wind_direction, optimal_wind_dir)
# Combined wind score
wind_score = (wind_speed_score + wind_dir_score) / 2
explanation = f"Wind {wind_speed}kt at {wind_direction}°"
if optimal_wind_dir:
explanation += f" (optimal: {optimal_wind_dir[0]}-{optimal_wind_dir[1]}°)"
return wind_score, explanation
def evaluate_swell(self, swell_direction: float, spot_prefs: Dict[str, Any]) -> Tuple[float, str]:
"""Evaluate swell direction compatibility."""
optimal_swell_dir = spot_prefs.get("swell_direction", [])
if not optimal_swell_dir:
return 0.7, "No swell direction preference specified"
score = self.calculate_direction_score(swell_direction, optimal_swell_dir)
explanation = f"Swell from {swell_direction}° (optimal: {optimal_swell_dir[0]}-{optimal_swell_dir[1]}°)"
return score, explanation
def calculate_skill_compatibility(self, spot: Dict[str, Any], user_prefs: Dict[str, Any]) -> Tuple[float, str]:
"""Check if spot matches user skill level."""
user_skill = user_prefs.get("skill_level", "intermediate")
spot_skills = spot.get("characteristics", {}).get("skill_level", ["intermediate"])
if user_skill in spot_skills:
score = 1.0
explanation = f"Perfect skill match for {user_skill} surfer"
elif user_skill == "beginner" and "intermediate" in spot_skills:
score = 0.7
explanation = "Spot suitable for progression to intermediate"
elif user_skill == "intermediate" and "advanced" in spot_skills:
score = 0.8
explanation = "Challenging spot for skill development"
else:
score = 0.3
explanation = f"Skill mismatch: {user_skill} vs {spot_skills}"
return score, explanation
async def run(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
"""Run comprehensive surf spot evaluation."""
data = self.SurfEvalInput(**input_data)
spot = data.spot
conditions = data.conditions
user_prefs = data.prefs
spot_name = spot.get("name", "Unknown Spot")
spot_prefs = spot.get("optimal_conditions", {})
# Extract conditions
wave_height = conditions.get("wave_height", 0)
wind_speed = conditions.get("wind_speed", 0)
wind_direction = conditions.get("wind_direction", 0)
swell_direction = conditions.get("swell_direction", 0)
# Individual evaluations
wave_score, wave_explanation = self.evaluate_wave_height(wave_height, spot_prefs, user_prefs)
wind_score, wind_explanation = self.evaluate_wind(wind_speed, wind_direction, spot_prefs)
swell_score, swell_explanation = self.evaluate_swell(swell_direction, spot_prefs)
skill_score, skill_explanation = self.calculate_skill_compatibility(spot, user_prefs)
# Weighted total score
weights = {
"wave": 0.35,
"wind": 0.25,
"swell": 0.25,
"skill": 0.15
}
total_score = (
wave_score * weights["wave"] +
wind_score * weights["wind"] +
swell_score * weights["swell"] +
skill_score * weights["skill"]
) * 100 # Scale to 0-100
# Detailed explanation
explanation = f"""
Evaluation for {spot_name}:
• Waves: {wave_explanation} (Score: {wave_score:.2f})
• Wind: {wind_explanation} (Score: {wind_score:.2f})
• Swell: {swell_explanation} (Score: {swell_score:.2f})
• Skill Match: {skill_explanation} (Score: {skill_score:.2f})
Overall conditions rating: {total_score:.1f}/100
""".strip()
return {
"spot": spot_name,
"score": round(total_score, 1),
"explanation": explanation,
"breakdown": {
"wave_score": round(wave_score, 2),
"wind_score": round(wind_score, 2),
"swell_score": round(swell_score, 2),
"skill_score": round(skill_score, 2)
}
}
|