Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 10 additions & 0 deletions src/accessiweather/models/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@
"notify_severe_risk_change",
"notify_minutely_precipitation_start",
"notify_minutely_precipitation_stop",
"precipitation_sensitivity",
# GitHub settings
"github_backend_url",
"github_app_id",
Expand Down Expand Up @@ -123,6 +124,8 @@ class AppSettings:
notify_severe_risk_change: bool = False
notify_minutely_precipitation_start: bool = True
notify_minutely_precipitation_stop: bool = True
# Minimum intensity level required to count as precipitation ("light", "moderate", "heavy")
precipitation_sensitivity: str = "light"
github_backend_url: str = ""
github_app_id: str = ""
github_app_private_key: str = ""
Expand Down Expand Up @@ -334,6 +337,11 @@ def validate_on_access(self, setting_name: str) -> bool:
if not isinstance(value, int) or value < 1:
setattr(self, setting_name, 180)

elif setting_name == "precipitation_sensitivity":
valid_levels = {"light", "moderate", "heavy"}
if value not in valid_levels:
setattr(self, setting_name, "light")

elif setting_name == "alert_radius_type":
valid_types = {"county", "point", "zone", "state"}
if value not in valid_types:
Expand Down Expand Up @@ -425,6 +433,7 @@ def to_dict(self) -> dict:
"notify_severe_risk_change": self.notify_severe_risk_change,
"notify_minutely_precipitation_start": self.notify_minutely_precipitation_start,
"notify_minutely_precipitation_stop": self.notify_minutely_precipitation_stop,
"precipitation_sensitivity": self.precipitation_sensitivity,
"github_backend_url": self.github_backend_url,
"alert_radius_type": self.alert_radius_type,
"alert_notifications_enabled": self.alert_notifications_enabled,
Expand Down Expand Up @@ -511,6 +520,7 @@ def from_dict(cls, data: dict) -> AppSettings:
notify_minutely_precipitation_stop=cls._as_bool(
data.get("notify_minutely_precipitation_stop"), True
),
precipitation_sensitivity=data.get("precipitation_sensitivity", "light"),
github_backend_url=data.get("github_backend_url", ""),
alert_radius_type=data.get("alert_radius_type", "county"),
alert_notifications_enabled=cls._as_bool(data.get("alert_notifications_enabled"), True),
Expand Down
69 changes: 58 additions & 11 deletions src/accessiweather/notifications/minutely_precipitation.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,19 @@

NO_TRANSITION_SIGNATURE = "__none__"

# Intensity thresholds (mm/h) for wet detection.
# Pirate Weather light rain is typically 0.01–0.1 mm/h; moderate 0.1–1.0 mm/h.
INTENSITY_THRESHOLD_LIGHT = 0.01
INTENSITY_THRESHOLD_MODERATE = 0.1
INTENSITY_THRESHOLD_HEAVY = 1.0

# Map setting values to thresholds
SENSITIVITY_THRESHOLDS: dict[str, float] = {
"light": INTENSITY_THRESHOLD_LIGHT,
"moderate": INTENSITY_THRESHOLD_MODERATE,
"heavy": INTENSITY_THRESHOLD_HEAVY,
}


