diff --git a/API.md b/API.md index 5cc2175..5e93758 100644 --- a/API.md +++ b/API.md @@ -37,11 +37,12 @@ Examples: ## Canonical schemas -weatherfeeder emits four canonical domain schemas: +weatherfeeder emits five canonical domain schemas: - `weather.observation.v1` - `weather.forecast.v1` - `weather.forecast_discussion.v1` +- `weather.weather_story.v1` - `weather.alert.v1` Each payload is described below using the JSON field names as the contract. @@ -49,11 +50,14 @@ Each payload is described below using the JSON field names as the contract. ### Raw upstream schemas weatherfeeder sources also emit provider-specific raw schemas before normalization. -For this feature, the raw source schema is: +Relevant raw source schemas include: - `raw.nws.forecast_discussion.v1` - payload type: string - payload contents: exact fetched HTML response body +- `raw.nws.weatherstories.v1` + - payload type: object + - payload contents: exact fetched JSON response body --- @@ -227,6 +231,38 @@ A run may contain zero, one, or many alerts. --- +## Schema: `weather.weather_story.v1` + +Payload type: `WeatherStoryRun` + +A `WeatherStoryRun` is a snapshot of NWS weather stories for an office as-of a point in time. +The run may contain zero, one, or many stories. + +### Fields + +| Field | Type | Required | Notes | +|---|---:|:---:|---| +| `officeId` | string | no | NWS office identifier, e.g. `LSX` | +| `asOf` | string (timestamp) | yes | Latest story update time or source fallback | +| `stories` | array | yes | Weather stories (order provider-dependent) | + +### Nested: `stories[]` (`WeatherStory`) + +| Field | Type | Required | Notes | +|---|---:|:---:|---| +| `officeId` | string | no | NWS office identifier | +| `startTime` | string (timestamp) | yes | Story validity start | +| `endTime` | string (timestamp) | yes | Story validity end | +| `updatedAt` | string (timestamp) | yes | Story update time | +| `title` | string | no | Human story title | +| `description` | string | no | Story narrative text | +| `altText` | string | no | Accessibility text for the provider graphic | +| `priority` | bool | yes | Provider priority flag | +| `order` | int | yes | Provider display order | +| `downloadUrl` | string | no | Provider download URL; weatherfeeder does not fetch the asset | + +--- + ## Schema: `weather.forecast_discussion.v1` Payload type: `WeatherForecastDiscussion` @@ -337,3 +373,32 @@ It is distinct from `weather.forecast.v1`, which is period-based. } } ``` + +### Weather story event (`weather.weather_story.v1`) + +```json +{ + "id": "nws:weatherstories:2026-05-30T09:00:34Z", + "schema": "weather.weather_story.v1", + "source": "nws_weatherstories", + "effectiveAt": "2026-05-30T09:00:34Z", + "payload": { + "officeId": "LSX", + "asOf": "2026-05-30T09:00:34Z", + "stories": [ + { + "officeId": "LSX", + "startTime": "2026-05-30T08:46:00Z", + "endTime": "2026-05-31T11:00:00Z", + "updatedAt": "2026-05-30T09:00:34Z", + "title": "Several Chances for Rain Through Monday", + "description": "Scattered showers and thunderstorms remain possible.", + "altText": "This slide shows the forecast for today through Tuesday.", + "priority": false, + "order": 1, + "downloadUrl": "https://api.weather.gov/offices/LSX/weatherstories/download/3228e499-2aae-45a8-9ff9-1c060311026f" + } + ] + } +} +``` diff --git a/README.md b/README.md index 81ec803..71a13ce 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ Canonical domain schemas emitted after normalization: - `weather.observation.v1` → `WeatherObservation` - `weather.forecast.v1` → `WeatherForecastRun` - `weather.forecast_discussion.v1` → `WeatherForecastDiscussion` +- `weather.weather_story.v1` → `WeatherStoryRun` - `weather.alert.v1` → `WeatherAlertRun` For the complete wire contract (event envelope + payload schemas, fields, units, and compatibility rules), see: @@ -23,7 +24,7 @@ For the complete wire contract (event envelope + payload schemas, fields, units, ## Upstream providers (current MVP) -- NWS: observations, hourly forecasts, narrative forecasts, forecast discussions, alerts +- NWS: observations, hourly forecasts, narrative forecasts, forecast discussions, weather stories, alerts - Open-Meteo: observations, hourly forecasts - OpenWeather: observations diff --git a/cmd/weatherfeeder/config.yml b/cmd/weatherfeeder/config.yml index c5501b0..9539e2a 100644 --- a/cmd/weatherfeeder/config.yml +++ b/cmd/weatherfeeder/config.yml @@ -72,6 +72,15 @@ sources: url: "https://forecast.weather.gov/product.php?site=LSX&issuedby=LSX&product=AFD&format=TXT&version=1&glossary=0" user_agent: "HomeOps (eric@maximumdirect.net)" + - name: NWSWeatherStoriesSTL + mode: poll + kinds: ["weather_story"] + driver: nws_weatherstories + every: 30m + params: + url: "https://api.weather.gov/offices/LSX/weatherstories" + user_agent: "HomeOps (eric@maximumdirect.net)" + - name: OpenMeteoHourlyForecastSTL mode: poll kinds: ["forecast"] @@ -117,13 +126,13 @@ sinks: routes: - sink: stdout - kinds: ["observation", "forecast", "forecast_discussion", "alert"] + kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"] - sink: nats_weatherfeeder - kinds: ["observation", "forecast", "forecast_discussion", "alert"] + kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"] # - sink: pg_weatherfeeder -# kinds: ["observation", "forecast", "forecast_discussion", "alert"] +# kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"] # - sink: logfile -# kinds: ["observation", "alert", "forecast", "forecast_discussion"] +# kinds: ["observation", "alert", "forecast", "forecast_discussion", "weather_story"] diff --git a/internal/normalizers/builtins_test.go b/internal/normalizers/builtins_test.go index 7c40ffb..11068fa 100644 --- a/internal/normalizers/builtins_test.go +++ b/internal/normalizers/builtins_test.go @@ -20,6 +20,7 @@ func TestRegisterBuiltinsOrder(t *testing.T) { nws.ObservationNormalizer{}, nws.ForecastNormalizer{}, nws.ForecastDiscussionNormalizer{}, + nws.WeatherStoriesNormalizer{}, nws.AlertsNormalizer{}, openmeteo.ObservationNormalizer{}, openmeteo.ForecastNormalizer{}, diff --git a/internal/normalizers/nws/register.go b/internal/normalizers/nws/register.go index 2d01a0a..80d9b4a 100644 --- a/internal/normalizers/nws/register.go +++ b/internal/normalizers/nws/register.go @@ -9,6 +9,7 @@ var builtins = []fknormalize.Normalizer{ ObservationNormalizer{}, ForecastNormalizer{}, ForecastDiscussionNormalizer{}, + WeatherStoriesNormalizer{}, AlertsNormalizer{}, } diff --git a/internal/normalizers/nws/types.go b/internal/normalizers/nws/types.go index af132f9..a4509d7 100644 --- a/internal/normalizers/nws/types.go +++ b/internal/normalizers/nws/types.go @@ -262,6 +262,25 @@ type nwsAlertProperties struct { References json.RawMessage `json:"references"` } +// nwsWeatherStoriesResponse is a minimal representation of the NWS /weatherstories +// payload needed for mapping into model.WeatherStoryRun. +type nwsWeatherStoriesResponse struct { + Stories []nwsWeatherStory `json:"stories"` +} + +type nwsWeatherStory struct { + OfficeID string `json:"officeId"` + StartTime string `json:"startTime"` + EndTime string `json:"endTime"` + UpdateTime string `json:"updateTime"` + Title string `json:"title"` + Description string `json:"description"` + AltText string `json:"altText"` + Priority bool `json:"priority"` + Order int `json:"order"` + Download string `json:"download"` +} + type nwsAlertReference struct { ID string `json:"id"` Identifier string `json:"identifier"` diff --git a/internal/normalizers/nws/weatherstories.go b/internal/normalizers/nws/weatherstories.go new file mode 100644 index 0000000..648a95f --- /dev/null +++ b/internal/normalizers/nws/weatherstories.go @@ -0,0 +1,107 @@ +package nws + +import ( + "context" + "fmt" + "strings" + "time" + + "gitea.maximumdirect.net/ejr/feedkit/event" + normcommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/common" + nwscommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/nws" + "gitea.maximumdirect.net/ejr/weatherfeeder/model" + "gitea.maximumdirect.net/ejr/weatherfeeder/standards" +) + +// WeatherStoriesNormalizer converts: +// +// standards.SchemaRawNWSWeatherStoriesV1 -> standards.SchemaWeatherStoryV1 +// +// It maps the NWS /weatherstories JSON response into a canonical story snapshot. +type WeatherStoriesNormalizer struct{} + +func (WeatherStoriesNormalizer) Match(e event.Event) bool { + return strings.TrimSpace(e.Schema) == standards.SchemaRawNWSWeatherStoriesV1 +} + +func (WeatherStoriesNormalizer) Normalize(ctx context.Context, in event.Event) (*event.Event, error) { + _ = ctx + + fallbackAsOf := in.EmittedAt.UTC() + if in.EffectiveAt != nil && !in.EffectiveAt.IsZero() { + fallbackAsOf = in.EffectiveAt.UTC() + } + + return normcommon.NormalizeJSON( + in, + "nws weatherstories", + standards.SchemaWeatherStoryV1, + func(parsed nwsWeatherStoriesResponse) (model.WeatherStoryRun, time.Time, error) { + return buildWeatherStories(parsed, fallbackAsOf) + }, + ) +} + +func buildWeatherStories(parsed nwsWeatherStoriesResponse, fallbackAsOf time.Time) (model.WeatherStoryRun, time.Time, error) { + stories := make([]model.WeatherStory, 0, len(parsed.Stories)) + + var officeID string + var asOf time.Time + for i, raw := range parsed.Stories { + startTime, err := parseRequiredNWSTime(raw.StartTime, fmt.Sprintf("stories[%d].startTime", i)) + if err != nil { + return model.WeatherStoryRun{}, time.Time{}, err + } + endTime, err := parseRequiredNWSTime(raw.EndTime, fmt.Sprintf("stories[%d].endTime", i)) + if err != nil { + return model.WeatherStoryRun{}, time.Time{}, err + } + updatedAt, err := parseRequiredNWSTime(raw.UpdateTime, fmt.Sprintf("stories[%d].updateTime", i)) + if err != nil { + return model.WeatherStoryRun{}, time.Time{}, err + } + + storyOfficeID := strings.TrimSpace(raw.OfficeID) + if officeID == "" && storyOfficeID != "" { + officeID = storyOfficeID + } + if asOf.IsZero() || updatedAt.After(asOf) { + asOf = updatedAt + } + + stories = append(stories, model.WeatherStory{ + OfficeID: storyOfficeID, + StartTime: startTime, + EndTime: endTime, + UpdatedAt: updatedAt, + Title: strings.TrimSpace(raw.Title), + Description: strings.TrimSpace(raw.Description), + AltText: strings.TrimSpace(raw.AltText), + Priority: raw.Priority, + Order: raw.Order, + DownloadURL: strings.TrimSpace(raw.Download), + }) + } + + if asOf.IsZero() { + asOf = fallbackAsOf.UTC() + } + + run := model.WeatherStoryRun{ + OfficeID: officeID, + AsOf: asOf, + Stories: stories, + } + return run, asOf, nil +} + +func parseRequiredNWSTime(raw, field string) (time.Time, error) { + if strings.TrimSpace(raw) == "" { + return time.Time{}, fmt.Errorf("%s is required", field) + } + t, err := nwscommon.ParseTime(raw) + if err != nil { + return time.Time{}, fmt.Errorf("%s: %w", field, err) + } + return t.UTC(), nil +} diff --git a/internal/normalizers/nws/weatherstories_test.go b/internal/normalizers/nws/weatherstories_test.go new file mode 100644 index 0000000..afaa33d --- /dev/null +++ b/internal/normalizers/nws/weatherstories_test.go @@ -0,0 +1,146 @@ +package nws + +import ( + "encoding/json" + "strings" + "testing" + "time" + + "gitea.maximumdirect.net/ejr/feedkit/event" + "gitea.maximumdirect.net/ejr/weatherfeeder/model" + "gitea.maximumdirect.net/ejr/weatherfeeder/standards" +) + +func TestWeatherStoriesNormalizerProducesCanonicalSchemaAndMapsSample(t *testing.T) { + out, err := (WeatherStoriesNormalizer{}).Normalize(nil, weatherStoriesRawEvent(weatherStoriesSamplePayload())) + if err != nil { + t.Fatalf("Normalize() error = %v", err) + } + if out == nil { + t.Fatalf("Normalize() returned nil output") + } + if out.Schema != standards.SchemaWeatherStoryV1 { + t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherStoryV1) + } + if out.Kind != event.Kind("weather_story") { + t.Fatalf("Kind = %q, want weather_story", out.Kind) + } + + payload, ok := out.Payload.(model.WeatherStoryRun) + if !ok { + t.Fatalf("Payload type = %T, want model.WeatherStoryRun", out.Payload) + } + if payload.OfficeID != "LSX" { + t.Fatalf("OfficeID = %q, want LSX", payload.OfficeID) + } + wantAsOf := time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC) + if !payload.AsOf.Equal(wantAsOf) { + t.Fatalf("AsOf = %s, want %s", payload.AsOf, wantAsOf) + } + if out.EffectiveAt == nil || !out.EffectiveAt.Equal(wantAsOf) { + t.Fatalf("EffectiveAt = %v, want %s", out.EffectiveAt, wantAsOf) + } + if len(payload.Stories) != 1 { + t.Fatalf("Stories len = %d, want 1", len(payload.Stories)) + } + story := payload.Stories[0] + if story.Title != "Several Chances for Rain Through Monday" { + t.Fatalf("Title = %q", story.Title) + } + if story.Description != "A stagnant weather pattern." { + t.Fatalf("Description = %q", story.Description) + } + if story.AltText != "This slide shows the forecast." { + t.Fatalf("AltText = %q", story.AltText) + } + if story.Priority { + t.Fatalf("Priority = true, want false") + } + if story.Order != 1 { + t.Fatalf("Order = %d, want 1", story.Order) + } + if story.DownloadURL != "https://api.weather.gov/offices/LSX/weatherstories/download/3228e499-2aae-45a8-9ff9-1c060311026f" { + t.Fatalf("DownloadURL = %q", story.DownloadURL) + } +} + +func TestWeatherStoriesNormalizerEmptyStoriesUsesFallbackAsOf(t *testing.T) { + effectiveAt := time.Date(2026, 5, 30, 12, 0, 0, 0, time.UTC) + in := weatherStoriesRawEvent(`{"stories":[]}`) + in.EffectiveAt = &effectiveAt + + out, err := (WeatherStoriesNormalizer{}).Normalize(nil, in) + if err != nil { + t.Fatalf("Normalize() error = %v", err) + } + payload, ok := out.Payload.(model.WeatherStoryRun) + if !ok { + t.Fatalf("Payload type = %T, want model.WeatherStoryRun", out.Payload) + } + if !payload.AsOf.Equal(effectiveAt) { + t.Fatalf("AsOf = %s, want fallback %s", payload.AsOf, effectiveAt) + } + if payload.Stories == nil { + t.Fatalf("Stories = nil, want empty slice") + } + if len(payload.Stories) != 0 { + t.Fatalf("Stories len = %d, want 0", len(payload.Stories)) + } +} + +func TestWeatherStoriesNormalizerRejectsInvalidRequiredStoryTime(t *testing.T) { + _, err := (WeatherStoriesNormalizer{}).Normalize(nil, weatherStoriesRawEvent(`{ + "stories": [{ + "officeId": "LSX", + "startTime": "bad", + "endTime": "2026-05-31T11:00:00+00:00", + "updateTime": "2026-05-30T09:00:34+00:00" + }] + }`)) + if err == nil { + t.Fatalf("Normalize() error = nil, want error") + } + if !strings.Contains(err.Error(), "stories[0].startTime") { + t.Fatalf("error = %q, want field context", err) + } +} + +func TestWeatherStoriesNormalizerMatch(t *testing.T) { + n := WeatherStoriesNormalizer{} + if !n.Match(event.Event{Schema: standards.SchemaRawNWSWeatherStoriesV1}) { + t.Fatalf("Match(raw weatherstories) = false, want true") + } + if n.Match(event.Event{Schema: standards.SchemaRawNWSAlertsV1}) { + t.Fatalf("Match(raw alerts) = true, want false") + } +} + +func weatherStoriesRawEvent(payload string) event.Event { + return event.Event{ + ID: "evt-weatherstories-1", + Kind: event.Kind("weather_story"), + Source: "nws-weatherstories-test", + EmittedAt: time.Date(2026, 5, 30, 9, 5, 0, 0, time.UTC), + Schema: standards.SchemaRawNWSWeatherStoriesV1, + Payload: json.RawMessage(payload), + } +} + +func weatherStoriesSamplePayload() string { + return `{ + "stories": [ + { + "officeId": " LSX ", + "startTime": "2026-05-30T08:46:00+00:00", + "endTime": "2026-05-31T11:00:00+00:00", + "updateTime": "2026-05-30T09:00:34+00:00", + "title": " Several Chances for Rain Through Monday ", + "description": " A stagnant weather pattern. ", + "altText": " This slide shows the forecast. ", + "priority": false, + "order": 1, + "download": " https://api.weather.gov/offices/LSX/weatherstories/download/3228e499-2aae-45a8-9ff9-1c060311026f " + } + ] + }` +} diff --git a/internal/sinks/postgres/doc.go b/internal/sinks/postgres/doc.go index 674064a..a71b804 100644 --- a/internal/sinks/postgres/doc.go +++ b/internal/sinks/postgres/doc.go @@ -8,11 +8,15 @@ // Canonical input schemas: // - weather.observation.v1 -> model.WeatherObservation // - weather.forecast.v1 -> model.WeatherForecastRun +// - weather.forecast_discussion.v1 -> model.WeatherForecastDiscussion +// - weather.weather_story.v1 -> model.WeatherStoryRun // - weather.alert.v1 -> model.WeatherAlertRun // // Parent/child relationships: // - observations.event_id -> observation_present_weather.event_id // - forecasts.event_id -> forecast_periods.run_event_id +// - forecast_discussions.event_id -> forecast_discussion_key_messages.run_event_id +// - weather_story_runs.event_id -> weather_stories.run_event_id // - alert_runs.event_id -> alerts.run_event_id // - alerts.(run_event_id, alert_index) -> alert_references.(run_event_id, alert_index) // @@ -24,13 +28,18 @@ // - observation_present_weather.observed_at // - forecasts.issued_at // - forecast_periods.issued_at +// - forecast_discussions.issued_at +// - forecast_discussion_key_messages.issued_at +// - weather_story_runs.as_of +// - weather_stories.as_of // - alert_runs.as_of // - alerts.as_of // - alert_references.as_of // // Envelope field mapping (shared parent columns) // -// These columns exist on observations, forecasts, and alert_runs: +// These columns exist on parent tables such as observations, forecasts, +// forecast_discussions, weather_story_runs, and alert_runs: // - event_id TEXT -> event.id // - event_kind TEXT -> event.kind // - event_source TEXT -> event.source @@ -120,7 +129,35 @@ // - snowfall_depth_mm DOUBLE PRECISION NULL -> payload.periods[i].snowfallDepthMm // - uv_index DOUBLE PRECISION NULL -> payload.periods[i].uvIndex // -// 5. alert_runs (PK: event_id) +// 5. weather_story_runs (PK: event_id) +// +// - event_id TEXT -> event.id +// - event_kind TEXT -> event.kind +// - event_source TEXT -> event.source +// - event_schema TEXT -> event.schema +// - event_emitted_at TIMESTAMPTZ -> event.emitted_at +// - event_effective_at TIMESTAMPTZ NULL -> event.effective_at +// - office_id TEXT NULL -> payload.officeId +// - as_of TIMESTAMPTZ -> payload.asOf +// - story_count INTEGER -> len(payload.stories) +// +// 6. weather_stories (PK: run_event_id, story_index) +// +// - run_event_id TEXT -> weather_story_runs.event_id / payload.stories[i] +// - story_index INTEGER -> i (array position in payload.stories) +// - as_of TIMESTAMPTZ -> payload.asOf (copied from parent) +// - office_id TEXT NULL -> payload.stories[i].officeId +// - start_time TIMESTAMPTZ -> payload.stories[i].startTime +// - end_time TIMESTAMPTZ -> payload.stories[i].endTime +// - updated_at TIMESTAMPTZ -> payload.stories[i].updatedAt +// - title TEXT NULL -> payload.stories[i].title +// - description TEXT NULL -> payload.stories[i].description +// - alt_text TEXT NULL -> payload.stories[i].altText +// - priority BOOLEAN -> payload.stories[i].priority +// - story_order INTEGER -> payload.stories[i].order +// - download_url TEXT NULL -> payload.stories[i].downloadUrl +// +// 7. alert_runs (PK: event_id) // // - event_id TEXT -> event.id // - event_kind TEXT -> event.kind @@ -135,7 +172,7 @@ // - longitude DOUBLE PRECISION NULL -> payload.longitude // - alert_count INTEGER -> len(payload.alerts) // -// 6. alerts (PK: run_event_id, alert_index) +// 8. alerts (PK: run_event_id, alert_index) // // - run_event_id TEXT -> alert_runs.event_id / payload.alerts[i] // - alert_index INTEGER -> i (array position in payload.alerts) @@ -160,7 +197,7 @@ // - sender_name TEXT NULL -> payload.alerts[i].senderName // - reference_count INTEGER -> len(payload.alerts[i].references) // -// 7. alert_references (PK: run_event_id, alert_index, reference_index) +// 9. alert_references (PK: run_event_id, alert_index, reference_index) // // - run_event_id TEXT -> alert_runs.event_id / payload.alerts[i].references[j] // - alert_index INTEGER -> i (array position in payload.alerts) @@ -181,6 +218,10 @@ // read one row from forecasts, then join forecast_periods by run_event_id // ordered by period_index to rebuild periods. // +// - WeatherStoryRun: +// read one row from weather_story_runs, then join weather_stories by +// run_event_id ordered by story_index to rebuild stories. +// // - WeatherAlertRun: // read one row from alert_runs, join alerts by run_event_id ordered by // alert_index, then join alert_references by (run_event_id, alert_index) diff --git a/internal/sinks/postgres/map.go b/internal/sinks/postgres/map.go index a8b6af1..9f6fcfd 100644 --- a/internal/sinks/postgres/map.go +++ b/internal/sinks/postgres/map.go @@ -22,6 +22,8 @@ func mapPostgresEvent(_ context.Context, e fkevent.Event) ([]fksinks.PostgresWri return mapForecastEvent(e) case standards.SchemaWeatherForecastDiscussionV1: return mapForecastDiscussionEvent(e) + case standards.SchemaWeatherStoryV1: + return mapWeatherStoryEvent(e) case standards.SchemaWeatherAlertV1: return mapAlertEvent(e) default: @@ -218,6 +220,59 @@ func mapForecastDiscussionEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error return writes, nil } +func mapWeatherStoryEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { + run, err := decodePayload[model.WeatherStoryRun](e.Payload) + if err != nil { + return nil, fmt.Errorf("decode weather story payload: %w", err) + } + if run.AsOf.IsZero() { + return nil, fmt.Errorf("decode weather story payload: asOf is required") + } + + asOf := run.AsOf.UTC() + writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Stories)) + writes = append(writes, fksinks.PostgresWrite{ + Table: tableWeatherStoryRuns, + Values: map[string]any{ + "event_id": e.ID, + "event_kind": string(e.Kind), + "event_source": e.Source, + "event_schema": e.Schema, + "event_emitted_at": e.EmittedAt.UTC(), + "event_effective_at": nullableTime(e.EffectiveAt), + "office_id": nullableString(run.OfficeID), + "as_of": asOf, + "story_count": len(run.Stories), + }, + }) + + for i, story := range run.Stories { + if story.StartTime.IsZero() || story.EndTime.IsZero() || story.UpdatedAt.IsZero() { + return nil, fmt.Errorf("decode weather story payload: stories[%d] startTime/endTime/updatedAt are required", i) + } + writes = append(writes, fksinks.PostgresWrite{ + Table: tableWeatherStories, + Values: map[string]any{ + "run_event_id": e.ID, + "story_index": i, + "as_of": asOf, + "office_id": nullableString(story.OfficeID), + "start_time": story.StartTime.UTC(), + "end_time": story.EndTime.UTC(), + "updated_at": story.UpdatedAt.UTC(), + "title": nullableString(story.Title), + "description": nullableString(story.Description), + "alt_text": nullableString(story.AltText), + "priority": story.Priority, + "story_order": story.Order, + "download_url": nullableString(story.DownloadURL), + }, + }) + } + + return writes, nil +} + func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) { run, err := decodePayload[model.WeatherAlertRun](e.Payload) if err != nil { diff --git a/internal/sinks/postgres/map_test.go b/internal/sinks/postgres/map_test.go index d4ec2d4..a4cd7ee 100644 --- a/internal/sinks/postgres/map_test.go +++ b/internal/sinks/postgres/map_test.go @@ -193,6 +193,76 @@ func TestMapPostgresEventForecastDiscussionStructPayload(t *testing.T) { assertAllWritesIncludeAllColumns(t, writes) } +func TestMapPostgresEventWeatherStoryStructPayload(t *testing.T) { + run := model.WeatherStoryRun{ + OfficeID: "LSX", + AsOf: time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC), + Stories: []model.WeatherStory{ + { + OfficeID: "LSX", + StartTime: time.Date(2026, 5, 30, 8, 46, 0, 0, time.UTC), + EndTime: time.Date(2026, 5, 31, 11, 0, 0, 0, time.UTC), + UpdatedAt: time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC), + Title: "Several Chances for Rain Through Monday", + Description: "Scattered showers and thunderstorms.", + AltText: "This slide shows the forecast.", + Priority: true, + Order: 1, + DownloadURL: "https://api.weather.gov/offices/LSX/weatherstories/download/story-1", + }, + }, + } + + writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", run)) + if err != nil { + t.Fatalf("mapPostgresEvent() error = %v", err) + } + if len(writes) != 2 { + t.Fatalf("mapPostgresEvent() writes len = %d, want 2", len(writes)) + } + if writes[0].Table != tableWeatherStoryRuns { + t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableWeatherStoryRuns) + } + if got := writes[0].Values["story_count"]; got != 1 { + t.Fatalf("weather_story_runs story_count = %#v, want 1", got) + } + if writes[1].Table != tableWeatherStories { + t.Fatalf("writes[1].Table = %q, want %q", writes[1].Table, tableWeatherStories) + } + if got := writes[1].Values["download_url"]; got != "https://api.weather.gov/offices/LSX/weatherstories/download/story-1" { + t.Fatalf("weather_stories download_url = %#v", got) + } + if got := writes[1].Values["story_order"]; got != 1 { + t.Fatalf("weather_stories story_order = %#v, want 1", got) + } + + assertAllWritesIncludeAllColumns(t, writes) +} + +func TestMapPostgresEventWeatherStoryRejectsMissingAsOf(t *testing.T) { + _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", model.WeatherStoryRun{})) + if err == nil { + t.Fatalf("mapPostgresEvent() error = nil, want missing asOf error") + } + if !strings.Contains(err.Error(), "asOf is required") { + t.Fatalf("error = %q, want asOf context", err) + } +} + +func TestMapPostgresEventWeatherStoryRejectsMissingStoryTimes(t *testing.T) { + run := model.WeatherStoryRun{ + AsOf: time.Date(2026, 5, 30, 9, 0, 34, 0, time.UTC), + Stories: []model.WeatherStory{{Title: "missing times"}}, + } + _, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", run)) + if err == nil { + t.Fatalf("mapPostgresEvent() error = nil, want missing story times error") + } + if !strings.Contains(err.Error(), "stories[0] startTime/endTime/updatedAt are required") { + t.Fatalf("error = %q, want story time context", err) + } +} + func TestMapPostgresEventMapPayload(t *testing.T) { run := model.WeatherForecastRun{ IssuedAt: time.Date(2026, 3, 16, 18, 0, 0, 0, time.UTC), diff --git a/internal/sinks/postgres/schema.go b/internal/sinks/postgres/schema.go index 3303090..1d98a87 100644 --- a/internal/sinks/postgres/schema.go +++ b/internal/sinks/postgres/schema.go @@ -11,6 +11,8 @@ const ( tableForecastPeriods = "forecast_periods" tableForecastDiscussions = "forecast_discussions" tableForecastDiscussionKeyMessages = "forecast_discussion_key_messages" + tableWeatherStoryRuns = "weather_story_runs" + tableWeatherStories = "weather_stories" tableAlertRuns = "alert_runs" tableAlerts = "alerts" tableAlertReferences = "alert_references" @@ -174,6 +176,51 @@ func PostgresSchema() fksinks.PostgresSchema { {Name: "idx_wf_discussion_message_issued_at", Columns: []string{"issued_at"}}, }, }, + { + Name: tableWeatherStoryRuns, + Columns: []fksinks.PostgresColumn{ + {Name: "event_id", Type: "TEXT", Nullable: false}, + {Name: "event_kind", Type: "TEXT", Nullable: false}, + {Name: "event_source", Type: "TEXT", Nullable: false}, + {Name: "event_schema", Type: "TEXT", Nullable: false}, + {Name: "event_emitted_at", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "event_effective_at", Type: "TIMESTAMPTZ", Nullable: true}, + {Name: "office_id", Type: "TEXT", Nullable: true}, + {Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "story_count", Type: "INTEGER", Nullable: false}, + }, + PrimaryKey: []string{"event_id"}, + PruneColumn: "as_of", + Indexes: []fksinks.PostgresIndex{ + {Name: "idx_wf_story_run_office_as_of", Columns: []string{"office_id", "as_of"}}, + {Name: "idx_wf_story_run_as_of", Columns: []string{"as_of"}}, + }, + }, + { + Name: tableWeatherStories, + Columns: []fksinks.PostgresColumn{ + {Name: "run_event_id", Type: "TEXT REFERENCES weather_story_runs(event_id) ON DELETE CASCADE", Nullable: false}, + {Name: "story_index", Type: "INTEGER", Nullable: false}, + {Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "office_id", Type: "TEXT", Nullable: true}, + {Name: "start_time", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "end_time", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "updated_at", Type: "TIMESTAMPTZ", Nullable: false}, + {Name: "title", Type: "TEXT", Nullable: true}, + {Name: "description", Type: "TEXT", Nullable: true}, + {Name: "alt_text", Type: "TEXT", Nullable: true}, + {Name: "priority", Type: "BOOLEAN", Nullable: false}, + {Name: "story_order", Type: "INTEGER", Nullable: false}, + {Name: "download_url", Type: "TEXT", Nullable: true}, + }, + PrimaryKey: []string{"run_event_id", "story_index"}, + PruneColumn: "as_of", + Indexes: []fksinks.PostgresIndex{ + {Name: "idx_wf_stories_start_time", Columns: []string{"start_time"}}, + {Name: "idx_wf_stories_end_time", Columns: []string{"end_time"}}, + {Name: "idx_wf_stories_updated_at", Columns: []string{"updated_at"}}, + }, + }, { Name: tableAlertRuns, Columns: []fksinks.PostgresColumn{ diff --git a/internal/sinks/postgres/schema_test.go b/internal/sinks/postgres/schema_test.go index cdeb919..6c92e29 100644 --- a/internal/sinks/postgres/schema_test.go +++ b/internal/sinks/postgres/schema_test.go @@ -15,6 +15,8 @@ func TestWeatherPostgresSchemaShape(t *testing.T) { tableForecastPeriods: true, tableForecastDiscussions: true, tableForecastDiscussionKeyMessages: true, + tableWeatherStoryRuns: true, + tableWeatherStories: true, tableAlertRuns: true, tableAlerts: true, tableAlertReferences: true, @@ -40,3 +42,38 @@ func TestWeatherPostgresSchemaShape(t *testing.T) { } } } + +func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) { + runColumns := columnsForTable(t, tableWeatherStoryRuns) + if !runColumns["as_of"] { + t.Fatalf("%s missing as_of column", tableWeatherStoryRuns) + } + if !runColumns["story_count"] { + t.Fatalf("%s missing story_count column", tableWeatherStoryRuns) + } + + storyColumns := columnsForTable(t, tableWeatherStories) + for _, col := range []string{"start_time", "end_time", "updated_at", "title", "description", "alt_text", "priority", "story_order", "download_url"} { + if !storyColumns[col] { + t.Fatalf("%s missing %s column", tableWeatherStories, col) + } + } +} + +func columnsForTable(t *testing.T, table string) map[string]bool { + t.Helper() + + schema := PostgresSchema() + for _, tbl := range schema.Tables { + if tbl.Name != table { + continue + } + cols := make(map[string]bool, len(tbl.Columns)) + for _, col := range tbl.Columns { + cols[col.Name] = true + } + return cols + } + t.Fatalf("missing table %q", table) + return nil +} diff --git a/internal/sources/builtins.go b/internal/sources/builtins.go index 5e7afdb..85f9127 100644 --- a/internal/sources/builtins.go +++ b/internal/sources/builtins.go @@ -22,6 +22,7 @@ var pollDriverRegistrations = []pollDriverRegistration{ {driver: "nws_forecast_discussion", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewForecastDiscussionSource(cfg) }}, + {driver: "nws_weatherstories", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return nws.NewWeatherStoriesSource(cfg) }}, {driver: "openmeteo_observation", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return openmeteo.NewObservationSource(cfg) }}, {driver: "openmeteo_forecast", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { return openmeteo.NewForecastSource(cfg) }}, {driver: "openweather_observation", factory: func(cfg config.SourceConfig) (fksource.PollSource, error) { diff --git a/internal/sources/builtins_test.go b/internal/sources/builtins_test.go index fc56020..2eec97e 100644 --- a/internal/sources/builtins_test.go +++ b/internal/sources/builtins_test.go @@ -47,6 +47,19 @@ func TestRegisterBuiltinsRegistersNWSForecastDiscussionDriver(t *testing.T) { } } +func TestRegisterBuiltinsRegistersNWSWeatherStoriesDriver(t *testing.T) { + reg := fksource.NewRegistry() + RegisterBuiltins(reg) + + in, err := reg.BuildInput(sourceConfigForDriver("nws_weatherstories")) + if err != nil { + t.Fatalf("BuildInput(nws_weatherstories) error = %v", err) + } + if _, ok := in.(fksource.PollSource); !ok { + t.Fatalf("BuildInput(nws_weatherstories) type = %T, want PollSource", in) + } +} + func TestRegisterBuiltinsDoesNotRegisterLegacyNWSForecastDriver(t *testing.T) { reg := fksource.NewRegistry() RegisterBuiltins(reg) @@ -70,6 +83,7 @@ func TestRegisterBuiltinsRegistersAllCurrentDrivers(t *testing.T) { "nws_forecast_hourly", "nws_forecast_narrative", "nws_forecast_discussion", + "nws_weatherstories", "openmeteo_observation", "openmeteo_forecast", "openweather_observation", diff --git a/internal/sources/nws/weatherstories.go b/internal/sources/nws/weatherstories.go new file mode 100644 index 0000000..e6a1949 --- /dev/null +++ b/internal/sources/nws/weatherstories.go @@ -0,0 +1,118 @@ +package nws + +import ( + "context" + "encoding/json" + "strings" + "time" + + "gitea.maximumdirect.net/ejr/feedkit/config" + "gitea.maximumdirect.net/ejr/feedkit/event" + fksources "gitea.maximumdirect.net/ejr/feedkit/sources" + nwscommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/nws" + "gitea.maximumdirect.net/ejr/weatherfeeder/standards" +) + +// WeatherStoriesSource polls an NWS weatherstories endpoint and emits a RAW weather story Event. +// +// Output schema: +// - standards.SchemaRawNWSWeatherStoriesV1 +type WeatherStoriesSource struct { + http *fksources.HTTPSource +} + +func NewWeatherStoriesSource(cfg config.SourceConfig) (*WeatherStoriesSource, error) { + const driver = "nws_weatherstories" + + hs, err := fksources.NewHTTPSource(driver, cfg, "application/geo+json, application/json") + if err != nil { + return nil, err + } + + return &WeatherStoriesSource{http: hs}, nil +} + +func (s *WeatherStoriesSource) Name() string { return s.http.Name } + +func (s *WeatherStoriesSource) Kinds() []event.Kind { + return []event.Kind{event.Kind("weather_story")} +} + +func (s *WeatherStoriesSource) Poll(ctx context.Context) ([]event.Event, error) { + raw, meta, changed, err := s.fetchRaw(ctx) + if err != nil { + return nil, err + } + if !changed { + return nil, nil + } + + var effectiveAt *time.Time + switch { + case !meta.ParsedLatestUpdateTime.IsZero(): + t := meta.ParsedLatestUpdateTime.UTC() + effectiveAt = &t + case !meta.ParsedLatestStartTime.IsZero(): + t := meta.ParsedLatestStartTime.UTC() + effectiveAt = &t + } + + emittedAt := time.Now().UTC() + eventID := fksources.DefaultEventID("", s.http.Name, effectiveAt, emittedAt) + + return fksources.SingleEvent( + event.Kind("weather_story"), + s.http.Name, + standards.SchemaRawNWSWeatherStoriesV1, + eventID, + emittedAt, + effectiveAt, + raw, + ) +} + +type weatherStoriesMeta struct { + Stories []struct { + StartTime string `json:"startTime"` + UpdateTime string `json:"updateTime"` + } `json:"stories"` + + ParsedLatestUpdateTime time.Time `json:"-"` + ParsedLatestStartTime time.Time `json:"-"` +} + +func (s *WeatherStoriesSource) fetchRaw(ctx context.Context) (json.RawMessage, weatherStoriesMeta, bool, error) { + raw, changed, err := s.http.FetchJSONIfChanged(ctx) + if err != nil { + return nil, weatherStoriesMeta{}, false, err + } + if !changed { + return nil, weatherStoriesMeta{}, false, nil + } + + var meta weatherStoriesMeta + if err := json.Unmarshal(raw, &meta); err != nil { + return raw, weatherStoriesMeta{}, true, nil + } + + for _, story := range meta.Stories { + if ts := strings.TrimSpace(story.UpdateTime); ts != "" { + if t, err := nwscommon.ParseTime(ts); err == nil { + t = t.UTC() + if meta.ParsedLatestUpdateTime.IsZero() || t.After(meta.ParsedLatestUpdateTime) { + meta.ParsedLatestUpdateTime = t + } + } + } + if ts := strings.TrimSpace(story.StartTime); ts != "" { + if t, err := nwscommon.ParseTime(ts); err == nil { + t = t.UTC() + if meta.ParsedLatestStartTime.IsZero() || t.After(meta.ParsedLatestStartTime) { + meta.ParsedLatestStartTime = t + } + } + } + } + + return raw, meta, true, nil +} diff --git a/internal/sources/nws/weatherstories_test.go b/internal/sources/nws/weatherstories_test.go new file mode 100644 index 0000000..4813013 --- /dev/null +++ b/internal/sources/nws/weatherstories_test.go @@ -0,0 +1,158 @@ +package nws + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" + + "gitea.maximumdirect.net/ejr/feedkit/config" + "gitea.maximumdirect.net/ejr/feedkit/event" + "gitea.maximumdirect.net/ejr/weatherfeeder/standards" +) + +func TestWeatherStoriesSourcePollEmitsExpectedEventAndPrefersLatestUpdateTime(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{ + "stories": [ + {"startTime":"2026-05-30T08:46:00+00:00","updateTime":"2026-05-30T09:00:34+00:00"}, + {"startTime":"2026-05-30T10:00:00+00:00","updateTime":"2026-05-30T11:00:34+00:00"} + ] + }`)) + })) + defer srv.Close() + + src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL)) + if err != nil { + t.Fatalf("NewWeatherStoriesSource() error = %v", err) + } + if got := src.Kinds(); len(got) != 1 || got[0] != event.Kind("weather_story") { + t.Fatalf("Kinds() = %#v, want [weather_story]", got) + } + + events, err := src.Poll(context.Background()) + if err != nil { + t.Fatalf("Poll() error = %v", err) + } + if len(events) != 1 { + t.Fatalf("Poll() len = %d, want 1", len(events)) + } + + got := events[0] + if got.Kind != event.Kind("weather_story") { + t.Fatalf("Kind = %q, want weather_story", got.Kind) + } + if got.Schema != standards.SchemaRawNWSWeatherStoriesV1 { + t.Fatalf("Schema = %q, want %q", got.Schema, standards.SchemaRawNWSWeatherStoriesV1) + } + wantEffectiveAt := time.Date(2026, 5, 30, 11, 0, 34, 0, time.UTC) + if got.EffectiveAt == nil || !got.EffectiveAt.Equal(wantEffectiveAt) { + t.Fatalf("EffectiveAt = %v, want %s", got.EffectiveAt, wantEffectiveAt.Format(time.RFC3339)) + } + if _, ok := got.Payload.(json.RawMessage); !ok { + t.Fatalf("Payload type = %T, want json.RawMessage", got.Payload) + } +} + +func TestWeatherStoriesSourcePollEffectiveAtFallsBackToLatestStartTime(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{ + "stories": [ + {"startTime":"2026-05-30T08:46:00+00:00","updateTime":"bad"}, + {"startTime":"2026-05-31T11:00:00+00:00","updateTime":""} + ] + }`)) + })) + defer srv.Close() + + src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL)) + if err != nil { + t.Fatalf("NewWeatherStoriesSource() error = %v", err) + } + + events, err := src.Poll(context.Background()) + if err != nil { + t.Fatalf("Poll() error = %v", err) + } + if len(events) != 1 { + t.Fatalf("Poll() len = %d, want 1", len(events)) + } + wantEffectiveAt := time.Date(2026, 5, 31, 11, 0, 0, 0, time.UTC) + if events[0].EffectiveAt == nil || !events[0].EffectiveAt.Equal(wantEffectiveAt) { + t.Fatalf("EffectiveAt = %v, want %s", events[0].EffectiveAt, wantEffectiveAt.Format(time.RFC3339)) + } +} + +func TestWeatherStoriesSourcePollReturnsNoEventsWhenUnchanged(t *testing.T) { + const etag = `"stories-v1"` + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("If-None-Match") == etag { + w.WriteHeader(http.StatusNotModified) + return + } + w.Header().Set("ETag", etag) + _, _ = w.Write([]byte(`{"stories":[]}`)) + })) + defer srv.Close() + + src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL)) + if err != nil { + t.Fatalf("NewWeatherStoriesSource() error = %v", err) + } + + first, err := src.Poll(context.Background()) + if err != nil { + t.Fatalf("first Poll() error = %v", err) + } + if len(first) != 1 { + t.Fatalf("first Poll() len = %d, want 1", len(first)) + } + + second, err := src.Poll(context.Background()) + if err != nil { + t.Fatalf("second Poll() error = %v", err) + } + if len(second) != 0 { + t.Fatalf("second Poll() len = %d, want 0", len(second)) + } +} + +func TestWeatherStoriesSourcePollMetadataDecodeFailureStillEmitsRawEvent(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`not-json`)) + })) + defer srv.Close() + + src, err := NewWeatherStoriesSource(weatherStoriesSourceConfig(srv.URL)) + if err != nil { + t.Fatalf("NewWeatherStoriesSource() error = %v", err) + } + + events, err := src.Poll(context.Background()) + if err != nil { + t.Fatalf("Poll() error = %v", err) + } + if len(events) != 1 { + t.Fatalf("Poll() len = %d, want 1", len(events)) + } + if events[0].EffectiveAt != nil { + t.Fatalf("EffectiveAt = %v, want nil", events[0].EffectiveAt) + } + if events[0].Schema != standards.SchemaRawNWSWeatherStoriesV1 { + t.Fatalf("Schema = %q, want %q", events[0].Schema, standards.SchemaRawNWSWeatherStoriesV1) + } +} + +func weatherStoriesSourceConfig(url string) config.SourceConfig { + return config.SourceConfig{ + Name: "test-weatherstories-source", + Driver: "nws_weatherstories", + Mode: config.SourceModePoll, + Params: map[string]any{ + "url": url, + "user_agent": "test-agent", + }, + } +} diff --git a/model/weather_story.go b/model/weather_story.go new file mode 100644 index 0000000..0890f24 --- /dev/null +++ b/model/weather_story.go @@ -0,0 +1,28 @@ +package model + +import "time" + +// WeatherStoryRun is a snapshot of NWS weather stories for an office as-of a point in time. +type WeatherStoryRun struct { + OfficeID string `json:"officeId,omitempty"` + AsOf time.Time `json:"asOf"` + + Stories []WeatherStory `json:"stories"` +} + +// WeatherStory is a provider-independent representation of a single story card. +type WeatherStory struct { + OfficeID string `json:"officeId,omitempty"` + + StartTime time.Time `json:"startTime"` + EndTime time.Time `json:"endTime"` + UpdatedAt time.Time `json:"updatedAt"` + + Title string `json:"title,omitempty"` + Description string `json:"description,omitempty"` + AltText string `json:"altText,omitempty"` + + Priority bool `json:"priority"` + Order int `json:"order"` + DownloadURL string `json:"downloadUrl,omitempty"` +} diff --git a/standards/schema.go b/standards/schema.go index 16ed4aa..ec5ca6f 100644 --- a/standards/schema.go +++ b/standards/schema.go @@ -18,6 +18,7 @@ const ( SchemaRawNWSHourlyForecastV1 = "raw.nws.hourly.forecast.v1" SchemaRawNWSNarrativeForecastV1 = "raw.nws.narrative.forecast.v1" SchemaRawNWSForecastDiscussionV1 = "raw.nws.forecast_discussion.v1" + SchemaRawNWSWeatherStoriesV1 = "raw.nws.weatherstories.v1" SchemaRawOpenMeteoHourlyForecastV1 = "raw.openmeteo.hourly.forecast.v1" SchemaRawOpenWeatherHourlyForecastV1 = "raw.openweather.hourly.forecast.v1" @@ -27,5 +28,6 @@ const ( SchemaWeatherObservationV1 = "weather.observation.v1" SchemaWeatherForecastV1 = "weather.forecast.v1" SchemaWeatherForecastDiscussionV1 = "weather.forecast_discussion.v1" + SchemaWeatherStoryV1 = "weather.weather_story.v1" SchemaWeatherAlertV1 = "weather.alert.v1" )