ejr/weatherfeeder
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(normalizers): deduplicate synthetic station ID generation

Browse Source 
- Add common SynthStationID helpers for coordinate-based providers
- Use shared helper for Open-Meteo and OpenWeather station ID synthesis
- Require both lat/lon when generating synthetic IDs to avoid misleading defaults
- Remove unused Open-Meteo normalizer wrapper code

This reduces cross-provider duplication while keeping provider-specific
mapping logic explicit and readable.
This commit is contained in:
Eric Rakestraw
2026-01-16 22:13:44 -06:00
parent 00e811f8f7
commit b8804d32d2
8 changed files with 111 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
cmd/weatherfeeder/config.yml
View File

@@ -1,75 +1,59 @@
--- ---
sources: sources:
# - name: NWSObservationKSTL - name: NWSObservationKSTL
# kind: observation kind: observation
# driver: nws_observation driver: nws_observation
# every: 12m every: 10m
# params: params:
# url: "https://api.weather.gov/stations/KSTL/observations/latest" url: "https://api.weather.gov/stations/KSTL/observations/latest"
# user_agent: "HomeOps (eric@maximumdirect.net)" user_agent: "HomeOps (eric@maximumdirect.net)"
# - name: OpenMeteoObservation - name: OpenMeteoObservation
# kind: observation kind: observation
# driver: openmeteo_observation driver: openmeteo_observation
# every: 12m every: 10m
# params: params:
# url: "https://api.open-meteo.com/v1/forecast?latitude=38.6239&longitude=-90.3571&current=temperature_2m,relative_humidity_2m,weather_code,wind_speed_10m,wind_direction_10m,precipitation,surface_pressure,rain,showers,snowfall,cloud_cover,apparent_temperature,is_day,wind_gusts_10m,pressure_msl&forecast_days=1" url: "https://api.open-meteo.com/v1/forecast?latitude=38.6239&longitude=-90.3571&current=temperature_2m,relative_humidity_2m,weather_code,wind_speed_10m,wind_direction_10m,precipitation,surface_pressure,rain,showers,snowfall,cloud_cover,apparent_temperature,is_day,wind_gusts_10m,pressure_msl&forecast_days=1"
# user_agent: "HomeOps (eric@maximumdirect.net)" user_agent: "HomeOps (eric@maximumdirect.net)"
# - name: OpenWeatherObservation - name: OpenWeatherObservation
# kind: observation kind: observation
# driver: openweather_observation driver: openweather_observation
# every: 12m every: 10m
# params: params:
# url: "https://api.openweathermap.org/data/2.5/weather?lat=38.6239&lon=-90.3571&appid=c954f2566cb7ccb56b43737b52e88fc6&units=metric" url: "https://api.openweathermap.org/data/2.5/weather?lat=38.6239&lon=-90.3571&appid=c954f2566cb7ccb56b43737b52e88fc6&units=metric"
# user_agent: "HomeOps (eric@maximumdirect.net)" user_agent: "HomeOps (eric@maximumdirect.net)"
# - name: NWSObservationKSUS - name: NWSObservationKSUS
# kind: observation kind: observation
# driver: nws_observation driver: nws_observation
# every: 18s every: 10m
# params: params:
# url: "https://api.weather.gov/stations/KSUS/observations/latest" url: "https://api.weather.gov/stations/KSUS/observations/latest"
# user_agent: "HomeOps (eric@maximumdirect.net)" user_agent: "HomeOps (eric@maximumdirect.net)"
# - name: NWSObservationKCPS - name: NWSObservationKCPS
# kind: observation kind: observation
# driver: nws_observation driver: nws_observation
# every: 12m every: 10m
# params: params:
# url: "https://api.weather.gov/stations/KCPS/observations/latest" url: "https://api.weather.gov/stations/KCPS/observations/latest"
# user_agent: "HomeOps (eric@maximumdirect.net)" user_agent: "HomeOps (eric@maximumdirect.net)"
# - name: NWSForecastSTL - name: NWSHourlyForecastSTL
# kind: forecast kind: forecast
# driver: nws_forecast driver: nws_forecast
# every: 1m every: 45m
# params: params:
# url: "https://api.weather.gov/gridpoints/LSX/90,74/forecast" url: "https://api.weather.gov/gridpoints/LSX/90,74/forecast/hourly"
# user_agent: "HomeOps (eric@maximumdirect.net)" user_agent: "HomeOps (eric@maximumdirect.net)"
# - name: NWSHourlyForecastSTL
# kind: forecast
# driver: nws_forecast
# every: 1m
# params:
# url: "https://api.weather.gov/gridpoints/LSX/90,74/forecast/hourly"
# user_agent: "HomeOps (eric@maximumdirect.net)"
- name: NWSAlertsSTL - name: NWSAlertsSTL
kind: alert kind: alert
driver: nws_alerts driver: nws_alerts
every: 1m every: 1m
params: params:
url: "https://api.weather.gov/alerts?point=38.6239,-90.3571&limit=500" url: "https://api.weather.gov/alerts?point=38.6239,-90.3571&limit=20"
user_agent: "HomeOps (eric@maximumdirect.net)"
- name: NWSAlertsIowa
kind: alert
driver: nws_alerts
every: 1m
params:
url: "https://api.weather.gov/alerts/active/zone/IAZ048"
user_agent: "HomeOps (eric@maximumdirect.net)" user_agent: "HomeOps (eric@maximumdirect.net)"
sinks: sinks:

