""" Base classes and standardized response models for external APIs. All provider implementations should return these standard models to ensure interoperability when swapping providers. """ from abc import ABC, abstractmethod from dataclasses import dataclass, field from datetime import datetime from typing import Optional from enum import Enum # ============================================================================= # Weather Models # ============================================================================= class WeatherCondition(Enum): """Standardized weather conditions across providers.""" CLEAR = "clear" PARTLY_CLOUDY = "partly_cloudy" CLOUDY = "cloudy" OVERCAST = "overcast" FOG = "fog" DRIZZLE = "drizzle" RAIN = "rain" HEAVY_RAIN = "heavy_rain" SNOW = "snow" HEAVY_SNOW = "heavy_snow" THUNDERSTORM = "thunderstorm" UNKNOWN = "unknown" @dataclass class CurrentWeather: """Standardized current weather response.""" temperature: float # Celsius feels_like: Optional[float] # Celsius humidity: int # Percentage 0-100 wind_speed: float # km/h wind_direction: Optional[int] # Degrees 0-360 condition: WeatherCondition condition_text: str # Human-readable description timestamp: datetime location: str # City/location name uv_index: Optional[float] = None # UV index 0-11+ def to_text(self) -> str: """Generate natural language description.""" parts = [ f"Currently {self.temperature:.1f}°C", f"({self.condition_text}) in {self.location}.", f"Humidity {self.humidity}%, wind {self.wind_speed:.0f} km/h." ] if self.uv_index is not None: parts.append(f"UV index: {self.uv_index:.0f}.") return " ".join(parts) @dataclass class DayForecast: """Standardized daily forecast.""" date: datetime temp_high: float # Celsius temp_low: float # Celsius condition: WeatherCondition condition_text: str precipitation_chance: Optional[int] # Percentage 0-100 precipitation_mm: Optional[float] uv_index_max: Optional[float] = None # Max UV index for the day def to_text(self) -> str: """Generate natural language description.""" date_str = self.date.strftime("%A") # Day name precip = f", {self.precipitation_chance}% rain" if self.precipitation_chance else "" uv = f", UV {self.uv_index_max:.0f}" if self.uv_index_max else "" return f"{date_str}: {self.temp_high:.0f}°/{self.temp_low:.0f}°C, {self.condition_text}{precip}{uv}" @dataclass class WeatherForecast: """Standardized forecast response.""" location: str current: CurrentWeather daily: list[DayForecast] = field(default_factory=list) @dataclass class GeoLocation: """Geocoding result.""" name: str latitude: float longitude: float country: Optional[str] = None admin_area: Optional[str] = None # State/province @dataclass class SunTimes: """Sunrise/sunset times for a location.""" location: str date: datetime sunrise: datetime sunset: datetime daylight_duration: int # seconds solar_noon: Optional[datetime] = None def to_text(self) -> str: """Generate natural language description.""" sunrise_str = self.sunrise.strftime("%H:%M") sunset_str = self.sunset.strftime("%H:%M") hours = self.daylight_duration // 3600 minutes = (self.daylight_duration % 3600) // 60 return ( f"Sun times for {self.location} on {self.date.strftime('%A %d %B')}: " f"Sunrise at {sunrise_str}, sunset at {sunset_str}. " f"Daylight duration: {hours}h {minutes}m." ) @dataclass class AirQuality: """Air quality measurements for a location.""" location: str timestamp: datetime aqi_european: Optional[int] # European AQI 0-500+ aqi_us: Optional[int] # US AQI 0-500+ pm2_5: Optional[float] # µg/m³ pm10: Optional[float] # µg/m³ ozone: Optional[float] # µg/m³ nitrogen_dioxide: Optional[float] # µg/m³ sulphur_dioxide: Optional[float] # µg/m³ carbon_monoxide: Optional[float] # µg/m³ # Pollen (European data only, seasonal) pollen_grass: Optional[float] = None pollen_birch: Optional[float] = None pollen_alder: Optional[float] = None def to_text(self) -> str: """Generate natural language description.""" parts = [f"Air quality in {self.location}:"] if self.aqi_european is not None: level = self._aqi_level(self.aqi_european) parts.append(f"European AQI {self.aqi_european} ({level}).") if self.pm2_5 is not None: parts.append(f"PM2.5: {self.pm2_5:.1f} µg/m³.") if self.pm10 is not None: parts.append(f"PM10: {self.pm10:.1f} µg/m³.") if self.ozone is not None: parts.append(f"Ozone: {self.ozone:.1f} µg/m³.") return " ".join(parts) @staticmethod def _aqi_level(aqi: int) -> str: """Convert AQI to human-readable level.""" if aqi <= 20: return "good" elif aqi <= 40: return "fair" elif aqi <= 60: return "moderate" elif aqi <= 80: return "poor" elif aqi <= 100: return "very poor" else: return "hazardous" # ============================================================================= # News Models # ============================================================================= @dataclass class NewsItem: """Standardized news article/item.""" title: str description: Optional[str] url: str published: Optional[datetime] source: str # e.g., "nos", "bbc" category: Optional[str] = None # e.g., "tech", "world" image_url: Optional[str] = None @dataclass class NewsFeed: """Standardized news feed response.""" source: str category: str items: list[NewsItem] = field(default_factory=list) fetched_at: datetime = field(default_factory=datetime.now) def to_text(self) -> str: """Generate natural language summary of headlines.""" if not self.items: return f"No news available from {self.source}." headlines = [f"- {item.title}" for item in self.items[:5]] return f"Headlines from {self.source} ({self.category}):\n" + "\n".join(headlines) # ============================================================================= # Financial Models # ============================================================================= @dataclass class StockQuote: """Standardized stock/crypto quote.""" symbol: str name: Optional[str] price: float currency: str # e.g., "USD", "EUR" change: Optional[float] # Absolute change change_percent: Optional[float] # Percentage change timestamp: datetime def to_text(self) -> str: """Generate natural language description.""" change_str = "" if self.change is not None and self.change_percent is not None: direction = "up" if self.change >= 0 else "down" change_str = f", {direction} {abs(self.change_percent):.2f}%" return f"{self.symbol}: {self.price:.2f} {self.currency}{change_str}" # ============================================================================= # Provider Interfaces # ============================================================================= class WeatherProvider(ABC): """Abstract base class for weather API providers.""" @abstractmethod async def geocode(self, city: str) -> Optional[GeoLocation]: """Convert city name to coordinates.""" pass @abstractmethod async def get_current(self, location: GeoLocation) -> CurrentWeather: """Get current weather for a location.""" pass @abstractmethod async def get_forecast(self, location: GeoLocation, days: int = 7) -> WeatherForecast: """Get weather forecast for a location.""" pass @abstractmethod async def get_sun_times(self, location: GeoLocation) -> SunTimes: """Get sunrise/sunset times for today.""" pass async def get_weather_for_city(self, city: str) -> CurrentWeather: """Convenience method: geocode and get current weather.""" location = await self.geocode(city) if not location: raise ValueError(f"Could not geocode city: {city}") return await self.get_current(location) class AirQualityProvider(ABC): """Abstract base class for air quality API providers.""" @abstractmethod async def get_air_quality(self, location: GeoLocation) -> AirQuality: """Get current air quality for a location.""" pass class NewsProvider(ABC): """Abstract base class for news API providers.""" @property @abstractmethod def source_name(self) -> str: """Provider name (e.g., 'nos', 'bbc').""" pass @property @abstractmethod def available_categories(self) -> list[str]: """List of available category keys.""" pass @abstractmethod async def get_feed(self, category: str, limit: int = 10) -> NewsFeed: """Get news feed for a category.""" pass async def get_headlines(self, categories: list[str], limit: int = 5) -> list[NewsFeed]: """Get headlines from multiple categories.""" feeds = [] for cat in categories: if cat in self.available_categories: feed = await self.get_feed(cat, limit) feeds.append(feed) return feeds class FinancialProvider(ABC): """Abstract base class for financial API providers.""" @abstractmethod async def get_quote(self, symbol: str) -> Optional[StockQuote]: """Get current quote for a stock/crypto symbol.""" pass @abstractmethod async def get_quotes(self, symbols: list[str]) -> list[StockQuote]: """Get quotes for multiple symbols.""" pass