@dataclass(frozen=True)
class MinutelyPrecipitationTransition:
Expand Down Expand Up @@ -86,54 +99,86 @@ def parse_pirate_weather_minutely_block(

def detect_minutely_precipitation_transition(
forecast: MinutelyPrecipitationForecast | None,
threshold: float = 0.0,
) -> MinutelyPrecipitationTransition | None:
"""Detect the first dry/wet transition in the next hour of minutely data."""
"""
Detect the first dry/wet transition in the next hour of minutely data.

Args:
forecast: Minutely precipitation forecast to analyse.
threshold: Minimum precipitation intensity (mm/h) to count as wet.
Defaults to ``0.0`` (any non-zero intensity). Use one of the
``INTENSITY_THRESHOLD_*`` constants or ``SENSITIVITY_THRESHOLDS``
to select a named sensitivity level.

"""
if forecast is None or not forecast.points:
return None

baseline_is_wet = is_wet(forecast.points[0])
baseline_is_wet = is_wet(forecast.points[0], threshold=threshold)
for idx, point in enumerate(forecast.points[1:], start=1):
if is_wet(point) == baseline_is_wet:
if is_wet(point, threshold=threshold) == baseline_is_wet:
continue
if baseline_is_wet:
return MinutelyPrecipitationTransition(
transition_type="stopping",
minutes_until=idx,
precipitation_type=_first_precipitation_type(forecast.points[:idx]),
precipitation_type=_first_precipitation_type(
forecast.points[:idx], threshold=threshold
),
)
return MinutelyPrecipitationTransition(
transition_type="starting",
minutes_until=idx,
precipitation_type=_first_precipitation_type(forecast.points[idx:]),
precipitation_type=_first_precipitation_type(
forecast.points[idx:], threshold=threshold
),
)

return None


def build_minutely_transition_signature(
forecast: MinutelyPrecipitationForecast | None,
threshold: float = 0.0,
) -> str | None:
"""
Return a stable signature for the current minutely transition state.

``None`` means the forecast was unavailable. ``NO_TRANSITION_SIGNATURE`` means
the forecast was available but no dry/wet transition was detected.

Args:
forecast: Minutely precipitation forecast to analyse.
threshold: Minimum precipitation intensity (mm/h) to count as wet.

"""
if forecast is None or not forecast.points:
return None

transition = detect_minutely_precipitation_transition(forecast)
transition = detect_minutely_precipitation_transition(forecast, threshold=threshold)
if transition is None:
return NO_TRANSITION_SIGNATURE

precip_type = transition.precipitation_type or "precipitation"
return f"{transition.transition_type}:{transition.minutes_until}:{precip_type}"


def is_wet(point: MinutelyPrecipitationPoint) -> bool:
"""Return True when a minutely point indicates precipitation."""
def is_wet(point: MinutelyPrecipitationPoint, threshold: float = 0.0) -> bool:
"""
Return True when a minutely point indicates precipitation.

Args:
point: A single minutely data point.
threshold: Minimum precipitation intensity (mm/h) required to be
considered wet. Defaults to ``0.0`` (any non-zero intensity).
Pass one of the ``INTENSITY_THRESHOLD_*`` constants to filter out
noise — e.g. ``INTENSITY_THRESHOLD_LIGHT`` (0.01 mm/h) ignores
trace/sensor-noise readings while still catching light rain.

"""
if point.precipitation_intensity is not None:
return point.precipitation_intensity > 0
return point.precipitation_intensity > threshold
if point.precipitation_probability is not None:
return point.precipitation_probability > 0
return False
Expand Down Expand Up @@ -169,8 +214,10 @@ def _normalize_precipitation_type(value: Any) -> str | None:
return normalized or None


def _first_precipitation_type(points: list[MinutelyPrecipitationPoint]) -> str | None:
def _first_precipitation_type(
points: list[MinutelyPrecipitationPoint], threshold: float = 0.0
) -> str | None:
for point in points:
if is_wet(point) and point.precipitation_type:
if is_wet(point, threshold=threshold) and point.precipitation_type:
return point.precipitation_type
return None
10 changes: 8 additions & 2 deletions src/accessiweather/notifications/notification_event_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@

from ..runtime_state import RuntimeStateManager
from .minutely_precipitation import (
SENSITIVITY_THRESHOLDS,
build_minutely_transition_signature,
detect_minutely_precipitation_transition,
)
Expand Down Expand Up @@ -543,7 +544,10 @@ def _check_minutely_precipitation_transition(
location_name: str,
) -> NotificationEvent | None:
"""Check for a new dry/wet transition in Pirate Weather minutely guidance."""
signature = build_minutely_transition_signature(minutely_precipitation)
sensitivity = getattr(settings, "precipitation_sensitivity", "light")
threshold = SENSITIVITY_THRESHOLDS.get(sensitivity, SENSITIVITY_THRESHOLDS["light"])

signature = build_minutely_transition_signature(minutely_precipitation, threshold=threshold)
if signature is None:
return None

Expand All @@ -556,7 +560,9 @@ def _check_minutely_precipitation_transition(
return None

self.state.last_minutely_transition_signature = signature
transition = detect_minutely_precipitation_transition(minutely_precipitation)
transition = detect_minutely_precipitation_transition(
minutely_precipitation, threshold=threshold
)
if transition is None:
return None

Expand Down
27 changes: 27 additions & 0 deletions src/accessiweather/ui/dialogs/settings_tabs/notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,9 @@
_RADIUS_TYPE_VALUES = ["county", "point", "zone", "state"]
_RADIUS_TYPE_MAP = {"county": 0, "point": 1, "zone": 2, "state": 3}

_SENSITIVITY_VALUES = ["light", "moderate", "heavy"]
_SENSITIVITY_MAP = {"light": 0, "moderate": 1, "heavy": 2}


class NotificationsTab:
"""Notifications tab: alert settings, severity levels, event notifications, rate limiting."""
Expand Down Expand Up @@ -118,6 +121,24 @@ def create(self):
)
sizer.Add(controls["notify_minutely_precipitation_stop"], 0, wx.LEFT | wx.BOTTOM, 10)

row_sensitivity = wx.BoxSizer(wx.HORIZONTAL)
row_sensitivity.Add(
wx.StaticText(panel, label="Notify for:"),
0,
wx.ALIGN_CENTER_VERTICAL | wx.RIGHT,
10,
)
controls["precipitation_sensitivity"] = wx.Choice(
panel,
choices=[
"Light rain and above (default, \u22650.01\u00a0mm/h)",
"Moderate rain and above (\u22650.1\u00a0mm/h)",
"Heavy rain only (\u22651.0\u00a0mm/h)",
],
)
row_sensitivity.Add(controls["precipitation_sensitivity"], 0)
sizer.Add(row_sensitivity, 0, wx.LEFT | wx.BOTTOM, 10)

# Hidden alert timing controls (values managed via Advanced dialog)
controls["global_cooldown"] = wx.SpinCtrl(panel, min=0, max=60, initial=5)
controls["global_cooldown"].Hide()
Expand Down Expand Up @@ -188,6 +209,8 @@ def load(self, settings):
controls["notify_minutely_precipitation_stop"].SetValue(
getattr(settings, "notify_minutely_precipitation_stop", False)
)
sensitivity = getattr(settings, "precipitation_sensitivity", "light")
controls["precipitation_sensitivity"].SetSelection(_SENSITIVITY_MAP.get(sensitivity, 0))

def save(self) -> dict:
"""Return Notifications tab settings as a dict."""
Expand All @@ -213,6 +236,9 @@ def save(self) -> dict:
"notify_minutely_precipitation_stop": controls[
"notify_minutely_precipitation_stop"
].GetValue(),
"precipitation_sensitivity": _SENSITIVITY_VALUES[
controls["precipitation_sensitivity"].GetSelection()
],
}