23
internal/normalizers/common/id.go Normal file
View File

@@ -0,0 +1,23 @@
// FILE: internal/normalizers/common/id.go
package common
import "fmt"
// SynthStationID formats a stable synthetic station identifier for providers that are
// coordinate-based rather than station-based.
//
// Example output:
//
// OPENMETEO(38.62700,-90.19940)
func SynthStationID(prefix string, lat, lon float64) string {
return fmt.Sprintf("%s(%.5f,%.5f)", prefix, lat, lon)
}
// SynthStationIDPtr is the pointer-friendly variant.
// If either coordinate is missing, it returns "" (unknown).
func SynthStationIDPtr(prefix string, lat, lon *float64) string {
if lat == nil || lon == nil {
return ""
}
return SynthStationID(prefix, *lat, *lon)
}

5
internal/normalizers/common/round.go
View File

@@ -56,14 +56,11 @@ func roundValue(v reflect.Value, decimals int) reflect.Value {
out.Set(elem) out.Set(elem)
return out return out
} }
if elem.IsValid() && elem.Type().AssignableTo(v.Type()) {
out.Set(elem)
return out
}
if elem.IsValid() && elem.Type().ConvertibleTo(v.Type()) { if elem.IsValid() && elem.Type().ConvertibleTo(v.Type()) {
out.Set(elem.Convert(v.Type())) out.Set(elem.Convert(v.Type()))
return out return out
} }
// If we can't sensibly re-wrap, just keep the original. // If we can't sensibly re-wrap, just keep the original.
return v return v
} }

62
internal/normalizers/nws/alerts.go
View File

