219 lines
11 KiB
Python
219 lines
11 KiB
Python
# services/data/app/services/weather_service.py - REVISED VERSION
|
|
|
|
"""Weather data service with repository pattern"""
|
|
|
|
from typing import List, Dict, Any, Optional
|
|
from datetime import datetime, timedelta
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
import structlog
|
|
|
|
from app.models.weather import WeatherData, WeatherForecast
|
|
from app.external.aemet import AEMETClient
|
|
from app.schemas.weather import WeatherDataResponse, WeatherForecastResponse, WeatherForecastAPIResponse, HourlyForecastResponse
|
|
from app.repositories.weather_repository import WeatherRepository
|
|
|
|
logger = structlog.get_logger()
|
|
from app.core.database import database_manager
|
|
|
|
class WeatherService:
|
|
|
|
def __init__(self):
|
|
self.aemet_client = AEMETClient()
|
|
self.database_manager = database_manager
|
|
|
|
async def get_current_weather(self, latitude: float, longitude: float) -> Optional[WeatherDataResponse]:
|
|
"""Get current weather for location with graceful failure handling"""
|
|
try:
|
|
logger.debug("Getting current weather", lat=latitude, lon=longitude)
|
|
weather_data = await self.aemet_client.get_current_weather(latitude, longitude)
|
|
|
|
if weather_data:
|
|
logger.debug("Weather data received", source=weather_data.get('source'))
|
|
return WeatherDataResponse(**weather_data)
|
|
else:
|
|
logger.warning("No weather data received from AEMET client - providing service unavailable response")
|
|
# Return a response indicating service unavailable rather than None
|
|
return WeatherDataResponse(
|
|
date=datetime.utcnow().isoformat(),
|
|
temperature=None,
|
|
precipitation=None,
|
|
humidity=None,
|
|
wind_speed=None,
|
|
pressure=None,
|
|
description="Servicio meteorológico temporalmente no disponible",
|
|
source="unavailable"
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get current weather", error=str(e), lat=latitude, lon=longitude)
|
|
# Return error response rather than None to prevent 404
|
|
return WeatherDataResponse(
|
|
date=datetime.utcnow().isoformat(),
|
|
temperature=None,
|
|
precipitation=None,
|
|
humidity=None,
|
|
wind_speed=None,
|
|
pressure=None,
|
|
description="Error al obtener datos meteorológicos",
|
|
source="error"
|
|
)
|
|
|
|
async def get_weather_forecast(self, latitude: float, longitude: float, days: int = 7) -> List[Dict[str, Any]]:
|
|
"""Get weather forecast for location - returns plain dicts"""
|
|
try:
|
|
logger.debug("Getting weather forecast", lat=latitude, lon=longitude, days=days)
|
|
forecast_data = await self.aemet_client.get_forecast(latitude, longitude, days)
|
|
|
|
if forecast_data:
|
|
logger.debug("Forecast data received", count=len(forecast_data))
|
|
# Validate and normalize each forecast item
|
|
valid_forecasts = []
|
|
for item in forecast_data:
|
|
try:
|
|
if isinstance(item, dict):
|
|
# Ensure required fields are present and convert to serializable format
|
|
forecast_date = item.get("forecast_date", datetime.now())
|
|
generated_at = item.get("generated_at", datetime.now())
|
|
|
|
forecast_item = {
|
|
"forecast_date": forecast_date.isoformat() if isinstance(forecast_date, datetime) else str(forecast_date),
|
|
"generated_at": generated_at.isoformat() if isinstance(generated_at, datetime) else str(generated_at),
|
|
"temperature": float(item.get("temperature", 15.0)),
|
|
"precipitation": float(item.get("precipitation", 0.0)),
|
|
"humidity": float(item.get("humidity", 50.0)),
|
|
"wind_speed": float(item.get("wind_speed", 10.0)),
|
|
"description": str(item.get("description", "Variable")),
|
|
"source": str(item.get("source", "unknown"))
|
|
}
|
|
valid_forecasts.append(forecast_item)
|
|
else:
|
|
logger.warning("Invalid forecast item type", item_type=type(item))
|
|
except Exception as item_error:
|
|
logger.warning("Error processing forecast item", error=str(item_error), item=item)
|
|
continue
|
|
|
|
logger.debug("Valid forecasts processed", count=len(valid_forecasts))
|
|
return valid_forecasts
|
|
else:
|
|
logger.warning("No forecast data received from AEMET client")
|
|
return []
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get weather forecast", error=str(e), lat=latitude, lon=longitude)
|
|
return []
|
|
|
|
async def get_hourly_forecast(self, latitude: float, longitude: float, hours: int = 48) -> List[HourlyForecastResponse]:
|
|
"""Get hourly weather forecast for location"""
|
|
try:
|
|
logger.debug("Getting hourly weather forecast", lat=latitude, lon=longitude, hours=hours)
|
|
hourly_data = await self.aemet_client.get_hourly_forecast(latitude, longitude, hours)
|
|
|
|
if hourly_data:
|
|
logger.debug("Hourly forecast data received", count=len(hourly_data))
|
|
# Validate each hourly forecast item before creating response
|
|
valid_forecasts = []
|
|
for item in hourly_data:
|
|
try:
|
|
if isinstance(item, dict):
|
|
# Ensure required fields are present
|
|
hourly_item = {
|
|
"forecast_datetime": item.get("forecast_datetime", datetime.now()),
|
|
"generated_at": item.get("generated_at", datetime.now()),
|
|
"temperature": float(item.get("temperature", 15.0)),
|
|
"precipitation": float(item.get("precipitation", 0.0)),
|
|
"humidity": float(item.get("humidity", 50.0)),
|
|
"wind_speed": float(item.get("wind_speed", 10.0)),
|
|
"description": str(item.get("description", "Variable")),
|
|
"source": str(item.get("source", "unknown")),
|
|
"hour": int(item.get("hour", 0))
|
|
}
|
|
valid_forecasts.append(HourlyForecastResponse(**hourly_item))
|
|
else:
|
|
logger.warning("Invalid hourly forecast item type", item_type=type(item))
|
|
except Exception as item_error:
|
|
logger.warning("Error processing hourly forecast item", error=str(item_error), item=item)
|
|
continue
|
|
|
|
logger.debug("Valid hourly forecasts processed", count=len(valid_forecasts))
|
|
return valid_forecasts
|
|
else:
|
|
logger.warning("No hourly forecast data received from AEMET client")
|
|
return []
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get hourly weather forecast", error=str(e), lat=latitude, lon=longitude)
|
|
return []
|
|
|
|
async def get_historical_weather(self,
|
|
latitude: float,
|
|
longitude: float,
|
|
start_date: datetime,
|
|
end_date: datetime) -> List[WeatherDataResponse]:
|
|
"""Get historical weather data"""
|
|
try:
|
|
logger.debug("Getting historical weather",
|
|
lat=latitude, lon=longitude,
|
|
start=start_date, end=end_date)
|
|
|
|
location_id = f"{latitude:.4f},{longitude:.4f}"
|
|
|
|
async with self.database_manager.get_session() as session:
|
|
weather_repository = WeatherRepository(session)
|
|
# Use the repository to get data from the database
|
|
db_records = await weather_repository.get_historical_weather(
|
|
location_id,
|
|
start_date,
|
|
end_date
|
|
)
|
|
|
|
if db_records:
|
|
logger.debug("Historical data found in database", count=len(db_records))
|
|
return [WeatherDataResponse(
|
|
date=record.date,
|
|
temperature=record.temperature,
|
|
precipitation=record.precipitation,
|
|
humidity=record.humidity,
|
|
wind_speed=record.wind_speed,
|
|
pressure=record.pressure,
|
|
description=record.description,
|
|
source=record.source
|
|
) for record in db_records]
|
|
|
|
# If not in database, fetch from API and store
|
|
logger.debug("Fetching historical data from AEMET API")
|
|
weather_data = await self.aemet_client.get_historical_weather(
|
|
latitude, longitude, start_date, end_date
|
|
)
|
|
|
|
if weather_data:
|
|
# Use the repository to store the new data
|
|
records_to_store = [{
|
|
"location_id": location_id,
|
|
"city": "Madrid", # Default city for AEMET data
|
|
"date": data.get('date', datetime.now()),
|
|
"temperature": data.get('temperature'),
|
|
"precipitation": data.get('precipitation'),
|
|
"humidity": data.get('humidity'),
|
|
"wind_speed": data.get('wind_speed'),
|
|
"pressure": data.get('pressure'),
|
|
"description": data.get('description'),
|
|
"source": "aemet",
|
|
"data_type": "historical",
|
|
"raw_data": data, # Pass as dict, not string
|
|
"tenant_id": None
|
|
} for data in weather_data]
|
|
|
|
async with self.database_manager.get_session() as session:
|
|
weather_repository = WeatherRepository(session)
|
|
await weather_repository.bulk_create_weather_data(records_to_store)
|
|
|
|
logger.debug("Historical data stored in database", count=len(weather_data))
|
|
|
|
return [WeatherDataResponse(**item) for item in weather_data]
|
|
else:
|
|
logger.warning("No historical weather data received")
|
|
return []
|
|
|
|
except Exception as e:
|
|
logger.error("Failed to get historical weather", error=str(e))
|
|
return [] |