init commit
This commit is contained in:
160
app/models/weather.py
Normal file
160
app/models/weather.py
Normal file
@@ -0,0 +1,160 @@
|
||||
"""Weather data models for VisualCrossing API responses."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class HourlyForecast:
|
||||
"""Represents a single hour's weather forecast."""
|
||||
|
||||
datetime_str: str
|
||||
datetime_epoch: int
|
||||
temp: float
|
||||
feelslike: float
|
||||
humidity: float
|
||||
precip: float
|
||||
precip_prob: float
|
||||
snow: float
|
||||
snow_depth: float
|
||||
wind_speed: float
|
||||
wind_gust: float
|
||||
wind_dir: float
|
||||
pressure: float
|
||||
visibility: float
|
||||
cloud_cover: float
|
||||
uv_index: float
|
||||
conditions: str
|
||||
icon: str
|
||||
|
||||
@property
|
||||
def datetime(self) -> datetime:
|
||||
"""Convert epoch to datetime."""
|
||||
return datetime.fromtimestamp(self.datetime_epoch)
|
||||
|
||||
@property
|
||||
def hour_key(self) -> str:
|
||||
"""Get a key representing this forecast hour (YYYY-MM-DD-HH)."""
|
||||
return self.datetime.strftime("%Y-%m-%d-%H")
|
||||
|
||||
@classmethod
|
||||
def from_api_data(cls, data: dict[str, Any]) -> "HourlyForecast":
|
||||
"""Create an HourlyForecast from VisualCrossing API data.
|
||||
|
||||
Args:
|
||||
data: The hourly data dict from the API response.
|
||||
|
||||
Returns:
|
||||
An HourlyForecast instance.
|
||||
"""
|
||||
return cls(
|
||||
datetime_str=data.get("datetime", ""),
|
||||
datetime_epoch=data.get("datetimeEpoch", 0),
|
||||
temp=float(data.get("temp", 0)),
|
||||
feelslike=float(data.get("feelslike", 0)),
|
||||
humidity=float(data.get("humidity", 0)),
|
||||
precip=float(data.get("precip") or 0),
|
||||
precip_prob=float(data.get("precipprob") or 0),
|
||||
snow=float(data.get("snow") or 0),
|
||||
snow_depth=float(data.get("snowdepth") or 0),
|
||||
wind_speed=float(data.get("windspeed") or 0),
|
||||
wind_gust=float(data.get("windgust") or 0),
|
||||
wind_dir=float(data.get("winddir") or 0),
|
||||
pressure=float(data.get("pressure") or 0),
|
||||
visibility=float(data.get("visibility") or 0),
|
||||
cloud_cover=float(data.get("cloudcover") or 0),
|
||||
uv_index=float(data.get("uvindex") or 0),
|
||||
conditions=data.get("conditions", ""),
|
||||
icon=data.get("icon", ""),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WeatherAlert:
|
||||
"""Represents a severe weather alert from the API."""
|
||||
|
||||
event: str
|
||||
headline: str
|
||||
description: str
|
||||
onset: Optional[str]
|
||||
ends: Optional[str]
|
||||
id: str
|
||||
language: str
|
||||
link: str
|
||||
|
||||
@classmethod
|
||||
def from_api_data(cls, data: dict[str, Any]) -> "WeatherAlert":
|
||||
"""Create a WeatherAlert from VisualCrossing API data.
|
||||
|
||||
Args:
|
||||
data: The alert data dict from the API response.
|
||||
|
||||
Returns:
|
||||
A WeatherAlert instance.
|
||||
"""
|
||||
return cls(
|
||||
event=data.get("event", "Unknown"),
|
||||
headline=data.get("headline", ""),
|
||||
description=data.get("description", ""),
|
||||
onset=data.get("onset"),
|
||||
ends=data.get("ends"),
|
||||
id=data.get("id", ""),
|
||||
language=data.get("language", "en"),
|
||||
link=data.get("link", ""),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class WeatherForecast:
|
||||
"""Complete weather forecast response."""
|
||||
|
||||
location: str
|
||||
resolved_address: str
|
||||
timezone: str
|
||||
hourly_forecasts: list[HourlyForecast] = field(default_factory=list)
|
||||
alerts: list[WeatherAlert] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def from_api_data(
|
||||
cls,
|
||||
data: dict[str, Any],
|
||||
hours_ahead: int = 24,
|
||||
) -> "WeatherForecast":
|
||||
"""Create a WeatherForecast from VisualCrossing API data.
|
||||
|
||||
Args:
|
||||
data: The full API response dict.
|
||||
hours_ahead: Number of hours of forecast to include.
|
||||
|
||||
Returns:
|
||||
A WeatherForecast instance.
|
||||
"""
|
||||
# Extract hourly forecasts from days
|
||||
hourly_forecasts: list[HourlyForecast] = []
|
||||
now = datetime.now()
|
||||
|
||||
for day in data.get("days", []):
|
||||
for hour_data in day.get("hours", []):
|
||||
forecast = HourlyForecast.from_api_data(hour_data)
|
||||
# Only include future hours up to hours_ahead
|
||||
if forecast.datetime > now:
|
||||
hourly_forecasts.append(forecast)
|
||||
if len(hourly_forecasts) >= hours_ahead:
|
||||
break
|
||||
if len(hourly_forecasts) >= hours_ahead:
|
||||
break
|
||||
|
||||
# Extract alerts
|
||||
alerts = [
|
||||
WeatherAlert.from_api_data(alert_data)
|
||||
for alert_data in data.get("alerts", [])
|
||||
]
|
||||
|
||||
return cls(
|
||||
location=data.get("address", ""),
|
||||
resolved_address=data.get("resolvedAddress", ""),
|
||||
timezone=data.get("timezone", ""),
|
||||
hourly_forecasts=hourly_forecasts,
|
||||
alerts=alerts,
|
||||
)
|
||||
Reference in New Issue
Block a user