From cefd4dfc7c43df8f45ee2fa74887f332b5763103 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 11 Jun 2026 00:15:40 +0000 Subject: [PATCH] Add raw SPC convective outlook source --- internal/sources/builtins.go | 4 + internal/sources/builtins_test.go | 14 +- internal/sources/spc/convective_outlook.go | 411 ++++++++++++++++++ .../sources/spc/convective_outlook_test.go | 359 +++++++++++++++ standards/schema.go | 2 + 5 files changed, 786 insertions(+), 4 deletions(-) create mode 100644 internal/sources/spc/convective_outlook.go create mode 100644 internal/sources/spc/convective_outlook_test.go diff --git a/internal/sources/builtins.go b/internal/sources/builtins.go index 85f9127..b7246bd 100644 --- a/internal/sources/builtins.go +++ b/internal/sources/builtins.go @@ -4,6 +4,7 @@ import ( "gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/nws" "gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/openmeteo" "gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/openweather" + "gitea.maximumdirect.net/ejr/weatherfeeder/internal/sources/spc" "gitea.maximumdirect.net/ejr/feedkit/config" fksource "gitea.maximumdirect.net/ejr/feedkit/sources" @@ -28,6 +29,9 @@ var pollDriverRegistrations = []pollDriverRegistration{ {driver: "openweather_observation", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return openweather.NewObservationSource(cfg) }}, + {driver: "spc_convective_outlook", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { + return spc.NewConvectiveOutlookSource(cfg) + }}, } // RegisterBuiltins registers the source drivers that ship with this binary. diff --git a/internal/sources/builtins_test.go b/internal/sources/builtins_test.go index 2eec97e..e1bd186 100644 --- a/internal/sources/builtins_test.go +++ b/internal/sources/builtins_test.go @@ -87,6 +87,7 @@ func TestRegisterBuiltinsRegistersAllCurrentDrivers(t *testing.T) { "openmeteo_observation", "openmeteo_forecast", "openweather_observation", + "spc_convective_outlook", } for _, driver := range drivers { @@ -105,13 +106,18 @@ func sourceConfigForDriver(driver string) config.SourceConfig { if driver == "openweather_observation" { url = "https://example.invalid?units=metric" } + params := map[string]any{ + "url": url, + "user_agent": "test-agent", + } + if driver == "spc_convective_outlook" { + params["latitude"] = 38.6239 + params["longitude"] = -90.3571 + } return config.SourceConfig{ Name: "test-source", Driver: driver, Mode: config.SourceModePoll, - Params: map[string]any{ - "url": url, - "user_agent": "test-agent", - }, + Params: params, } } diff --git a/internal/sources/spc/convective_outlook.go b/internal/sources/spc/convective_outlook.go new file mode 100644 index 0000000..89c2c44 --- /dev/null +++ b/internal/sources/spc/convective_outlook.go @@ -0,0 +1,411 @@ +package spc + +import ( + "context" + "crypto/sha256" + "encoding/json" + "fmt" + "math" + "net/http" + "strconv" + "strings" + "time" + + "gitea.maximumdirect.net/ejr/feedkit/config" + "gitea.maximumdirect.net/ejr/feedkit/event" + fksources "gitea.maximumdirect.net/ejr/feedkit/sources" + "gitea.maximumdirect.net/ejr/feedkit/transport" + spcprovider "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/spc" + "gitea.maximumdirect.net/ejr/weatherfeeder/standards" +) + +const ( + driverConvectiveOutlook = "spc_convective_outlook" + + acceptGeoJSON = "application/geo+json, application/json" + acceptDiscussion = "text/html, application/xhtml+xml" + acceptRSS = "application/rss+xml, application/xml, text/xml" +) + +type fetchProduct struct { + Key string + Day int + OutlookType string + URL string + Accept string +} + +// ConvectiveOutlookSource polls SPC Day 1-3 convective outlook products and +// emits one raw outlook bundle event. +type ConvectiveOutlookSource struct { + name string + userAgent string + locationID string + locationName string + latitude float64 + longitude float64 + client *http.Client + bodyLimit int64 + + geoJSONProducts []fetchProduct + discussions []fetchProduct + rssURL string + + lastHash [sha256.Size]byte + hasHash bool +} + +func NewConvectiveOutlookSource(cfg config.SourceConfig) (*ConvectiveOutlookSource, error) { + name := strings.TrimSpace(cfg.Name) + if name == "" { + return nil, fmt.Errorf("%s: name is required", driverConvectiveOutlook) + } + if cfg.Params == nil { + return nil, fmt.Errorf("%s %q: params are required", driverConvectiveOutlook, name) + } + + userAgent, ok := cfg.ParamString("user_agent", "userAgent") + if !ok { + return nil, fmt.Errorf("%s %q: params.user_agent is required", driverConvectiveOutlook, name) + } + + latitude, err := requireFloatParam(cfg, "latitude") + if err != nil { + return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err) + } + longitude, err := requireFloatParam(cfg, "longitude") + if err != nil { + return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err) + } + + timeout := transport.DefaultHTTPTimeout + if _, exists := cfg.Params["http_timeout"]; exists { + var ok bool + timeout, ok = cfg.ParamDuration("http_timeout") + if !ok || timeout <= 0 { + return nil, fmt.Errorf("source %q: params.http_timeout must be a positive duration", name) + } + } + + bodyLimit := transport.DefaultHTTPResponseBodyLimitBytes + if _, exists := cfg.Params["http_response_body_limit_bytes"]; exists { + rawLimit, ok := cfg.ParamInt("http_response_body_limit_bytes") + if !ok || rawLimit <= 0 { + return nil, fmt.Errorf("source %q: params.http_response_body_limit_bytes must be a positive integer", name) + } + bodyLimit = int64(rawLimit) + } + + geoJSONProducts, err := configuredGeoJSONProducts(cfg) + if err != nil { + return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err) + } + discussions, err := configuredDiscussionProducts(cfg) + if err != nil { + return nil, fmt.Errorf("%s %q: %w", driverConvectiveOutlook, name, err) + } + + rssURL := "" + if rawRSSURL, ok := cfg.ParamString("rss_url", "rssURL"); ok { + rssURL = rawRSSURL + } + + locationID, _ := cfg.ParamString("location_id", "locationID") + locationName, _ := cfg.ParamString("location_name", "locationName") + + return &ConvectiveOutlookSource{ + name: name, + userAgent: userAgent, + locationID: locationID, + locationName: locationName, + latitude: latitude, + longitude: longitude, + client: transport.NewHTTPClient(timeout), + bodyLimit: bodyLimit, + geoJSONProducts: geoJSONProducts, + discussions: discussions, + rssURL: rssURL, + }, nil +} + +func (s *ConvectiveOutlookSource) Name() string { return s.name } + +func (s *ConvectiveOutlookSource) Kinds() []event.Kind { + return []event.Kind{event.Kind("outlook")} +} + +func (s *ConvectiveOutlookSource) Poll(ctx context.Context) ([]event.Event, error) { + fetchedAt := time.Now().UTC() + bundle := spcprovider.RawConvectiveOutlookBundle{ + LocationID: s.locationID, + LocationName: s.locationName, + Latitude: s.latitude, + Longitude: s.longitude, + FetchedAt: fetchedAt, + Products: make([]spcprovider.RawOutlookProduct, 0, len(s.geoJSONProducts)), + Discussions: make([]spcprovider.RawDiscussionPage, 0, len(s.discussions)), + } + hash := sha256.New() + + var latestIssue time.Time + for _, product := range s.geoJSONProducts { + body, err := s.fetch(ctx, product.URL, product.Accept) + if err != nil { + return nil, fmt.Errorf("fetch geojson %s: %w", product.Key, err) + } + addHashPart(hash, product.Key, product.URL, body) + + bundle.Products = append(bundle.Products, spcprovider.RawOutlookProduct{ + Key: product.Key, + Day: product.Day, + OutlookType: product.OutlookType, + URL: product.URL, + FetchedAt: fetchedAt, + Body: json.RawMessage(body), + }) + if t := latestIssueTime(body); !t.IsZero() && (latestIssue.IsZero() || t.After(latestIssue)) { + latestIssue = t + } + } + + var latestUpdated time.Time + for _, product := range s.discussions { + body, err := s.fetch(ctx, product.URL, product.Accept) + if err != nil { + return nil, fmt.Errorf("fetch discussion %s: %w", product.Key, err) + } + addHashPart(hash, product.Key, product.URL, body) + + bodyText := string(body) + bundle.Discussions = append(bundle.Discussions, spcprovider.RawDiscussionPage{ + Key: product.Key, + Day: product.Day, + URL: product.URL, + FetchedAt: fetchedAt, + Body: bodyText, + }) + if t := discussionUpdatedTime(bodyText); !t.IsZero() && (latestUpdated.IsZero() || t.After(latestUpdated)) { + latestUpdated = t + } + } + + var rssBuild time.Time + if s.rssURL != "" { + body, err := s.fetch(ctx, s.rssURL, acceptRSS) + if err != nil { + return nil, fmt.Errorf("fetch rss: %w", err) + } + addHashPart(hash, "rss", s.rssURL, body) + + bodyText := string(body) + bundle.RSS = &spcprovider.RawRSSFeed{ + URL: s.rssURL, + FetchedAt: fetchedAt, + Body: bodyText, + } + if feed, err := spcprovider.ParseRSSFeed(bodyText); err == nil && feed.LastBuildDate != nil { + rssBuild = feed.LastBuildDate.UTC() + } + } + + var currentHash [sha256.Size]byte + copy(currentHash[:], hash.Sum(nil)) + if s.hasHash && currentHash == s.lastHash { + return nil, nil + } + s.lastHash = currentHash + s.hasHash = true + + effectiveAt := chooseEffectiveTime(latestIssue, latestUpdated, rssBuild, fetchedAt) + emittedAt := time.Now().UTC() + eventID := fksources.DefaultEventID("", s.name, &effectiveAt, emittedAt) + + return fksources.SingleEvent( + event.Kind("outlook"), + s.name, + standards.SchemaRawSPCConvectiveOutlookV1, + eventID, + emittedAt, + &effectiveAt, + bundle, + ) +} + +func (s *ConvectiveOutlookSource) fetch(ctx context.Context, url, accept string) ([]byte, error) { + return transport.FetchBodyWithLimit(ctx, s.client, url, s.userAgent, accept, s.bodyLimit) +} + +func configuredGeoJSONProducts(cfg config.SourceConfig) ([]fetchProduct, error) { + overrides, err := optionalStringMap(cfg, "geojson_urls") + if err != nil { + return nil, err + } + out := make([]fetchProduct, 0, len(spcprovider.GeoJSONProducts())) + for _, product := range spcprovider.GeoJSONProducts() { + url := product.URL + if override := strings.TrimSpace(overrides[product.Key]); override != "" { + url = override + } + out = append(out, fetchProduct{ + Key: product.Key, + Day: product.Day, + OutlookType: product.OutlookType, + URL: url, + Accept: acceptGeoJSON, + }) + } + return out, nil +} + +func configuredDiscussionProducts(cfg config.SourceConfig) ([]fetchProduct, error) { + overrides, err := optionalStringMap(cfg, "discussion_urls") + if err != nil { + return nil, err + } + out := make([]fetchProduct, 0, len(spcprovider.DiscussionProducts())) + for _, product := range spcprovider.DiscussionProducts() { + url := product.URL + if override := strings.TrimSpace(overrides[product.Key]); override != "" { + url = override + } + out = append(out, fetchProduct{ + Key: product.Key, + Day: product.Day, + URL: url, + Accept: acceptDiscussion, + }) + } + return out, nil +} + +func optionalStringMap(cfg config.SourceConfig, key string) (map[string]string, error) { + raw, ok := cfg.Params[key] + if !ok || raw == nil { + return map[string]string{}, nil + } + + out := map[string]string{} + switch typed := raw.(type) { + case map[string]string: + for k, v := range typed { + if strings.TrimSpace(k) != "" && strings.TrimSpace(v) != "" { + out[strings.TrimSpace(k)] = strings.TrimSpace(v) + } + } + case map[string]any: + for k, v := range typed { + s, ok := v.(string) + if !ok { + return nil, fmt.Errorf("params.%s[%q] must be a string", key, k) + } + if strings.TrimSpace(k) != "" && strings.TrimSpace(s) != "" { + out[strings.TrimSpace(k)] = strings.TrimSpace(s) + } + } + default: + return nil, fmt.Errorf("params.%s must be a string map", key) + } + return out, nil +} + +func requireFloatParam(cfg config.SourceConfig, key string) (float64, error) { + raw, ok := cfg.Params[key] + if !ok || raw == nil { + return 0, fmt.Errorf("params.%s is required", key) + } + v, ok := numberFromAny(raw) + if !ok { + return 0, fmt.Errorf("params.%s must be a number", key) + } + if math.IsNaN(v) || math.IsInf(v, 0) { + return 0, fmt.Errorf("params.%s must be finite", key) + } + return v, nil +} + +func numberFromAny(raw any) (float64, bool) { + switch v := raw.(type) { + case float64: + return v, true + case float32: + return float64(v), true + case int: + return float64(v), true + case int8: + return float64(v), true + case int16: + return float64(v), true + case int32: + return float64(v), true + case int64: + return float64(v), true + case uint: + return float64(v), true + case uint8: + return float64(v), true + case uint16: + return float64(v), true + case uint32: + return float64(v), true + case uint64: + return float64(v), true + case string: + parsed, err := strconv.ParseFloat(strings.TrimSpace(v), 64) + return parsed, err == nil + default: + return 0, false + } +} + +func latestIssueTime(raw []byte) time.Time { + collection, err := spcprovider.DecodeGeoJSON(raw) + if err != nil { + return time.Time{} + } + var latest time.Time + for _, feature := range collection.Features { + t, err := spcprovider.ParseISOTimestamp(feature.Properties.IssueISO) + if err != nil { + continue + } + t = t.UTC() + if latest.IsZero() || t.After(latest) { + latest = t + } + } + return latest +} + +func discussionUpdatedTime(rawHTML string) time.Time { + text, err := spcprovider.ExtractProductText(rawHTML) + if err != nil { + return time.Time{} + } + t := spcprovider.ParseUpdatedTimestamp(text) + if t == nil { + return time.Time{} + } + return t.UTC() +} + +func chooseEffectiveTime(issue time.Time, updated time.Time, rss time.Time, fetched time.Time) time.Time { + switch { + case !issue.IsZero(): + return issue.UTC() + case !updated.IsZero(): + return updated.UTC() + case !rss.IsZero(): + return rss.UTC() + default: + return fetched.UTC() + } +} + +func addHashPart(hash interface{ Write([]byte) (int, error) }, key, url string, body []byte) { + _, _ = hash.Write([]byte(key)) + _, _ = hash.Write([]byte{0}) + _, _ = hash.Write([]byte(url)) + _, _ = hash.Write([]byte{0}) + _, _ = hash.Write(body) + _, _ = hash.Write([]byte{0}) +} diff --git a/internal/sources/spc/convective_outlook_test.go b/internal/sources/spc/convective_outlook_test.go new file mode 100644 index 0000000..507bb92 --- /dev/null +++ b/internal/sources/spc/convective_outlook_test.go @@ -0,0 +1,359 @@ +package spc + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/ejr/feedkit/config" + "gitea.maximumdirect.net/ejr/feedkit/event" + spcprovider "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/spc" + "gitea.maximumdirect.net/ejr/weatherfeeder/standards" +) + +func TestConvectiveOutlookSourceKinds(t *testing.T) { + src, err := NewConvectiveOutlookSource(convectiveOutlookConfig(map[string]any{})) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource() error = %v", err) + } + got := src.Kinds() + if len(got) != 1 || got[0] != event.Kind("outlook") { + t.Fatalf("Kinds() = %#v, want [outlook]", got) + } +} + +func TestConvectiveOutlookSourceRequiresLatitudeAndLongitude(t *testing.T) { + for _, key := range []string{"latitude", "longitude"} { + cfg := convectiveOutlookConfig(map[string]any{}) + delete(cfg.Params, key) + + _, err := NewConvectiveOutlookSource(cfg) + if err == nil { + t.Fatalf("NewConvectiveOutlookSource() without %s error = nil, want error", key) + } + if !strings.Contains(err.Error(), "params."+key+" is required") { + t.Fatalf("error = %q, want missing %s", err, key) + } + } +} + +func TestConvectiveOutlookSourcePollEmitsRawBundle(t *testing.T) { + srv := newSPCTestServer(t, spcServerOptions{}) + src, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, false))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource() error = %v", err) + } + + events, err := src.Poll(t.Context()) + if err != nil { + t.Fatalf("Poll() error = %v", err) + } + if len(events) != 1 { + t.Fatalf("Poll() returned %d events, want 1", len(events)) + } + got := events[0] + if got.Kind != event.Kind("outlook") { + t.Fatalf("Kind = %q, want outlook", got.Kind) + } + if got.Schema != standards.SchemaRawSPCConvectiveOutlookV1 { + t.Fatalf("Schema = %q, want %q", got.Schema, standards.SchemaRawSPCConvectiveOutlookV1) + } + wantEffective := time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC) + if got.EffectiveAt == nil || !got.EffectiveAt.Equal(wantEffective) { + t.Fatalf("EffectiveAt = %v, want %s", got.EffectiveAt, wantEffective) + } + + bundle, ok := got.Payload.(spcprovider.RawConvectiveOutlookBundle) + if !ok { + t.Fatalf("Payload type = %T, want RawConvectiveOutlookBundle", got.Payload) + } + if bundle.LocationID != "stl" || bundle.LocationName != "St. Louis, MO" { + t.Fatalf("location metadata = %q/%q", bundle.LocationID, bundle.LocationName) + } + if bundle.Latitude != 38.6239 || bundle.Longitude != -90.3571 { + t.Fatalf("coordinates = %v,%v", bundle.Latitude, bundle.Longitude) + } + if len(bundle.Products) != 12 { + t.Fatalf("Products length = %d, want 12", len(bundle.Products)) + } + if len(bundle.Discussions) != 3 { + t.Fatalf("Discussions length = %d, want 3", len(bundle.Discussions)) + } + if bundle.RSS != nil { + t.Fatalf("RSS = %#v, want nil", bundle.RSS) + } +} + +func TestConvectiveOutlookSourceEffectiveAtFallsBackToDiscussionUpdated(t *testing.T) { + srv := newSPCTestServer(t, spcServerOptions{blankIssueISO: true}) + src, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, false))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource() error = %v", err) + } + + events, err := src.Poll(t.Context()) + if err != nil { + t.Fatalf("Poll() error = %v", err) + } + want := time.Date(2026, 6, 11, 20, 0, 0, 0, time.UTC) + if events[0].EffectiveAt == nil || !events[0].EffectiveAt.Equal(want) { + t.Fatalf("EffectiveAt = %v, want %s", events[0].EffectiveAt, want) + } +} + +func TestConvectiveOutlookSourceIncludesRSSOnlyWhenConfigured(t *testing.T) { + srv := newSPCTestServer(t, spcServerOptions{}) + + withoutRSS, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, false))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource(without RSS) error = %v", err) + } + events, err := withoutRSS.Poll(t.Context()) + if err != nil { + t.Fatalf("Poll(without RSS) error = %v", err) + } + if events[0].Payload.(spcprovider.RawConvectiveOutlookBundle).RSS != nil { + t.Fatalf("RSS present without rss_url") + } + + withRSS, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, true))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource(with RSS) error = %v", err) + } + events, err = withRSS.Poll(t.Context()) + if err != nil { + t.Fatalf("Poll(with RSS) error = %v", err) + } + if events[0].Payload.(spcprovider.RawConvectiveOutlookBundle).RSS == nil { + t.Fatalf("RSS missing with rss_url") + } +} + +func TestConvectiveOutlookSourceUnchangedResponseEmitsNoEvents(t *testing.T) { + srv := newSPCTestServer(t, spcServerOptions{}) + src, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, false))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource() error = %v", err) + } + + events, err := src.Poll(t.Context()) + if err != nil { + t.Fatalf("first Poll() error = %v", err) + } + if len(events) != 1 { + t.Fatalf("first Poll() events = %d, want 1", len(events)) + } + + events, err = src.Poll(t.Context()) + if err != nil { + t.Fatalf("second Poll() error = %v", err) + } + if len(events) != 0 { + t.Fatalf("second Poll() events = %d, want 0", len(events)) + } +} + +func TestConvectiveOutlookSourceGeoJSONFailureReturnsError(t *testing.T) { + srv := newSPCTestServer(t, spcServerOptions{failGeoJSONKey: "day2_wind"}) + src, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, false))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource() error = %v", err) + } + + events, err := src.Poll(t.Context()) + if err == nil { + t.Fatalf("Poll() error = nil, want error") + } + if len(events) != 0 { + t.Fatalf("Poll() events = %d, want 0", len(events)) + } + if !strings.Contains(err.Error(), "fetch geojson day2_wind") { + t.Fatalf("error = %q", err) + } +} + +func TestConvectiveOutlookSourceDiscussionFailureReturnsError(t *testing.T) { + srv := newSPCTestServer(t, spcServerOptions{failDiscussionKey: "day2"}) + src, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, false))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource() error = %v", err) + } + + events, err := src.Poll(t.Context()) + if err == nil { + t.Fatalf("Poll() error = nil, want error") + } + if len(events) != 0 { + t.Fatalf("Poll() events = %d, want 0", len(events)) + } + if !strings.Contains(err.Error(), "fetch discussion day2") { + t.Fatalf("error = %q", err) + } +} + +func TestConvectiveOutlookSourcePayloadJSONShape(t *testing.T) { + srv := newSPCTestServer(t, spcServerOptions{}) + src, err := NewConvectiveOutlookSource(convectiveOutlookConfig(serverOverrideParams(srv.URL, false))) + if err != nil { + t.Fatalf("NewConvectiveOutlookSource() error = %v", err) + } + + events, err := src.Poll(t.Context()) + if err != nil { + t.Fatalf("Poll() error = %v", err) + } + raw, err := json.Marshal(events[0].Payload) + if err != nil { + t.Fatalf("Marshal(payload) error = %v", err) + } + if !strings.Contains(string(raw), `"products"`) || !strings.Contains(string(raw), `"discussions"`) { + t.Fatalf("payload JSON missing raw bundle fields: %s", raw) + } +} + +type spcServerOptions struct { + blankIssueISO bool + failGeoJSONKey string + failDiscussionKey string +} + +func newSPCTestServer(t *testing.T, opts spcServerOptions) *httptest.Server { + t.Helper() + + mux := http.NewServeMux() + for _, product := range spcprovider.GeoJSONProducts() { + product := product + mux.HandleFunc("/geojson/"+product.Key, func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Accept") != acceptGeoJSON { + t.Errorf("geojson Accept = %q, want %q", r.Header.Get("Accept"), acceptGeoJSON) + } + if product.Key == opts.failGeoJSONKey { + http.Error(w, "failed", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/geo+json") + _, _ = w.Write(geoJSONFixtureForProduct(t, product.Key, opts.blankIssueISO)) + }) + } + for _, product := range spcprovider.DiscussionProducts() { + product := product + mux.HandleFunc("/discussion/"+product.Key, func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Accept") != acceptDiscussion { + t.Errorf("discussion Accept = %q, want %q", r.Header.Get("Accept"), acceptDiscussion) + } + if product.Key == opts.failDiscussionKey { + http.Error(w, "failed", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "text/html") + _, _ = w.Write(discussionFixtureForProduct(t, product.Key)) + }) + } + mux.HandleFunc("/rss", func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Accept") != acceptRSS { + t.Errorf("rss Accept = %q, want %q", r.Header.Get("Accept"), acceptRSS) + } + w.Header().Set("Content-Type", "application/rss+xml") + _, _ = w.Write([]byte(testRSS)) + }) + return httptest.NewServer(mux) +} + +func convectiveOutlookConfig(extra map[string]any) config.SourceConfig { + params := map[string]any{ + "latitude": 38.6239, + "longitude": -90.3571, + "location_id": "stl", + "location_name": "St. Louis, MO", + "user_agent": "test-agent", + } + for k, v := range extra { + params[k] = v + } + return config.SourceConfig{ + Name: "spc-test", + Driver: driverConvectiveOutlook, + Mode: config.SourceModePoll, + Params: params, + } +} + +func serverOverrideParams(baseURL string, includeRSS bool) map[string]any { + geoJSONURLs := map[string]any{} + for _, product := range spcprovider.GeoJSONProducts() { + geoJSONURLs[product.Key] = baseURL + "/geojson/" + product.Key + } + discussionURLs := map[string]any{} + for _, product := range spcprovider.DiscussionProducts() { + discussionURLs[product.Key] = baseURL + "/discussion/" + product.Key + } + + out := map[string]any{ + "geojson_urls": geoJSONURLs, + "discussion_urls": discussionURLs, + } + if includeRSS { + out["rss_url"] = baseURL + "/rss" + } + return out +} + +func geoJSONFixtureForProduct(t *testing.T, key string, blankIssueISO bool) []byte { + t.Helper() + var name string + switch { + case strings.HasPrefix(key, "day1_"): + name = "day1_cat.geojson" + case strings.HasPrefix(key, "day2_"): + name = "day2_torn.geojson" + case strings.HasPrefix(key, "day3_"): + name = "day3_wind.geojson" + default: + t.Fatalf("unknown product key %q", key) + } + raw := readSPCTestFixture(t, name) + if blankIssueISO { + raw = []byte(strings.ReplaceAll(string(raw), `"ISSUE_ISO": "2026-06-11T12:34:56Z"`, `"ISSUE_ISO": ""`)) + raw = []byte(strings.ReplaceAll(string(raw), `"ISSUE_ISO": "2026-06-11T17:30:00Z"`, `"ISSUE_ISO": ""`)) + raw = []byte(strings.ReplaceAll(string(raw), `"ISSUE_ISO": "2026-06-11T19:45:00Z"`, `"ISSUE_ISO": ""`)) + } + return raw +} + +func discussionFixtureForProduct(t *testing.T, key string) []byte { + t.Helper() + switch key { + case "day1": + return readSPCTestFixture(t, "day1_prt.html") + case "day2": + return readSPCTestFixture(t, "day2_prt_corr.html") + case "day3": + return readSPCTestFixture(t, "day3_prt.html") + default: + t.Fatalf("unknown discussion key %q", key) + return nil + } +} + +func readSPCTestFixture(t *testing.T, name string) []byte { + t.Helper() + path := filepath.Join("..", "..", "providers", "spc", "testdata", name) + raw, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read fixture %s: %v", path, err) + } + return raw +} + +const testRSS = ` + + + SPC AC RSS + Thu, 11 Jun 2026 21:00:00 +0000 + +` diff --git a/standards/schema.go b/standards/schema.go index ec5ca6f..15565a5 100644 --- a/standards/schema.go +++ b/standards/schema.go @@ -24,6 +24,8 @@ const ( SchemaRawNWSAlertsV1 = "raw.nws.alerts.v1" + SchemaRawSPCConvectiveOutlookV1 = "raw.spc.convective_outlook.v1" + // Canonical domain schemas (emitted after normalization). SchemaWeatherObservationV1 = "weather.observation.v1" SchemaWeatherForecastV1 = "weather.forecast.v1"