Spaces:
Sleeping
Sleeping
| """ | |
| Professional Itinerary Builder | |
| This module creates beautiful, realistic day-by-day itineraries from agent outputs. | |
| It implements sophisticated travel planning patterns used by professional travel planners. | |
| Key Features: | |
| 1. Smart activity sequencing with logical flow | |
| 2. Time management with travel time calculations | |
| 3. Energy and pace management for realistic schedules | |
| 4. Meal planning and logistics integration | |
| 5. Itinerary optimization for maximum value | |
| 6. Multiple itinerary variants for different travel styles | |
| """ | |
| import asyncio | |
| import uuid | |
| from abc import ABC, abstractmethod | |
| from dataclasses import dataclass, field | |
| from datetime import datetime, timedelta, time | |
| from enum import Enum | |
| from typing import Dict, List, Optional, Any, Tuple, Set, Union | |
| from decimal import Decimal | |
| import math | |
| import random | |
| from pydantic import BaseModel, Field, validator | |
| from ..models.flight_models import FlightOption | |
| from ..models.hotel_models import HotelOption | |
| from ..models.poi_models import POI, ActivityLevel, ActivityCategory, TimeSlot | |
| class ItineraryStyle(str, Enum): | |
| """Different itinerary styles for various travel preferences.""" | |
| CONSERVATIVE = "conservative" # Relaxed pace, fewer activities | |
| ADVENTUROUS = "adventurous" # Fast pace, many activities | |
| CULTURAL = "cultural" # Museums, historical sites | |
| OUTDOOR = "outdoor" # Nature, hiking, outdoor activities | |
| FOODIE = "foodie" # Restaurants, food tours, markets | |
| LUXURY = "luxury" # High-end experiences, premium venues | |
| BUDGET = "budget" # Free/cheap activities, cost-conscious | |
| FAMILY = "family" # Child-friendly activities | |
| BUSINESS = "business" # Work-friendly schedule | |
| ROMANTIC = "romantic" # Couple-focused experiences | |
| class MealType(str, Enum): | |
| """Types of meals to plan.""" | |
| BREAKFAST = "breakfast" | |
| LUNCH = "lunch" | |
| DINNER = "dinner" | |
| SNACK = "snack" | |
| COFFEE = "coffee" | |
| class TimeBlock(str, Enum): | |
| """Time blocks for scheduling.""" | |
| EARLY_MORNING = "early_morning" # 6:00-9:00 | |
| MORNING = "morning" # 9:00-12:00 | |
| AFTERNOON = "afternoon" # 12:00-17:00 | |
| EVENING = "evening" # 17:00-21:00 | |
| NIGHT = "night" # 21:00-24:00 | |
| class EnergyLevel(str, Enum): | |
| """Energy levels for activity planning.""" | |
| LOW = "low" # Relaxing, minimal physical activity | |
| MODERATE = "moderate" # Some walking, moderate activity | |
| HIGH = "high" # Active, lots of walking/activity | |
| EXTREME = "extreme" # Very active, strenuous activities | |
| class ActivitySlot: | |
| """Represents a time slot for an activity.""" | |
| start_time: time | |
| end_time: time | |
| activity: POI | |
| travel_time_minutes: int = 0 | |
| energy_level: EnergyLevel = EnergyLevel.MODERATE | |
| meal_break: Optional[MealType] = None | |
| notes: str = "" | |
| def duration_hours(self) -> float: | |
| """Calculate duration in hours.""" | |
| start_dt = datetime.combine(datetime.today(), self.start_time) | |
| end_dt = datetime.combine(datetime.today(), self.end_time) | |
| return (end_dt - start_dt).total_seconds() / 3600 | |
| def to_dict(self) -> Dict[str, Any]: | |
| """Convert to dictionary for serialization.""" | |
| return { | |
| "start_time": self.start_time.isoformat(), | |
| "end_time": self.end_time.isoformat(), | |
| "activity": { | |
| "name": self.activity.name, | |
| "category": self.activity.category.value if hasattr(self.activity.category, 'value') else str(self.activity.category), | |
| "location": self.activity.location, | |
| "duration_hours": self.activity.duration_hours | |
| }, | |
| "travel_time_minutes": self.travel_time_minutes, | |
| "energy_level": self.energy_level.value, | |
| "meal_break": self.meal_break.value if self.meal_break else None, | |
| "notes": self.notes | |
| } | |
| class DayItinerary: | |
| """Represents a complete day itinerary.""" | |
| date: datetime | |
| day_number: int | |
| activities: List[ActivitySlot] = field(default_factory=list) | |
| meals: Dict[MealType, Dict[str, Any]] = field(default_factory=dict) | |
| total_energy_level: EnergyLevel = EnergyLevel.MODERATE | |
| estimated_cost: float = 0.0 | |
| notes: str = "" | |
| def total_duration_hours(self) -> float: | |
| """Calculate total activity duration for the day.""" | |
| return sum(activity.duration_hours for activity in self.activities) | |
| def activity_count(self) -> int: | |
| """Get number of activities.""" | |
| return len(self.activities) | |
| def to_dict(self) -> Dict[str, Any]: | |
| """Convert to dictionary for serialization.""" | |
| return { | |
| "date": self.date.isoformat(), | |
| "day_number": self.day_number, | |
| "activities": [activity.to_dict() for activity in self.activities], | |
| "meals": {meal.value: details for meal, details in self.meals.items()}, | |
| "total_energy_level": self.total_energy_level.value, | |
| "estimated_cost": self.estimated_cost, | |
| "total_duration_hours": self.total_duration_hours, | |
| "activity_count": self.activity_count, | |
| "notes": self.notes | |
| } | |
| class Itinerary: | |
| """Complete trip itinerary.""" | |
| itinerary_id: str = field(default_factory=lambda: str(uuid.uuid4())) | |
| trip_name: str = "" | |
| style: ItineraryStyle = ItineraryStyle.CONSERVATIVE | |
| days: List[DayItinerary] = field(default_factory=list) | |
| flight_info: Optional[Dict[str, Any]] = None | |
| hotel_info: Optional[Dict[str, Any]] = None | |
| total_estimated_cost: float = 0.0 | |
| total_activities: int = 0 | |
| average_daily_energy: EnergyLevel = EnergyLevel.MODERATE | |
| created_at: datetime = field(default_factory=datetime.now) | |
| def to_dict(self) -> Dict[str, Any]: | |
| """Convert to dictionary for serialization.""" | |
| return { | |
| "itinerary_id": self.itinerary_id, | |
| "trip_name": self.trip_name, | |
| "style": self.style.value, | |
| "days": [day.to_dict() for day in self.days], | |
| "flight_info": self.flight_info, | |
| "hotel_info": self.hotel_info, | |
| "total_estimated_cost": self.total_estimated_cost, | |
| "total_activities": self.total_activities, | |
| "average_daily_energy": self.average_daily_energy.value, | |
| "created_at": self.created_at.isoformat(), | |
| "trip_duration_days": len(self.days) | |
| } | |
| class TravelTimeCalculator: | |
| """Calculates travel time between locations.""" | |
| def __init__(self): | |
| # Base travel speeds (km/h) | |
| self.walking_speed = 5.0 # km/h | |
| self.driving_speed = 30.0 # km/h (city driving) | |
| self.public_transport_speed = 20.0 # km/h | |
| # Time buffers for different activities | |
| self.buffer_times = { | |
| "museum": 30, # 30 minutes buffer | |
| "restaurant": 15, # 15 minutes buffer | |
| "outdoor": 10, # 10 minutes buffer | |
| "shopping": 20, # 20 minutes buffer | |
| } | |
| def calculate_travel_time(self, | |
| from_location: Tuple[float, float], | |
| to_location: Tuple[float, float], | |
| travel_mode: str = "walking") -> int: | |
| """Calculate travel time between two locations in minutes.""" | |
| # Calculate distance using simple Euclidean distance (in real app, use proper geodetic calculation) | |
| lat1, lon1 = from_location | |
| lat2, lon2 = to_location | |
| # Rough conversion to km (1 degree ≈ 111 km) | |
| distance_km = math.sqrt((lat2 - lat1)**2 + (lon2 - lon1)**2) * 111 | |
| # Select speed based on travel mode | |
| if travel_mode == "walking": | |
| speed = self.walking_speed | |
| elif travel_mode == "driving": | |
| speed = self.driving_speed | |
| else: # public transport | |
| speed = self.public_transport_speed | |
| # Calculate time in minutes | |
| time_hours = distance_km / speed | |
| time_minutes = int(time_hours * 60) | |
| # Add minimum travel time | |
| return max(time_minutes, 5) # Minimum 5 minutes | |
| def get_buffer_time(self, activity_type: str) -> int: | |
| """Get buffer time for an activity type.""" | |
| return self.buffer_times.get(activity_type.lower(), 15) | |
| class EnergyManager: | |
| """Manages energy levels and pacing for itineraries.""" | |
| def __init__(self): | |
| # Energy level mappings | |
| self.activity_energy_levels = { | |
| "museum": EnergyLevel.LOW, | |
| "shopping": EnergyLevel.LOW, | |
| "restaurant": EnergyLevel.LOW, | |
| "spa": EnergyLevel.LOW, | |
| "park": EnergyLevel.MODERATE, | |
| "walking_tour": EnergyLevel.MODERATE, | |
| "market": EnergyLevel.MODERATE, | |
| "hiking": EnergyLevel.HIGH, | |
| "sports": EnergyLevel.HIGH, | |
| "adventure": EnergyLevel.EXTREME, | |
| "climbing": EnergyLevel.EXTREME, | |
| } | |
| # Energy recovery rates | |
| self.recovery_rates = { | |
| EnergyLevel.LOW: 1.0, # No recovery needed | |
| EnergyLevel.MODERATE: 0.8, # 80% recovery | |
| EnergyLevel.HIGH: 0.6, # 60% recovery | |
| EnergyLevel.EXTREME: 0.4, # 40% recovery | |
| } | |
| def get_activity_energy_level(self, activity: POI) -> EnergyLevel: | |
| """Determine energy level for an activity.""" | |
| # Check activity type first | |
| activity_type = str(activity.activity_type).lower() | |
| if activity_type in self.activity_energy_levels: | |
| return self.activity_energy_levels[activity_type] | |
| # Check activity level from POI | |
| if hasattr(activity, 'activity_level'): | |
| activity_level = activity.activity_level.value if hasattr(activity.activity_level, 'value') else str(activity.activity_level) | |
| return EnergyLevel(activity_level.lower()) | |
| # Default to moderate | |
| return EnergyLevel.MODERATE | |
| def calculate_day_energy_level(self, activities: List[ActivitySlot]) -> EnergyLevel: | |
| """Calculate overall energy level for a day.""" | |
| if not activities: | |
| return EnergyLevel.LOW | |
| # Get energy levels for all activities | |
| energy_levels = [activity.energy_level for activity in activities] | |
| # Count each energy level | |
| energy_counts = {} | |
| for level in EnergyLevel: | |
| energy_counts[level] = energy_levels.count(level) | |
| # Determine dominant energy level | |
| if energy_counts[EnergyLevel.EXTREME] > 0: | |
| return EnergyLevel.EXTREME | |
| elif energy_counts[EnergyLevel.HIGH] >= 2: | |
| return EnergyLevel.HIGH | |
| elif energy_counts[EnergyLevel.HIGH] == 1 and energy_counts[EnergyLevel.MODERATE] >= 2: | |
| return EnergyLevel.HIGH | |
| elif energy_counts[EnergyLevel.MODERATE] >= 2: | |
| return EnergyLevel.MODERATE | |
| else: | |
| return EnergyLevel.LOW | |
| def should_add_break(self, current_energy: EnergyLevel, next_energy: EnergyLevel) -> bool: | |
| """Determine if a break should be added between activities.""" | |
| if current_energy == EnergyLevel.LOW: | |
| return False | |
| elif current_energy == EnergyLevel.MODERATE and next_energy in [EnergyLevel.HIGH, EnergyLevel.EXTREME]: | |
| return True | |
| elif current_energy == EnergyLevel.HIGH and next_energy == EnergyLevel.EXTREME: | |
| return True | |
| elif current_energy == EnergyLevel.EXTREME: | |
| return True | |
| return False | |
| class MealPlanner: | |
| """Handles meal planning and restaurant recommendations.""" | |
| def __init__(self): | |
| self.meal_times = { | |
| MealType.BREAKFAST: time(8, 0), # 8:00 AM | |
| MealType.LUNCH: time(13, 0), # 1:00 PM | |
| MealType.DINNER: time(19, 0), # 7:00 PM | |
| MealType.SNACK: time(15, 0), # 3:00 PM | |
| MealType.COFFEE: time(10, 0), # 10:00 AM | |
| } | |
| self.meal_durations = { | |
| MealType.BREAKFAST: 45, # 45 minutes | |
| MealType.LUNCH: 60, # 1 hour | |
| MealType.DINNER: 90, # 1.5 hours | |
| MealType.SNACK: 20, # 20 minutes | |
| MealType.COFFEE: 30, # 30 minutes | |
| } | |
| def get_meal_recommendations(self, | |
| meal_type: MealType, | |
| location: str, | |
| style: ItineraryStyle) -> Dict[str, Any]: | |
| """Get meal recommendations for a location and style.""" | |
| # Base recommendations (in real app, integrate with restaurant APIs) | |
| recommendations = { | |
| MealType.BREAKFAST: { | |
| "type": "Café or Hotel Restaurant", | |
| "duration_minutes": self.meal_durations[meal_type], | |
| "cost_range": "$15-25" if style == ItineraryStyle.LUXURY else "$8-15", | |
| "notes": "Start the day with a good breakfast" | |
| }, | |
| MealType.LUNCH: { | |
| "type": "Local Restaurant" if style == ItineraryStyle.FOODIE else "Quick Bite", | |
| "duration_minutes": self.meal_durations[meal_type], | |
| "cost_range": "$25-40" if style == ItineraryStyle.LUXURY else "$12-25", | |
| "notes": "Refuel for afternoon activities" | |
| }, | |
| MealType.DINNER: { | |
| "type": "Fine Dining" if style == ItineraryStyle.LUXURY else "Local Cuisine", | |
| "duration_minutes": self.meal_durations[meal_type], | |
| "cost_range": "$60-120" if style == ItineraryStyle.LUXURY else "$20-45", | |
| "notes": "Relax and enjoy local flavors" | |
| }, | |
| MealType.SNACK: { | |
| "type": "Street Food" if style == ItineraryStyle.FOODIE else "Café", | |
| "duration_minutes": self.meal_durations[meal_type], | |
| "cost_range": "$5-15", | |
| "notes": "Quick energy boost" | |
| }, | |
| MealType.COFFEE: { | |
| "type": "Local Coffee Shop", | |
| "duration_minutes": self.meal_durations[meal_type], | |
| "cost_range": "$4-8", | |
| "notes": "Perfect for a morning break" | |
| } | |
| } | |
| return recommendations.get(meal_type, { | |
| "type": "Local Restaurant", | |
| "duration_minutes": 30, | |
| "cost_range": "$10-20", | |
| "notes": "Enjoy local cuisine" | |
| }) | |
| def plan_meals_for_day(self, | |
| day_activities: List[ActivitySlot], | |
| style: ItineraryStyle) -> Dict[MealType, Dict[str, Any]]: | |
| """Plan meals for a day based on activities and style.""" | |
| meals = {} | |
| # Always include breakfast | |
| meals[MealType.BREAKFAST] = self.get_meal_recommendations( | |
| MealType.BREAKFAST, "hotel", style | |
| ) | |
| # Add lunch if there are afternoon activities | |
| afternoon_activities = [a for a in day_activities | |
| if a.start_time >= time(12, 0) and a.end_time <= time(18, 0)] | |
| if afternoon_activities: | |
| meals[MealType.LUNCH] = self.get_meal_recommendations( | |
| MealType.LUNCH, "activity_area", style | |
| ) | |
| # Add dinner if there are evening activities or it's a long day | |
| evening_activities = [a for a in day_activities if a.start_time >= time(17, 0)] | |
| if evening_activities or len(day_activities) >= 3: | |
| meals[MealType.DINNER] = self.get_meal_recommendations( | |
| MealType.DINNER, "activity_area", style | |
| ) | |
| # Add coffee break if there are many activities | |
| if len(day_activities) >= 3: | |
| meals[MealType.COFFEE] = self.get_meal_recommendations( | |
| MealType.COFFEE, "activity_area", style | |
| ) | |
| return meals | |
| class ItineraryOptimizer: | |
| """Optimizes itineraries for maximum value and efficiency.""" | |
| def __init__(self): | |
| self.travel_calculator = TravelTimeCalculator() | |
| self.energy_manager = EnergyManager() | |
| def optimize_activity_sequence(self, | |
| activities: List[POI], | |
| hotel_location: Tuple[float, float], | |
| style: ItineraryStyle) -> List[POI]: | |
| """Optimize the sequence of activities to minimize travel time.""" | |
| if not activities: | |
| return activities | |
| # Group activities by category for logical flow | |
| categorized_activities = self._categorize_activities(activities) | |
| # Apply style-specific optimization | |
| if style == ItineraryStyle.CULTURAL: | |
| return self._optimize_for_cultural_flow(categorized_activities, hotel_location) | |
| elif style == ItineraryStyle.OUTDOOR: | |
| return self._optimize_for_outdoor_flow(categorized_activities, hotel_location) | |
| elif style == ItineraryStyle.FOODIE: | |
| return self._optimize_for_food_flow(categorized_activities, hotel_location) | |
| else: | |
| return self._optimize_for_general_flow(categorized_activities, hotel_location) | |
| def _categorize_activities(self, activities: List[POI]) -> Dict[str, List[POI]]: | |
| """Categorize activities by type.""" | |
| categories = { | |
| "morning": [], | |
| "afternoon": [], | |
| "evening": [], | |
| "indoor": [], | |
| "outdoor": [], | |
| "cultural": [], | |
| "food": [] | |
| } | |
| for activity in activities: | |
| category = activity.category.value if hasattr(activity.category, 'value') else str(activity.category) | |
| # Categorize by time preference | |
| if hasattr(activity, 'best_time_to_visit'): | |
| best_time = activity.best_time_to_visit | |
| if best_time and "morning" in best_time.lower(): | |
| categories["morning"].append(activity) | |
| elif best_time and "evening" in best_time.lower(): | |
| categories["evening"].append(activity) | |
| else: | |
| categories["afternoon"].append(activity) | |
| else: | |
| categories["afternoon"].append(activity) | |
| # Categorize by type | |
| if "cultural" in category.lower() or "museum" in category.lower(): | |
| categories["cultural"].append(activity) | |
| elif "outdoor" in category.lower() or "park" in category.lower(): | |
| categories["outdoor"].append(activity) | |
| elif "food" in category.lower() or "restaurant" in category.lower(): | |
| categories["food"].append(activity) | |
| # Categorize by indoor/outdoor | |
| if hasattr(activity, 'weather_dependency'): | |
| weather_dep = activity.weather_dependency.value if hasattr(activity.weather_dependency, 'value') else str(activity.weather_dependency) | |
| if "indoor" in weather_dep.lower(): | |
| categories["indoor"].append(activity) | |
| else: | |
| categories["outdoor"].append(activity) | |
| return categories | |
| def _optimize_for_cultural_flow(self, | |
| categorized: Dict[str, List[POI]], | |
| hotel_location: Tuple[float, float]) -> List[POI]: | |
| """Optimize for cultural activities with logical flow.""" | |
| optimized = [] | |
| # Start with morning cultural activities | |
| optimized.extend(categorized["morning"]) | |
| optimized.extend(categorized["cultural"]) | |
| # Add afternoon activities | |
| optimized.extend(categorized["afternoon"]) | |
| # End with evening activities | |
| optimized.extend(categorized["evening"]) | |
| return optimized | |
| def _optimize_for_outdoor_flow(self, | |
| categorized: Dict[str, List[POI]], | |
| hotel_location: Tuple[float, float]) -> List[POI]: | |
| """Optimize for outdoor activities with weather considerations.""" | |
| optimized = [] | |
| # Prioritize outdoor activities in morning/afternoon | |
| optimized.extend(categorized["outdoor"]) | |
| optimized.extend(categorized["morning"]) | |
| optimized.extend(categorized["afternoon"]) | |
| # Add indoor activities as backup | |
| optimized.extend(categorized["indoor"]) | |
| # End with evening activities | |
| optimized.extend(categorized["evening"]) | |
| return optimized | |
| def _optimize_for_food_flow(self, | |
| categorized: Dict[str, List[POI]], | |
| hotel_location: Tuple[float, float]) -> List[POI]: | |
| """Optimize for food-focused itinerary.""" | |
| optimized = [] | |
| # Start with food markets or morning food activities | |
| optimized.extend(categorized["food"]) | |
| optimized.extend(categorized["morning"]) | |
| # Add other activities | |
| optimized.extend(categorized["afternoon"]) | |
| # End with evening dining | |
| optimized.extend(categorized["evening"]) | |
| return optimized | |
| def _optimize_for_general_flow(self, | |
| categorized: Dict[str, List[POI]], | |
| hotel_location: Tuple[float, float]) -> List[POI]: | |
| """General optimization for balanced itinerary.""" | |
| optimized = [] | |
| # Balanced approach | |
| optimized.extend(categorized["morning"]) | |
| optimized.extend(categorized["cultural"]) | |
| optimized.extend(categorized["afternoon"]) | |
| optimized.extend(categorized["outdoor"]) | |
| optimized.extend(categorized["evening"]) | |
| return optimized | |
| class ProfessionalItineraryBuilder: | |
| """ | |
| Main itinerary builder that creates professional-grade travel itineraries. | |
| This class orchestrates all the components to create realistic, well-paced | |
| itineraries that feel professionally planned. | |
| """ | |
| def __init__(self): | |
| self.travel_calculator = TravelTimeCalculator() | |
| self.energy_manager = EnergyManager() | |
| self.meal_planner = MealPlanner() | |
| self.optimizer = ItineraryOptimizer() | |
| # Default scheduling parameters | |
| self.default_start_time = time(9, 0) # 9:00 AM | |
| self.default_end_time = time(21, 0) # 9:00 PM | |
| self.max_activities_per_day = 4 | |
| self.min_break_time = 30 # minutes | |
| async def build_itinerary(self, | |
| flight: FlightOption, | |
| hotel: HotelOption, | |
| pois: List[POI], | |
| trip_context: Dict[str, Any]) -> Itinerary: | |
| """ | |
| Build a complete professional itinerary from agent outputs. | |
| """ | |
| # Extract trip information | |
| trip_name = trip_context.get("trip_name", "Amazing Trip") | |
| style = ItineraryStyle(trip_context.get("itinerary_style", "conservative")) | |
| duration_days = trip_context.get("trip_duration_days", 3) | |
| start_date = trip_context.get("start_date", datetime.now().date()) | |
| # Create itinerary | |
| itinerary = Itinerary( | |
| trip_name=trip_name, | |
| style=style, | |
| flight_info=self._extract_flight_info(flight), | |
| hotel_info=self._extract_hotel_info(hotel) | |
| ) | |
| # Optimize activity sequence | |
| optimized_pois = self.optimizer.optimize_activity_sequence( | |
| pois, (hotel.latitude, hotel.longitude), style | |
| ) | |
| # Create day-by-day itineraries | |
| days = await self._create_daily_itineraries( | |
| optimized_pois, hotel, duration_days, start_date, style, trip_context | |
| ) | |
| itinerary.days = days | |
| itinerary.total_activities = sum(day.activity_count for day in days) | |
| itinerary.total_estimated_cost = sum(day.estimated_cost for day in days) | |
| # Calculate average daily energy | |
| if days: | |
| energy_levels = [day.total_energy_level for day in days] | |
| itinerary.average_daily_energy = max(energy_levels, key=energy_levels.count) | |
| return itinerary | |
| async def create_multiple_variants(self, | |
| flight: FlightOption, | |
| hotel: HotelOption, | |
| pois: List[POI], | |
| trip_context: Dict[str, Any]) -> Dict[ItineraryStyle, Itinerary]: | |
| """ | |
| Create multiple itinerary variants for different travel styles. | |
| """ | |
| variants = {} | |
| # Define variant styles | |
| variant_styles = [ | |
| ItineraryStyle.CONSERVATIVE, | |
| ItineraryStyle.ADVENTUROUS, | |
| ItineraryStyle.CULTURAL, | |
| ItineraryStyle.OUTDOOR, | |
| ItineraryStyle.FOODIE | |
| ] | |
| # Create each variant | |
| for style in variant_styles: | |
| variant_context = trip_context.copy() | |
| variant_context["itinerary_style"] = style.value | |
| variant_context["trip_name"] = f"{trip_context.get('trip_name', 'Trip')} - {style.value.title()}" | |
| itinerary = await self.build_itinerary(flight, hotel, pois, variant_context) | |
| variants[style] = itinerary | |
| return variants | |
| async def _create_daily_itineraries(self, | |
| pois: List[POI], | |
| hotel: HotelOption, | |
| duration_days: int, | |
| start_date: datetime, | |
| style: ItineraryStyle, | |
| trip_context: Dict[str, Any]) -> List[DayItinerary]: | |
| """Create day-by-day itineraries.""" | |
| days = [] | |
| # Distribute activities across days | |
| activities_per_day = self._distribute_activities(pois, duration_days, style) | |
| for day_num in range(duration_days): | |
| day_date = start_date + timedelta(days=day_num) | |
| day_pois = activities_per_day[day_num] | |
| # Create day itinerary | |
| day_itinerary = await self._create_single_day_itinerary( | |
| day_pois, hotel, day_date, day_num + 1, style | |
| ) | |
| days.append(day_itinerary) | |
| return days | |
| def _distribute_activities(self, | |
| pois: List[POI], | |
| duration_days: int, | |
| style: ItineraryStyle) -> List[List[POI]]: | |
| """Distribute activities across days based on style and constraints.""" | |
| # Calculate activities per day based on style | |
| if style == ItineraryStyle.CONSERVATIVE: | |
| max_per_day = 2 | |
| elif style == ItineraryStyle.ADVENTUROUS: | |
| max_per_day = 5 | |
| elif style == ItineraryStyle.CULTURAL: | |
| max_per_day = 3 | |
| elif style == ItineraryStyle.FOODIE: | |
| max_per_day = 4 | |
| else: | |
| max_per_day = 3 | |
| # Distribute activities | |
| activities_per_day = [] | |
| remaining_pois = pois.copy() | |
| for day in range(duration_days): | |
| day_activities = [] | |
| day_limit = min(max_per_day, len(remaining_pois)) | |
| for _ in range(day_limit): | |
| if remaining_pois: | |
| day_activities.append(remaining_pois.pop(0)) | |
| activities_per_day.append(day_activities) | |
| return activities_per_day | |
| async def _create_single_day_itinerary(self, | |
| day_pois: List[POI], | |
| hotel: HotelOption, | |
| date: datetime, | |
| day_number: int, | |
| style: ItineraryStyle) -> DayItinerary: | |
| """Create a single day's itinerary.""" | |
| day_itinerary = DayItinerary( | |
| date=date, | |
| day_number=day_number | |
| ) | |
| # Create activity slots | |
| activity_slots = await self._schedule_activities( | |
| day_pois, hotel, style | |
| ) | |
| day_itinerary.activities = activity_slots | |
| day_itinerary.total_energy_level = self.energy_manager.calculate_day_energy_level(activity_slots) | |
| # Plan meals | |
| day_itinerary.meals = self.meal_planner.plan_meals_for_day(activity_slots, style) | |
| # Calculate estimated cost | |
| day_itinerary.estimated_cost = self._calculate_day_cost(activity_slots, day_itinerary.meals) | |
| # Add day notes | |
| day_itinerary.notes = self._generate_day_notes(day_itinerary, style) | |
| return day_itinerary | |
| async def _schedule_activities(self, | |
| day_pois: List[POI], | |
| hotel: HotelOption, | |
| style: ItineraryStyle) -> List[ActivitySlot]: | |
| """Schedule activities for a day with proper timing.""" | |
| activity_slots = [] | |
| current_time = self.default_start_time | |
| hotel_location = (hotel.latitude, hotel.longitude) | |
| previous_location = hotel_location | |
| for i, poi in enumerate(day_pois): | |
| # Calculate travel time from previous location | |
| poi_location = (poi.latitude, poi.longitude) | |
| travel_time = self.travel_calculator.calculate_travel_time( | |
| previous_location, poi_location | |
| ) | |
| # Add travel time to current time | |
| if i > 0: # Don't add travel time for first activity | |
| travel_minutes = travel_time | |
| else: | |
| travel_minutes = 0 | |
| # Calculate activity duration | |
| activity_duration_hours = poi.duration_hours | |
| activity_duration_minutes = int(activity_duration_hours * 60) | |
| # Create activity slot | |
| start_time = current_time | |
| end_time = self._add_minutes_to_time(start_time, activity_duration_minutes + travel_minutes) | |
| # Determine energy level | |
| energy_level = self.energy_manager.get_activity_energy_level(poi) | |
| # Check if we need a meal break | |
| meal_break = None | |
| if self._should_add_meal_break(start_time, style): | |
| meal_break = self._get_appropriate_meal_type(start_time) | |
| activity_slot = ActivitySlot( | |
| start_time=start_time, | |
| end_time=end_time, | |
| activity=poi, | |
| travel_time_minutes=travel_minutes, | |
| energy_level=energy_level, | |
| meal_break=meal_break, | |
| notes=self._generate_activity_notes(poi, travel_minutes, meal_break) | |
| ) | |
| activity_slots.append(activity_slot) | |
| # Update for next iteration | |
| current_time = end_time | |
| previous_location = poi_location | |
| # Add break time if needed | |
| if self._should_add_break(energy_level, day_pois, i, style): | |
| current_time = self._add_minutes_to_time(current_time, self.min_break_time) | |
| return activity_slots | |
| def _calculate_day_cost(self, | |
| activity_slots: List[ActivitySlot], | |
| meals: Dict[MealType, Dict[str, Any]]) -> float: | |
| """Calculate estimated cost for a day.""" | |
| total_cost = 0.0 | |
| # Calculate activity costs | |
| for slot in activity_slots: | |
| if hasattr(slot.activity, 'adult_price') and slot.activity.adult_price: | |
| total_cost += float(slot.activity.adult_price) | |
| # Add meal costs (rough estimates) | |
| for meal_type, meal_info in meals.items(): | |
| cost_range = meal_info.get("cost_range", "$10-20") | |
| # Extract average cost from range (simplified) | |
| if "$" in cost_range: | |
| try: | |
| costs = cost_range.replace("$", "").split("-") | |
| avg_cost = (float(costs[0]) + float(costs[1])) / 2 | |
| total_cost += avg_cost | |
| except: | |
| total_cost += 15 # Default estimate | |
| return total_cost | |
| def _generate_day_notes(self, day_itinerary: DayItinerary, style: ItineraryStyle) -> str: | |
| """Generate helpful notes for the day.""" | |
| notes = [] | |
| # Add style-specific notes | |
| if style == ItineraryStyle.CULTURAL: | |
| notes.append("Perfect day for exploring local culture and history.") | |
| elif style == ItineraryStyle.OUTDOOR: | |
| notes.append("Great day for outdoor activities and nature.") | |
| elif style == ItineraryStyle.FOODIE: | |
| notes.append("Foodie paradise - don't forget to try local specialties!") | |
| # Add energy level notes | |
| if day_itinerary.total_energy_level == EnergyLevel.HIGH: | |
| notes.append("High-energy day - wear comfortable shoes and stay hydrated.") | |
| elif day_itinerary.total_energy_level == EnergyLevel.EXTREME: | |
| notes.append("Very active day - make sure you're well-rested and prepared.") | |
| # Add weather notes | |
| outdoor_activities = [a for a in day_itinerary.activities | |
| if hasattr(a.activity, 'weather_dependency') and | |
| 'outdoor' in str(a.activity.weather_dependency).lower()] | |
| if outdoor_activities: | |
| notes.append("Check weather forecast - some activities are weather-dependent.") | |
| return " ".join(notes) | |
| def _generate_activity_notes(self, poi: POI, travel_time: int, meal_break: Optional[MealType]) -> str: | |
| """Generate notes for a specific activity.""" | |
| notes = [] | |
| if travel_time > 15: | |
| notes.append(f"Allow {travel_time} minutes for travel.") | |
| if meal_break: | |
| notes.append(f"Perfect time for {meal_break.value}.") | |
| if hasattr(poi, 'advance_booking_required') and poi.advance_booking_required: | |
| notes.append("Advance booking required.") | |
| if hasattr(poi, 'dress_code') and poi.dress_code: | |
| notes.append(f"Dress code: {poi.dress_code}") | |
| return " ".join(notes) | |
| def _should_add_meal_break(self, current_time: time, style: ItineraryStyle) -> bool: | |
| """Determine if a meal break should be added.""" | |
| if style == ItineraryStyle.FOODIE: | |
| return True | |
| # Add meal breaks at appropriate times | |
| if current_time.hour in [12, 13, 19, 20]: # Lunch or dinner time | |
| return True | |
| return False | |
| def _get_appropriate_meal_type(self, current_time: time) -> MealType: | |
| """Get appropriate meal type for current time.""" | |
| if current_time.hour < 11: | |
| return MealType.BREAKFAST | |
| elif current_time.hour < 15: | |
| return MealType.LUNCH | |
| elif current_time.hour < 17: | |
| return MealType.COFFEE | |
| else: | |
| return MealType.DINNER | |
| def _should_add_break(self, | |
| current_energy: EnergyLevel, | |
| day_pois: List[POI], | |
| current_index: int, | |
| style: ItineraryStyle) -> bool: | |
| """Determine if a break should be added.""" | |
| if current_index >= len(day_pois) - 1: # Last activity | |
| return False | |
| next_poi = day_pois[current_index + 1] | |
| next_energy = self.energy_manager.get_activity_energy_level(next_poi) | |
| return self.energy_manager.should_add_break(current_energy, next_energy) | |
| def _add_minutes_to_time(self, current_time: time, minutes: int) -> time: | |
| """Add minutes to a time object.""" | |
| current_dt = datetime.combine(datetime.today(), current_time) | |
| new_dt = current_dt + timedelta(minutes=minutes) | |
| return new_dt.time() | |
| def _extract_flight_info(self, flight: FlightOption) -> Dict[str, Any]: | |
| """Extract flight information for itinerary.""" | |
| return { | |
| "airline": flight.airline, | |
| "flight_number": flight.flight_number, | |
| "route": f"{flight.departure_city} → {flight.arrival_city}", | |
| "departure_time": flight.departure_time.isoformat(), | |
| "arrival_time": flight.arrival_time.isoformat(), | |
| "duration_hours": flight.duration_hours, | |
| "price": float(flight.price) | |
| } | |
| def _extract_hotel_info(self, hotel: HotelOption) -> Dict[str, Any]: | |
| """Extract hotel information for itinerary.""" | |
| return { | |
| "name": hotel.name, | |
| "chain": hotel.chain.value if hasattr(hotel.chain, 'value') else str(hotel.chain), | |
| "address": hotel.address, | |
| "rating": hotel.rating, | |
| "price_per_night": float(hotel.price_per_night), | |
| "location_type": hotel.location_type.value if hasattr(hotel.location_type, 'value') else str(hotel.location_type), | |
| "check_in": hotel.check_in_date.isoformat(), | |
| "check_out": hotel.check_out_date.isoformat() | |
| } | |