weather-alerts/app/services/rule_engine.py

"""Rule engine for evaluating weather conditions against alert rules."""

from typing import Optional

from app.models.alerts import AlertRules, AlertType, TriggeredAlert
from app.models.weather import HourlyForecast, WeatherAlert, WeatherForecast
from app.utils.logging_config import get_logger


class RuleEngine:
    """Evaluates weather forecasts against configured alert rules."""

    def __init__(self, rules: AlertRules) -> None:
        """Initialize the rule engine.

        Args:
            rules: The alert rules configuration.
        """
        self.rules = rules
        self.logger = get_logger(__name__)

    def evaluate(self, forecast: WeatherForecast) -> list[TriggeredAlert]:
        """Evaluate a forecast against all enabled rules.

        Args:
            forecast: The weather forecast to evaluate.

        Returns:
            List of triggered alerts.
        """
        alerts: list[TriggeredAlert] = []

        # Evaluate hourly forecasts
        for hourly in forecast.hourly_forecasts:
            alerts.extend(self._evaluate_hourly(hourly))

        # Evaluate severe weather alerts from API
        if self.rules.severe_weather.enabled:
            alerts.extend(self._evaluate_severe_alerts(forecast.alerts))

        self.logger.info(
            "rules_evaluated",
            hourly_count=len(forecast.hourly_forecasts),
            triggered_count=len(alerts),
        )

        return alerts

    def _evaluate_hourly(self, hourly: HourlyForecast) -> list[TriggeredAlert]:
        """Evaluate a single hourly forecast against rules.

        Args:
            hourly: The hourly forecast data.

        Returns:
            List of triggered alerts for this hour.
        """
        alerts: list[TriggeredAlert] = []

        # Temperature rules
        if self.rules.temperature.enabled:
            alert = self._check_temperature(hourly)
            if alert:
                alerts.append(alert)

        # Precipitation rules
        if self.rules.precipitation.enabled:
            alert = self._check_precipitation(hourly)
            if alert:
                alerts.append(alert)

        # Wind rules
        if self.rules.wind.enabled:
            wind_alerts = self._check_wind(hourly)
            alerts.extend(wind_alerts)

        return alerts

    def _check_temperature(
        self,
        hourly: HourlyForecast,
    ) -> Optional[TriggeredAlert]:
        """Check temperature thresholds.

        Args:
            hourly: The hourly forecast data.

        Returns:
            TriggeredAlert if threshold exceeded, None otherwise.
        """
        temp_rule = self.rules.temperature

        # Check low temperature
        if temp_rule.below is not None and hourly.temp < temp_rule.below:
            return TriggeredAlert(
                alert_type=AlertType.TEMPERATURE_LOW,
                title="Low Temperature Alert",
                message=(
                    f"Temperature expected to drop to {hourly.temp:.0f}°F "
                    f"at {hourly.datetime.strftime('%I:%M %p on %b %d')}. "
                    f"Threshold: {temp_rule.below:.0f}°F"
                ),
                forecast_hour=hourly.hour_key,
                value=hourly.temp,
                threshold=temp_rule.below,
            )

        # Check high temperature
        if temp_rule.above is not None and hourly.temp > temp_rule.above:
            return TriggeredAlert(
                alert_type=AlertType.TEMPERATURE_HIGH,
                title="High Temperature Alert",
                message=(
                    f"Temperature expected to reach {hourly.temp:.0f}°F "
                    f"at {hourly.datetime.strftime('%I:%M %p on %b %d')}. "
                    f"Threshold: {temp_rule.above:.0f}°F"
                ),
                forecast_hour=hourly.hour_key,
                value=hourly.temp,
                threshold=temp_rule.above,
            )

        return None

    def _check_precipitation(
        self,
        hourly: HourlyForecast,
    ) -> Optional[TriggeredAlert]:
        """Check precipitation probability threshold.

        Args:
            hourly: The hourly forecast data.

        Returns:
            TriggeredAlert if threshold exceeded, None otherwise.
        """
        precip_rule = self.rules.precipitation
        threshold = precip_rule.probability_above

        if hourly.precip_prob > threshold:
            return TriggeredAlert(
                alert_type=AlertType.PRECIPITATION,
                title="Precipitation Alert",
                message=(
                    f"{hourly.precip_prob:.0f}% chance of precipitation "
                    f"at {hourly.datetime.strftime('%I:%M %p on %b %d')}. "
                    f"Conditions: {hourly.conditions}"
                ),
                forecast_hour=hourly.hour_key,
                value=hourly.precip_prob,
                threshold=threshold,
            )

        return None

    def _check_wind(self, hourly: HourlyForecast) -> list[TriggeredAlert]:
        """Check wind speed and gust thresholds.

        Args:
            hourly: The hourly forecast data.

        Returns:
            List of triggered wind alerts.
        """
        alerts: list[TriggeredAlert] = []
        wind_rule = self.rules.wind

        # Check sustained wind speed
        if hourly.wind_speed > wind_rule.speed_above:
            alerts.append(
                TriggeredAlert(
                    alert_type=AlertType.WIND_SPEED,
                    title="High Wind Alert",
                    message=(
                        f"Sustained winds of {hourly.wind_speed:.0f} mph expected "
                        f"at {hourly.datetime.strftime('%I:%M %p on %b %d')}. "
                        f"Threshold: {wind_rule.speed_above:.0f} mph"
                    ),
                    forecast_hour=hourly.hour_key,
                    value=hourly.wind_speed,
                    threshold=wind_rule.speed_above,
                )
            )

        # Check wind gusts
        if hourly.wind_gust > wind_rule.gust_above:
            alerts.append(
                TriggeredAlert(
                    alert_type=AlertType.WIND_GUST,
                    title="Wind Gust Alert",
                    message=(
                        f"Wind gusts up to {hourly.wind_gust:.0f} mph expected "
                        f"at {hourly.datetime.strftime('%I:%M %p on %b %d')}. "
                        f"Threshold: {wind_rule.gust_above:.0f} mph"
                    ),
                    forecast_hour=hourly.hour_key,
                    value=hourly.wind_gust,
                    threshold=wind_rule.gust_above,
                )
            )

        return alerts

    def _evaluate_severe_alerts(
        self,
        api_alerts: list[WeatherAlert],
    ) -> list[TriggeredAlert]:
        """Convert API severe weather alerts to triggered alerts.

        Args:
            api_alerts: List of WeatherAlert from the API.

        Returns:
            List of triggered severe weather alerts.
        """
        triggered: list[TriggeredAlert] = []

        for api_alert in api_alerts:
            # Use alert ID as the hour key for deduplication
            triggered.append(
                TriggeredAlert(
                    alert_type=AlertType.SEVERE_WEATHER,
                    title=f"Severe Weather: {api_alert.event}",
                    message=api_alert.headline or api_alert.description[:200],
                    forecast_hour=api_alert.id or api_alert.event,
                    value=1.0,  # Placeholder - severe alerts don't have numeric values
                    threshold=0.0,
                )
            )

        return triggered