import logging from datetime import datetime from typing import Dict, List, Optional, Any import httpx from fastapi import HTTPException from app.core.config import settings from app.schemas.weather import WeatherData, ForecastItem, Forecast, CurrentWeather logger = logging.getLogger(__name__) class WeatherService: """Service for interacting with the OpenWeatherMap API.""" def __init__(self): self.api_key = settings.OPENWEATHER_API_KEY self.base_url = settings.OPENWEATHER_API_URL if not self.api_key: logger.warning("OpenWeatherMap API key not set. Weather data will not be available.") async def get_current_weather(self, lat: float, lon: float, location_name: Optional[str] = None) -> CurrentWeather: """ Get current weather data for a specific location. Args: lat: Latitude coordinate lon: Longitude coordinate location_name: Optional name of the location Returns: CurrentWeather object with location name and weather data """ if not self.api_key: raise HTTPException(status_code=503, detail="Weather service not available. API key not configured.") url = f"{self.base_url}/weather" params = { "lat": lat, "lon": lon, "appid": self.api_key, "units": "metric", # Use metric units (Celsius) } try: async with httpx.AsyncClient() as client: response = await client.get(url, params=params) response.raise_for_status() data = response.json() # Extract location name from response if not provided if not location_name: location_name = data.get("name", f"Location ({lat}, {lon})") # Parse weather data weather_data = self._parse_current_weather(data) return CurrentWeather( location_name=location_name, weather=weather_data ) except httpx.HTTPStatusError as e: logger.error(f"HTTP error from OpenWeatherMap API: {e}") if e.response.status_code == 401: raise HTTPException(status_code=503, detail="Weather service authentication failed.") elif e.response.status_code == 404: raise HTTPException(status_code=404, detail="Weather data not found for this location.") else: raise HTTPException(status_code=503, detail=f"Weather service error: {e}") except Exception as e: logger.error(f"Error fetching weather data: {e}") raise HTTPException(status_code=503, detail="Failed to fetch weather data.") async def get_forecast(self, lat: float, lon: float, location_name: Optional[str] = None) -> Forecast: """ Get 5-day weather forecast for a specific location. Args: lat: Latitude coordinate lon: Longitude coordinate location_name: Optional name of the location Returns: Forecast object with location name and forecast items """ if not self.api_key: raise HTTPException(status_code=503, detail="Weather service not available. API key not configured.") url = f"{self.base_url}/forecast" params = { "lat": lat, "lon": lon, "appid": self.api_key, "units": "metric", # Use metric units (Celsius) } try: async with httpx.AsyncClient() as client: response = await client.get(url, params=params) response.raise_for_status() data = response.json() # Extract location name from response if not provided if not location_name: city_data = data.get("city", {}) location_name = city_data.get("name", f"Location ({lat}, {lon})") # Parse forecast data forecast_items = self._parse_forecast(data) return Forecast( location_name=location_name, items=forecast_items ) except httpx.HTTPStatusError as e: logger.error(f"HTTP error from OpenWeatherMap API: {e}") if e.response.status_code == 401: raise HTTPException(status_code=503, detail="Weather service authentication failed.") elif e.response.status_code == 404: raise HTTPException(status_code=404, detail="Weather forecast not found for this location.") else: raise HTTPException(status_code=503, detail=f"Weather service error: {e}") except Exception as e: logger.error(f"Error fetching weather forecast: {e}") raise HTTPException(status_code=503, detail="Failed to fetch weather forecast.") def _parse_current_weather(self, data: Dict[str, Any]) -> WeatherData: """Parse current weather data from OpenWeatherMap API response.""" try: weather = data.get("weather", [{}])[0] main = data.get("main", {}) wind = data.get("wind", {}) clouds = data.get("clouds", {}) sys = data.get("sys", {}) return WeatherData( temperature=main.get("temp", 0), feels_like=main.get("feels_like", 0), humidity=main.get("humidity", 0), pressure=main.get("pressure", 0), wind_speed=wind.get("speed", 0), wind_direction=wind.get("deg", 0), weather_condition=weather.get("main", "Unknown"), weather_description=weather.get("description", "Unknown"), weather_icon=weather.get("icon", ""), clouds=clouds.get("all", 0), timestamp=datetime.fromtimestamp(data.get("dt", 0)), sunrise=datetime.fromtimestamp(sys.get("sunrise", 0)), sunset=datetime.fromtimestamp(sys.get("sunset", 0)), ) except Exception as e: logger.error(f"Error parsing current weather data: {e}") raise HTTPException(status_code=500, detail="Failed to parse weather data.") def _parse_forecast(self, data: Dict[str, Any]) -> List[ForecastItem]: """Parse forecast data from OpenWeatherMap API response.""" try: forecast_items = [] for item in data.get("list", []): weather = item.get("weather", [{}])[0] main = item.get("main", {}) wind = item.get("wind", {}) clouds = item.get("clouds", {}) forecast_item = ForecastItem( timestamp=datetime.fromtimestamp(item.get("dt", 0)), temperature=main.get("temp", 0), feels_like=main.get("feels_like", 0), humidity=main.get("humidity", 0), pressure=main.get("pressure", 0), wind_speed=wind.get("speed", 0), wind_direction=wind.get("deg", 0), weather_condition=weather.get("main", "Unknown"), weather_description=weather.get("description", "Unknown"), weather_icon=weather.get("icon", ""), clouds=clouds.get("all", 0), probability_of_precipitation=item.get("pop", 0), ) forecast_items.append(forecast_item) return forecast_items except Exception as e: logger.error(f"Error parsing forecast data: {e}") raise HTTPException(status_code=500, detail="Failed to parse forecast data.")