RSK World - Weather Chatbot - Project Files Browser | RSK World - Free Programming Resources & Source Code

requirements.txt manifest.ini weather_maps.py advanced_nlp.py

requirements.txt

# Weather Chatbot Dependencies
# ==============================
#
# Author: RSK World (https://rskworld.in)
# Founded by: Molla Samser
# Designer & Tester: Rima Khatun
# Contact: +91 93305 39277, hello@rskworld.in, support@rskworld.in
# Location: Nutanhat, Mongolkote, Purba Burdwan, West Bengal, India, 713147
# Year: 2026
#
# Description: Required Python packages for the Weather Chatbot application
# providing forecasts, alerts, and weather information with advanced features.

# Web Framework
Flask==2.3.3
Werkzeug==2.3.7
Flask-CORS==4.0.0
Flask-Limiter==3.5.0

# HTTP Requests
requests==2.31.0

# Environment Variables
python-dotenv==1.0.0

# OpenAI API
openai>=1.0.0

# Database
SQLAlchemy==2.0.21

# Authentication & Security
PyJWT==2.8.0
cryptography==41.0.4
Werkzeug==2.3.7

# Natural Language Processing
textblob==0.17.1
spacy==3.6.1
nltk==3.8.1

# Data Processing & Analytics
pandas==2.1.1
numpy==1.25.2
scipy==1.11.3

# Data Visualization
matplotlib==3.7.2
plotly==5.17.0
seaborn==0.12.2

# Email & Notifications
email-validator==2.0.0
# Note: smtplib is built-in to Python, no need to install

# SMS (Optional)
twilio==8.8.0

# Push Notifications (Optional)
firebase-admin==6.2.0

# Caching
redis>=4.5.0
Flask-Caching==2.1.0

# Rate Limiting
flask-limiter==3.5.0

# Data Validation
marshmallow==3.20.1

# Logging
loguru==0.7.2

# Date/Time Processing
pytz==2023.3
python-dateutil==2.8.2

# Configuration Management
configparser==6.0.0

# Testing
pytest==7.4.2
pytest-flask==1.2.0
pytest-cov==4.1.0

# Code Quality
black==23.9.1
flake8==6.1.0
mypy==1.5.1

# Documentation
sphinx==7.2.6

# Development Tools
jupyter==1.0.0
ipython==8.15.0

# JSON Processing (built-in)
# json - included with Python

# Date/Time Processing (built-in)
# datetime - included with Python

# Operating System Interface (built-in)
# os - included with Python

107 lines•1.9 KB

text

utils/weather_maps.py

Raw Download

#!/usr/bin/env python3
"""
Weather Chatbot Maps & Radar Module
===================================

Author: RSK World (https://rskworld.in)
Founded by: Molla Samser
Designer & Tester: Rima Khatun
Contact: +91 93305 39277, hello@rskworld.in, support@rskworld.in
Location: Nutanhat, Mongolkote, Purba Burdwan, West Bengal, India, 713147
Year: 2026

Description: Weather maps and radar integration services
"""

import requests
import json
import base64
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
import io

# Optional dependencies for weather maps - handle gracefully if not installed
try:
    import matplotlib
    matplotlib.use('Agg')  # Use non-interactive backend (must be before pyplot import)
    import matplotlib.pyplot as plt
    import matplotlib.patches as patches
    from matplotlib.backends.backend_agg import FigureCanvasAgg
    from matplotlib.patches import Patch as MplPatch
    MATPLOTLIB_AVAILABLE = True
except (ImportError, AttributeError) as e:
    matplotlib = None
    plt = None
    patches = None
    FigureCanvasAgg = None
    MplPatch = None
    MATPLOTLIB_AVAILABLE = False
    print(f"Warning: Matplotlib not available. Weather map image generation will be disabled. Error: {e}")

try:
    import numpy as np
    NUMPY_AVAILABLE = True
except ImportError:
    np = None
    NUMPY_AVAILABLE = False
    print("Warning: NumPy not available. Some map calculations will be limited.")

