From 3126edf2092ea868ab210bc5cceaeddf27452902 Mon Sep 17 00:00:00 2001 From: Harshvardhan-91 Date: Mon, 20 Apr 2026 23:34:29 +0530 Subject: [PATCH] feat: :sparkles: add context enrichment module for weather and geography --- src/context_enrichment.py | 158 +++++++++++++++++++++++++++++++ tests/test_context_enrichment.py | 106 +++++++++++++++++++++ 2 files changed, 264 insertions(+) create mode 100644 src/context_enrichment.py create mode 100644 tests/test_context_enrichment.py diff --git a/src/context_enrichment.py b/src/context_enrichment.py new file mode 100644 index 0000000..ca57186 --- /dev/null +++ b/src/context_enrichment.py @@ -0,0 +1,158 @@ +"""Trusted context enrichment for incident reports. + +This module defines a small provider-agnostic layer for enriching extracted +incident data with deterministic weather and geographic context. Real NOAA, +OpenStreetMap, or local boundary-data adapters can implement these interfaces +later without changing the orchestration logic. +""" + +from dataclasses import asdict, dataclass +from datetime import datetime +from typing import Protocol + + +@dataclass(frozen=True) +class WeatherContext: + """Weather fields resolved from a trusted provider.""" + + temperature: str | None = None + humidity: str | None = None + wind_speed: str | None = None + source: str = "unknown" + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass(frozen=True) +class GeographicContext: + """Geographic fields resolved from a trusted provider.""" + + latitude: float | None = None + longitude: float | None = None + jurisdiction: str | None = None + district: str | None = None + source: str = "unknown" + + def to_dict(self) -> dict: + return asdict(self) + + +@dataclass(frozen=True) +class ContextEnrichmentResult: + """Combined enrichment output plus non-fatal warnings.""" + + weather: WeatherContext | None = None + geography: GeographicContext | None = None + warnings: list[str] | None = None + + def to_dict(self) -> dict: + return { + "weather": self.weather.to_dict() if self.weather else None, + "geography": self.geography.to_dict() if self.geography else None, + "warnings": list(self.warnings or []), + } + + +class WeatherProvider(Protocol): + """Provider interface for incident-time weather lookup.""" + + def get_weather( + self, + latitude: float, + longitude: float, + incident_time: str | datetime, + ) -> WeatherContext | None: + ... + + +class GeocodingProvider(Protocol): + """Provider interface for geocoding and jurisdiction lookup.""" + + def geocode(self, address: str) -> GeographicContext | None: + ... + + +class ContextEnricher: + """Enrich incident data using deterministic provider interfaces.""" + + def __init__( + self, + weather_provider: WeatherProvider | None = None, + geocoding_provider: GeocodingProvider | None = None, + ): + self.weather_provider = weather_provider + self.geocoding_provider = geocoding_provider + + def enrich( + self, + incident_address: str | None = None, + incident_time: str | datetime | None = None, + ) -> ContextEnrichmentResult: + """Return weather/geographic context without raising provider errors.""" + warnings: list[str] = [] + geography = self._resolve_geography(incident_address, warnings) + weather = self._resolve_weather(geography, incident_time, warnings) + + return ContextEnrichmentResult( + weather=weather, + geography=geography, + warnings=warnings, + ) + + def _resolve_geography( + self, + incident_address: str | None, + warnings: list[str], + ) -> GeographicContext | None: + if not incident_address or not incident_address.strip(): + warnings.append("incident_address is required for geographic enrichment") + return None + + if self.geocoding_provider is None: + warnings.append("geocoding provider is not configured") + return None + + try: + geography = self.geocoding_provider.geocode(incident_address.strip()) + except Exception as exc: + warnings.append(f"geocoding enrichment failed: {exc}") + return None + + if geography is None: + warnings.append("geocoding provider returned no result") + + return geography + + def _resolve_weather( + self, + geography: GeographicContext | None, + incident_time: str | datetime | None, + warnings: list[str], + ) -> WeatherContext | None: + if self.weather_provider is None: + warnings.append("weather provider is not configured") + return None + + if geography is None or geography.latitude is None or geography.longitude is None: + warnings.append("weather enrichment requires latitude and longitude") + return None + + if incident_time is None: + warnings.append("incident_time is required for weather enrichment") + return None + + try: + weather = self.weather_provider.get_weather( + latitude=geography.latitude, + longitude=geography.longitude, + incident_time=incident_time, + ) + except Exception as exc: + warnings.append(f"weather enrichment failed: {exc}") + return None + + if weather is None: + warnings.append("weather provider returned no result") + + return weather diff --git a/tests/test_context_enrichment.py b/tests/test_context_enrichment.py new file mode 100644 index 0000000..e1762e0 --- /dev/null +++ b/tests/test_context_enrichment.py @@ -0,0 +1,106 @@ +"""Tests for trusted weather and geographic context enrichment.""" + +from src.context_enrichment import ( + ContextEnricher, + GeographicContext, + WeatherContext, +) + + +class FakeGeocodingProvider: + def geocode(self, address): + assert address == "142 Oak Street" + return GeographicContext( + latitude=37.7749, + longitude=-122.4194, + jurisdiction="Santa Cruz County", + district="CAL FIRE CZU", + source="mock-osm", + ) + + +class FakeWeatherProvider: + def get_weather(self, latitude, longitude, incident_time): + assert latitude == 37.7749 + assert longitude == -122.4194 + assert incident_time == "2026-04-20T18:40:00" + return WeatherContext( + temperature="72 F", + humidity="31%", + wind_speed="14 mph", + source="mock-noaa", + ) + + +class FailingWeatherProvider: + def get_weather(self, latitude, longitude, incident_time): + raise RuntimeError("provider unavailable") + + +def test_context_enricher_combines_weather_and_geographic_context(): + enricher = ContextEnricher( + weather_provider=FakeWeatherProvider(), + geocoding_provider=FakeGeocodingProvider(), + ) + + result = enricher.enrich( + incident_address="142 Oak Street", + incident_time="2026-04-20T18:40:00", + ) + + assert result.to_dict() == { + "weather": { + "temperature": "72 F", + "humidity": "31%", + "wind_speed": "14 mph", + "source": "mock-noaa", + }, + "geography": { + "latitude": 37.7749, + "longitude": -122.4194, + "jurisdiction": "Santa Cruz County", + "district": "CAL FIRE CZU", + "source": "mock-osm", + }, + "warnings": [], + } + + +def test_context_enricher_handles_missing_address_without_crashing(): + enricher = ContextEnricher(weather_provider=FakeWeatherProvider()) + + result = enricher.enrich(incident_time="2026-04-20T18:40:00") + + assert result.weather is None + assert result.geography is None + assert "incident_address is required" in result.warnings[0] + assert "weather enrichment requires latitude and longitude" in result.warnings[1] + + +def test_context_enricher_keeps_geography_when_weather_provider_fails(): + enricher = ContextEnricher( + weather_provider=FailingWeatherProvider(), + geocoding_provider=FakeGeocodingProvider(), + ) + + result = enricher.enrich( + incident_address="142 Oak Street", + incident_time="2026-04-20T18:40:00", + ) + + assert result.geography.jurisdiction == "Santa Cruz County" + assert result.weather is None + assert result.warnings == ["weather enrichment failed: provider unavailable"] + + +def test_context_enricher_warns_when_incident_time_is_missing(): + enricher = ContextEnricher( + weather_provider=FakeWeatherProvider(), + geocoding_provider=FakeGeocodingProvider(), + ) + + result = enricher.enrich(incident_address="142 Oak Street") + + assert result.geography.source == "mock-osm" + assert result.weather is None + assert result.warnings == ["incident_time is required for weather enrichment"]