Create part1_data.py

part1_data.py

@@ -0,0 +1,175 @@
+# part1_data.py
+
+import os
+import numpy as np
+import pandas as pd
+from datetime import datetime, timedelta
+import requests
+from geopy.geocoders import Nominatim
+from geopy.exc import GeocoderTimedOut
+from scipy import stats
+
+# Get API key from environment variable
+OPENWEATHER_API_KEY = os.getenv('OPENWEATHER_API_KEY', 'default_key')
+
+class TobaccoAnalyzer:
+    def __init__(self):
+        self.api_key = OPENWEATHER_API_KEY
+        self.optimal_conditions = {
+            'temperature': {'min': 20, 'max': 30},
+            'humidity': {'min': 60, 'max': 80},
+            'rainfall': {'min': 500/365, 'max': 1200/365}
+        }
+        self.geolocator = Nominatim(user_agent="tobacco_analyzer")
+        self.seasons = {
+            1: 'Winter', 2: 'Winter', 3: 'Spring',
+            4: 'Spring', 5: 'Spring', 6: 'Summer',
+            7: 'Summer', 8: 'Summer', 9: 'Fall',
+            10: 'Fall', 11: 'Fall', 12: 'Winter'
+        }
+
+    def geocode_location(self, location_name):
+        """Convert location name to coordinates"""
+        try:
+            location = self.geolocator.geocode(location_name)
+            if location:
+                return {
+                    'lat': location.latitude,
+                    'lon': location.longitude,
+                    'address': location.address
+                }
+            return None
+        except GeocoderTimedOut:
+            return None
+
+    def get_weather_data(self, lat, lon, historical_days=90, forecast_days=90):
+        """Get historical and forecast weather data"""
+        historical_data = []
+        
+        # Get historical data
+        for day in range(historical_days):
+            date = datetime.now() - timedelta(days=day)
+            url = f"https://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={self.api_key}&units=metric&dt={int(date.timestamp())}"
+            try:
+                response = requests.get(url)
+                if response.status_code == 200:
+                    data = response.json()
+                    weather_data = {
+                        'date': date,
+                        'temperature': data['main']['temp'],
+                        'humidity': data['main']['humidity'],
+                        'rainfall': data.get('rain', {}).get('1h', 0) * 24,
+                        'type': 'historical',
+                        'description': data['weather'][0]['description']
+                    }
+                    historical_data.append(weather_data)
+            except Exception as e:
+                print(f"Error fetching historical data: {e}")
+
+        # Get forecast data
+        forecast_data = []
+        try:
+            # Get 5-day forecast
+            forecast_url = f"https://api.openweathermap.org/data/2.5/forecast?lat={lat}&lon={lon}&appid={self.api_key}&units=metric"
+            response = requests.get(forecast_url)
+            if response.status_code == 200:
+                data = response.json()
+                for item in data['list']:
+                    forecast = {
+                        'date': datetime.fromtimestamp(item['dt']),
+                        'temperature': item['main']['temp'],
+                        'humidity': item['main']['humidity'],
+                        'rainfall': item.get('rain', {}).get('3h', 0) * 8,
+                        'type': 'forecast_5day',
+                        'description': item['weather'][0]['description']
+                    }
+                    forecast_data.append(forecast)
+
+                # Generate extended forecast
+                last_date = max(d['date'] for d in forecast_data)
+                historical_df = pd.DataFrame(historical_data)
+                
+                for day in range(1, forecast_days - 5):
+                    date = last_date + timedelta(days=day)
+                    
+                    # Calculate trends from historical data
+                    if not historical_df.empty:
+                        temp_trend = stats.linregress(range(len(historical_df)), historical_df['temperature'])[0]
+                        humidity_trend = stats.linregress(range(len(historical_df)), historical_df['humidity'])[0]
+                        rainfall_trend = stats.linregress(range(len(historical_df)), historical_df['rainfall'])[0]
+                    else:
+                        temp_trend = humidity_trend = rainfall_trend = 0
+                    
+                    # Get recent averages
+                    recent_temps = [d['temperature'] for d in forecast_data[-5:]]
+                    recent_humidity = [d['humidity'] for d in forecast_data[-5:]]
+                    recent_rainfall = [d['rainfall'] for d in forecast_data[-5:]]
+                    
+                    extended_forecast = {
+                        'date': date,
+                        'temperature': np.mean(recent_temps) + temp_trend * day,
+                        'humidity': np.mean(recent_humidity) + humidity_trend * day,
+                        'rainfall': np.mean(recent_rainfall) + rainfall_trend * day,
+                        'type': 'forecast_extended',
+                        'description': 'Extended Forecast'
+                    }
+                    forecast_data.append(extended_forecast)
+                    
+        except Exception as e:
+            print(f"Error fetching forecast data: {e}")
+
+        # Combine and process all data
+        all_data = pd.DataFrame(historical_data + forecast_data)
+        all_data = all_data.sort_values('date')
+        
+        # Add analysis columns
+        all_data['month'] = all_data['date'].dt.month
+        all_data['season'] = all_data['month'].map(self.seasons)
+        
+        # Calculate rolling averages
+        all_data['temp_7day_avg'] = all_data['temperature'].rolling(window=7, min_periods=1).mean()
+        all_data['humidity_7day_avg'] = all_data['humidity'].rolling(window=7, min_periods=1).mean()
+        all_data['rainfall_7day_avg'] = all_data['rainfall'].rolling(window=7, min_periods=1).mean()
+        
+        return all_data
+
+    def analyze_trends(self, df):
+        """Analyze weather trends and patterns"""
+        historical = df[df['type'] == 'historical']
+        forecast = df[df['type'].isin(['forecast_5day', 'forecast_extended'])]
+        
+        analysis = {
+            'historical': {
+                'temperature': {
+                    'mean': historical['temperature'].mean(),
+                    'std': historical['temperature'].std(),
+                    'trend': stats.linregress(range(len(historical)), historical['temperature'])[0]
+                },
+                'humidity': {
+                    'mean': historical['humidity'].mean(),
+                    'std': historical['humidity'].std(),
+                    'trend': stats.linregress(range(len(historical)), historical['humidity'])[0]
+                },
+                'rainfall': {
+                    'mean': historical['rainfall'].mean(),
+                    'std': historical['rainfall'].std(),
+                    'trend': stats.linregress(range(len(historical)), historical['rainfall'])[0]
+                }
+            },
+            'forecast': {
+                'temperature': {
+                    'mean': forecast['temperature'].mean(),
+                    'std': forecast['temperature'].std(),
+                },
+                'humidity': {
+                    'mean': forecast['humidity'].mean(),
+                    'std': forecast['humidity'].std(),
+                },
+                'rainfall': {
+                    'mean': forecast['rainfall'].mean(),
+                    'std': forecast['rainfall'].std(),
+                }
+            }
+        }
+        
+        return analysis