class WeatherMapsService:
    """Weather maps and radar integration service"""
    
    def __init__(self, openweather_api_key: str = None, mapbox_api_key: str = None):
        self.openweather_api_key = openweather_api_key
        self.mapbox_api_key = mapbox_api_key
        self.openweather_layers_url = "https://maps.openweathermap.org/maps/2.0"
        self.mapbox_static_url = "https://api.mapbox.com/styles/v1"
        self.radar_url = "https://api.openweathermap.org/data/2.5"
    
    def get_weather_map_layers(self) -> Dict:
        """
        Get available weather map layers.
        
        Returns:
            Available map layers information
        """
        layers = {
            'precipitation': {
                'name': 'Precipitation',
                'description': 'Current precipitation radar',
                'url': f"{self.openweather_layers_url}/precipitation/0/{datetime.now().strftime('%Y-%m-%d')}",
                'opacity_range': [0.1, 1.0],
                'default_opacity': 0.8
            },
            'precipitation_new': {
                'name': 'Precipitation (New)',
                'description': 'Enhanced precipitation radar',
                'url': f"{self.openweather_layers_url}/precipitation_new/0/{datetime.now().strftime('%Y-%m-%d')}",
                'opacity_range': [0.1, 1.0],
                'default_opacity': 0.8
            },
            'clouds': {
                'name': 'Clouds',
                'description': 'Cloud cover satellite imagery',
                'url': f"{self.openweather_layers_url}/clouds/0/{datetime.now().strftime('%Y-%m-%d')}",
                'opacity_range': [0.1, 1.0],
                'default_opacity': 0.7
            },
            'pressure': {
                'name': 'Pressure',
                'description': 'Atmospheric pressure contours',
                'url': f"{self.openweather_layers_url}/pressure/0/{datetime.now().strftime('%Y-%m-%d')}",
                'opacity_range': [0.1, 1.0],
                'default_opacity': 0.6
            },
            'wind': {
                'name': 'Wind Speed',
                'description': 'Wind speed and direction',
                'url': f"{self.openweather_layers_url}/wind/0/{datetime.now().strftime('%Y-%m-%d')}",
                'opacity_range': [0.1, 1.0],
                'default_opacity': 0.7
            },
            'temperature': {
                'name': 'Temperature',
                'description': 'Surface temperature',
                'url': f"{self.openweather_layers_url}/temp/0/{datetime.now().strftime('%Y-%m-%d')}",
                'opacity_range': [0.1, 1.0],
                'default_opacity': 0.6
            }
        }
        
        return {
            'available_layers': layers,
            'base_url': self.openweather_layers_url,
            'api_key_required': bool(self.openweather_api_key)
        }
    
    def generate_weather_map_url(self, layer: str, bounds: List[float], 
                               zoom: int = 5, opacity: float = 0.8) -> Dict:
        """
        Generate weather map URL for specific layer and bounds.
        
        Args:
            layer: Weather layer type
            bounds: Map bounds [lat_min, lon_min, lat_max, lon_max]
            zoom: Zoom level
            opacity: Layer opacity
            
        Returns:
            Map URL and metadata
        """
        try:
            if not self.openweather_api_key:
                return {'error': 'OpenWeatherMap API key required'}
            
            # Validate bounds
            if len(bounds) != 4:
                return {'error': 'Invalid bounds format. Use [lat_min, lon_min, lat_max, lon_max]'}
            
            lat_min, lon_min, lat_max, lon_max = bounds
            
            # Validate coordinates
            if not (-90 <= lat_min <= 90 and -90 <= lat_max <= 90):
                return {'error': 'Invalid latitude bounds'}
            if not (-180 <= lon_min <= 180 and -180 <= lon_max <= 180):
                return {'error': 'Invalid longitude bounds'}
            
            # Calculate center point
            center_lat = (lat_min + lat_max) / 2
            center_lon = (lon_min + lon_max) / 2
            
            # Generate map URL
            map_url = f"{self.openweather_layers_url}/{layer}/{zoom}/{center_lat:.4f}/{center_lon:.4f}"
            
            params = {
                'appid': self.openweather_api_key,
                'opacity': opacity,
                'bbox': f"{lon_min},{lat_min},{lon_max},{lat_max}"
            }
            
            # Build full URL with parameters
            param_string = '&'.join([f"{k}={v}" for k, v in params.items()])
            full_url = f"{map_url}?{param_string}"
            
            return {
                'success': True,
                'map_url': full_url,
                'layer': layer,
                'bounds': bounds,
                'center': [center_lat, center_lon],
                'zoom': zoom,
                'opacity': opacity,
                'generated_at': datetime.now().isoformat()
            }
            
        except Exception as e:
            return {'error': f'Failed to generate map URL: {str(e)}'}
    
    def get_radar_data(self, lat: float, lon: float, radius: int = 50) -> Dict:
        """
        Get radar data for a specific location.
        
        Args:
            lat: Latitude
            lon: Longitude
            radius: Search radius in kilometers
            
        Returns:
            Radar data and imagery
        """
        try:
            if not self.openweather_api_key:
                return {'error': 'OpenWeatherMap API key required'}
            
            # Get current weather with radar information
            url = f"{self.radar_url}/weather"
            params = {
                'lat': lat,
                'lon': lon,
                'appid': self.openweather_api_key,
                'units': 'metric'
            }
            
            response = requests.get(url, params=params, timeout=10)
            response.raise_for_status()
            
            weather_data = response.json()
            
            # Extract radar-relevant information
            radar_info = {
                'location': {
                    'lat': lat,
                    'lon': lon,
                    'city': weather_data.get('name', 'Unknown'),
                    'country': weather_data.get('sys', {}).get('country', '')
                },
                'current_conditions': {
                    'temperature': weather_data.get('main', {}).get('temp', 0),
                    'humidity': weather_data.get('main', {}).get('humidity', 0),
                    'pressure': weather_data.get('main', {}).get('pressure', 0),
                    'visibility': weather_data.get('visibility', 0) / 1000,  # Convert to km
                    'description': weather_data.get('weather', [{}])[0].get('description', ''),
                    'icon': weather_data.get('weather', [{}])[0].get('icon', '')
                },
                'wind': {
                    'speed': weather_data.get('wind', {}).get('speed', 0),
                    'direction': weather_data.get('wind', {}).get('deg', 0),
                    'gust': weather_data.get('wind', {}).get('gust', 0)
                },
                'precipitation': {
                    'rain_1h': weather_data.get('rain', {}).get('1h', 0),
                    'rain_3h': weather_data.get('rain', {}).get('3h', 0),
                    'snow_1h': weather_data.get('snow', {}).get('1h', 0),
                    'snow_3h': weather_data.get('snow', {}).get('3h', 0)
                },
                'clouds': {
                    'coverage': weather_data.get('clouds', {}).get('all', 0)
                },
                'radar_url': self._generate_radar_url(lat, lon, radius),
                'timestamp': datetime.now().isoformat()
            }
            
            return radar_info
            
        except requests.exceptions.RequestException as e:
            return {'error': f'Failed to get radar data: {str(e)}'}
        except Exception as e:
            return {'error': f'Error processing radar data: {str(e)}'}
    
    def _generate_radar_url(self, lat: float, lon: float, radius: int) -> str:
        """Generate radar URL for the location"""
        if not self.openweather_api_key:
            return ""
        
        # Calculate bounds based on radius
        # Rough approximation: 1 degree latitude ≈ 111 km
        import math
        lat_delta = radius / 111
        lon_delta = radius / (111 * math.cos(math.radians(lat)))
        
        lat_min = lat - lat_delta
        lat_max = lat + lat_delta
        lon_min = lon - lon_delta
        lon_max = lon + lon_delta
        
        # Generate precipitation radar URL
        radar_url = f"{self.openweather_layers_url}/precipitation_new/6/{lat:.4f}/{lon:.4f}"
        params = f"appid={self.openweather_api_key}&opacity=0.8&bbox={lon_min},{lat_min},{lon_max},{lat_max}"
        
        return f"{radar_url}?{params}"
    
    def create_weather_map_image(self, cities_data: Dict[str, Dict], 
                               map_type: str = 'temperature', 
                               width: int = 800, height: int = 600) -> Dict:
        """
        Create a custom weather map image with city data.
        
        Args:
            cities_data: Dictionary of city weather data
            map_type: Type of weather map (temperature, precipitation, pressure, wind)
            width: Image width
            height: Image height
            
        Returns:
            Base64 encoded image and metadata
        """
        try:
            if not MATPLOTLIB_AVAILABLE or plt is None:
                return {'error': 'Matplotlib is required for map image generation. Please install matplotlib.'}
            
            # Create matplotlib figure
            fig, ax = plt.subplots(figsize=(width/100, height/100), dpi=100)
            
            # Set up the map
            ax.set_xlim(-180, 180)
            ax.set_ylim(-90, 90)
            ax.set_aspect('equal')
            
            # Add world map background (simplified)
            self._add_world_map_background(ax)
            
            # Plot cities with weather data
            for city, data in cities_data.items():
                if 'error' in data:
                    continue
                
                lat = data.get('latitude', 0)
                lon = data.get('longitude', 0)
                
                if lat == 0 and lon == 0:
                    # Try to get coordinates from city name
                    coords = self._get_city_coordinates(city)
                    if coords:
                        lat, lon = coords
                    else:
                        continue
                
                # Get weather value based on map type
                value = self._get_weather_value(data, map_type)
                color = self._get_weather_color(value, map_type)
                
                # Plot city marker
                ax.scatter(lon, lat, c=color, s=100, alpha=0.8, edgecolors='black', linewidth=1)
                
                # Add city label
                ax.annotate(city, (lon, lat), xytext=(5, 5), 
                           textcoords='offset points', fontsize=8, alpha=0.8)
                
                # Add value label
                value_text = self._format_value_text(value, map_type)
                ax.annotate(value_text, (lon, lat), xytext=(5, -15), 
                           textcoords='offset points', fontsize=7, alpha=0.7)
            
            # Set title and labels
            ax.set_title(f'Weather Map - {map_type.title()}', fontsize=16, fontweight='bold')
            ax.set_xlabel('Longitude', fontsize=12)
            ax.set_ylabel('Latitude', fontsize=12)
            
            # Add grid
            ax.grid(True, alpha=0.3)
            
            # Add colorbar if applicable
            if map_type in ['temperature', 'pressure', 'wind']:
                self._add_colorbar(ax, map_type)
            
            # Add legend
            self._add_map_legend(ax, map_type)
            
            # Add attribution
            ax.text(0.02, 0.02, 'Weather Chatbot - RSK World (https://rskworld.in)', 
                   transform=ax.transAxes, fontsize=8, alpha=0.7)
            
            # Save figure to buffer
            buffer = io.BytesIO()
            plt.savefig(buffer, format='png', dpi=100, bbox_inches='tight')
            buffer.seek(0)
            
            # Convert to base64
            image_base64 = base64.b64encode(buffer.getvalue()).decode()
            
            plt.close()
            
            return {
                'success': True,
                'image_base64': image_base64,
                'map_type': map_type,
                'cities_count': len(cities_data),
                'dimensions': {'width': width, 'height': height},
                'generated_at': datetime.now().isoformat()
            }
            
        except Exception as e:
            return {'error': f'Failed to create weather map image: {str(e)}'}
    
    def _add_world_map_background(self, ax):
        """Add simplified world map background"""
        if not MATPLOTLIB_AVAILABLE or patches is None:
            return
        
        # Create a simple world map outline
        world_coords = [
            # Simplified continent outlines (very basic)
            [-100, 40, -80, 50],  # North America (simplified)
            [-10, 35, 40, 70],   # Europe/Africa (simplified)
            [70, 10, 140, 50],   # Asia (simplified)
            [110, -45, 155, -10], # Australia (simplified)
            [-80, -55, -35, -15], # South America (simplified)
        ]
        
        for coords in world_coords:
            rect = patches.Rectangle((coords[0], coords[2]), 
                                   coords[1] - coords[0], 
                                   coords[3] - coords[2],
                                   linewidth=0.5, edgecolor='gray', 
                                   facecolor='lightgray', alpha=0.3)
            ax.add_patch(rect)
    
    def _get_city_coordinates(self, city: str) -> Optional[Tuple[float, float]]:
        """Get city coordinates (simplified - would use geocoding API in production)"""
        # This is a very simplified version - in production, use a proper geocoding service
        city_coords = {
            'london': (51.5074, -0.1278),
            'new york': (40.7128, -74.0060),
            'paris': (48.8566, 2.3522),
            'tokyo': (35.6762, 139.6503),
            'delhi': (28.7041, 77.1025),
            'mumbai': (19.0760, 72.8777),
            'kolkata': (22.5726, 88.3639),
            'sydney': (-33.8688, 151.2093),
            'moscow': (55.7558, 37.6173),
            'beijing': (39.9042, 116.4074),
            'dubai': (25.2048, 55.2708),
            'singapore': (1.3521, 103.8198)
        }
        
        return city_coords.get(city.lower())
    
    def _get_weather_value(self, data: Dict, map_type: str) -> float:
        """Get weather value based on map type"""
        if map_type == 'temperature':
            return data.get('temperature', 0)
        elif map_type == 'precipitation':
            return data.get('rain_1h', 0) + data.get('snow_1h', 0)
        elif map_type == 'pressure':
            return data.get('pressure', 1013)
        elif map_type == 'wind':
            return data.get('wind_speed', 0)
        elif map_type == 'humidity':
            return data.get('humidity', 0)
        else:
            return 0
    
    def _get_weather_color(self, value: float, map_type: str) -> str:
        """Get color based on weather value and type"""
        if map_type == 'temperature':
            if value < -10:
                return 'darkblue'
            elif value < 0:
                return 'blue'
            elif value < 10:
                return 'lightblue'
            elif value < 20:
                return 'green'
            elif value < 30:
                return 'yellow'
            elif value < 40:
                return 'orange'
            else:
                return 'red'
        
        elif map_type == 'precipitation':
            if value == 0:
                return 'white'
            elif value < 2.5:
                return 'lightblue'
            elif value < 10:
                return 'blue'
            else:
                return 'darkblue'
        
        elif map_type == 'pressure':
            if value < 1000:
                return 'red'
            elif value < 1010:
                return 'orange'
            elif value < 1020:
                return 'green'
            else:
                return 'blue'
        
        elif map_type == 'wind':
            if value < 5:
                return 'green'
            elif value < 15:
                return 'yellow'
            elif value < 25:
                return 'orange'
            else:
                return 'red'
        
        elif map_type == 'humidity':
            if value < 30:
                return 'brown'
            elif value < 60:
                return 'green'
            else:
                return 'blue'
        
        else:
            return 'gray'
    
    def _format_value_text(self, value: float, map_type: str) -> str:
        """Format value text for display"""
        if map_type == 'temperature':
            return f"{value:.1f}°C"
        elif map_type == 'precipitation':
            return f"{value:.1f}mm"
        elif map_type == 'pressure':
            return f"{value:.0f}hPa"
        elif map_type == 'wind':
            return f"{value:.1f}m/s"
        elif map_type == 'humidity':
            return f"{value:.0f}%"
        else:
            return f"{value:.1f}"
    
    def _add_colorbar(self, ax, map_type: str):
        """Add colorbar to the map"""
        if not MATPLOTLIB_AVAILABLE or plt is None:
            return
        # This is a simplified colorbar implementation
        # In production, use proper matplotlib colorbar
        # For now, we skip colorbar as it requires additional matplotlib configuration
        pass
    
    def _add_map_legend(self, ax, map_type: str):
        """Add legend to the map"""
        if not MATPLOTLIB_AVAILABLE or patches is None:
            return
        
        legend_elements = []
        
        if map_type == 'temperature':
            legend_elements = [
                ('< -10°C', 'darkblue'),
                ('-10°C to 0°C', 'blue'),
                ('0°C to 10°C', 'lightblue'),
                ('10°C to 20°C', 'green'),
                ('20°C to 30°C', 'yellow'),
                ('30°C to 40°C', 'orange'),
                ('> 40°C', 'red')
            ]
        elif map_type == 'precipitation':
            legend_elements = [
                ('No precipitation', 'white'),
                ('Light (< 2.5mm)', 'lightblue'),
                ('Moderate (2.5-10mm)', 'blue'),
                ('Heavy (> 10mm)', 'darkblue')
            ]
        elif map_type == 'wind':
            legend_elements = [
                ('Calm (< 5m/s)', 'green'),
                ('Moderate (5-15m/s)', 'yellow'),
                ('Strong (15-25m/s)', 'orange'),
                ('Severe (> 25m/s)', 'red')
            ]
        
        # Add legend to plot
        if legend_elements and MATPLOTLIB_AVAILABLE and MplPatch is not None:
            try:
                legend_patches = [MplPatch(facecolor=color, label=label) 
                                for label, color in legend_elements]
                
                ax.legend(handles=legend_patches, loc='upper right', fontsize=8)
            except Exception as e:
                # Fallback if legend creation fails
                print(f"Warning: Could not create legend: {e}")
    
    def get_weather_animation_frames(self, city: str, hours: int = 24) -> Dict:
        """
        Get weather animation frames for a city.
        
        Args:
            city: City name
            hours: Number of hours for animation
            
        Returns:
            Animation frame URLs
        """
        try:
            if not self.openweather_api_key:
                return {'error': 'OpenWeatherMap API key required'}
            
            # Get forecast data
            forecast_url = f"{self.radar_url}/forecast"
            params = {
                'q': city,
                'appid': self.openweather_api_key,
                'units': 'metric',
                'cnt': hours // 3  # 3-hour intervals
            }
            
            response = requests.get(forecast_url, params=params, timeout=10)
            response.raise_for_status()
            
            forecast_data = response.json()
            
            # Generate animation frames
            frames = []
            for i, item in enumerate(forecast_data['list'][:hours // 3]):
                frame_data = {
                    'timestamp': item['dt_txt'],
                    'temperature': item['main']['temp'],
                    'description': item['weather'][0]['description'],
                    'icon': item['weather'][0]['icon'],
                    'humidity': item['main']['humidity'],
                    'pressure': item['main']['pressure'],
                    'wind_speed': item['wind']['speed'],
                    'frame_number': i + 1
                }
                
                # Generate map URL for this frame
                city_coords = self._get_city_coordinates(city)
                if city_coords:
                    lat, lon = city_coords
                    map_url = self._generate_radar_url(lat, lon, 100)
                    frame_data['map_url'] = map_url
                
                frames.append(frame_data)
            
            return {
                'success': True,
                'city': city,
                'total_frames': len(frames),
                'hours_covered': hours,
                'frames': frames,
                'generated_at': datetime.now().isoformat()
            }
            
        except requests.exceptions.RequestException as e:
            return {'error': f'Failed to get animation frames: {str(e)}'}
        except Exception as e:
            return {'error': f'Error processing animation: {str(e)}'}
    
    def create_weather_comparison_map(self, cities_comparison: Dict) -> Dict:
        """
        Create a weather comparison map between multiple cities.
        
        Args:
            cities_comparison: Comparison data from analytics module
            
        Returns:
            Comparison map image
        """
        try:
            if 'error' in cities_comparison:
                return cities_comparison
            
            # Extract cities and their data
            cities_data = {}
            for city in cities_comparison.get('cities_compared', []):
                # Get current weather for each city (simplified)
                cities_data[city] = {
                    'temperature': cities_comparison.get('metrics', {}).get('temperature', {}).get(city, 20),
                    'humidity': cities_comparison.get('metrics', {}).get('humidity', {}).get(city, 50),
                    'wind_speed': cities_comparison.get('metrics', {}).get('wind_speed', {}).get(city, 5),
                    'pressure': cities_comparison.get('metrics', {}).get('pressure', {}).get(city, 1013)
                }
            
            # Create comparison map
            return self.create_weather_map_image(cities_data, map_type='temperature')
            
        except Exception as e:
            return {'error': f'Failed to create comparison map: {str(e)}'}
    
    def get_map_tile_url(self, lat: float, lon: float, zoom: int, 
                        layer: str = 'precipitation') -> Dict:
        """
        Get map tile URL for specific coordinates.
        
        Args:
            lat: Latitude
            lon: Longitude
            zoom: Zoom level
            layer: Map layer
            
        Returns:
            Tile URL
        """
        try:
            if not self.openweather_api_key:
                return {'error': 'OpenWeatherMap API key required'}
            
            # Convert lat/lon to tile coordinates
            import math
            x = int((lon + 180) / 360 * (2 ** zoom))
            y = int((1 - math.log(math.tan(math.radians(lat)) + 1 / math.cos(math.radians(lat))) / math.pi) / 2 * (2 ** zoom))
            
            # Generate tile URL
            tile_url = f"{self.openweather_layers_url}/{layer}/{zoom}/{x}/{y}"
            params = f"appid={self.openweather_api_key}&opacity=0.8"
            
            return {
                'success': True,
                'tile_url': f"{tile_url}?{params}",
                'tile_coords': {'x': x, 'y': y, 'z': zoom},
                'lat_lon': {'lat': lat, 'lon': lon},
                'layer': layer
            }
            
        except Exception as e:
            return {'error': f'Failed to generate tile URL: {str(e)}'}
    
    def export_map_data(self, map_data: Dict, format_type: str = 'json') -> str:
        """
        Export map data in different formats.
        
        Args:
            map_data: Map data dictionary
            format_type: Export format (json, csv, geojson)
            
        Returns:
            Exported data string
        """
        if format_type == 'json':
            return json.dumps(map_data, indent=2, default=str)
        elif format_type == 'csv':
            return self._export_to_csv(map_data)
        elif format_type == 'geojson':
            return self._export_to_geojson(map_data)
        else:
            return json.dumps(map_data, indent=2, default=str)
    
    def _export_to_csv(self, map_data: Dict) -> str:
        """Export map data to CSV format"""
        import csv
        import io
        
        output = io.StringIO()
        writer = csv.writer(output)
        
        # Write header
        writer.writerow(['City', 'Latitude', 'Longitude', 'Temperature', 'Humidity', 'Wind Speed', 'Pressure'])
        
        # Write data (simplified - would need actual city data)
        if 'cities_data' in map_data:
            for city, data in map_data['cities_data'].items():
                writer.writerow([
                    city,
                    data.get('latitude', ''),
                    data.get('longitude', ''),
                    data.get('temperature', ''),
                    data.get('humidity', ''),
                    data.get('wind_speed', ''),
                    data.get('pressure', '')
                ])
        
        return output.getvalue()
    
    def _export_to_geojson(self, map_data: Dict) -> str:
        """Export map data to GeoJSON format"""
        features = []
        
        if 'cities_data' in map_data:
            for city, data in map_data['cities_data'].items():
                if 'latitude' in data and 'longitude' in data:
                    feature = {
                        "type": "Feature",
                        "geometry": {
                            "type": "Point",
                            "coordinates": [data['longitude'], data['latitude']]
                        },
                        "properties": {
                            "city": city,
                            "temperature": data.get('temperature', 0),
                            "humidity": data.get('humidity', 0),
                            "wind_speed": data.get('wind_speed', 0),
                            "pressure": data.get('pressure', 0)
                        }
                    }
                    features.append(feature)
        
        geojson = {
            "type": "FeatureCollection",
            "features": features
        }
        
        return json.dumps(geojson, indent=2)

768 lines•29.6 KB

python

utils/advanced_nlp.py

Raw Download

#!/usr/bin/env python3
"""
Weather Chatbot Advanced NLP Module
====================================

Author: RSK World (https://rskworld.in)
Founded by: Molla Samser
Designer & Tester: Rima Khatun
Contact: +91 93305 39277, hello@rskworld.in, support@rskworld.in
Location: Nutanhat, Mongolkote, Purba Burdwan, West Bengal, India, 713147
Year: 2026

Description: Advanced natural language processing with sentiment analysis
"""

import re
import json
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
import openai

# Optional dependencies - handle gracefully if not installed
try:
    from textblob import TextBlob
    TEXTBLOB_AVAILABLE = True
except ImportError:
    TextBlob = None
    TEXTBLOB_AVAILABLE = False
    print("Warning: TextBlob not available. Sentiment analysis will be limited.")

try:
    import spacy
    SPACY_AVAILABLE = True
except ImportError:
    spacy = None
    SPACY_AVAILABLE = False
    print("Warning: spaCy not available. Advanced NLP features will be limited.")

# Try to load spaCy model
nlp = None
if SPACY_AVAILABLE:
    try:
        nlp = spacy.load("en_core_web_sm")
    except OSError:
        nlp = None
        print("Warning: spaCy model 'en_core_web_sm' not found. Some NLP features will be limited.")

class AdvancedNLP:
    """Advanced natural language processing for weather queries"""
    
    def __init__(self, openai_api_key: str = None):
        self.openai_api_key = openai_api_key
        self.openai_client = None
        if openai_api_key:
            try:
                self.openai_client = openai.OpenAI(api_key=openai_api_key)
            except Exception:
                # Fallback to old API if new version not available
                openai.api_key = openai_api_key
        
        # Weather-related vocabulary
        self.weather_entities = {
            'cities': [
                'london', 'new york', 'paris', 'tokyo', 'delhi', 'mumbai', 'kolkata',
                'chennai', 'bangalore', 'sydney', 'moscow', 'beijing', 'dubai',
                'singapore', 'hong kong', 'los angeles', 'chicago', 'boston',
                'san francisco', 'seattle', 'toronto', 'vancouver', 'montreal'
            ],
            'weather_conditions': [
                'sunny', 'cloudy', 'rainy', 'snowy', 'stormy', 'windy', 'foggy',
                'clear', 'overcast', 'partly cloudy', 'mostly cloudy', 'drizzle',
                'thunderstorm', 'blizzard', 'hail', 'sleet', 'mist', 'haze'
            ],
            'time_expressions': [
                'today', 'tomorrow', 'yesterday', 'now', 'current', 'right now',
                'morning', 'afternoon', 'evening', 'night', 'tonight',
                'this week', 'next week', 'last week', 'weekend',
                'monday', 'tuesday', 'wednesday', 'thursday', 'friday', 'saturday', 'sunday',
                'in 1 hour', 'in 2 hours', 'in 3 hours', 'in 6 hours', 'in 12 hours',
                'in 1 day', 'in 2 days', 'in 3 days', 'in 5 days', 'in a week'
            ],
            'query_types': [
                'weather', 'forecast', 'temperature', 'rain', 'snow', 'wind',
                'humidity', 'pressure', 'visibility', 'alerts', 'warnings',
                'conditions', 'climate', 'outlook', 'prediction'
            ],
            'comparative_words': [
                'compare', 'versus', 'vs', 'difference', 'better', 'worse',
                'hotter', 'colder', 'warmer', 'cooler', 'drier', 'wetter',
                'windier', 'calmer', 'sunnier', 'cloudier'
            ]
        }
        
        # Intent patterns
        self.intent_patterns = {
            'current_weather': [
                r'weather (?:in|at|for) (.+)',
                r'what(?:\'s| is) the weather (?:in|at|for) (.+)',
                r'current weather (?:in|for) (.+)',
                r'how(?:\'s| is) the weather (?:in|at) (.+)'
            ],
            'forecast': [
                r'forecast (?:for|in) (.+)',
                r'weather forecast (?:for|in) (.+)',
                r'what(?:\'s| is) the forecast (?:for|in) (.+)',
                r'will it (?:rain|snow) (?:in|at) (.+)',
                r'(?:temperature|temp) (?:in|for) (.+) (?:tomorrow|today|tonight)'
            ],
            'alerts': [
                r'weather alerts (?:for|in) (.+)',
                r'any (?:alerts|warnings) (?:for|in) (.+)',
                'severe weather (.+)',
                r'weather warnings (?:for|in) (.+)'
            ],
            'comparison': [
                r'compare (.+) (?:and|vs|versus) (.+)',
                r'(.+) (?:vs|versus) (.+) weather',
                r'difference between (.+) and (.+) weather',
                r'which (?:is|has) (?:better|worse) weather (.+) or (.+)'
            ],
            'historical': [
                r'weather history (?:for|in) (.+)',
                r'past weather (?:in|for) (.+)',
                r'weather (?:last|yesterday) (?:in|for) (.+)'
            ]
        }
    
    def analyze_query(self, query: str) -> Dict:
        """
        Comprehensive analysis of user query.
        
        Args:
            query: User's natural language query
            
        Returns:
            Detailed analysis result
        """
        try:
            # Basic preprocessing
            clean_query = self._preprocess_query(query)
            
            # Extract entities
            entities = self._extract_entities(clean_query)
            
            # Determine intent
            intent = self._determine_intent(clean_query)
            
            # Extract temporal information
            temporal_info = self._extract_temporal_info(clean_query)
            
            # Sentiment analysis
            sentiment = self._analyze_sentiment(clean_query)
            
            # Generate query context
            context = self._generate_query_context(entities, intent, temporal_info)
            
            # Confidence scoring
            confidence = self._calculate_confidence(entities, intent, temporal_info)
            
            return {
                'original_query': query,
                'cleaned_query': clean_query,
                'intent': intent,
                'entities': entities,
                'temporal_info': temporal_info,
                'sentiment': sentiment,
                'context': context,
                'confidence': confidence,
                'suggestions': self._generate_suggestions(entities, intent),
                'processed_at': datetime.now().isoformat()
            }
            
        except Exception as e:
            return {
                'error': f'Query analysis failed: {str(e)}',
                'original_query': query,
                'fallback_processed': True
            }
    
    def _preprocess_query(self, query: str) -> str:
        """Preprocess and clean the query"""
        # Convert to lowercase
        clean = query.lower().strip()
        
        # Remove extra whitespace
        clean = re.sub(r'\s+', ' ', clean)
        
        # Handle contractions
        contractions = {
            "what's": "what is",
            "how's": "how is",
            "it's": "it is",
            "that's": "that is",
            "there's": "there is",
            "where's": "where is",
            "when's": "when is",
            "why's": "why is",
            "who's": "who is",
            "let's": "let us",
            "don't": "do not",
            "doesn't": "does not",
            "didn't": "did not",
            "won't": "will not",
            "wouldn't": "would not",
            "can't": "cannot",
            "couldn't": "could not",
            "shouldn't": "should not",
            "mightn't": "might not",
            "mustn't": "must not"
        }
        
        for contraction, expansion in contractions.items():
            clean = clean.replace(contraction, expansion)
        
        # Remove punctuation except for meaningful ones
        clean = re.sub(r'[^\w\s\?\.\!\,\-]', '', clean)
        
        return clean
    
    def _extract_entities(self, query: str) -> Dict:
        """Extract entities from the query"""
        entities = {
            'cities': [],
            'weather_conditions': [],
            'time_expressions': [],
            'query_types': [],
            'comparative_words': [],
            'numbers': [],
            'unknown_locations': []
        }
        
        # Extract cities
        for city in self.weather_entities['cities']:
            if city in query:
                entities['cities'].append(city)
        
        # Extract weather conditions
        for condition in self.weather_entities['weather_conditions']:
            if condition in query:
                entities['weather_conditions'].append(condition)
        
        # Extract time expressions
        for time_expr in self.weather_entities['time_expressions']:
            if time_expr in query:
                entities['time_expressions'].append(time_expr)
        
        # Extract query types
        for query_type in self.weather_entities['query_types']:
            if query_type in query:
                entities['query_types'].append(query_type)
        
        # Extract comparative words
        for comp_word in self.weather_entities['comparative_words']:
            if comp_word in query:
                entities['comparative_words'].append(comp_word)
        
        # Extract numbers
        numbers = re.findall(r'\b\d+\b', query)
        entities['numbers'] = [int(n) for n in numbers]
        
        # Use spaCy for advanced entity extraction if available
        if SPACY_AVAILABLE and nlp is not None:
            try:
                doc = nlp(query)
                
                # Extract GPE (Geopolitical Entity) for cities
                for ent in doc.ents:
                    if ent.label_ == 'GPE' and ent.text.lower() not in entities['cities']:
                        entities['unknown_locations'].append(ent.text)
            except Exception as e:
                # Silently fail if spaCy processing fails
                pass
        
        return entities
    
    def _determine_intent(self, query: str) -> Dict:
        """Determine the primary intent of the query"""
        intent_scores = {}
        
        for intent_name, patterns in self.intent_patterns.items():
            score = 0
            for pattern in patterns:
                if re.search(pattern, query, re.IGNORECASE):
                    score += 1
            intent_scores[intent_name] = score
        
        # Find the intent with highest score
        if intent_scores:
            primary_intent = max(intent_scores, key=intent_scores.get)
            confidence = intent_scores[primary_intent] / len(self.intent_patterns[primary_intent])
        else:
            primary_intent = 'general_query'
            confidence = 0.5
        
        return {
            'primary_intent': primary_intent,
            'confidence': confidence,
            'all_scores': intent_scores
        }
    
    def _extract_temporal_info(self, query: str) -> Dict:
        """Extract temporal information from the query"""
        temporal_info = {
            'time_reference': 'current',
            'specific_time': None,
            'relative_time': None,
            'duration': None,
            'time_range': None
        }
        
        # Check for time references
        if any(word in query for word in ['current', 'now', 'right now', 'today']):
            temporal_info['time_reference'] = 'current'
        elif 'tomorrow' in query:
            temporal_info['time_reference'] = 'tomorrow'
        elif 'yesterday' in query:
            temporal_info['time_reference'] = 'yesterday'
        elif any(word in query for word in ['forecast', 'future', 'next']):
            temporal_info['time_reference'] = 'forecast'
        elif any(word in query for word in ['past', 'history', 'last', 'previous']):
            temporal_info['time_reference'] = 'historical'
        
        # Extract specific times
        time_patterns = [
            (r'(\d{1,2})\s*(am|pm)', 'specific_time'),
            (r'(\d{1,2}):(\d{2})\s*(am|pm)?', 'specific_time'),
            (r'(morning|afternoon|evening|night|tonight)', 'specific_time'),
        ]
        
        for pattern, time_type in time_patterns:
            match = re.search(pattern, query)
            if match:
                temporal_info[time_type] = match.group(0)
                break
        
        # Extract durations
        duration_patterns = [
            (r'(\d+)\s*(hour|hours|hr|hrs)', 'hours'),
            (r'(\d+)\s*(day|days)', 'days'),
            (r'(\d+)\s*(week|weeks)', 'weeks')
        ]
        
        for pattern, unit in duration_patterns:
            match = re.search(pattern, query)
            if match:
                temporal_info['duration'] = {
                    'value': int(match.group(1)),
                    'unit': unit
                }
                break
        
        return temporal_info
    
    def _analyze_sentiment(self, query: str) -> Dict:
        """Analyze sentiment of the query"""
        try:
            if not TEXTBLOB_AVAILABLE or TextBlob is None:
                # Fallback sentiment analysis without TextBlob
                return {
                    'polarity': 0.0,
                    'subjectivity': 0.5,
                    'sentiment_label': 'neutral',
                    'emotional_tone': 'neutral',
                    'confidence': 0.5
                }
            
            # Use TextBlob for sentiment analysis
            blob = TextBlob(query)
            polarity = blob.sentiment.polarity
            subjectivity = blob.sentiment.subjectivity
            
            # Categorize sentiment
            if polarity > 0.1:
                sentiment_label = 'positive'
            elif polarity < -0.1:
                sentiment_label = 'negative'
            else:
                sentiment_label = 'neutral'
            
            # Determine emotional tone
            emotional_tone = self._determine_emotional_tone(query, polarity)
            
            return {
                'polarity': polarity,
                'subjectivity': subjectivity,
                'sentiment_label': sentiment_label,
                'emotional_tone': emotional_tone,
                'confidence': abs(polarity)
            }
            
        except Exception as e:
            return {
                'error': f'Sentiment analysis failed: {str(e)}',
                'polarity': 0,
                'subjectivity': 0,
                'sentiment_label': 'neutral',
                'emotional_tone': 'neutral',
                'confidence': 0
            }
    
    def _determine_emotional_tone(self, query: str, polarity: float) -> str:
        """Determine emotional tone based on keywords and polarity"""
        positive_words = ['great', 'excellent', 'perfect', 'beautiful', 'nice', 'good', 'wonderful', 'amazing']
        negative_words = ['terrible', 'awful', 'horrible', 'bad', 'worst', 'hate', 'dislike', 'ugly']
        urgent_words = ['urgent', 'emergency', 'immediately', 'asap', 'critical', 'important']
        
        query_lower = query.lower()
        
        if any(word in query_lower for word in urgent_words):
            return 'urgent'
        elif polarity > 0.3 and any(word in query_lower for word in positive_words):
            return 'enthusiastic'
        elif polarity < -0.3 and any(word in query_lower for word in negative_words):
            return 'concerned'
        elif polarity > 0.1:
            return 'positive'
        elif polarity < -0.1:
            return 'negative'
        else:
            return 'neutral'
    
    def _generate_query_context(self, entities: Dict, intent: Dict, temporal_info: Dict) -> Dict:
        """Generate comprehensive query context"""
        context = {
            'primary_location': None,
            'secondary_locations': [],
            'weather_focus': None,
            'time_frame': temporal_info['time_reference'],
            'query_complexity': 'simple',
            'expected_response_type': 'weather_data'
        }
        
        # Determine primary location
        if entities['cities']:
            context['primary_location'] = entities['cities'][0]
            context['secondary_locations'] = entities['cities'][1:]
        elif entities['unknown_locations']:
            context['primary_location'] = entities['unknown_locations'][0]
        
        # Determine weather focus
        if entities['weather_conditions']:
            context['weather_focus'] = entities['weather_conditions'][0]
        
        # Determine query complexity
        if len(entities['cities']) > 1 or entities['comparative_words']:
            context['query_complexity'] = 'complex'
            context['expected_response_type'] = 'comparison'
        elif intent['primary_intent'] == 'forecast':
            context['query_complexity'] = 'moderate'
            context['expected_response_type'] = 'forecast'
        elif intent['primary_intent'] == 'alerts':
            context['query_complexity'] = 'moderate'
            context['expected_response_type'] = 'alerts'
        
        return context
    
    def _calculate_confidence(self, entities: Dict, intent: Dict, temporal_info: Dict) -> float:
        """Calculate overall confidence in the analysis"""
        confidence_factors = []
        
        # Entity extraction confidence
        entity_confidence = 0
        total_entities = sum(len(v) for v in entities.values())
        if total_entities > 0:
            entity_confidence = min(total_entities / 3, 1.0)  # Cap at 1.0
        confidence_factors.append(entity_confidence)
        
        # Intent confidence
        confidence_factors.append(intent['confidence'])
        
        # Temporal information confidence
        temporal_confidence = 0.5  # Base confidence
        if temporal_info['time_reference'] != 'current':
            temporal_confidence = 0.8
        confidence_factors.append(temporal_confidence)
        
        # Overall confidence (average of factors)
        overall_confidence = sum(confidence_factors) / len(confidence_factors)
        
        return round(overall_confidence, 2)
    
    def _generate_suggestions(self, entities: Dict, intent: Dict) -> List[str]:
        """Generate suggestions for improving the query"""
        suggestions = []
        
        # Location suggestions
        if not entities['cities'] and not entities['unknown_locations']:
            suggestions.append("Please specify a city name for weather information")
        
        # Intent suggestions
        if intent['confidence'] < 0.5:
            suggestions.append("Try being more specific about what weather information you need")
        
        # Time suggestions
        if intent['primary_intent'] == 'forecast' and not any(
            word in ' '.join(entities['time_expressions']) 
            for word in ['tomorrow', 'today', 'week', 'day']
        ):
            suggestions.append("Specify when you want the forecast (e.g., 'tomorrow', 'this week')")
        
        return suggestions
    
    def enhance_with_openai(self, query: str, analysis: Dict) -> Dict:
        """Enhance analysis using OpenAI GPT"""
        if not self.openai_api_key:
            return analysis
        
        try:
            prompt = f"""
            Analyze this weather query and provide enhanced understanding:
            
            Query: "{query}"
            
            Current Analysis: {json.dumps(analysis, indent=2)}
            
            Please provide:
            1. Enhanced intent classification
            2. Better entity extraction
            3. Contextual understanding
            4. Suggested follow-up questions
            5. Query improvement suggestions
            
            Respond in JSON format with these keys:
            - enhanced_intent
            - refined_entities
            - contextual_understanding
            - follow_up_questions
            - improvement_suggestions
            """
            
            # Use new API client if available, otherwise fallback to old API
            if self.openai_client:
                response = self.openai_client.chat.completions.create(
                    model="gpt-3.5-turbo",
                    messages=[
                        {"role": "system", "content": "You are an expert weather query analyzer."},
                        {"role": "user", "content": prompt}
                    ],
                    max_tokens=500,
                    temperature=0.3
                )
                content = response.choices[0].message.content.strip()
            else:
                # Fallback to old OpenAI API format
                try:
                    response = openai.ChatCompletion.create(
                        model="gpt-3.5-turbo",
                        messages=[
                            {"role": "system", "content": "You are an expert weather query analyzer."},
                            {"role": "user", "content": prompt}
                        ],
                        max_tokens=500,
                        temperature=0.3
                    )
                    content = response.choices[0].message.content.strip()
                except AttributeError:
                    # Old API format might be different
                    analysis['enhancement_error'] = 'OpenAI API format not supported'
                    return analysis
            
            # Try to parse JSON response
            try:
                enhanced_data = json.loads(content)
                
                # Merge enhanced data with original analysis
                analysis['enhanced'] = enhanced_data
                analysis['enhanced_at'] = datetime.now().isoformat()
            except json.JSONDecodeError:
                # If response is not JSON, store raw response
                analysis['enhancement_error'] = 'Could not parse OpenAI response as JSON'
                analysis['enhanced_raw'] = content
            
            return analysis
            
        except Exception as e:
            analysis['enhancement_error'] = str(e)
            return analysis
    
    def generate_conversation_response(self, query: str, weather_data: Dict, analysis: Dict) -> str:
        """Generate natural conversation response based on analysis"""
        try:
            intent = analysis.get('intent', {}).get('primary_intent', 'general_query')
            sentiment = analysis.get('sentiment', {})
            emotional_tone = sentiment.get('emotional_tone', 'neutral')
            
            # Base response based on intent
            if intent == 'current_weather':
                response = self._generate_current_weather_response(weather_data, emotional_tone)
            elif intent == 'forecast':
                response = self._generate_forecast_response(weather_data, emotional_tone)
            elif intent == 'alerts':
                response = self._generate_alerts_response(weather_data, emotional_tone)
            elif intent == 'comparison':
                response = self._generate_comparison_response(weather_data, emotional_tone)
            else:
                response = self._generate_general_response(weather_data, emotional_tone)
            
            # Add contextual elements
            if emotional_tone == 'enthusiastic':
                response = "Great question! " + response
            elif emotional_tone == 'concerned':
                response = "I understand your concern. " + response
            elif emotional_tone == 'urgent':
                response = "I'll help you right away. " + response
            
            # Add follow-up suggestions
            suggestions = self._generate_follow_up_suggestions(analysis, weather_data)
            if suggestions:
                response += "\n\n" + " ".join(suggestions)
            
            return response
            
        except Exception as e:
            return f"I apologize, but I encountered an error while generating my response: {str(e)}"
    
    def _generate_current_weather_response(self, weather_data: Dict, emotional_tone: str) -> str:
        """Generate response for current weather queries"""
        if 'error' in weather_data:
            return f"I'm sorry, I couldn't get the weather information. {weather_data['error']}"
        
        city = weather_data.get('city', 'Unknown')
        temp = weather_data.get('temperature', 0)
        description = weather_data.get('description', 'unknown conditions')
        humidity = weather_data.get('humidity', 0)
        wind_speed = weather_data.get('wind_speed', 0)
        
        response = f"Currently in {city}, it's {temp}°C with {description}. "
        response += f"The humidity is {humidity}% and wind speed is {wind_speed} m/s."
        
        return response
    
    def _generate_forecast_response(self, weather_data: Dict, emotional_tone: str) -> str:
        """Generate response for forecast queries"""
        if 'forecasts' not in weather_data:
            return "I'm sorry, I couldn't get the forecast information."
        
        city = weather_data.get('city', 'Unknown')
        forecasts = weather_data['forecasts'][:3]  # First 3 forecasts
        
        response = f"Here's the forecast for {city}:\n"
        
        for forecast in forecasts:
            datetime_str = forecast.get('datetime', '')
            temp = forecast.get('temperature', 0)
            desc = forecast.get('description', '')
            
            # Format datetime
            try:
                dt = datetime.fromisoformat(datetime_str.replace('Z', '+00:00'))
                time_str = dt.strftime('%A, %I %p')
            except:
                time_str = datetime_str
            
            response += f"• {time_str}: {temp}°C, {desc}\n"
        
        return response
    
    def _generate_alerts_response(self, weather_data: Dict, emotional_tone: str) -> str:
        """Generate response for weather alerts queries"""
        if not weather_data.get('alerts'):
            return f"Good news! There are no active weather alerts."
        
        alerts = weather_data['alerts']
        response = f"There are {len(alerts)} active weather alerts:\n"
        
        for alert in alerts:
            event = alert.get('event', 'Weather Alert')
            description = alert.get('description', '')
            response += f"• {event}: {description}\n"
        
        return response
    
    def _generate_comparison_response(self, weather_data: Dict, emotional_tone: str) -> str:
        """Generate response for comparison queries"""
        # This would need to be implemented based on comparison data structure
        return "Here's the weather comparison between the cities you requested."
    
    def _generate_general_response(self, weather_data: Dict, emotional_tone: str) -> str:
        """Generate general weather response"""
        if 'error' in weather_data:
            return f"I'm sorry, I couldn't process your weather request. {weather_data['error']}"
        
        return self._generate_current_weather_response(weather_data, emotional_tone)
    
    def _generate_follow_up_suggestions(self, analysis: Dict, weather_data: Dict) -> List[str]:
        """Generate follow-up suggestions based on context"""
        suggestions = []
        
        intent = analysis.get('intent', {}).get('primary_intent', 'general_query')
        entities = analysis.get('entities', {})
        
        if intent == 'current_weather':
            suggestions.append("Would you like to see the forecast for tomorrow?")
            suggestions.append("Are there any weather alerts for this area?")
        elif intent == 'forecast':
            suggestions.append("Would you like to compare this with another city?")
            suggestions.append("Do you need historical weather data?")
        elif entities['cities']:
            city = entities['cities'][0] if entities['cities'] else 'this location'
            suggestions.append(f"Would you like to save {city.title()} to your favorites?")
        
        return suggestions[:2]  # Limit to 2 suggestions

703 lines•28.4 KB

python

Theme Settings

Color Scheme

Display Options

Font Size