# app/services/green_space_service.py from typing import List, Optional, Tuple, Dict, Any from app.models.green_space import GreenSpace, ScoringRequest from app.models.response import DetailedAnalysis, DiscoveryResult from app.services.scoring_engine import ScoringEngine from app.services.berlin_data_service import BerlinDataService class GreenSpaceService: """Service for managing green space operations and scoring.""" def __init__(self): self.scoring_engine = ScoringEngine() self.berlin_data = BerlinDataService() async def search_and_score(self, request: ScoringRequest) -> List[GreenSpace]: """Search for green spaces and score them based on personality preferences.""" try: # Get green spaces based on search criteria green_spaces = await self.berlin_data.search_green_spaces( location=request.location, radius=request.radius, neighborhood=request.neighborhood, filters=request.filters ) # Score each green space for the personality scored_spaces = [] for space in green_spaces: personality_score = await self.scoring_engine.score_green_space( space, request.personality, request.location ) # Only include if meets minimum score if personality_score.score >= request.min_score: space.current_personality_score = personality_score # Include amenities if requested if request.include_amenities: space.nearby_amenities = await self.berlin_data.get_amenities_near_point( space.coordinates.lat, space.coordinates.lng, 500 # 500m radius for amenities ) scored_spaces.append(space) # Sort by score (highest first) and limit results scored_spaces.sort( key=lambda x: x.current_personality_score.score if x.current_personality_score else 0, reverse=True ) return scored_spaces[:request.limit] except Exception as e: raise Exception(f"Failed to search and score green spaces: {str(e)}") async def discover_nearby( self, lat: float, lng: float, max_distance: int, personality: str ) -> List[DiscoveryResult]: """Discover highly-rated green spaces near a location.""" try: # Get nearby green spaces nearby_spaces = await self.berlin_data.get_green_spaces_within_radius( lat, lng, max_distance ) discoveries = [] for space in nearby_spaces: # Score the space personality_score = await self.scoring_engine.score_green_space( space, personality, (lat, lng) ) # Only include high-scoring spaces if personality_score.score >= 70: space.current_personality_score = personality_score # Calculate travel info distance = await self.berlin_data.calculate_distance( lat, lng, space.coordinates.lat, space.coordinates.lng ) discovery = DiscoveryResult( green_space=space, distance_meters=distance, travel_time_walking=max(5, distance // 80), # ~5 km/h walking speed travel_time_cycling=max(2, distance // 250), # ~15 km/h cycling speed why_recommended=personality_score.explanation, best_route_description=await self._get_route_description( lat, lng, space.coordinates.lat, space.coordinates.lng ) ) discoveries.append(discovery) # Sort by score and distance discoveries.sort( key=lambda x: (x.green_space.current_personality_score.score, -x.distance_meters), reverse=True ) return discoveries[:10] # Top 10 discoveries except Exception as e: raise Exception(f"Failed to discover green spaces: {str(e)}") async def get_detailed_analysis( self, space_id: str, focus_personality: Optional[str] = None ) -> Optional[DetailedAnalysis]: """Get detailed analysis of a specific green space.""" try: # Get the green space green_space = await self.berlin_data.get_green_space_by_id(space_id) if not green_space: return None # Score for all personalities or just the focused one personality_breakdown = {} personalities = [focus_personality] if focus_personality else [ "little_adventurers", "date_night", "squad_goals", "zen_masters", "active_lifestyle", "wildlife_lover", "art_nerd", "history_geek" ] for personality in personalities: if personality: # Skip None values score = await self.scoring_engine.score_green_space( green_space, personality ) personality_breakdown[personality] = score # Find best locations within the space best_locations = await self.scoring_engine.find_best_locations_within( green_space, focus_personality or "zen_masters" ) # Generate seasonal recommendations seasonal_recommendations = self._generate_seasonal_recommendations( green_space, focus_personality ) # Find similar spaces similar_spaces = await self.berlin_data.find_similar_green_spaces( green_space, limit=5 ) return DetailedAnalysis( green_space=green_space, personality_breakdown=personality_breakdown, best_locations_within=best_locations, seasonal_recommendations=seasonal_recommendations, optimal_visit_times=self._get_optimal_visit_times(green_space), similar_spaces=[space.id for space in similar_spaces] ) except Exception as e: raise Exception(f"Failed to analyze green space: {str(e)}") def _generate_seasonal_recommendations( self, green_space: GreenSpace, personality: Optional[str] ) -> Dict[str, str]: """Generate seasonal visit recommendations.""" recommendations = {} # Base recommendations on green space features if green_space.environmental.water_features: recommendations["summer"] = "Perfect for cooling off by the water features" recommendations["spring"] = "Beautiful reflections in the water as nature awakens" if green_space.environmental.tree_coverage_percent > 70: recommendations["autumn"] = "Stunning fall foliage display" recommendations["spring"] = "Fresh green canopy and blooming trees" if green_space.recreation.sports_facilities: recommendations["summer"] = "Great weather for outdoor sports and activities" recommendations["winter"] = "Crisp air perfect for winter sports if available" # Fill in any missing seasons with generic recommendations seasons = ["spring", "summer", "autumn", "winter"] for season in seasons: if season not in recommendations: recommendations[season] = f"Enjoy the peaceful {season} atmosphere" return recommendations def _get_optimal_visit_times(self, green_space: GreenSpace) -> List[str]: """Get optimal visit times based on green space characteristics.""" times = [] if green_space.environmental.noise_level.value <= 2: times.append("Early morning (7-9 AM) for peaceful solitude") if green_space.recreation.playground_quality > 50: times.append("Mid-morning (9-11 AM) for family activities") if green_space.accessibility.lighting_quality >= 4: times.append("Evening (6-8 PM) for romantic walks") if green_space.recreation.sports_facilities: times.append("Late afternoon (4-6 PM) for sports activities") # Always include at least one recommendation if not times: times.append("Midday (11 AM - 2 PM) for general enjoyment") return times async def _get_route_description( self, from_lat: float, from_lng: float, to_lat: float, to_lng: float ) -> str: """Get a simple route description.""" # This is a simplified version - in a real app you'd use a routing service distance = await self.berlin_data.calculate_distance(from_lat, from_lng, to_lat, to_lng) if distance < 500: return "Short walk through the neighborhood" elif distance < 1500: return "Pleasant walk or quick bike ride" elif distance < 3000: return "Good cycling distance or longer walk" else: return "Best reached by bike or public transport"