def setup_accessibility(self):
Expand All @@ -231,6 +257,7 @@ def setup_accessibility(self):
"notify_severe_risk_change": "Notify when severe weather risk level changes (Visual Crossing only)",
"notify_minutely_precipitation_start": "Notify when precipitation is expected to start soon (Pirate Weather)",
"notify_minutely_precipitation_stop": "Notify when precipitation is expected to stop soon (Pirate Weather)",
"precipitation_sensitivity": "Notify for: precipitation sensitivity level",
"global_cooldown": "Minimum time between any alert notifications (minutes)",
"per_alert_cooldown": "Re-notify for same alert after (minutes)",
"freshness_window": "Only notify for alerts issued within (minutes)",
Expand Down
22 changes: 22 additions & 0 deletions tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,28 @@ def test_minutely_notification_settings_round_trip(self):
assert restored.notify_minutely_precipitation_start is True
assert restored.notify_minutely_precipitation_stop is False

def test_precipitation_sensitivity_round_trip(self):
"""precipitation_sensitivity should round-trip through to_dict/from_dict."""
for level in ("light", "moderate", "heavy"):
settings = AppSettings(precipitation_sensitivity=level)
restored = AppSettings.from_dict(settings.to_dict())
assert restored.precipitation_sensitivity == level

def test_precipitation_sensitivity_default(self):
"""precipitation_sensitivity should default to 'light'."""
settings = AppSettings()
assert settings.precipitation_sensitivity == "light"

restored = AppSettings.from_dict({})
assert restored.precipitation_sensitivity == "light"

def test_precipitation_sensitivity_validation(self):
"""Invalid precipitation_sensitivity values should fall back to 'light'."""
settings = AppSettings()
settings.precipitation_sensitivity = "invalid"
assert settings.validate_on_access("precipitation_sensitivity") is True
assert settings.precipitation_sensitivity == "light"

def test_forecast_time_reference_validation(self):
"""Ensure invalid forecast_time_reference values fall back to location."""
settings = AppSettings()
Expand Down
Loading
Loading