weather-alerts/app/models/alerts.py

"""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,
        }