"""Alert rule and triggered alert models.""" from dataclasses import dataclass, field from datetime import datetime from enum import Enum from typing import Any, Optional class AlertType(Enum): """Types of weather alerts that can be triggered.""" TEMPERATURE_LOW = "temperature_low" TEMPERATURE_HIGH = "temperature_high" PRECIPITATION = "precipitation" WIND_SPEED = "wind_speed" WIND_GUST = "wind_gust" SEVERE_WEATHER = "severe_weather" @dataclass class TemperatureRule: """Temperature alert rule configuration.""" enabled: bool = True below: Optional[float] = 32 above: Optional[float] = 100 @dataclass class PrecipitationRule: """Precipitation alert rule configuration.""" enabled: bool = True probability_above: float = 70 @dataclass class WindRule: """Wind alert rule configuration.""" enabled: bool = True speed_above: float = 25 gust_above: float = 40 @dataclass class SevereWeatherRule: """Severe weather alert rule configuration.""" enabled: bool = True @dataclass class AlertRules: """Collection of all alert rules.""" temperature: TemperatureRule = field(default_factory=TemperatureRule) precipitation: PrecipitationRule = field(default_factory=PrecipitationRule) wind: WindRule = field(default_factory=WindRule) severe_weather: SevereWeatherRule = field(default_factory=SevereWeatherRule) @classmethod def from_dict(cls, data: dict[str, Any]) -> "AlertRules": """Create AlertRules from a configuration dict. Args: data: The rules configuration dict. Returns: An AlertRules instance. """ temp_data = data.get("temperature", {}) precip_data = data.get("precipitation", {}) wind_data = data.get("wind", {}) severe_data = data.get("severe_weather", {}) return cls( temperature=TemperatureRule( enabled=temp_data.get("enabled", True), below=temp_data.get("below", 32), above=temp_data.get("above", 100), ), precipitation=PrecipitationRule( enabled=precip_data.get("enabled", True), probability_above=precip_data.get("probability_above", 70), ), wind=WindRule( enabled=wind_data.get("enabled", True), speed_above=wind_data.get("speed_above", 25), gust_above=wind_data.get("gust_above", 40), ), severe_weather=SevereWeatherRule( enabled=severe_data.get("enabled", True), ), ) @dataclass class TriggeredAlert: """Represents an alert that was triggered by a rule evaluation.""" alert_type: AlertType title: str message: str forecast_hour: str value: float threshold: float created_at: datetime = field(default_factory=datetime.now) @property def dedup_key(self) -> str: """Generate a deduplication key for this alert. Format: {alert_type}:{forecast_hour} This allows re-alerting for different time periods while preventing duplicate alerts for the same hour. """ return f"{self.alert_type.value}:{self.forecast_hour}" def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "alert_type": self.alert_type.value, "title": self.title, "message": self.message, "forecast_hour": self.forecast_hour, "value": self.value, "threshold": self.threshold, "created_at": self.created_at.isoformat(), "dedup_key": self.dedup_key, } @dataclass class AggregatedAlert: """Represents multiple alerts of the same type aggregated into one notification.""" alert_type: AlertType title: str message: str triggered_hours: list[str] start_time: str end_time: str extreme_value: float extreme_hour: str threshold: float created_at: datetime = field(default_factory=datetime.now) @property def dedup_key(self) -> str: """Generate a deduplication key for this aggregated alert. Format: {alert_type}:{date} Day-level deduplication prevents re-sending aggregated alerts for the same alert type on the same day. """ # Extract date from the first triggered hour (format: YYYY-MM-DD-HH) date_part = self.start_time.rsplit("-", 1)[0] if self.start_time else "" return f"{self.alert_type.value}:{date_part}" @property def hour_count(self) -> int: """Number of hours that triggered this alert.""" return len(self.triggered_hours) def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "alert_type": self.alert_type.value, "title": self.title, "message": self.message, "triggered_hours": self.triggered_hours, "start_time": self.start_time, "end_time": self.end_time, "extreme_value": self.extreme_value, "extreme_hour": self.extreme_hour, "threshold": self.threshold, "created_at": self.created_at.isoformat(), "dedup_key": self.dedup_key, "hour_count": self.hour_count, }