@@ -39,34 +39,26 @@ func (AlertsNormalizer) Match(e event.Event) bool {
func (AlertsNormalizer) Normalize(ctx context.Context, in event.Event) (*event.Event, error) { func (AlertsNormalizer) Normalize(ctx context.Context, in event.Event) (*event.Event, error) {
_ = ctx // normalization is pure/CPU; keep ctx for future expensive steps _ = ctx // normalization is pure/CPU; keep ctx for future expensive steps
parsed, err := normcommon.DecodeJSONPayload[nwsAlertsResponse](in)
if err != nil {
return nil, fmt.Errorf("nws alerts normalize: %w", err)
}
// If we can't derive AsOf from the payload, fall back to the existing event envelope. // If we can't derive AsOf from the payload, fall back to the existing event envelope.
fallbackAsOf := in.EmittedAt.UTC() fallbackAsOf := in.EmittedAt.UTC()
if in.EffectiveAt != nil && !in.EffectiveAt.IsZero() { if in.EffectiveAt != nil && !in.EffectiveAt.IsZero() {
fallbackAsOf = in.EffectiveAt.UTC() fallbackAsOf = in.EffectiveAt.UTC()
} }
payload, effectiveAt, err := buildAlerts(parsed, fallbackAsOf) return normcommon.NormalizeJSON(
if err != nil { in,
return nil, fmt.Errorf("nws alerts normalize: build: %w", err) "nws alerts",
} standards.SchemaWeatherAlertV1,
func(parsed nwsAlertsResponse) (model.WeatherAlertRun, time.Time, error) {
out, err := normcommon.Finalize(in, standards.SchemaWeatherAlertV1, payload, effectiveAt) return buildAlerts(parsed, fallbackAsOf)
if err != nil { },
return nil, fmt.Errorf("nws alerts normalize: %w", err) )
}
return out, nil
} }
// buildAlerts contains the domain mapping logic (provider -> canonical model). // buildAlerts contains the domain mapping logic (provider -> canonical model).
func buildAlerts(parsed nwsAlertsResponse, fallbackAsOf time.Time) (model.WeatherAlertRun, time.Time, error) { func buildAlerts(parsed nwsAlertsResponse, fallbackAsOf time.Time) (model.WeatherAlertRun, time.Time, error) {
// 1) Determine AsOf (required by canonical model; also used as EffectiveAt). // 1) Determine AsOf (required by canonical model; also used as EffectiveAt).
asOf := parseNWSTimeUTC(parsed.Updated) asOf := nwscommon.ParseTimeBestEffort(parsed.Updated)
if asOf.IsZero() { if asOf.IsZero() {
asOf = latestAlertTimestamp(parsed.Features) asOf = latestAlertTimestamp(parsed.Features)
} }
@@ -98,14 +90,14 @@ func buildAlerts(parsed nwsAlertsResponse, fallbackAsOf time.Time) (model.Weathe
id = fmt.Sprintf("nws:alert:%s:%d", asOf.UTC().Format(time.RFC3339Nano), i) id = fmt.Sprintf("nws:alert:%s:%d", asOf.UTC().Format(time.RFC3339Nano), i)
} }
sent := parseNWSTimePtr(p.Sent) sent := nwscommon.ParseTimePtr(p.Sent)
effective := parseNWSTimePtr(p.Effective) effective := nwscommon.ParseTimePtr(p.Effective)
onset := parseNWSTimePtr(p.Onset) onset := nwscommon.ParseTimePtr(p.Onset)
// Expires: prefer "expires"; fall back to "ends" if present. // Expires: prefer "expires"; fall back to "ends" if present.
expires := parseNWSTimePtr(p.Expires) expires := nwscommon.ParseTimePtr(p.Expires)
if expires == nil { if expires == nil {
expires = parseNWSTimePtr(p.Ends) expires = nwscommon.ParseTimePtr(p.Ends)
} }
refs := parseNWSAlertReferences(p.References) refs := parseNWSAlertReferences(p.References)
@@ -168,7 +160,7 @@ func latestAlertTimestamp(features []nwsAlertFeature) time.Time {
p.Onset, p.Onset,
} }
for _, s := range candidates { for _, s := range candidates {
t := parseNWSTimeUTC(s) t := nwscommon.ParseTimeBestEffort(s)
if t.IsZero() { if t.IsZero() {
continue continue
} }
@@ -180,28 +172,6 @@ func latestAlertTimestamp(features []nwsAlertFeature) time.Time {
return latest return latest
} }
// parseNWSTimeUTC parses an NWS timestamp string into UTC time.Time.
// Returns zero time on empty/unparseable input (best-effort; alerts should be resilient).
func parseNWSTimeUTC(s string) time.Time {
s = strings.TrimSpace(s)
if s == "" {
return time.Time{}
}
t, err := nwscommon.ParseTime(s)
if err != nil {
return time.Time{}
}
return t.UTC()
}
func parseNWSTimePtr(s string) *time.Time {
t := parseNWSTimeUTC(s)
if t.IsZero() {
return nil
}
return &t
}
func firstNonEmpty(vals ...string) string { func firstNonEmpty(vals ...string) string {
for _, v := range vals { for _, v := range vals {
if strings.TrimSpace(v) != "" { if strings.TrimSpace(v) != "" {
@@ -228,7 +198,7 @@ func parseNWSAlertReferences(raw json.RawMessage) []model.AlertReference {
ID: strings.TrimSpace(r.ID), ID: strings.TrimSpace(r.ID),
Identifier: strings.TrimSpace(r.Identifier), Identifier: strings.TrimSpace(r.Identifier),
Sender: strings.TrimSpace(r.Sender), Sender: strings.TrimSpace(r.Sender),
Sent: parseNWSTimePtr(r.Sent), Sent: nwscommon.ParseTimePtr(r.Sent),
} }
// If only Identifier is present, preserve it as ID too (useful downstream). // If only Identifier is present, preserve it as ID too (useful downstream).
if ref.ID == "" && ref.Identifier != "" { if ref.ID == "" && ref.Identifier != "" {

19
internal/normalizers/openmeteo/common.go
View File

@@ -1,19 +0,0 @@
// FILE: ./internal/normalizers/openmeteo/common.go
package openmeteo
import (
"time"
openmeteo "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/openmeteo"
)
// parseOpenMeteoTime parses Open-Meteo timestamps.
//
// The actual parsing logic lives in internal/providers/openmeteo so both the
// source (envelope EffectiveAt / event ID) and normalizer (canonical payload)
// can share identical timestamp behavior.
//
// We keep this thin wrapper to avoid churn in the normalizer package.
func parseOpenMeteoTime(s string, tz string, utcOffsetSeconds int) (time.Time, error) {
return openmeteo.ParseTime(s, tz, utcOffsetSeconds)
}

14
internal/normalizers/openmeteo/observation.go
View File

@@ -78,18 +78,8 @@ func buildObservation(parsed omResponse) (model.WeatherObservation, time.Time, e
canonicalText := standards.WMOText(wmo, isDay) canonicalText := standards.WMOText(wmo, isDay)
// Station identity: Open-Meteo is not a station feed; synthesize from coordinates. // Station identity: Open-Meteo is not a station feed; synthesize from coordinates.
stationID := "" // Require BOTH lat/lon to avoid misleading OPENMETEO(0.00000,...) IDs.
if parsed.Latitude != nil || parsed.Longitude != nil { stationID := normcommon.SynthStationIDPtr("OPENMETEO", parsed.Latitude, parsed.Longitude)
lat := 0.0
lon := 0.0
if parsed.Latitude != nil {
lat = *parsed.Latitude
}
if parsed.Longitude != nil {
lon = *parsed.Longitude
}
stationID = fmt.Sprintf("OPENMETEO(%.5f,%.5f)", lat, lon)
}
obs := model.WeatherObservation{ obs := model.WeatherObservation{
StationID: stationID, StationID: stationID,

4
internal/normalizers/openweather/common.go
View File

@@ -4,6 +4,8 @@ package openweather
import ( import (
"fmt" "fmt"
"strings" "strings"
normcommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/common"
) )
// This file holds provider-specific helpers that are shared across multiple // This file holds provider-specific helpers that are shared across multiple
@@ -67,5 +69,5 @@ func openWeatherStationID(parsed owmResponse) string {
return fmt.Sprintf("OPENWEATHER(%d)", parsed.ID) return fmt.Sprintf("OPENWEATHER(%d)", parsed.ID)
} }
// Fallback: synthesize from coordinates. // Fallback: synthesize from coordinates.
return fmt.Sprintf("OPENWEATHER(%.5f,%.5f)", parsed.Coord.Lat, parsed.Coord.Lon) return normcommon.SynthStationID("OPENWEATHER", parsed.Coord.Lat, parsed.Coord.Lon)
} }

23
internal/providers/nws/time.go
View File

@@ -25,3 +25,26 @@ func ParseTime(s string) (time.Time, error) {
return time.Time{}, fmt.Errorf("unsupported NWS timestamp format: %q", s) return time.Time{}, fmt.Errorf("unsupported NWS timestamp format: %q", s)
} }
// ParseTimeBestEffort parses an NWS timestamp and returns it in UTC.
//
// This is a convenience for normalizers that want "best effort" parsing:
// invalid/empty strings do not fail the entire normalization; they return zero time.
func ParseTimeBestEffort(s string) time.Time {
t, err := ParseTime(s)
if err != nil {
return time.Time{}
}
return t.UTC()
}
// ParseTimePtr parses an NWS timestamp and returns a UTC *time.Time.
//
// Empty/unparseable input returns nil. This is useful for optional CAP fields.
func ParseTimePtr(s string) *time.Time {
t := ParseTimeBestEffort(s)
if t.IsZero() {
return nil
}
return &t
}