9 Commits

Author SHA1 Message Date
fba519cab0 Update documentation for SPC outlook support
All checks were successful
ci/woodpecker/push/build-image Pipeline was successful
2026-06-11 00:30:49 +00:00
da8ff81692 Verify SPC outlook implementation 2026-06-11 00:28:30 +00:00
f91a185f9d Document SPC outlook support 2026-06-11 00:27:23 +00:00
e966276c40 Add Postgres mapping for SPC outlooks 2026-06-11 00:24:12 +00:00
1e2db468ea Add SPC outlook normalization 2026-06-11 00:20:31 +00:00
cefd4dfc7c Add raw SPC convective outlook source 2026-06-11 00:15:40 +00:00
42c646c328 Add GeoJSON point containment helper 2026-06-11 00:10:01 +00:00
b2c429983c Add SPC provider parsing helpers 2026-06-11 00:06:55 +00:00
0f20d1e4cb Add AGENTS.md 2026-06-10 23:59:45 +00:00
52 changed files with 3756 additions and 33 deletions

4
AGENTS.md Normal file
View File

@@ -0,0 +1,4 @@
Please carefully review the documents in `docs/policy` before making any changes to this repository.
- `architecture.md` provides the canonical high-level architecture policy for this repository.
- `development.md` provides more granular development policy for this repository.
- `documentation.md` provides the canonical documentation policy for this repository.

View File

@@ -5,9 +5,9 @@ provider-specific responses into canonical weather events, and routes those
events to configured sinks.
It currently supports NWS observations, alerts, hourly forecasts, narrative
forecasts, forecast discussions, and weather stories; Open-Meteo observations
and hourly forecasts; and OpenWeather observations. Implemented sinks are
stdout, NATS, and Postgres.
forecasts, forecast discussions, and weather stories; SPC Day 1-3 convective
outlooks; Open-Meteo observations and hourly forecasts; and OpenWeather
observations. Implemented sinks are stdout, NATS, and Postgres.
## Quickstart
@@ -32,6 +32,7 @@ current working directory.
- [Event wire contract](docs/integrations/events.md)
- [Postgres table contract](docs/integrations/postgres.md)
- [NWS integration notes](docs/integrations/nws.md)
- [SPC integration notes](docs/integrations/spc.md)
- [Open-Meteo integration notes](docs/integrations/openmeteo.md)
- [OpenWeather integration notes](docs/integrations/openweather.md)
- [Architecture policy](docs/policy/architecture.md)

View File

@@ -99,6 +99,18 @@ sources:
url: "https://api.weather.gov/alerts?point=38.6239,-90.3571&limit=20"
user_agent: "HomeOps (eric@maximumdirect.net)"
- name: SPCConvectiveOutlookSTL
mode: poll
kinds: ["outlook"]
driver: spc_convective_outlook
every: 30m
params:
latitude: 38.6239
longitude: -90.3571
location_id: "stl"
location_name: "St. Louis, MO"
user_agent: "HomeOps (eric@maximumdirect.net)"
sinks:
- name: stdout
driver: stdout
@@ -126,13 +138,13 @@ sinks:
routes:
- sink: stdout
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"]
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert", "outlook"]
- sink: nats_weatherfeeder
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"]
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert", "outlook"]
# - sink: pg_weatherfeeder
# kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"]
# kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert", "outlook"]
# - sink: logfile
# kinds: ["observation", "alert", "forecast", "forecast_discussion", "weather_story"]
# kinds: ["observation", "alert", "forecast", "forecast_discussion", "weather_story", "outlook"]

View File

@@ -52,6 +52,18 @@ sources:
url: "https://api.weather.gov/alerts?point=38.6239,-90.3571&limit=20"
user_agent: "Example weatherfeeder operator (ops@example.com)"
- name: SPCConvectiveOutlookLocal
mode: poll
driver: spc_convective_outlook
every: 30m
kinds: ["outlook"]
params:
latitude: 38.6239
longitude: -90.3571
location_id: "local"
location_name: "Configured point"
user_agent: "Example weatherfeeder operator (ops@example.com)"
sinks:
- name: nats_weather
driver: nats
@@ -69,10 +81,10 @@ sinks:
routes:
- sink: nats_weather
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"]
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert", "outlook"]
- sink: pg_weather
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert"]
kinds: ["observation", "forecast", "forecast_discussion", "weather_story", "alert", "outlook"]
```
Do not commit real API keys, database passwords, or personal contact addresses in
@@ -95,10 +107,10 @@ copyable configs.
| `mode` | no | `poll`, `stream`, or omitted for auto. Current weatherfeeder drivers are polling drivers. |
| `every` | yes | Poll interval for current weatherfeeder source drivers. |
| `kinds` | no | Expected event kinds. If present, startup verifies they match the source driver. |
| `params` | driver-specific | Driver parameters. Current source drivers require HTTP params. |
| `params` | driver-specific | Driver parameters. See the source-specific sections below. |
Current event kinds are `observation`, `forecast`, `forecast_discussion`,
`weather_story`, and `alert`.
`weather_story`, `alert`, and `outlook`.
## Source Drivers
@@ -113,10 +125,11 @@ Current event kinds are `observation`, `forecast`, `forecast_discussion`,
| `openmeteo_observation` | `observation` | Open-Meteo current conditions. |
| `openmeteo_forecast` | `forecast` | Open-Meteo hourly forecast. |
| `openweather_observation` | `observation` | OpenWeather current weather. |
| `spc_convective_outlook` | `outlook` | SPC Day 1-3 convective outlooks. |
## HTTP Source Params
All current source drivers use the shared HTTP polling helper.
Most source drivers use the shared HTTP polling helper.
| Param | Required | Description |
|---|:---:|---|
@@ -132,6 +145,44 @@ source emits no events for that poll.
OpenWeather observation URLs must include `units=metric`. Startup fails if the
URL omits it or sets another unit system.
## SPC Convective Outlook Params
`spc_convective_outlook` fetches the twelve required Day 1-3 GeoJSON outlook
products and the three required Day 1-3 print pages as one atomic bundle.
| Param | Required | Description |
|---|:---:|---|
| `latitude` | yes | Location latitude in decimal degrees. |
| `longitude` | yes | Location longitude in decimal degrees. |
| `user_agent` | yes | User-Agent sent to SPC. `userAgent` is also accepted. |
| `location_id` | no | Operator-defined location identifier copied into canonical outlook runs. |
| `location_name` | no | Human location label copied into canonical outlook runs. |
| `geojson_urls` | no | Map of product key to override URL. Used for tests and upstream URL changes. |
| `discussion_urls` | no | Map of day key to override print-page URL. Used for tests and upstream URL changes. |
| `rss_url` | no | Optional RSS URL. RSS is not fetched unless this is configured. |
| `http_timeout` | no | Positive duration for the HTTP client timeout. |
| `http_response_body_limit_bytes` | no | Positive integer response body limit in bytes. |
GeoJSON product keys are `day1_categorical`, `day1_tornado`, `day1_hail`,
`day1_wind`, `day2_categorical`, `day2_tornado`, `day2_hail`, `day2_wind`,
`day3_categorical`, `day3_tornado`, `day3_hail`, and `day3_wind`.
Discussion keys are `day1`, `day2`, and `day3`.
```yaml
sources:
- name: SPCConvectiveOutlookSTL
mode: poll
kinds: ["outlook"]
driver: spc_convective_outlook
every: 30m
params:
latitude: 38.6239
longitude: -90.3571
location_id: "stl"
location_name: "St. Louis, MO"
user_agent: "Example weatherfeeder operator (ops@example.com)"
```
## Sink Fields
| Field | Required | Description |

View File

@@ -37,6 +37,7 @@ Canonical schemas emitted after normalization:
| `forecast_discussion` | `weather.forecast_discussion.v1` | `WeatherForecastDiscussion` |
| `weather_story` | `weather.weather_story.v1` | `WeatherStoryRun` |
| `alert` | `weather.alert.v1` | `WeatherAlertRun` |
| `outlook` | `weather.outlook.v1` | `WeatherOutlookRun` |
Raw upstream schemas emitted by current sources:
@@ -51,6 +52,7 @@ Raw upstream schemas emitted by current sources:
| `weather_story` | `raw.nws.weatherstories.v1` | NWS weather stories JSON |
| `forecast` | `raw.openmeteo.hourly.forecast.v1` | Open-Meteo hourly forecast JSON |
| `alert` | `raw.nws.alerts.v1` | NWS alerts JSON |
| `outlook` | `raw.spc.convective_outlook.v1` | SPC convective outlook raw bundle |
`standards.SchemaRawOpenWeatherHourlyForecastV1` exists in code, but no current
registered source emits it.
@@ -218,6 +220,55 @@ Payload type: `WeatherAlertRun`.
`references[]` entries contain optional `id`, `identifier`, `sender`, and
`sent` fields.
## `weather.outlook.v1`
Payload type: `WeatherOutlookRun`.
The current producer is the SPC convective outlook normalizer. It emits Day 1-3
convective outlook polygons for categorical, tornado, hail, and wind products.
All timestamps are UTC.
| Field | Type | Required | Notes |
|---|---|:---:|---|
| `locationId` | string | no | Operator-configured location identifier. |
| `locationName` | string | no | Operator-configured location label. |
| `latitude` | number | no | Configured point latitude in decimal degrees. |
| `longitude` | number | no | Configured point longitude in decimal degrees. |
| `asOf` | timestamp | yes | Snapshot time. For SPC, this is the latest outlook issue time when available. |
| `issuedAt` | timestamp | no | Latest issue time across outlook features when any feature exists. |
| `outlooks` | array | yes | Ordered outlook polygons. |
`outlooks[]` entries:
| Field | Type | Required | Notes |
|---|---|:---:|---|
| `id` | string | yes | Deterministic weatherfeeder outlook identifier. |
| `provider` | string | yes | Current value is `spc`. |
| `product` | string | yes | Current value is `convective`. |
| `day` | integer | yes | SPC outlook day, currently `1`, `2`, or `3`. |
| `outlookType` | string | yes | `categorical`, `tornado`, `hail`, or `wind`. |
| `label` | string | yes | SPC outlook label such as `SLGT` or `15`. |
| `labelText` | string | no | Human label text from SPC, when present. |
| `severityRank` | integer | no | SPC `DN` value, when present. |
| `validFrom` | timestamp | yes | Valid period start. |
| `validTo` | timestamp | yes | Valid period end. |
| `issuedAt` | timestamp | yes | Feature issue time. |
| `expiresAt` | timestamp | yes | Expiration time; currently equal to `validTo`. |
| `forecaster` | string | no | SPC forecaster text, when present. |
| `headline` | string | no | Matching Day 1-3 print-page product title. |
| `summary` | string | no | Text from the print-page `...SUMMARY...` section. |
| `discussion` | string | no | Cleaned full print-page product text. |
| `sourceUrl` | string | no | GeoJSON product URL for this outlook feature. |
| `imageUrl` | string | no | Reserved for provider image URLs; currently empty. |
| `containsLocation` | boolean | yes | Whether the configured point is inside or on the boundary of the polygon. |
| `geometry` | object | yes | Compact GeoJSON `Polygon` or `MultiPolygon` geometry. |
`geometry` preserves the SPC feature geometry as compact GeoJSON using
`[longitude, latitude]` coordinate order. `containsLocation` is computed with
that geometry and the configured source `latitude`/`longitude`; boundary points
count as contained. All outlook polygons are emitted, including polygons that do
not contain the configured point.
## Compact Example
```json

View File

@@ -23,6 +23,7 @@ Events are mapped only for canonical weather schemas:
- `weather.forecast_discussion.v1`
- `weather.weather_story.v1`
- `weather.alert.v1`
- `weather.outlook.v1`
Unsupported schemas produce no writes for this sink. Mapped events are inserted
transactionally. Inserts use ordinary `INSERT`; duplicate primary keys fail the
@@ -56,6 +57,8 @@ Parent tables store the feed event envelope:
| `alert_runs` | `event_id` | `as_of` |
| `alerts` | `run_event_id`, `alert_index` | `as_of` |
| `alert_references` | `run_event_id`, `alert_index`, `reference_index` | `as_of` |
| `outlook_runs` | `event_id` | `as_of` |
| `outlooks` | `run_event_id`, `outlook_index` | `as_of` |
## Table Contract
@@ -379,6 +382,72 @@ Indexes:
| `sender` | `TEXT` | yes | `payload.alerts[].references[].sender` |
| `sent` | `TIMESTAMPTZ` | yes | `payload.alerts[].references[].sent` |
### `outlook_runs`
Primary key: `event_id`
Prune column: `as_of`
Indexes:
- `idx_wf_outlook_run_location_as_of` on `location_id`, `as_of`
- `idx_wf_outlook_run_as_of` on `as_of`
| Column | Type | Null | Source |
|---|---|:---:|---|
| `event_id` | `TEXT` | no | `event.id` |
| `event_kind` | `TEXT` | no | `event.kind` |
| `event_source` | `TEXT` | no | `event.source` |
| `event_schema` | `TEXT` | no | `event.schema` |
| `event_emitted_at` | `TIMESTAMPTZ` | no | `event.emitted_at` |
| `event_effective_at` | `TIMESTAMPTZ` | yes | `event.effective_at` |
| `location_id` | `TEXT` | yes | `payload.locationId` |
| `location_name` | `TEXT` | yes | `payload.locationName` |
| `latitude` | `DOUBLE PRECISION` | yes | `payload.latitude` |
| `longitude` | `DOUBLE PRECISION` | yes | `payload.longitude` |
| `as_of` | `TIMESTAMPTZ` | no | `payload.asOf` |
| `issued_at` | `TIMESTAMPTZ` | yes | `payload.issuedAt` |
| `outlook_count` | `INTEGER` | no | `len(payload.outlooks)` |
### `outlooks`
Primary key: `run_event_id`, `outlook_index`
Prune column: `as_of`
Foreign key: `run_event_id` references `outlook_runs(event_id)` with cascade
delete.
Indexes:
- `idx_wf_outlooks_contains_valid` on `contains_location`, `valid_from`, `valid_to`
- `idx_wf_outlooks_day_type_label` on `day`, `outlook_type`, `label`
- `idx_wf_outlooks_valid` on `valid_from`, `valid_to`
| Column | Type | Null | Source |
|---|---|:---:|---|
| `run_event_id` | `TEXT REFERENCES outlook_runs(event_id) ON DELETE CASCADE` | no | Parent event ID. |
| `outlook_index` | `INTEGER` | no | `payload.outlooks[]` index. |
| `as_of` | `TIMESTAMPTZ` | no | Parent `payload.asOf` |
| `product` | `TEXT` | no | `payload.outlooks[].product` |
| `day` | `INTEGER` | no | `payload.outlooks[].day` |
| `outlook_type` | `TEXT` | no | `payload.outlooks[].outlookType` |
| `label` | `TEXT` | no | `payload.outlooks[].label` |
| `label_text` | `TEXT` | yes | `payload.outlooks[].labelText` |
| `severity_rank` | `INTEGER` | yes | `payload.outlooks[].severityRank` |
| `valid_from` | `TIMESTAMPTZ` | no | `payload.outlooks[].validFrom` |
| `valid_to` | `TIMESTAMPTZ` | no | `payload.outlooks[].validTo` |
| `issued_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].issuedAt` |
| `expires_at` | `TIMESTAMPTZ` | no | `payload.outlooks[].expiresAt` |
| `forecaster` | `TEXT` | yes | `payload.outlooks[].forecaster` |
| `headline` | `TEXT` | yes | `payload.outlooks[].headline` |
| `summary` | `TEXT` | yes | `payload.outlooks[].summary` |
| `discussion` | `TEXT` | yes | `payload.outlooks[].discussion` |
| `source_url` | `TEXT` | yes | `payload.outlooks[].sourceUrl` |
| `image_url` | `TEXT` | yes | `payload.outlooks[].imageUrl` |
| `contains_location` | `BOOLEAN` | no | `payload.outlooks[].containsLocation` |
| `geometry_json` | `TEXT` | no | Compact JSON from `payload.outlooks[].geometry` |
## Retention
When sink param `prune` is set, every successful write transaction deletes rows
@@ -401,3 +470,5 @@ binary does not provide CLI commands for them.
- `WeatherAlertRun`: read `alert_runs`, join `alerts` by `run_event_id` ordered
by `alert_index`, then join `alert_references` by `run_event_id` and
`alert_index` ordered by `reference_index`.
- `WeatherOutlookRun`: read `outlook_runs`, then join `outlooks` by
`run_event_id` ordered by `outlook_index`.

134
docs/integrations/spc.md Normal file
View File

@@ -0,0 +1,134 @@
# SPC Integration Notes
## Purpose
This document describes the Storm Prediction Center convective outlook usage
implemented by `weatherfeeder`. It is for developers and operators maintaining
SPC source configuration, provider helpers, fixtures, and tests.
General config syntax belongs in [configuration](../config.md). Emitted JSON
events are documented in [event wire contract](events.md).
## Implemented Driver
| Driver | Kind | Raw schema | Canonical schema |
| --- | --- | --- | --- |
| `spc_convective_outlook` | `outlook` | `raw.spc.convective_outlook.v1` | `weather.outlook.v1` |
## Config Requirements
The driver requires:
- `latitude`
- `longitude`
- `user_agent`
Optional params are:
- `location_id`
- `location_name`
- `geojson_urls`
- `discussion_urls`
- `rss_url`
- `http_timeout`
- `http_response_body_limit_bytes`
RSS is not fetched unless `rss_url` is configured. URL override maps are intended
for tests and upstream URL changes; the default driver configuration owns the
current Day 1-3 SPC product URLs.
## Upstream Products Used
The source fetches twelve required GeoJSON products every poll:
- Day 1 categorical, tornado, hail, and wind
- Day 2 categorical, tornado, hail, and wind
- Day 3 categorical, tornado, hail, and wind
It also fetches three required print pages:
- Day 1 convective outlook print page
- Day 2 convective outlook print page
- Day 3 convective outlook print page
GeoJSON products are authoritative for outlook polygons, valid windows, issue
times, labels, and severity rank. Print pages are authoritative for headline,
summary, and discussion text.
## Accept Headers
GeoJSON requests use:
```text
application/geo+json, application/json
```
Print-page requests use:
```text
text/html, application/xhtml+xml
```
RSS requests, when configured, use:
```text
application/rss+xml, application/xml, text/xml
```
## Polling And Raw Events
The source polls all required GeoJSON and print-page products as one bundle. If
any required request fails or returns a non-2xx response, the poll returns an
error and emits no partial event.
The raw payload contains fetched bodies plus configured location metadata and
per-product metadata. The source parses only the timestamp metadata needed for
event effective time selection; canonical mapping belongs to the normalizer.
The source emits no event when a complete fetched bundle is unchanged from the
previous successful poll. It does this with a source-local hash of the fetched
document bodies.
## Time Handling
Raw source `effective_at` prefers:
1. the latest valid GeoJSON `ISSUE_ISO`;
2. the latest print-page `Updated:` timestamp;
3. RSS `lastBuildDate` when RSS was fetched and parseable;
4. fetch time.
The normalizer sets canonical `asOf` and normalized event `effective_at` from
the latest valid outlook feature `issuedAt`, with fallback to print-page update
time and then the incoming event envelope.
## Mapping Notes
Each GeoJSON feature becomes one canonical outlook. Products are ordered by day,
then categorical, tornado, hail, and wind. Feature order is preserved within
each product.
The normalizer computes `containsLocation` with the configured latitude and
longitude against compact GeoJSON `Polygon` or `MultiPolygon` geometry.
Coordinates use GeoJSON order, `[longitude, latitude]`, and boundary points
count as contained.
All outlook polygons are preserved, including polygons that do not contain the
configured point. Matching print-page headline, summary, and discussion text is
attached to every outlook for the same day.
## Failure Behavior
Constructor validation failures stop daemon startup. Polling failures are
returned to the scheduler and emit no event for that poll.
Normalization fails when required GeoJSON timestamps, labels, geometry, or
configured coordinates are missing or invalid. Print-page extraction failures
also fail normalization because print pages are required inputs.
## Tests To Inspect
- `internal/providers/spc/*_test.go`
- `internal/sources/spc/*_test.go`
- `internal/normalizers/spc/*_test.go`
- fixtures under `internal/providers/spc/testdata`

View File

@@ -27,6 +27,7 @@ Current mappings:
| `raw.nws.forecast_discussion.v1` | `weather.forecast_discussion.v1` |
| `raw.nws.weatherstories.v1` | `weather.weather_story.v1` |
| `raw.nws.alerts.v1` | `weather.alert.v1` |
| `raw.spc.convective_outlook.v1` | `weather.outlook.v1` |
## Boundaries
@@ -54,9 +55,18 @@ Weatherfeeder registers normalizers in a stable order:
1. NWS
2. Open-Meteo
3. OpenWeather
4. SPC
The current normalizers avoid ambiguous matches by using schema equality.
The SPC outlook normalizer decodes the raw multi-document bundle, maps each
GeoJSON feature to a canonical outlook, and enriches all outlooks for a day with
the matching print-page headline, summary, and discussion. It preserves compact
GeoJSON feature geometry and computes `containsLocation` with
`internal/geo.ContainsPoint` using the source-configured point. Boundary points
count as contained, and all polygons are preserved whether or not they contain
the point.
## State
Normalizers should be stateless. Shared helpers should be deterministic and free
@@ -78,6 +88,7 @@ to four digits after the decimal point.
- provider normalizer tests under `internal/normalizers/nws`
- provider normalizer tests under `internal/normalizers/openmeteo`
- provider normalizer tests under `internal/normalizers/openweather`
- provider normalizer tests under `internal/normalizers/spc`
- common helper tests under `internal/normalizers/common`
## Invariants

View File

@@ -19,6 +19,7 @@ Inputs are canonical feed events. The mapper currently handles these schemas:
- `weather.forecast_discussion.v1`
- `weather.weather_story.v1`
- `weather.alert.v1`
- `weather.outlook.v1`
Outputs are feedkit `PostgresWrite` values for weatherfeeder-owned tables.
Unsupported schemas produce no writes and no error.
@@ -80,6 +81,7 @@ Child rows use positional indexes to preserve canonical array order:
- `story_index`
- `alert_index`
- `reference_index`
- `outlook_index`
Required canonical fields are validated before writes are returned:
@@ -89,10 +91,14 @@ Required canonical fields are validated before writes are returned:
- forecast discussions require `issuedAt` and `product`;
- weather story runs require `asOf`, and each story requires `startTime`,
`endTime`, and `updatedAt`;
- alert runs require `asOf`, and each alert requires `id`.
- alert runs require `asOf`, and each alert requires `id`;
- outlook runs require `asOf`, and each outlook requires `id`, `provider`,
`product`, `day`, `outlookType`, `label`, `validFrom`, `validTo`, `issuedAt`,
`expiresAt`, and `geometry`.
Nullable canonical values are converted to SQL nulls by mapper helpers.
Observation present-weather raw values are stored as compact JSON text.
Observation present-weather raw values and outlook geometry values are stored as
compact JSON text.
## Failure Behavior

View File

@@ -74,7 +74,7 @@ Weatherfeeder-owned runtime state is in process:
- event channel contents;
- the bounded dedupe key set;
- source instances and their HTTP conditional validators;
- source instances and their in-memory unchanged-content state;
- scheduler and dispatcher goroutines.
There is no weatherfeeder-owned durable scheduler state, checkpoint, replay log,

View File

@@ -28,6 +28,7 @@ Current drivers:
| `openmeteo_observation` | `observation` | `raw.openmeteo.current.v1` |
| `openmeteo_forecast` | `forecast` | `raw.openmeteo.hourly.forecast.v1` |
| `openweather_observation` | `observation` | `raw.openweather.current.v1` |
| `spc_convective_outlook` | `outlook` | `raw.spc.convective_outlook.v1` |
## Boundaries
@@ -40,7 +41,7 @@ Current drivers:
## Config Fields Used
All current source drivers use feedkit `HTTPSource`.
Most source drivers use feedkit `HTTPSource`.
Required params:
@@ -56,12 +57,18 @@ Optional params:
OpenWeather observation sources additionally require the configured URL to use
metric units. This is enforced by `internal/providers/openweather`.
The SPC convective outlook source is a multi-document poll source rather than a
single-URL `HTTPSource`. It requires `latitude`, `longitude`, and `user_agent`;
accepts optional `location_id`, `location_name`, `geojson_urls`,
`discussion_urls`, and `rss_url`; and supports `http_timeout` and
`http_response_body_limit_bytes`.
Source-level `kinds`, when configured, are validated against the source's
advertised `Kinds()`.
## External Adapters Used
Sources use feedkit's HTTP helper for:
Most sources use feedkit's HTTP helper for:
- request construction;
- `User-Agent` and `Accept` headers;
@@ -71,11 +78,15 @@ Sources use feedkit's HTTP helper for:
NWS helpers parse NWS timestamps. Open-Meteo helpers parse provider-local times
with timezone or UTC-offset data. OpenWeather helpers enforce metric-unit URLs.
SPC helpers define Day 1-3 product metadata, parse GeoJSON timestamps, extract
cleaned print-page discussion text, and parse optional RSS metadata.
## State
HTTP conditional validators are held in each source instance. They are not
persisted across process restarts.
HTTP conditional validators are held in each single-document HTTP source
instance. They are not persisted across process restarts. The SPC outlook source
keeps only a source-local hash of the most recent complete required product
bundle and emits no event when a later complete bundle is unchanged.
## Failure Behavior
@@ -88,12 +99,18 @@ to default ID/effective-time behavior from feedkit source helpers.
Unchanged conditional responses return no events and no error.
The SPC outlook source fetches all configured Day 1-3 GeoJSON products and print
pages atomically. If any required GeoJSON or print-page request fails, the poll
returns an error and emits no partial bundle. RSS is fetched only when `rss_url`
is configured.
## Tests To Inspect
- `internal/sources/builtins_test.go`
- provider source tests under `internal/sources/nws`
- provider source tests under `internal/sources/openmeteo`
- provider source tests under `internal/sources/openweather`
- provider source tests under `internal/sources/spc`
- provider helper tests under `internal/providers`
## Invariants

View File

@@ -64,16 +64,20 @@ and before each interval tick. If no jitter is configured in code, feedkit uses
Poll failures are logged and do not stop the daemon. A failed poll emits no
events for that source until a subsequent poll succeeds.
## Conditional HTTP Fetches
## Unchanged Upstream Content
All current sources use feedkit's HTTP polling helper. By default,
`params.conditional` is `true`, so the helper keeps ETag and Last-Modified
validators in memory for each source instance.
Most sources use feedkit's single-document HTTP polling helper. By default,
`params.conditional` is `true` for those sources, so the helper keeps ETag and
Last-Modified validators in memory for each source instance.
If the upstream returns `304 Not Modified`, the source emits no events for that
poll. Validator state is in memory only; restarting the process starts with no
cached validators.
The SPC convective outlook source polls multiple required documents as one
bundle. It emits no event when a later complete bundle has the same fetched
document bodies as the previous successful poll.
## Processing And Dedupe
Every event passes through normalization first and dedupe second.
@@ -128,6 +132,7 @@ table in that same transaction. See the
- scheduler goroutines and timers;
- HTTP conditional request validators;
- source-local unchanged-content state;
- event channel buffers;
- per-sink fanout queues;
- the dedupe ID set.

View File

@@ -58,7 +58,8 @@ Diagnostic: compare the configured driver with the source driver table in
Safe fix: correct the driver name. Current drivers include `nws_observation`,
`nws_alerts`, `nws_forecast_hourly`, `nws_forecast_narrative`,
`nws_forecast_discussion`, `nws_weatherstories`, `openmeteo_observation`,
`openmeteo_forecast`, and `openweather_observation`.
`openmeteo_forecast`, `openweather_observation`, and
`spc_convective_outlook`.
## `unknown sink driver`
@@ -94,6 +95,20 @@ Diagnostic: inspect the named source in the error and check its `params`.
Safe fix: add non-empty `url` and `user_agent` values. See
[HTTP source params](config.md#http-source-params).
SPC convective outlook sources do not use `params.url`; they require
`latitude`, `longitude`, and `user_agent`. See
[SPC convective outlook params](config.md#spc-convective-outlook-params).
## `params.latitude is required` Or `params.longitude is required`
Symptom: startup exits for an `spc_convective_outlook` source.
Likely cause: the source is missing one of the configured point coordinates.
Diagnostic: inspect the named SPC source in the error and check its `params`.
Safe fix: add numeric `latitude` and `longitude` values in decimal degrees.
## `url must include units=metric`
Symptom: startup exits for an `openweather_observation` source.

View File

@@ -18,6 +18,18 @@ sources:
url: "https://api.weather.gov/alerts?point=38.6239,-90.3571&limit=20"
user_agent: "weatherfeeder example (operator@example.com)"
- name: SPCConvectiveOutlookSTL
mode: poll
kinds: ["outlook"]
driver: spc_convective_outlook
every: 30m
params:
latitude: 38.6239
longitude: -90.3571
location_id: "stl"
location_name: "St. Louis, MO"
user_agent: "weatherfeeder example (operator@example.com)"
sinks:
- name: nats_weather
driver: nats
@@ -27,4 +39,4 @@ sinks:
routes:
- sink: nats_weather
kinds: ["observation", "alert"]
kinds: ["observation", "alert", "outlook"]

View File

@@ -18,6 +18,18 @@ sources:
url: "https://api.open-meteo.com/v1/forecast?latitude=38.6239&longitude=-90.3571&hourly=temperature_2m,relative_humidity_2m,dew_point_2m,apparent_temperature,precipitation_probability,precipitation,snowfall,weather_code,surface_pressure,wind_speed_10m,wind_direction_10m&forecast_days=3"
user_agent: "weatherfeeder example (operator@example.com)"
- name: SPCConvectiveOutlookSTL
mode: poll
kinds: ["outlook"]
driver: spc_convective_outlook
every: 30m
params:
latitude: 38.6239
longitude: -90.3571
location_id: "stl"
location_name: "St. Louis, MO"
user_agent: "weatherfeeder example (operator@example.com)"
sinks:
- name: pg_weather
driver: postgres
@@ -29,4 +41,4 @@ sinks:
routes:
- sink: pg_weather
kinds: ["observation", "forecast"]
kinds: ["observation", "forecast", "outlook"]

107
internal/geo/geojson.go Normal file
View File

@@ -0,0 +1,107 @@
package geo
import (
"encoding/json"
"fmt"
)
type geometry struct {
Type string `json:"type"`
Coordinates json.RawMessage `json:"coordinates"`
}
// ContainsPoint reports whether a GeoJSON Polygon or MultiPolygon contains p.
// GeoJSON coordinate order is [longitude, latitude].
func ContainsPoint(raw []byte, p Point) (bool, error) {
if len(raw) == 0 {
return false, fmt.Errorf("geojson geometry is empty")
}
var geom geometry
if err := json.Unmarshal(raw, &geom); err != nil {
return false, fmt.Errorf("decode geojson geometry: %w", err)
}
switch geom.Type {
case "Polygon":
polygon, err := decodePolygon(geom.Coordinates)
if err != nil {
return false, fmt.Errorf("decode polygon: %w", err)
}
return polygonContainsPoint(polygon, p), nil
case "MultiPolygon":
multiPolygon, err := decodeMultiPolygon(geom.Coordinates)
if err != nil {
return false, fmt.Errorf("decode multipolygon: %w", err)
}
for _, polygon := range multiPolygon {
if polygonContainsPoint(polygon, p) {
return true, nil
}
}
return false, nil
case "":
return false, fmt.Errorf("geojson geometry type is required")
default:
return false, fmt.Errorf("unsupported geojson geometry type %q", geom.Type)
}
}
func decodePolygon(raw json.RawMessage) (Polygon, error) {
var coords [][][]float64
if err := json.Unmarshal(raw, &coords); err != nil {
return nil, err
}
return polygonFromCoordinates(coords)
}
func decodeMultiPolygon(raw json.RawMessage) ([]Polygon, error) {
var coords [][][][]float64
if err := json.Unmarshal(raw, &coords); err != nil {
return nil, err
}
if len(coords) == 0 {
return nil, fmt.Errorf("multipolygon has no polygons")
}
out := make([]Polygon, 0, len(coords))
for i, polygonCoords := range coords {
polygon, err := polygonFromCoordinates(polygonCoords)
if err != nil {
return nil, fmt.Errorf("polygons[%d]: %w", i, err)
}
out = append(out, polygon)
}
return out, nil
}
func polygonFromCoordinates(coords [][][]float64) (Polygon, error) {
if len(coords) == 0 {
return nil, fmt.Errorf("polygon has no rings")
}
polygon := make(Polygon, 0, len(coords))
for i, ringCoords := range coords {
ring, err := ringFromCoordinates(ringCoords)
if err != nil {
return nil, fmt.Errorf("rings[%d]: %w", i, err)
}
polygon = append(polygon, ring)
}
return polygon, nil
}
func ringFromCoordinates(coords [][]float64) (Ring, error) {
if len(coords) == 0 {
return nil, fmt.Errorf("ring has no points")
}
ring := make(Ring, 0, len(coords))
for i, pair := range coords {
if len(pair) < 2 {
return nil, fmt.Errorf("points[%d] has %d values, need longitude and latitude", i, len(pair))
}
ring = append(ring, Point{Longitude: pair[0], Latitude: pair[1]})
}
return ring, nil
}

105
internal/geo/point.go Normal file
View File

@@ -0,0 +1,105 @@
package geo
import "math"
const epsilon = 1e-9
// Point is a geographic coordinate in decimal degrees.
type Point struct {
Longitude float64
Latitude float64
}
// Ring is one GeoJSON linear ring.
type Ring []Point
// Polygon is a GeoJSON polygon. The first ring is the exterior ring; subsequent
// rings are holes.
type Polygon []Ring
func polygonContainsPoint(polygon Polygon, p Point) bool {
if len(polygon) == 0 {
return false
}
if pointOnRing(polygon[0], p) {
return true
}
if !ringContainsPoint(polygon[0], p) {
return false
}
for _, hole := range polygon[1:] {
if pointOnRing(hole, p) {
return true
}
if ringContainsPoint(hole, p) {
return false
}
}
return true
}
func ringContainsPoint(ring Ring, p Point) bool {
inside := false
n := len(ring)
if n == 0 {
return false
}
for i, j := 0, n-1; i < n; j, i = i, i+1 {
a := ring[j]
b := ring[i]
if pointOnSegment(p, a, b) {
return true
}
intersects := (a.Latitude > p.Latitude) != (b.Latitude > p.Latitude)
if intersects {
x := (b.Longitude-a.Longitude)*(p.Latitude-a.Latitude)/(b.Latitude-a.Latitude) + a.Longitude
if almostEqual(x, p.Longitude) {
return true
}
if x > p.Longitude {
inside = !inside
}
}
}
return inside
}
func pointOnRing(ring Ring, p Point) bool {
n := len(ring)
if n == 0 {
return false
}
for i, j := 0, n-1; i < n; j, i = i, i+1 {
if pointOnSegment(p, ring[j], ring[i]) {
return true
}
}
return false
}
func pointOnSegment(p, a, b Point) bool {
cross := (p.Latitude-a.Latitude)*(b.Longitude-a.Longitude) - (p.Longitude-a.Longitude)*(b.Latitude-a.Latitude)
if math.Abs(cross) > epsilon {
return false
}
minLon, maxLon := minMax(a.Longitude, b.Longitude)
minLat, maxLat := minMax(a.Latitude, b.Latitude)
return p.Longitude >= minLon-epsilon &&
p.Longitude <= maxLon+epsilon &&
p.Latitude >= minLat-epsilon &&
p.Latitude <= maxLat+epsilon
}
func minMax(a, b float64) (float64, float64) {
if a < b {
return a, b
}
return b, a
}
func almostEqual(a, b float64) bool {
return math.Abs(a-b) <= epsilon
}

181
internal/geo/point_test.go Normal file
View File

@@ -0,0 +1,181 @@
package geo
import (
"encoding/json"
"strings"
"testing"
)
const squarePolygon = `{
"type": "Polygon",
"coordinates": [[
[-91.0, 38.0],
[-90.0, 38.0],
[-90.0, 39.0],
[-91.0, 39.0],
[-91.0, 38.0]
]]
}`
func TestContainsPointInsideSimplePolygon(t *testing.T) {
got, err := ContainsPoint([]byte(squarePolygon), Point{Longitude: -90.5, Latitude: 38.5})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if !got {
t.Fatalf("ContainsPoint() = false, want true")
}
}
func TestContainsPointAcceptsRawMessage(t *testing.T) {
got, err := ContainsPoint(json.RawMessage(squarePolygon), Point{Longitude: -90.5, Latitude: 38.5})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if !got {
t.Fatalf("ContainsPoint() = false, want true")
}
}
func TestContainsPointOutsideSimplePolygon(t *testing.T) {
got, err := ContainsPoint([]byte(squarePolygon), Point{Longitude: -89.5, Latitude: 38.5})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if got {
t.Fatalf("ContainsPoint() = true, want false")
}
}
func TestContainsPointOnBoundary(t *testing.T) {
got, err := ContainsPoint([]byte(squarePolygon), Point{Longitude: -91.0, Latitude: 38.5})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if !got {
t.Fatalf("ContainsPoint() = false, want true")
}
}
func TestContainsPointInHoleReturnsFalse(t *testing.T) {
const polygonWithHole = `{
"type": "Polygon",
"coordinates": [
[[0,0],[10,0],[10,10],[0,10],[0,0]],
[[4,4],[6,4],[6,6],[4,6],[4,4]]
]
}`
got, err := ContainsPoint([]byte(polygonWithHole), Point{Longitude: 5, Latitude: 5})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if got {
t.Fatalf("ContainsPoint() = true, want false")
}
}
func TestContainsPointOnHoleBoundaryReturnsTrue(t *testing.T) {
const polygonWithHole = `{
"type": "Polygon",
"coordinates": [
[[0,0],[10,0],[10,10],[0,10],[0,0]],
[[4,4],[6,4],[6,6],[4,6],[4,4]]
]
}`
got, err := ContainsPoint([]byte(polygonWithHole), Point{Longitude: 4, Latitude: 5})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if !got {
t.Fatalf("ContainsPoint() = false, want true")
}
}
func TestContainsPointInsideOneMultiPolygonMember(t *testing.T) {
const multiPolygon = `{
"type": "MultiPolygon",
"coordinates": [
[[[0,0],[1,0],[1,1],[0,1],[0,0]]],
[[[10,10],[12,10],[12,12],[10,12],[10,10]]]
]
}`
got, err := ContainsPoint([]byte(multiPolygon), Point{Longitude: 11, Latitude: 11})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if !got {
t.Fatalf("ContainsPoint() = false, want true")
}
}
func TestContainsPointUsesLongitudeLatitudeOrder(t *testing.T) {
const narrowPolygon = `{
"type": "Polygon",
"coordinates": [[
[-91.0, 38.0],
[-90.0, 38.0],
[-90.0, 39.0],
[-91.0, 39.0],
[-91.0, 38.0]
]]
}`
got, err := ContainsPoint([]byte(narrowPolygon), Point{Longitude: -90.5, Latitude: 38.5})
if err != nil {
t.Fatalf("ContainsPoint() error = %v", err)
}
if !got {
t.Fatalf("ContainsPoint() = false, want true")
}
got, err = ContainsPoint([]byte(narrowPolygon), Point{Longitude: 38.5, Latitude: -90.5})
if err != nil {
t.Fatalf("ContainsPoint() reversed error = %v", err)
}
if got {
t.Fatalf("ContainsPoint() with reversed coordinate values = true, want false")
}
}
func TestContainsPointUnsupportedGeometryError(t *testing.T) {
_, err := ContainsPoint([]byte(`{"type":"Point","coordinates":[-90,38]}`), Point{Longitude: -90, Latitude: 38})
if err == nil {
t.Fatalf("ContainsPoint() error = nil, want error")
}
if !strings.Contains(err.Error(), `unsupported geojson geometry type "Point"`) {
t.Fatalf("ContainsPoint() error = %q", err)
}
}
func TestContainsPointInvalidJSONError(t *testing.T) {
_, err := ContainsPoint([]byte(`{"type":"Polygon"`), Point{Longitude: -90, Latitude: 38})
if err == nil {
t.Fatalf("ContainsPoint() error = nil, want error")
}
if !strings.Contains(err.Error(), "decode geojson geometry") {
t.Fatalf("ContainsPoint() error = %q", err)
}
}
func TestContainsPointMalformedCoordinatesError(t *testing.T) {
_, err := ContainsPoint([]byte(`{"type":"Polygon","coordinates":[[[1]]]}`), Point{Longitude: 1, Latitude: 1})
if err == nil {
t.Fatalf("ContainsPoint() error = nil, want error")
}
if !strings.Contains(err.Error(), "need longitude and latitude") {
t.Fatalf("ContainsPoint() error = %q", err)
}
}
func TestContainsPointEmptyRingError(t *testing.T) {
_, err := ContainsPoint([]byte(`{"type":"Polygon","coordinates":[[]]}`), Point{Longitude: 1, Latitude: 1})
if err == nil {
t.Fatalf("ContainsPoint() error = nil, want error")
}
if !strings.Contains(err.Error(), "ring has no points") {
t.Fatalf("ContainsPoint() error = %q", err)
}
}

View File

@@ -7,12 +7,14 @@ import (
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/nws"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/openmeteo"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/openweather"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/spc"
)
var builtinRegistrations = []func([]fknormalize.Normalizer) []fknormalize.Normalizer{
nws.Register,
openmeteo.Register,
openweather.Register,
spc.Register,
}
// RegisterBuiltins registers all normalizers shipped with this binary.

View File

@@ -8,6 +8,7 @@ import (
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/nws"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/openmeteo"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/openweather"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/spc"
)
func TestRegisterBuiltinsOrder(t *testing.T) {
@@ -25,6 +26,7 @@ func TestRegisterBuiltinsOrder(t *testing.T) {
openmeteo.ObservationNormalizer{},
openmeteo.ForecastNormalizer{},
openweather.ObservationNormalizer{},
spc.ConvectiveOutlookNormalizer{},
}
if len(got) != len(want) {

View File

@@ -0,0 +1,314 @@
package spc
import (
"context"
"encoding/json"
"fmt"
"math"
"regexp"
"sort"
"strings"
"time"
"gitea.maximumdirect.net/ejr/feedkit/event"
"gitea.maximumdirect.net/ejr/weatherfeeder/internal/geo"
normcommon "gitea.maximumdirect.net/ejr/weatherfeeder/internal/normalizers/common"
spcprovider "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/spc"
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
)
const (
providerSPC = "spc"
productConvective = "convective"
outlookNormalizer = "spc convective outlook"
outlookKind = "outlook"
outlookTypeUnknown = 99
)
var idTokenRE = regexp.MustCompile(`[^a-z0-9]+`)
// ConvectiveOutlookNormalizer converts:
//
// standards.SchemaRawSPCConvectiveOutlookV1 -> standards.SchemaWeatherOutlookV1
//
// It maps SPC GeoJSON outlook features into canonical outlook polygons and
// enriches each day with the matching required print-page discussion.
type ConvectiveOutlookNormalizer struct{}
func (ConvectiveOutlookNormalizer) Match(e event.Event) bool {
return strings.TrimSpace(e.Schema) == standards.SchemaRawSPCConvectiveOutlookV1
}
func (ConvectiveOutlookNormalizer) 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,
outlookNormalizer,
standards.SchemaWeatherOutlookV1,
func(parsed spcprovider.RawConvectiveOutlookBundle) (model.WeatherOutlookRun, time.Time, error) {
return buildConvectiveOutlook(parsed, fallbackAsOf)
},
)
}
func buildConvectiveOutlook(bundle spcprovider.RawConvectiveOutlookBundle, fallbackAsOf time.Time) (model.WeatherOutlookRun, time.Time, error) {
if err := validateCoordinates(bundle.Latitude, bundle.Longitude); err != nil {
return model.WeatherOutlookRun{}, time.Time{}, err
}
discussions, latestDiscussionUpdated, err := parseDiscussions(bundle.Discussions)
if err != nil {
return model.WeatherOutlookRun{}, time.Time{}, err
}
products := orderedProducts(bundle.Products)
point := geo.Point{Latitude: bundle.Latitude, Longitude: bundle.Longitude}
outlooks := make([]model.WeatherOutlook, 0)
var latestIssue time.Time
for _, product := range products {
if err := validateProductMetadata(product); err != nil {
return model.WeatherOutlookRun{}, time.Time{}, err
}
discussion, ok := discussions[product.Day]
if !ok {
return model.WeatherOutlookRun{}, time.Time{}, fmt.Errorf("product %s: discussion for day %d is required", product.Key, product.Day)
}
collection, err := spcprovider.DecodeGeoJSON(product.Body)
if err != nil {
return model.WeatherOutlookRun{}, time.Time{}, fmt.Errorf("product %s: %w", product.Key, err)
}
for i, feature := range collection.Features {
outlook, err := mapFeature(product, feature, i, point, discussion)
if err != nil {
return model.WeatherOutlookRun{}, time.Time{}, err
}
if latestIssue.IsZero() || outlook.IssuedAt.After(latestIssue) {
latestIssue = outlook.IssuedAt
}
outlooks = append(outlooks, outlook)
}
}
asOf := latestIssue
if asOf.IsZero() {
asOf = latestDiscussionUpdated
}
if asOf.IsZero() {
asOf = fallbackAsOf.UTC()
}
var issuedAt *time.Time
if !latestIssue.IsZero() {
t := latestIssue.UTC()
issuedAt = &t
}
lat := bundle.Latitude
lon := bundle.Longitude
run := model.WeatherOutlookRun{
LocationID: strings.TrimSpace(bundle.LocationID),
LocationName: strings.TrimSpace(bundle.LocationName),
Latitude: &lat,
Longitude: &lon,
AsOf: asOf.UTC(),
IssuedAt: issuedAt,
Outlooks: outlooks,
}
return run, run.AsOf, nil
}
type parsedDiscussion struct {
Headline string
Summary string
Discussion string
UpdatedAt *time.Time
}
func parseDiscussions(pages []spcprovider.RawDiscussionPage) (map[int]parsedDiscussion, time.Time, error) {
out := map[int]parsedDiscussion{}
var latestUpdated time.Time
for _, page := range pages {
text, err := spcprovider.ExtractProductText(page.Body)
if err != nil {
return nil, time.Time{}, fmt.Errorf("discussion %s: %w", page.Key, err)
}
parsed := spcprovider.ParseDiscussionText(text)
day := page.Day
if day == 0 {
if meta, ok := spcprovider.DiscussionProductByKey(page.Key); ok {
day = meta.Day
}
}
if day < 1 || day > 3 {
return nil, time.Time{}, fmt.Errorf("discussion %s: day must be 1, 2, or 3, got %d", page.Key, page.Day)
}
disc := parsedDiscussion{
Headline: strings.TrimSpace(parsed.Headline),
Summary: strings.TrimSpace(parsed.Summary),
Discussion: strings.TrimSpace(parsed.Discussion),
UpdatedAt: parsed.UpdatedAt,
}
out[day] = disc
if parsed.UpdatedAt != nil && (latestUpdated.IsZero() || parsed.UpdatedAt.After(latestUpdated)) {
latestUpdated = parsed.UpdatedAt.UTC()
}
}
return out, latestUpdated, nil
}
func orderedProducts(products []spcprovider.RawOutlookProduct) []spcprovider.RawOutlookProduct {
out := make([]spcprovider.RawOutlookProduct, len(products))
copy(out, products)
sort.SliceStable(out, func(i, j int) bool {
if out[i].Day != out[j].Day {
return out[i].Day < out[j].Day
}
left := outlookTypeOrder(out[i].OutlookType)
right := outlookTypeOrder(out[j].OutlookType)
if left != right {
return left < right
}
return out[i].Key < out[j].Key
})
return out
}
func outlookTypeOrder(outlookType string) int {
switch strings.TrimSpace(outlookType) {
case spcprovider.OutlookTypeCategorical:
return 0
case spcprovider.OutlookTypeTornado:
return 1
case spcprovider.OutlookTypeHail:
return 2
case spcprovider.OutlookTypeWind:
return 3
default:
return outlookTypeUnknown
}
}
func validateProductMetadata(product spcprovider.RawOutlookProduct) error {
if product.Day < 1 || product.Day > 3 {
return fmt.Errorf("product %s: day must be 1, 2, or 3, got %d", product.Key, product.Day)
}
switch strings.TrimSpace(product.OutlookType) {
case spcprovider.OutlookTypeCategorical, spcprovider.OutlookTypeTornado, spcprovider.OutlookTypeHail, spcprovider.OutlookTypeWind:
return nil
default:
return fmt.Errorf("product %s: unsupported outlook type %q", product.Key, product.OutlookType)
}
}
func mapFeature(product spcprovider.RawOutlookProduct, feature spcprovider.GeoJSONFeature, index int, point geo.Point, discussion parsedDiscussion) (model.WeatherOutlook, error) {
fieldPrefix := fmt.Sprintf("product %s feature %d", product.Key, index)
props := feature.Properties
validFrom, err := parseRequiredSPCTime(props.ValidISO, fieldPrefix+".VALID_ISO")
if err != nil {
return model.WeatherOutlook{}, err
}
validTo, err := parseRequiredSPCTime(props.ExpireISO, fieldPrefix+".EXPIRE_ISO")
if err != nil {
return model.WeatherOutlook{}, err
}
issuedAt, err := parseRequiredSPCTime(props.IssueISO, fieldPrefix+".ISSUE_ISO")
if err != nil {
return model.WeatherOutlook{}, err
}
label := strings.TrimSpace(props.Label)
if label == "" {
return model.WeatherOutlook{}, fmt.Errorf("%s.LABEL is required", fieldPrefix)
}
if len(feature.Geometry) == 0 {
return model.WeatherOutlook{}, fmt.Errorf("%s.geometry is required", fieldPrefix)
}
containsLocation, err := geo.ContainsPoint(feature.Geometry, point)
if err != nil {
return model.WeatherOutlook{}, fmt.Errorf("%s.geometry: %w", fieldPrefix, err)
}
geometry := make(json.RawMessage, len(feature.Geometry))
copy(geometry, feature.Geometry)
return model.WeatherOutlook{
ID: outlookID(product.Day, product.OutlookType, label, issuedAt, validFrom, index),
Provider: providerSPC,
Product: productConvective,
Day: product.Day,
OutlookType: strings.TrimSpace(product.OutlookType),
Label: label,
LabelText: strings.TrimSpace(props.Label2),
SeverityRank: props.DN,
ValidFrom: validFrom,
ValidTo: validTo,
IssuedAt: issuedAt,
ExpiresAt: validTo,
Forecaster: strings.TrimSpace(props.Forecaster),
Headline: discussion.Headline,
Summary: discussion.Summary,
Discussion: discussion.Discussion,
SourceURL: strings.TrimSpace(product.URL),
ImageURL: "",
ContainsLocation: containsLocation,
Geometry: geometry,
}, nil
}
func parseRequiredSPCTime(value, field string) (time.Time, error) {
if strings.TrimSpace(value) == "" {
return time.Time{}, fmt.Errorf("%s is required", field)
}
t, err := spcprovider.ParseISOTimestamp(value)
if err != nil {
return time.Time{}, fmt.Errorf("%s: %w", field, err)
}
return t.UTC(), nil
}
func outlookID(day int, outlookType, label string, issuedAt time.Time, validFrom time.Time, index int) string {
return fmt.Sprintf(
"spc-convective-day%d-%s-%s-%s-%s-%d",
day,
safeIDToken(outlookType),
safeIDToken(label),
issuedAt.UTC().Format(time.RFC3339),
validFrom.UTC().Format(time.RFC3339),
index,
)
}
func safeIDToken(value string) string {
value = strings.ToLower(strings.TrimSpace(value))
value = idTokenRE.ReplaceAllString(value, "-")
value = strings.Trim(value, "-")
if value == "" {
return "unknown"
}
return value
}
func validateCoordinates(latitude, longitude float64) error {
switch {
case math.IsNaN(latitude) || math.IsInf(latitude, 0):
return fmt.Errorf("latitude must be finite")
case math.IsNaN(longitude) || math.IsInf(longitude, 0):
return fmt.Errorf("longitude must be finite")
case latitude < -90 || latitude > 90:
return fmt.Errorf("latitude must be between -90 and 90, got %v", latitude)
case longitude < -180 || longitude > 180:
return fmt.Errorf("longitude must be between -180 and 180, got %v", longitude)
default:
return nil
}
}

View File

@@ -0,0 +1,354 @@
package spc
import (
"encoding/json"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/ejr/feedkit/event"
spcprovider "gitea.maximumdirect.net/ejr/weatherfeeder/internal/providers/spc"
"gitea.maximumdirect.net/ejr/weatherfeeder/model"
"gitea.maximumdirect.net/ejr/weatherfeeder/standards"
)
func TestConvectiveOutlookNormalizerMatch(t *testing.T) {
n := ConvectiveOutlookNormalizer{}
if !n.Match(event.Event{Schema: standards.SchemaRawSPCConvectiveOutlookV1}) {
t.Fatalf("Match(raw SPC outlook) = false, want true")
}
if n.Match(event.Event{Schema: standards.SchemaRawNWSAlertsV1}) {
t.Fatalf("Match(raw NWS alerts) = true, want false")
}
}
func TestConvectiveOutlookNormalizerProducesCanonicalSchemaAndMapsSample(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 38.5, -90.5)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
if out.Schema != standards.SchemaWeatherOutlookV1 {
t.Fatalf("Schema = %q, want %q", out.Schema, standards.SchemaWeatherOutlookV1)
}
if out.Kind != event.Kind("outlook") {
t.Fatalf("Kind = %q, want outlook", out.Kind)
}
run, ok := out.Payload.(model.WeatherOutlookRun)
if !ok {
t.Fatalf("Payload type = %T, want model.WeatherOutlookRun", out.Payload)
}
wantAsOf := time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC)
if !run.AsOf.Equal(wantAsOf) {
t.Fatalf("AsOf = %s, want %s", run.AsOf, wantAsOf)
}
if run.IssuedAt == nil || !run.IssuedAt.Equal(wantAsOf) {
t.Fatalf("IssuedAt = %v, want %s", run.IssuedAt, wantAsOf)
}
if out.EffectiveAt == nil || !out.EffectiveAt.Equal(wantAsOf) {
t.Fatalf("EffectiveAt = %v, want %s", out.EffectiveAt, wantAsOf)
}
if run.LocationID != "stl" || run.LocationName != "St. Louis, MO" {
t.Fatalf("location metadata = %q/%q", run.LocationID, run.LocationName)
}
if run.Latitude == nil || *run.Latitude != 38.5 || run.Longitude == nil || *run.Longitude != -90.5 {
t.Fatalf("coordinates = %v,%v", run.Latitude, run.Longitude)
}
if len(run.Outlooks) != 12 {
t.Fatalf("Outlooks length = %d, want 12", len(run.Outlooks))
}
got := run.Outlooks[0]
if got.Provider != "spc" || got.Product != "convective" {
t.Fatalf("provider/product = %q/%q", got.Provider, got.Product)
}
if got.Day != 1 || got.OutlookType != spcprovider.OutlookTypeCategorical {
t.Fatalf("day/type = %d/%q", got.Day, got.OutlookType)
}
if got.Label != "SLGT" || got.LabelText != "Slight Risk" {
t.Fatalf("label fields = %q/%q", got.Label, got.LabelText)
}
if got.SeverityRank == nil || *got.SeverityRank != 3 {
t.Fatalf("SeverityRank = %v, want 3", got.SeverityRank)
}
assertTime(t, "ValidFrom", got.ValidFrom, 2026, 6, 11, 13, 0, 0)
assertTime(t, "ValidTo", got.ValidTo, 2026, 6, 12, 12, 0, 0)
assertTime(t, "IssuedAt", got.IssuedAt, 2026, 6, 11, 12, 34, 56)
assertTime(t, "ExpiresAt", got.ExpiresAt, 2026, 6, 12, 12, 0, 0)
if got.Forecaster != "SMITH" {
t.Fatalf("Forecaster = %q, want SMITH", got.Forecaster)
}
if got.SourceURL != "https://example.invalid/day1_categorical.geojson" {
t.Fatalf("SourceURL = %q", got.SourceURL)
}
wantGeometry := `{"type":"Polygon","coordinates":[[[-91.0,38.0],[-90.0,38.0],[-90.0,39.0],[-91.0,39.0],[-91.0,38.0]]]}`
if string(got.Geometry) != wantGeometry {
t.Fatalf("Geometry = %s, want %s", got.Geometry, wantGeometry)
}
if !got.ContainsLocation {
t.Fatalf("ContainsLocation = false, want true")
}
if got.Headline != "Day 1 Convective Outlook" {
t.Fatalf("Headline = %q", got.Headline)
}
if !strings.Contains(got.Summary, "central Plains") {
t.Fatalf("Summary = %q", got.Summary)
}
if !strings.Contains(got.Discussion, "...DISCUSSION...") {
t.Fatalf("Discussion missing product text: %q", got.Discussion)
}
if got.ID != "spc-convective-day1-categorical-slgt-2026-06-11T12:34:56Z-2026-06-11T13:00:00Z-0" {
t.Fatalf("ID = %q", got.ID)
}
}
func TestConvectiveOutlookNormalizerOrdersProductsByDayAndType(t *testing.T) {
bundle := spcBundle(t, 0, 0)
for i, j := 0, len(bundle.Products)-1; i < j; i, j = i+1, j-1 {
bundle.Products[i], bundle.Products[j] = bundle.Products[j], bundle.Products[i]
}
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
got := []string{
run.Outlooks[0].OutlookType,
run.Outlooks[1].OutlookType,
run.Outlooks[2].OutlookType,
run.Outlooks[3].OutlookType,
}
want := []string{
spcprovider.OutlookTypeCategorical,
spcprovider.OutlookTypeTornado,
spcprovider.OutlookTypeHail,
spcprovider.OutlookTypeWind,
}
for i := range want {
if got[i] != want[i] || run.Outlooks[i].Day != 1 {
t.Fatalf("outlook[%d] = day %d type %q, want day 1 type %q", i, run.Outlooks[i].Day, got[i], want[i])
}
}
}
func TestConvectiveOutlookNormalizerMapsProbabilisticOutlookTypes(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
for _, outlookType := range []string{
spcprovider.OutlookTypeTornado,
spcprovider.OutlookTypeHail,
spcprovider.OutlookTypeWind,
} {
if findOutlook(run.Outlooks, 1, outlookType) == nil {
t.Fatalf("missing day 1 outlook type %q", outlookType)
}
}
}
func TestConvectiveOutlookNormalizerContainsLocationFalseOutsidePolygon(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
if run.Outlooks[0].ContainsLocation {
t.Fatalf("ContainsLocation = true, want false")
}
}
func TestConvectiveOutlookNormalizerPreservesCorrectionMarker(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 0, 0)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
run := out.Payload.(model.WeatherOutlookRun)
got := findOutlook(run.Outlooks, 2, spcprovider.OutlookTypeTornado)
if got == nil {
t.Fatalf("missing day 2 tornado outlook")
}
if !strings.Contains(got.Headline, "CORR 1") {
t.Fatalf("Headline = %q, want correction marker", got.Headline)
}
if !strings.Contains(got.Discussion, "CORR 1") {
t.Fatalf("Discussion = %q, want correction marker", got.Discussion)
}
}
func TestConvectiveOutlookNormalizerMissingRSSNormalizes(t *testing.T) {
bundle := spcBundle(t, 0, 0)
bundle.RSS = nil
if _, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle)); err != nil {
t.Fatalf("Normalize() error = %v", err)
}
}
func TestConvectiveOutlookNormalizerInvalidRequiredTimestampFailsWithContext(t *testing.T) {
bundle := spcBundle(t, 0, 0)
bundle.Products[0].Body = json.RawMessage(strings.Replace(
string(bundle.Products[0].Body),
`"ISSUE_ISO": "2026-06-11T12:34:56Z"`,
`"ISSUE_ISO": "bad"`,
1,
))
_, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle))
if err == nil {
t.Fatalf("Normalize() error = nil, want error")
}
if !strings.Contains(err.Error(), "product day1_categorical feature 0.ISSUE_ISO") {
t.Fatalf("error = %q, want product and feature context", err)
}
}
func TestConvectiveOutlookNormalizerInvalidGeometryFailsWithContext(t *testing.T) {
bundle := spcBundle(t, 0, 0)
bundle.Products[0].Body = json.RawMessage(strings.Replace(
string(bundle.Products[0].Body),
`"geometry": {`,
`"geometry": {"type":"LineString","coordinates":[[-91,38],[-90,39]]}, "oldGeometry": {`,
1,
))
_, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle))
if err == nil {
t.Fatalf("Normalize() error = nil, want error")
}
if !strings.Contains(err.Error(), "product day1_categorical feature 0.geometry") {
t.Fatalf("error = %q, want product and feature context", err)
}
}
func TestConvectiveOutlookNormalizerRejectsMissingLabel(t *testing.T) {
bundle := spcBundle(t, 0, 0)
bundle.Products[0].Body = json.RawMessage(strings.Replace(
string(bundle.Products[0].Body),
`"LABEL": "SLGT"`,
`"LABEL": ""`,
1,
))
_, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, bundle))
if err == nil {
t.Fatalf("Normalize() error = nil, want error")
}
if !strings.Contains(err.Error(), "product day1_categorical feature 0.LABEL") {
t.Fatalf("error = %q, want label context", err)
}
}
func TestConvectiveOutlookNormalizerOutputJSONShape(t *testing.T) {
out, err := (ConvectiveOutlookNormalizer{}).Normalize(nil, spcRawEvent(t, spcBundle(t, 38.5, -90.5)))
if err != nil {
t.Fatalf("Normalize() error = %v", err)
}
raw, err := json.Marshal(out.Payload)
if err != nil {
t.Fatalf("Marshal(payload) error = %v", err)
}
got := string(raw)
for _, want := range []string{`"asOf"`, `"outlooks"`, `"containsLocation"`, `"geometry"`} {
if !strings.Contains(got, want) {
t.Fatalf("payload JSON missing %s: %s", want, got)
}
}
for _, unwanted := range []string{`"products"`, `"discussions"`, `"fetchedAt"`, `"body"`} {
if strings.Contains(got, unwanted) {
t.Fatalf("payload JSON exposed raw key %s: %s", unwanted, got)
}
}
}
func spcRawEvent(t *testing.T, bundle spcprovider.RawConvectiveOutlookBundle) event.Event {
t.Helper()
raw, err := json.Marshal(bundle)
if err != nil {
t.Fatalf("Marshal(bundle) error = %v", err)
}
effectiveAt := time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC)
return event.Event{
ID: "evt-spc-outlook-1",
Kind: event.Kind("outlook"),
Source: "spc-test",
EmittedAt: time.Date(2026, 6, 11, 20, 5, 0, 0, time.UTC),
EffectiveAt: &effectiveAt,
Schema: standards.SchemaRawSPCConvectiveOutlookV1,
Payload: json.RawMessage(raw),
}
}
func spcBundle(t *testing.T, latitude, longitude float64) spcprovider.RawConvectiveOutlookBundle {
t.Helper()
fetchedAt := time.Date(2026, 6, 11, 20, 0, 0, 0, time.UTC)
products := make([]spcprovider.RawOutlookProduct, 0, len(spcprovider.GeoJSONProducts()))
for _, product := range spcprovider.GeoJSONProducts() {
products = append(products, spcprovider.RawOutlookProduct{
Key: product.Key,
Day: product.Day,
OutlookType: product.OutlookType,
URL: "https://example.invalid/" + product.Key + ".geojson",
FetchedAt: fetchedAt,
Body: json.RawMessage(geoJSONFixtureForProduct(t, product.Key)),
})
}
return spcprovider.RawConvectiveOutlookBundle{
LocationID: "stl",
LocationName: "St. Louis, MO",
Latitude: latitude,
Longitude: longitude,
FetchedAt: fetchedAt,
Products: products,
Discussions: []spcprovider.RawDiscussionPage{
{Key: "day1", Day: 1, URL: "https://example.invalid/day1.html", FetchedAt: fetchedAt, Body: string(readSPCTestFixture(t, "day1_prt.html"))},
{Key: "day2", Day: 2, URL: "https://example.invalid/day2.html", FetchedAt: fetchedAt, Body: string(readSPCTestFixture(t, "day2_prt_corr.html"))},
{Key: "day3", Day: 3, URL: "https://example.invalid/day3.html", FetchedAt: fetchedAt, Body: string(readSPCTestFixture(t, "day3_prt.html"))},
},
}
}
func geoJSONFixtureForProduct(t *testing.T, key string) []byte {
t.Helper()
switch {
case strings.HasPrefix(key, "day1_"):
return readSPCTestFixture(t, "day1_cat.geojson")
case strings.HasPrefix(key, "day2_"):
return readSPCTestFixture(t, "day2_torn.geojson")
case strings.HasPrefix(key, "day3_"):
return readSPCTestFixture(t, "day3_wind.geojson")
default:
t.Fatalf("unknown product 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
}
func findOutlook(outlooks []model.WeatherOutlook, day int, outlookType string) *model.WeatherOutlook {
for i := range outlooks {
if outlooks[i].Day == day && outlooks[i].OutlookType == outlookType {
return &outlooks[i]
}
}
return nil
}
func assertTime(t *testing.T, name string, got time.Time, year int, month time.Month, day int, hour int, minute int, second int) {
t.Helper()
want := time.Date(year, month, day, hour, minute, second, 0, time.UTC)
if !got.Equal(want) {
t.Fatalf("%s = %s, want %s", name, got, want)
}
}

View File

@@ -0,0 +1,14 @@
package spc
import (
fknormalize "gitea.maximumdirect.net/ejr/feedkit/processors/normalize"
)
var builtins = []fknormalize.Normalizer{
ConvectiveOutlookNormalizer{},
}
// Register appends SPC normalizers in stable order.
func Register(in []fknormalize.Normalizer) []fknormalize.Normalizer {
return append(in, builtins...)
}

View File

@@ -0,0 +1,158 @@
package spc
import (
"fmt"
"html"
"regexp"
"strings"
"time"
)
var (
scriptBlockRE = regexp.MustCompile(`(?is)<script\b[^>]*>.*?</script>`)
preBlockRE = regexp.MustCompile(`(?is)<pre\b[^>]*>(.*?)</pre>`)
tagRE = regexp.MustCompile(`(?is)<[^>]+>`)
updatedRE = regexp.MustCompile(`(?im)^\s*Updated:\s*(.+?)\s*$`)
sectionRE = regexp.MustCompile(`^\s*\.\.\.[A-Z0-9 /-]+\.{3}\s*$`)
)
// DiscussionText contains parsed text from an SPC print page.
type DiscussionText struct {
ProductTitle string
Headline string
Summary string
Discussion string
UpdatedAt *time.Time
}
// ExtractProductText extracts and cleans the first useful preformatted SPC
// product text block from a print page.
func ExtractProductText(rawHTML string) (string, error) {
matches := preBlockRE.FindAllStringSubmatch(rawHTML, -1)
for _, match := range matches {
if len(match) < 2 {
continue
}
text := cleanHTMLText(match[1])
if strings.TrimSpace(text) != "" {
return text, nil
}
}
return "", fmt.Errorf("no useful pre block found")
}
// ParseDiscussionText extracts common SPC narrative metadata from cleaned
// product text.
func ParseDiscussionText(text string) DiscussionText {
text = trimBlankLines(normalizeNewlines(text))
title := ParseProductTitle(text)
return DiscussionText{
ProductTitle: title,
Headline: title,
Summary: ExtractSummary(text),
Discussion: text,
UpdatedAt: ParseUpdatedTimestamp(text),
}
}
// ParseUpdatedTimestamp parses an SPC print-page Updated line when present.
func ParseUpdatedTimestamp(text string) *time.Time {
match := updatedRE.FindStringSubmatch(normalizeNewlines(text))
if len(match) != 2 {
return nil
}
return parseUpdatedValue(match[1])
}
// ParseProductTitle returns the first non-empty product line from cleaned text.
func ParseProductTitle(text string) string {
for _, line := range strings.Split(normalizeNewlines(text), "\n") {
line = strings.TrimSpace(line)
if line == "" || strings.HasPrefix(line, "Updated:") {
continue
}
return line
}
return ""
}
// ParseHeadline returns the human-facing headline from cleaned text.
func ParseHeadline(text string) string {
return ParseProductTitle(text)
}
// ExtractSummary returns text under the ...SUMMARY... section through the next
// SPC section heading.
func ExtractSummary(text string) string {
lines := strings.Split(normalizeNewlines(text), "\n")
start := -1
for i, line := range lines {
if strings.EqualFold(strings.TrimSpace(line), "...SUMMARY...") {
start = i + 1
break
}
}
if start < 0 {
return ""
}
var out []string
for _, line := range lines[start:] {
if sectionRE.MatchString(line) {
break
}
out = append(out, line)
}
return trimBlankLines(strings.Join(out, "\n"))
}
func cleanHTMLText(raw string) string {
raw = scriptBlockRE.ReplaceAllString(raw, "")
raw = tagRE.ReplaceAllString(raw, "")
raw = html.UnescapeString(raw)
raw = normalizeNewlines(raw)
return trimBlankLines(raw)
}
func normalizeNewlines(text string) string {
text = strings.ReplaceAll(text, "\r\n", "\n")
text = strings.ReplaceAll(text, "\r", "\n")
return text
}
func trimBlankLines(text string) string {
lines := strings.Split(normalizeNewlines(text), "\n")
start := 0
for start < len(lines) && strings.TrimSpace(lines[start]) == "" {
start++
}
end := len(lines)
for end > start && strings.TrimSpace(lines[end-1]) == "" {
end--
}
return strings.Join(lines[start:end], "\n")
}
func parseUpdatedValue(value string) *time.Time {
value = strings.TrimSpace(value)
if value == "" {
return nil
}
if t := parseOptionalISOTimestamp(value); t != nil {
return t
}
for _, layout := range []string{
"1504 UTC Mon Jan 2 2006",
"1504Z Mon Jan 2 2006",
"3:04 PM UTC Mon Jan 2 2006",
time.RFC1123,
time.RFC1123Z,
} {
t, err := time.Parse(layout, value)
if err == nil {
tt := t.UTC()
return &tt
}
}
return nil
}

View File

@@ -0,0 +1,90 @@
package spc
import (
"strings"
"testing"
"time"
)
func TestExtractProductTextCleansPreBlock(t *testing.T) {
raw := string(readTestFile(t, "day1_prt.html"))
got, err := ExtractProductText(raw)
if err != nil {
t.Fatalf("ExtractProductText() error = %v", err)
}
if strings.Contains(got, "<script") || strings.Contains(got, "<pre") {
t.Fatalf("ExtractProductText() retained HTML: %q", got)
}
if strings.Contains(got, "ignore me") {
t.Fatalf("ExtractProductText() retained script content: %q", got)
}
if !strings.Contains(got, "Day 1 Convective Outlook") {
t.Fatalf("ExtractProductText() missing headline: %q", got)
}
if strings.HasPrefix(got, "\n") || strings.HasSuffix(got, "\n") {
t.Fatalf("ExtractProductText() retained surrounding blank lines: %q", got)
}
}
func TestParseDiscussionTextExtractsHeadlineSummaryAndUpdated(t *testing.T) {
text, err := ExtractProductText(string(readTestFile(t, "day1_prt.html")))
if err != nil {
t.Fatalf("ExtractProductText() error = %v", err)
}
got := ParseDiscussionText(text)
if got.ProductTitle != "Day 1 Convective Outlook" {
t.Fatalf("ProductTitle = %q", got.ProductTitle)
}
if got.Headline != "Day 1 Convective Outlook" {
t.Fatalf("Headline = %q", got.Headline)
}
wantSummary := "Severe thunderstorms are possible across parts of the central Plains\nand mid Mississippi Valley this afternoon and evening."
if got.Summary != wantSummary {
t.Fatalf("Summary = %q, want %q", got.Summary, wantSummary)
}
if !strings.Contains(got.Discussion, "...DISCUSSION...") {
t.Fatalf("Discussion missing full text: %q", got.Discussion)
}
wantUpdated := time.Date(2026, 6, 11, 12, 45, 0, 0, time.UTC)
if got.UpdatedAt == nil || !got.UpdatedAt.Equal(wantUpdated) {
t.Fatalf("UpdatedAt = %v, want %s", got.UpdatedAt, wantUpdated)
}
}
func TestParseDiscussionTextPreservesCorrectionMarker(t *testing.T) {
text, err := ExtractProductText(string(readTestFile(t, "day2_prt_corr.html")))
if err != nil {
t.Fatalf("ExtractProductText() error = %v", err)
}
got := ParseDiscussionText(text)
if !strings.Contains(got.Headline, "CORR 1") {
t.Fatalf("Headline = %q, want correction marker", got.Headline)
}
if !strings.Contains(got.Discussion, "CORR 1") {
t.Fatalf("Discussion = %q, want correction marker", got.Discussion)
}
if got.UpdatedAt != nil {
t.Fatalf("UpdatedAt = %v, want nil", got.UpdatedAt)
}
}
func TestParseUpdatedTimestampReturnsNilWhenAbsent(t *testing.T) {
text, err := ExtractProductText(string(readTestFile(t, "day2_prt_corr.html")))
if err != nil {
t.Fatalf("ExtractProductText() error = %v", err)
}
if got := ParseUpdatedTimestamp(text); got != nil {
t.Fatalf("ParseUpdatedTimestamp() = %v, want nil", got)
}
}
func TestParseUpdatedTimestampAcceptsSPCUTCFormat(t *testing.T) {
got := ParseUpdatedTimestamp("Updated: 1945 UTC Thu Jun 11 2026")
want := time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC)
if got == nil || !got.Equal(want) {
t.Fatalf("ParseUpdatedTimestamp() = %v, want %s", got, want)
}
}

View File

@@ -0,0 +1,8 @@
// Package spc contains deterministic helper code for Storm Prediction Center
// products used by sources and normalizers.
//
// Rules:
// - No network I/O here.
// - Keep helpers deterministic and easy to unit test.
// - Preserve upstream payload fragments needed for canonical mapping.
package spc

View File

@@ -0,0 +1,105 @@
package spc
import (
"bytes"
"encoding/json"
"fmt"
"strconv"
"strings"
)
// GeoJSONFeatureCollection is the minimal SPC outlook FeatureCollection shape
// needed by weatherfeeder.
type GeoJSONFeatureCollection struct {
Type string `json:"type"`
Features []GeoJSONFeature `json:"features"`
}
// GeoJSONFeature preserves typed SPC properties and compact raw geometry.
type GeoJSONFeature struct {
Type string `json:"type"`
Properties GeoJSONProperties `json:"properties"`
Geometry json.RawMessage `json:"geometry"`
}
// GeoJSONProperties contains the SPC fields used by canonical mapping.
type GeoJSONProperties struct {
ValidISO string `json:"VALID_ISO"`
ExpireISO string `json:"EXPIRE_ISO"`
IssueISO string `json:"ISSUE_ISO"`
Forecaster string `json:"FORECASTER"`
Label string `json:"LABEL"`
Label2 string `json:"LABEL2"`
DN *int `json:"DN"`
}
// DecodeGeoJSON decodes an SPC GeoJSON outlook product and compacts feature
// geometry JSON for stable downstream storage.
func DecodeGeoJSON(raw []byte) (GeoJSONFeatureCollection, error) {
var collection GeoJSONFeatureCollection
if err := json.Unmarshal(raw, &collection); err != nil {
return GeoJSONFeatureCollection{}, fmt.Errorf("decode geojson: %w", err)
}
for i := range collection.Features {
geom, err := compactJSON(collection.Features[i].Geometry)
if err != nil {
return GeoJSONFeatureCollection{}, fmt.Errorf("features[%d].geometry: %w", i, err)
}
collection.Features[i].Geometry = geom
}
return collection, nil
}
func (p *GeoJSONProperties) UnmarshalJSON(raw []byte) error {
type alias GeoJSONProperties
var aux struct {
alias
DN any `json:"DN"`
}
if err := json.Unmarshal(raw, &aux); err != nil {
return err
}
*p = GeoJSONProperties(aux.alias)
dn, err := parseSeverityRank(aux.DN)
if err != nil {
return err
}
p.DN = dn
return nil
}
func parseSeverityRank(value any) (*int, error) {
switch v := value.(type) {
case nil:
return nil, nil
case float64:
rank := int(v)
if float64(rank) != v {
return nil, fmt.Errorf("DN must be an integer, got %v", v)
}
return &rank, nil
case string:
v = strings.TrimSpace(v)
if v == "" {
return nil, nil
}
rank, err := strconv.Atoi(v)
if err != nil {
return nil, fmt.Errorf("DN must be an integer, got %q", v)
}
return &rank, nil
default:
return nil, fmt.Errorf("DN must be an integer or string, got %T", value)
}
}
func compactJSON(raw json.RawMessage) (json.RawMessage, error) {
if len(raw) == 0 {
return nil, fmt.Errorf("missing")
}
var buf bytes.Buffer
if err := json.Compact(&buf, raw); err != nil {
return nil, err
}
return json.RawMessage(buf.Bytes()), nil
}

View File

@@ -0,0 +1,90 @@
package spc
import (
"os"
"path/filepath"
"strings"
"testing"
"time"
)
func TestDecodeGeoJSONExposesSPCPropertiesAndCompactGeometry(t *testing.T) {
raw := readTestFile(t, "day1_cat.geojson")
got, err := DecodeGeoJSON(raw)
if err != nil {
t.Fatalf("DecodeGeoJSON() error = %v", err)
}
if got.Type != "FeatureCollection" {
t.Fatalf("Type = %q, want FeatureCollection", got.Type)
}
if len(got.Features) != 1 {
t.Fatalf("Features length = %d, want 1", len(got.Features))
}
feature := got.Features[0]
props := feature.Properties
if props.ValidISO != "2026-06-11T13:00:00Z" {
t.Fatalf("VALID_ISO = %q", props.ValidISO)
}
if props.ExpireISO != "2026-06-12T12:00:00Z" {
t.Fatalf("EXPIRE_ISO = %q", props.ExpireISO)
}
if props.IssueISO != "2026-06-11T12:34:56Z" {
t.Fatalf("ISSUE_ISO = %q", props.IssueISO)
}
if props.Forecaster != "SMITH" {
t.Fatalf("FORECASTER = %q", props.Forecaster)
}
if props.Label != "SLGT" {
t.Fatalf("LABEL = %q", props.Label)
}
if props.Label2 != "Slight Risk" {
t.Fatalf("LABEL2 = %q", props.Label2)
}
if props.DN == nil || *props.DN != 3 {
t.Fatalf("DN = %v, want 3", props.DN)
}
wantGeometry := `{"type":"Polygon","coordinates":[[[-91.0,38.0],[-90.0,38.0],[-90.0,39.0],[-91.0,39.0],[-91.0,38.0]]]}`
if string(feature.Geometry) != wantGeometry {
t.Fatalf("Geometry = %s, want %s", feature.Geometry, wantGeometry)
}
if strings.Contains(string(feature.Geometry), "\n") || strings.Contains(string(feature.Geometry), " ") {
t.Fatalf("Geometry is not compact: %q", feature.Geometry)
}
}
func TestDecodeGeoJSONParsesSeverityRankString(t *testing.T) {
raw := readTestFile(t, "day2_torn.geojson")
got, err := DecodeGeoJSON(raw)
if err != nil {
t.Fatalf("DecodeGeoJSON() error = %v", err)
}
props := got.Features[0].Properties
if props.DN == nil || *props.DN != 5 {
t.Fatalf("DN = %v, want 5", props.DN)
}
}
func TestParseISOTimestampTrimsAndReturnsUTC(t *testing.T) {
got, err := ParseISOTimestamp(" 2026-06-11T12:34:56Z ")
if err != nil {
t.Fatalf("ParseISOTimestamp() error = %v", err)
}
want := time.Date(2026, 6, 11, 12, 34, 56, 0, time.UTC)
if !got.Equal(want) {
t.Fatalf("ParseISOTimestamp() = %s, want %s", got, want)
}
}
func readTestFile(t *testing.T, name string) []byte {
t.Helper()
path := filepath.Join("testdata", name)
raw, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read %s: %v", path, err)
}
return raw
}

View File

@@ -0,0 +1,89 @@
package spc
import "fmt"
const (
OutlookTypeCategorical = "categorical"
OutlookTypeTornado = "tornado"
OutlookTypeHail = "hail"
OutlookTypeWind = "wind"
)
// GeoJSONProduct describes one required SPC convective outlook GeoJSON product.
type GeoJSONProduct struct {
Key string
Day int
OutlookType string
URL string
}
// DiscussionProduct describes one required SPC convective outlook print page.
type DiscussionProduct struct {
Key string
Day int
URL string
}
var geoJSONProducts = []GeoJSONProduct{
{Key: "day1_categorical", Day: 1, OutlookType: OutlookTypeCategorical, URL: "https://www.spc.noaa.gov/products/outlook/day1otlk_cat.nolyr.geojson"},
{Key: "day1_tornado", Day: 1, OutlookType: OutlookTypeTornado, URL: "https://www.spc.noaa.gov/products/outlook/day1otlk_torn.nolyr.geojson"},
{Key: "day1_hail", Day: 1, OutlookType: OutlookTypeHail, URL: "https://www.spc.noaa.gov/products/outlook/day1otlk_hail.nolyr.geojson"},
{Key: "day1_wind", Day: 1, OutlookType: OutlookTypeWind, URL: "https://www.spc.noaa.gov/products/outlook/day1otlk_wind.nolyr.geojson"},
{Key: "day2_categorical", Day: 2, OutlookType: OutlookTypeCategorical, URL: "https://www.spc.noaa.gov/products/outlook/day2otlk_cat.nolyr.geojson"},
{Key: "day2_tornado", Day: 2, OutlookType: OutlookTypeTornado, URL: "https://www.spc.noaa.gov/products/outlook/day2otlk_torn.nolyr.geojson"},
{Key: "day2_hail", Day: 2, OutlookType: OutlookTypeHail, URL: "https://www.spc.noaa.gov/products/outlook/day2otlk_hail.nolyr.geojson"},
{Key: "day2_wind", Day: 2, OutlookType: OutlookTypeWind, URL: "https://www.spc.noaa.gov/products/outlook/day2otlk_wind.nolyr.geojson"},
{Key: "day3_categorical", Day: 3, OutlookType: OutlookTypeCategorical, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_cat.nolyr.geojson"},
{Key: "day3_tornado", Day: 3, OutlookType: OutlookTypeTornado, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_torn.nolyr.geojson"},
{Key: "day3_hail", Day: 3, OutlookType: OutlookTypeHail, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_hail.nolyr.geojson"},
{Key: "day3_wind", Day: 3, OutlookType: OutlookTypeWind, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_wind.nolyr.geojson"},
}
var discussionProducts = []DiscussionProduct{
{Key: "day1", Day: 1, URL: "https://www.spc.noaa.gov/products/outlook/day1otlk_prt.html"},
{Key: "day2", Day: 2, URL: "https://www.spc.noaa.gov/products/outlook/day2otlk_prt.html"},
{Key: "day3", Day: 3, URL: "https://www.spc.noaa.gov/products/outlook/day3otlk_prt.html"},
}
// GeoJSONProducts returns the required SPC convective outlook GeoJSON products
// in stable day/type order.
func GeoJSONProducts() []GeoJSONProduct {
out := make([]GeoJSONProduct, len(geoJSONProducts))
copy(out, geoJSONProducts)
return out
}
// DiscussionProducts returns the required SPC convective outlook print pages in
// stable day order.
func DiscussionProducts() []DiscussionProduct {
out := make([]DiscussionProduct, len(discussionProducts))
copy(out, discussionProducts)
return out
}
// GeoJSONProductByKey returns product metadata for a configured product key.
func GeoJSONProductByKey(key string) (GeoJSONProduct, bool) {
for _, product := range geoJSONProducts {
if product.Key == key {
return product, true
}
}
return GeoJSONProduct{}, false
}
// DiscussionProductByKey returns discussion metadata for a configured day key.
func DiscussionProductByKey(key string) (DiscussionProduct, bool) {
for _, product := range discussionProducts {
if product.Key == key {
return product, true
}
}
return DiscussionProduct{}, false
}
func validateProductDay(day int) error {
if day < 1 || day > 3 {
return fmt.Errorf("day must be 1, 2, or 3, got %d", day)
}
return nil
}

View File

@@ -0,0 +1,56 @@
package spc
import "testing"
func TestGeoJSONProductsStableOrder(t *testing.T) {
got := GeoJSONProducts()
if len(got) != 12 {
t.Fatalf("GeoJSONProducts() length = %d, want 12", len(got))
}
wantKeys := []string{
"day1_categorical",
"day1_tornado",
"day1_hail",
"day1_wind",
"day2_categorical",
"day2_tornado",
"day2_hail",
"day2_wind",
"day3_categorical",
"day3_tornado",
"day3_hail",
"day3_wind",
}
for i, want := range wantKeys {
if got[i].Key != want {
t.Fatalf("GeoJSONProducts()[%d].Key = %q, want %q", i, got[i].Key, want)
}
if err := validateProductDay(got[i].Day); err != nil {
t.Fatalf("GeoJSONProducts()[%d].Day invalid: %v", i, err)
}
if got[i].URL == "" {
t.Fatalf("GeoJSONProducts()[%d].URL is empty", i)
}
}
}
func TestDiscussionProductsStableOrder(t *testing.T) {
got := DiscussionProducts()
if len(got) != 3 {
t.Fatalf("DiscussionProducts() length = %d, want 3", len(got))
}
wantKeys := []string{"day1", "day2", "day3"}
for i, want := range wantKeys {
if got[i].Key != want {
t.Fatalf("DiscussionProducts()[%d].Key = %q, want %q", i, got[i].Key, want)
}
if got[i].Day != i+1 {
t.Fatalf("DiscussionProducts()[%d].Day = %d, want %d", i, got[i].Day, i+1)
}
if got[i].URL == "" {
t.Fatalf("DiscussionProducts()[%d].URL is empty", i)
}
}
}

View File

@@ -0,0 +1,45 @@
package spc
import (
"encoding/json"
"time"
)
// RawConvectiveOutlookBundle is the provider payload shape for SPC convective
// outlook fetch bundles.
type RawConvectiveOutlookBundle struct {
LocationID string `json:"locationId,omitempty"`
LocationName string `json:"locationName,omitempty"`
Latitude float64 `json:"latitude"`
Longitude float64 `json:"longitude"`
FetchedAt time.Time `json:"fetchedAt"`
Products []RawOutlookProduct `json:"products"`
Discussions []RawDiscussionPage `json:"discussions"`
RSS *RawRSSFeed `json:"rss,omitempty"`
}
// RawOutlookProduct contains one fetched SPC GeoJSON product.
type RawOutlookProduct struct {
Key string `json:"key"`
Day int `json:"day"`
OutlookType string `json:"outlookType"`
URL string `json:"url"`
FetchedAt time.Time `json:"fetchedAt"`
Body json.RawMessage `json:"body"`
}
// RawDiscussionPage contains one fetched SPC print page.
type RawDiscussionPage struct {
Key string `json:"key"`
Day int `json:"day"`
URL string `json:"url"`
FetchedAt time.Time `json:"fetchedAt"`
Body string `json:"body"`
}
// RawRSSFeed contains optional fetched SPC RSS metadata.
type RawRSSFeed struct {
URL string `json:"url"`
FetchedAt time.Time `json:"fetchedAt"`
Body string `json:"body"`
}

View File

@@ -0,0 +1,51 @@
package spc
import (
"encoding/json"
"testing"
"time"
)
func TestRawConvectiveOutlookBundleJSONShape(t *testing.T) {
fetchedAt := time.Date(2026, 6, 11, 20, 0, 0, 0, time.UTC)
bundle := RawConvectiveOutlookBundle{
LocationID: "stl",
LocationName: "St. Louis, MO",
Latitude: 38.6239,
Longitude: -90.3571,
FetchedAt: fetchedAt,
Products: []RawOutlookProduct{{
Key: "day1_categorical",
Day: 1,
OutlookType: OutlookTypeCategorical,
URL: "https://example.invalid/day1.geojson",
FetchedAt: fetchedAt,
Body: json.RawMessage(`{"type":"FeatureCollection","features":[]}`),
}},
Discussions: []RawDiscussionPage{{
Key: "day1",
Day: 1,
URL: "https://example.invalid/day1.html",
FetchedAt: fetchedAt,
Body: "Day 1 Convective Outlook",
}},
}
raw, err := json.Marshal(bundle)
if err != nil {
t.Fatalf("Marshal() error = %v", err)
}
var got map[string]any
if err := json.Unmarshal(raw, &got); err != nil {
t.Fatalf("Unmarshal() error = %v", err)
}
for _, key := range []string{"locationId", "locationName", "latitude", "longitude", "fetchedAt", "products", "discussions"} {
if _, ok := got[key]; !ok {
t.Fatalf("marshaled bundle missing key %q in %s", key, raw)
}
}
if _, ok := got["rss"]; ok {
t.Fatalf("marshaled bundle included empty rss: %s", raw)
}
}

View File

@@ -0,0 +1,81 @@
package spc
import (
"encoding/xml"
"fmt"
"strings"
"time"
)
// RSSFeed is a minimal view of the optional SPC RSS feed.
type RSSFeed struct {
Title string
Link string
Description string
LastBuildDate *time.Time
Items []RSSItem
}
// RSSItem is a minimal view of one optional SPC RSS item.
type RSSItem struct {
Title string
Link string
Description string
PubDate string
GUID string
}
// ParseRSSFeed decodes supplemental SPC RSS metadata.
func ParseRSSFeed(raw string) (RSSFeed, error) {
var doc struct {
Channel struct {
Title string `xml:"title"`
Link string `xml:"link"`
Description string `xml:"description"`
LastBuildDate string `xml:"lastBuildDate"`
Items []struct {
Title string `xml:"title"`
Link string `xml:"link"`
Description string `xml:"description"`
PubDate string `xml:"pubDate"`
GUID string `xml:"guid"`
} `xml:"item"`
} `xml:"channel"`
}
if err := xml.Unmarshal([]byte(raw), &doc); err != nil {
return RSSFeed{}, fmt.Errorf("decode rss: %w", err)
}
feed := RSSFeed{
Title: strings.TrimSpace(doc.Channel.Title),
Link: strings.TrimSpace(doc.Channel.Link),
Description: strings.TrimSpace(doc.Channel.Description),
LastBuildDate: parseRSSDate(doc.Channel.LastBuildDate),
Items: make([]RSSItem, 0, len(doc.Channel.Items)),
}
for _, item := range doc.Channel.Items {
feed.Items = append(feed.Items, RSSItem{
Title: strings.TrimSpace(item.Title),
Link: strings.TrimSpace(item.Link),
Description: strings.TrimSpace(item.Description),
PubDate: strings.TrimSpace(item.PubDate),
GUID: strings.TrimSpace(item.GUID),
})
}
return feed, nil
}
func parseRSSDate(value string) *time.Time {
value = strings.TrimSpace(value)
if value == "" {
return nil
}
for _, layout := range []string{time.RFC1123Z, time.RFC1123} {
t, err := time.Parse(layout, value)
if err == nil {
tt := t.UTC()
return &tt
}
}
return nil
}

View File

@@ -0,0 +1,43 @@
package spc
import (
"testing"
"time"
)
func TestParseRSSFeed(t *testing.T) {
const raw = `<?xml version="1.0"?>
<rss version="2.0">
<channel>
<title>SPC AC RSS</title>
<link>https://www.spc.noaa.gov/products/</link>
<description>SPC products</description>
<lastBuildDate>Thu, 11 Jun 2026 19:00:00 +0000</lastBuildDate>
<item>
<title>Day 1 Convective Outlook</title>
<link>https://www.spc.noaa.gov/products/outlook/day1otlk.html</link>
<description>Outlook text</description>
<pubDate>Thu, 11 Jun 2026 18:55:00 +0000</pubDate>
<guid>day1</guid>
</item>
</channel>
</rss>`
got, err := ParseRSSFeed(raw)
if err != nil {
t.Fatalf("ParseRSSFeed() error = %v", err)
}
if got.Title != "SPC AC RSS" {
t.Fatalf("Title = %q", got.Title)
}
wantBuild := time.Date(2026, 6, 11, 19, 0, 0, 0, time.UTC)
if got.LastBuildDate == nil || !got.LastBuildDate.Equal(wantBuild) {
t.Fatalf("LastBuildDate = %v, want %s", got.LastBuildDate, wantBuild)
}
if len(got.Items) != 1 {
t.Fatalf("Items length = %d, want 1", len(got.Items))
}
if got.Items[0].GUID != "day1" {
t.Fatalf("Item GUID = %q", got.Items[0].GUID)
}
}

View File

@@ -0,0 +1,29 @@
{
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"properties": {
"VALID_ISO": "2026-06-11T13:00:00Z",
"EXPIRE_ISO": "2026-06-12T12:00:00Z",
"ISSUE_ISO": "2026-06-11T12:34:56Z",
"FORECASTER": "SMITH",
"LABEL": "SLGT",
"LABEL2": "Slight Risk",
"DN": 3
},
"geometry": {
"type": "Polygon",
"coordinates": [
[
[-91.0, 38.0],
[-90.0, 38.0],
[-90.0, 39.0],
[-91.0, 39.0],
[-91.0, 38.0]
]
]
}
}
]
}

View File

@@ -0,0 +1,19 @@
<!doctype html>
<html>
<head><title>Day 1 Convective Outlook</title></head>
<body>
<pre>
<script>window.bad = "<b>ignore me</b>";</script>
Day 1 Convective Outlook
NWS Storm Prediction Center Norman OK
Updated: 2026-06-11T12:45:00Z
...SUMMARY...
Severe thunderstorms are possible across parts of the central Plains
and mid Mississippi Valley this afternoon and evening.
...DISCUSSION...
The primary threats will be damaging wind and large hail.
</pre>
</body>
</html>

View File

@@ -0,0 +1,15 @@
<!doctype html>
<html>
<body>
<pre>
Day 2 Convective Outlook CORR 1
NWS Storm Prediction Center Norman OK
...SUMMARY...
Scattered severe thunderstorms remain possible across the southern Plains.
...DISCUSSION...
Corrected outlook text remains otherwise unchanged.
</pre>
</body>
</html>

View File

@@ -0,0 +1,29 @@
{
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"properties": {
"VALID_ISO": "2026-06-12T12:00:00Z",
"EXPIRE_ISO": "2026-06-13T12:00:00Z",
"ISSUE_ISO": "2026-06-11T17:30:00Z",
"FORECASTER": "DOE",
"LABEL": "5",
"LABEL2": "5% Tornado",
"DN": "5"
},
"geometry": {
"type": "Polygon",
"coordinates": [
[
[-100.0, 35.0],
[-98.0, 35.0],
[-98.0, 37.0],
[-100.0, 37.0],
[-100.0, 35.0]
]
]
}
}
]
}

View File

@@ -0,0 +1,16 @@
<!doctype html>
<html>
<body>
<pre>
Day 3 Convective Outlook
NWS Storm Prediction Center Norman OK
Updated: 2026-06-11T20:00:00Z
...SUMMARY...
A corridor of strong to severe storms may develop near a frontal zone.
...DISCUSSION...
Confidence remains moderate for organized storms.
</pre>
</body>
</html>

View File

@@ -0,0 +1,31 @@
{
"type": "FeatureCollection",
"features": [
{
"type": "Feature",
"properties": {
"VALID_ISO": "2026-06-13T12:00:00Z",
"EXPIRE_ISO": "2026-06-14T12:00:00Z",
"ISSUE_ISO": "2026-06-11T19:45:00Z",
"FORECASTER": "LEE",
"LABEL": "15",
"LABEL2": "15% Wind",
"DN": 15
},
"geometry": {
"type": "MultiPolygon",
"coordinates": [
[
[
[-103.0, 34.0],
[-101.0, 34.0],
[-101.0, 36.0],
[-103.0, 36.0],
[-103.0, 34.0]
]
]
]
}
}
]
}

View File

@@ -0,0 +1,24 @@
package spc
import (
"strings"
"time"
)
// ParseISOTimestamp parses SPC ISO timestamps from GeoJSON properties.
func ParseISOTimestamp(value string) (time.Time, error) {
return time.Parse(time.RFC3339, strings.TrimSpace(value))
}
func parseOptionalISOTimestamp(value string) *time.Time {
value = strings.TrimSpace(value)
if value == "" {
return nil
}
t, err := ParseISOTimestamp(value)
if err != nil {
return nil
}
tt := t.UTC()
return &tt
}

View File

@@ -11,6 +11,7 @@
// - weather.forecast_discussion.v1 -> model.WeatherForecastDiscussion
// - weather.weather_story.v1 -> model.WeatherStoryRun
// - weather.alert.v1 -> model.WeatherAlertRun
// - weather.outlook.v1 -> model.WeatherOutlookRun
//
// Parent/child relationships:
// - observations.event_id -> observation_present_weather.event_id
@@ -19,9 +20,10 @@
// - 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)
// - outlook_runs.event_id -> outlooks.run_event_id
//
// Dedupe and retention behavior:
// - Parent primary keys (event_id): observations, forecasts, alert_runs.
// - Parent primary keys (event_id): observations, forecasts, alert_runs, outlook_runs.
// - Child primary keys use positional indexes to preserve payload order.
// - Prune columns:
// - observations.observed_at
@@ -35,11 +37,13 @@
// - alert_runs.as_of
// - alerts.as_of
// - alert_references.as_of
// - outlook_runs.as_of
// - outlooks.as_of
//
// Envelope field mapping (shared parent columns)
//
// These columns exist on parent tables such as observations, forecasts,
// forecast_discussions, weather_story_runs, and alert_runs:
// forecast_discussions, weather_story_runs, alert_runs, and outlook_runs:
// - event_id TEXT -> event.id
// - event_kind TEXT -> event.kind
// - event_source TEXT -> event.source
@@ -208,6 +212,46 @@
// - sender TEXT NULL -> payload.alerts[i].references[j].sender
// - sent TIMESTAMPTZ NULL -> payload.alerts[i].references[j].sent
//
// 10. outlook_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
// - location_id TEXT NULL -> payload.locationId
// - location_name TEXT NULL -> payload.locationName
// - latitude DOUBLE PRECISION NULL -> payload.latitude
// - longitude DOUBLE PRECISION NULL -> payload.longitude
// - as_of TIMESTAMPTZ -> payload.asOf
// - issued_at TIMESTAMPTZ NULL -> payload.issuedAt
// - outlook_count INTEGER -> len(payload.outlooks)
//
// 11. outlooks (PK: run_event_id, outlook_index)
//
// - run_event_id TEXT -> outlook_runs.event_id / payload.outlooks[i]
// - outlook_index INTEGER -> i (array position in payload.outlooks)
// - as_of TIMESTAMPTZ -> payload.asOf (copied from parent)
// - product TEXT -> payload.outlooks[i].product
// - day INTEGER -> payload.outlooks[i].day
// - outlook_type TEXT -> payload.outlooks[i].outlookType
// - label TEXT -> payload.outlooks[i].label
// - label_text TEXT NULL -> payload.outlooks[i].labelText
// - severity_rank INTEGER NULL -> payload.outlooks[i].severityRank
// - valid_from TIMESTAMPTZ -> payload.outlooks[i].validFrom
// - valid_to TIMESTAMPTZ -> payload.outlooks[i].validTo
// - issued_at TIMESTAMPTZ -> payload.outlooks[i].issuedAt
// - expires_at TIMESTAMPTZ -> payload.outlooks[i].expiresAt
// - forecaster TEXT NULL -> payload.outlooks[i].forecaster
// - headline TEXT NULL -> payload.outlooks[i].headline
// - summary TEXT NULL -> payload.outlooks[i].summary
// - discussion TEXT NULL -> payload.outlooks[i].discussion
// - source_url TEXT NULL -> payload.outlooks[i].sourceUrl
// - image_url TEXT NULL -> payload.outlooks[i].imageUrl
// - contains_location BOOLEAN -> payload.outlooks[i].containsLocation
// - geometry_json TEXT -> compact JSON payload.outlooks[i].geometry
//
// Reconstructing canonical JSON payloads
//
// - WeatherObservation:
@@ -226,4 +270,8 @@
// 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)
// ordered by reference_index to rebuild references per alert.
//
// - WeatherOutlookRun:
// read one row from outlook_runs, then join outlooks by run_event_id ordered
// by outlook_index to rebuild outlooks.
package postgres

View File

@@ -1,6 +1,7 @@
package postgres
import (
"bytes"
"context"
"encoding/json"
"fmt"
@@ -26,6 +27,8 @@ func mapPostgresEvent(_ context.Context, e fkevent.Event) ([]fksinks.PostgresWri
return mapWeatherStoryEvent(e)
case standards.SchemaWeatherAlertV1:
return mapAlertEvent(e)
case standards.SchemaWeatherOutlookV1:
return mapOutlookEvent(e)
default:
return nil, nil
}
@@ -356,6 +359,107 @@ func mapAlertEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
return writes, nil
}
func mapOutlookEvent(e fkevent.Event) ([]fksinks.PostgresWrite, error) {
run, err := decodePayload[model.WeatherOutlookRun](e.Payload)
if err != nil {
return nil, fmt.Errorf("decode outlook payload: %w", err)
}
if run.AsOf.IsZero() {
return nil, fmt.Errorf("decode outlook payload: asOf is required")
}
asOf := run.AsOf.UTC()
writes := make([]fksinks.PostgresWrite, 0, 1+len(run.Outlooks))
writes = append(writes, fksinks.PostgresWrite{
Table: tableOutlookRuns,
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),
"location_id": nullableString(run.LocationID),
"location_name": nullableString(run.LocationName),
"latitude": nullableFloat64(run.Latitude),
"longitude": nullableFloat64(run.Longitude),
"as_of": asOf,
"issued_at": nullableTime(run.IssuedAt),
"outlook_count": len(run.Outlooks),
},
})
for i, outlook := range run.Outlooks {
if err := validateOutlook(outlook, i); err != nil {
return nil, err
}
geometryJSON, err := requiredCompactJSONText(outlook.Geometry)
if err != nil {
return nil, fmt.Errorf("decode outlook payload: outlooks[%d].geometry: %w", i, err)
}
writes = append(writes, fksinks.PostgresWrite{
Table: tableOutlooks,
Values: map[string]any{
"run_event_id": e.ID,
"outlook_index": i,
"as_of": asOf,
"product": outlook.Product,
"day": outlook.Day,
"outlook_type": outlook.OutlookType,
"label": outlook.Label,
"label_text": nullableString(outlook.LabelText),
"severity_rank": nullableInt(outlook.SeverityRank),
"valid_from": outlook.ValidFrom.UTC(),
"valid_to": outlook.ValidTo.UTC(),
"issued_at": outlook.IssuedAt.UTC(),
"expires_at": outlook.ExpiresAt.UTC(),
"forecaster": nullableString(outlook.Forecaster),
"headline": nullableString(outlook.Headline),
"summary": nullableString(outlook.Summary),
"discussion": nullableString(outlook.Discussion),
"source_url": nullableString(outlook.SourceURL),
"image_url": nullableString(outlook.ImageURL),
"contains_location": outlook.ContainsLocation,
"geometry_json": geometryJSON,
},
})
}
return writes, nil
}
func validateOutlook(outlook model.WeatherOutlook, index int) error {
if strings.TrimSpace(outlook.ID) == "" {
return fmt.Errorf("decode outlook payload: outlooks[%d].id is required", index)
}
if strings.TrimSpace(outlook.Provider) == "" {
return fmt.Errorf("decode outlook payload: outlooks[%d].provider is required", index)
}
if strings.TrimSpace(outlook.Product) == "" {
return fmt.Errorf("decode outlook payload: outlooks[%d].product is required", index)
}
if outlook.Day == 0 {
return fmt.Errorf("decode outlook payload: outlooks[%d].day is required", index)
}
if strings.TrimSpace(outlook.OutlookType) == "" {
return fmt.Errorf("decode outlook payload: outlooks[%d].outlookType is required", index)
}
if strings.TrimSpace(outlook.Label) == "" {
return fmt.Errorf("decode outlook payload: outlooks[%d].label is required", index)
}
if outlook.ValidFrom.IsZero() || outlook.ValidTo.IsZero() {
return fmt.Errorf("decode outlook payload: outlooks[%d] validFrom/validTo are required", index)
}
if outlook.IssuedAt.IsZero() || outlook.ExpiresAt.IsZero() {
return fmt.Errorf("decode outlook payload: outlooks[%d] issuedAt/expiresAt are required", index)
}
if len(outlook.Geometry) == 0 {
return fmt.Errorf("decode outlook payload: outlooks[%d].geometry is required", index)
}
return nil
}
func decodePayload[T any](payload any) (T, error) {
var out T
if payload == nil {
@@ -418,6 +522,13 @@ func nullableBool(v *bool) any {
return *v
}
func nullableInt(v *int) any {
if v == nil {
return nil
}
return *v
}
func nullableTime(v *time.Time) any {
if v == nil || v.IsZero() {
return nil
@@ -445,3 +556,19 @@ func compactJSONText(v any) (any, error) {
}
return string(b), nil
}
func requiredCompactJSONText(v any) (string, error) {
compact, err := compactJSONText(v)
if err != nil {
return "", err
}
s, ok := compact.(string)
if !ok || strings.TrimSpace(s) == "" || strings.TrimSpace(s) == "null" {
return "", fmt.Errorf("is required")
}
var buf bytes.Buffer
if err := json.Compact(&buf, []byte(s)); err != nil {
return "", err
}
return buf.String(), nil
}

View File

@@ -239,6 +239,149 @@ func TestMapPostgresEventWeatherStoryStructPayload(t *testing.T) {
assertAllWritesIncludeAllColumns(t, writes)
}
func TestMapPostgresEventOutlookStructPayload(t *testing.T) {
lat := 38.6239
lon := -90.3571
issuedAt := time.Date(2026, 6, 11, 19, 45, 0, 0, time.FixedZone("UTC-5", -5*60*60))
severity := 3
run := model.WeatherOutlookRun{
LocationID: "stl",
LocationName: "St. Louis, MO",
Latitude: &lat,
Longitude: &lon,
AsOf: time.Date(2026, 6, 12, 0, 45, 0, 0, time.UTC),
IssuedAt: &issuedAt,
Outlooks: []model.WeatherOutlook{
{
ID: "outlook-1",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "categorical",
Label: "SLGT",
LabelText: "Slight Risk",
SeverityRank: &severity,
ValidFrom: time.Date(2026, 6, 11, 13, 0, 0, 0, time.FixedZone("UTC-5", -5*60*60)),
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: issuedAt,
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
Forecaster: "SMITH",
Headline: "Day 1 Convective Outlook",
Summary: "Severe thunderstorms are possible.",
Discussion: "Full discussion text.",
SourceURL: "https://example.invalid/day1.geojson",
ContainsLocation: true,
Geometry: json.RawMessage(`{ "type" : "Polygon", "coordinates" : [ [ [ -91.0, 38.0 ], [ -90.0, 38.0 ], [ -90.0, 39.0 ], [ -91.0, 39.0 ], [ -91.0, 38.0 ] ] ] }`),
},
{
ID: "outlook-2",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "wind",
Label: "15",
ValidFrom: time.Date(2026, 6, 11, 13, 0, 0, 0, time.UTC),
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
ContainsLocation: false,
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-100,35],[-98,35],[-98,37],[-100,37],[-100,35]]]}`),
},
},
}
writes, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run))
if err != nil {
t.Fatalf("mapPostgresEvent() error = %v", err)
}
if len(writes) != 3 {
t.Fatalf("mapPostgresEvent() writes len = %d, want 3", len(writes))
}
if writes[0].Table != tableOutlookRuns {
t.Fatalf("writes[0].Table = %q, want %q", writes[0].Table, tableOutlookRuns)
}
if got := writes[0].Values["outlook_count"]; got != 2 {
t.Fatalf("outlook_runs outlook_count = %#v, want 2", got)
}
if got := writes[0].Values["issued_at"]; got != issuedAt.UTC() {
t.Fatalf("outlook_runs issued_at = %#v, want UTC %s", got, issuedAt.UTC())
}
if writes[1].Table != tableOutlooks || writes[2].Table != tableOutlooks {
t.Fatalf("outlook writes not in expected order")
}
if got := writes[1].Values["outlook_index"]; got != 0 {
t.Fatalf("first outlook index = %#v, want 0", got)
}
if got := writes[1].Values["valid_from"]; got != run.Outlooks[0].ValidFrom.UTC() {
t.Fatalf("first valid_from = %#v, want UTC %s", got, run.Outlooks[0].ValidFrom.UTC())
}
if got := writes[1].Values["geometry_json"]; got != `{"type":"Polygon","coordinates":[[[-91.0,38.0],[-90.0,38.0],[-90.0,39.0],[-91.0,39.0],[-91.0,38.0]]]}` {
t.Fatalf("first geometry_json = %#v", got)
}
if got := writes[2].Values["contains_location"]; got != false {
t.Fatalf("second contains_location = %#v, want false", got)
}
assertAllWritesIncludeAllColumns(t, writes)
}
func TestMapPostgresEventOutlookRejectsMissingAsOf(t *testing.T) {
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", model.WeatherOutlookRun{}))
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 TestMapPostgresEventOutlookRejectsMissingRequiredTimes(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Outlooks: []model.WeatherOutlook{{
ID: "outlook-1",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "categorical",
Label: "SLGT",
Geometry: json.RawMessage(`{"type":"Polygon","coordinates":[[[-91,38],[-90,38],[-90,39],[-91,39],[-91,38]]]}`),
}},
}
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run))
if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want missing time error")
}
if !strings.Contains(err.Error(), "outlooks[0] validFrom/validTo are required") {
t.Fatalf("error = %q, want outlook time context", err)
}
}
func TestMapPostgresEventOutlookRejectsEmptyGeometry(t *testing.T) {
run := model.WeatherOutlookRun{
AsOf: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
Outlooks: []model.WeatherOutlook{{
ID: "outlook-1",
Provider: "spc",
Product: "convective",
Day: 1,
OutlookType: "categorical",
Label: "SLGT",
ValidFrom: time.Date(2026, 6, 11, 13, 0, 0, 0, time.UTC),
ValidTo: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
IssuedAt: time.Date(2026, 6, 11, 19, 45, 0, 0, time.UTC),
ExpiresAt: time.Date(2026, 6, 12, 12, 0, 0, 0, time.UTC),
}},
}
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherOutlookV1, "outlook", run))
if err == nil {
t.Fatalf("mapPostgresEvent() error = nil, want geometry error")
}
if !strings.Contains(err.Error(), "outlooks[0].geometry is required") {
t.Fatalf("error = %q, want geometry context", err)
}
}
func TestMapPostgresEventWeatherStoryRejectsMissingAsOf(t *testing.T) {
_, err := mapPostgresEvent(context.Background(), testEvent(standards.SchemaWeatherStoryV1, "weather_story", model.WeatherStoryRun{}))
if err == nil {

View File

@@ -16,6 +16,8 @@ const (
tableAlertRuns = "alert_runs"
tableAlerts = "alerts"
tableAlertReferences = "alert_references"
tableOutlookRuns = "outlook_runs"
tableOutlooks = "outlooks"
)
// PostgresSchema returns weatherfeeder's Postgres schema definition.
@@ -297,6 +299,63 @@ func PostgresSchema() fksinks.PostgresSchema {
{Name: "idx_wf_alert_refs_sent", Columns: []string{"sent"}},
},
},
{
Name: tableOutlookRuns,
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: "location_id", Type: "TEXT", Nullable: true},
{Name: "location_name", Type: "TEXT", Nullable: true},
{Name: "latitude", Type: "DOUBLE PRECISION", Nullable: true},
{Name: "longitude", Type: "DOUBLE PRECISION", Nullable: true},
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: true},
{Name: "outlook_count", Type: "INTEGER", Nullable: false},
},
PrimaryKey: []string{"event_id"},
PruneColumn: "as_of",
Indexes: []fksinks.PostgresIndex{
{Name: "idx_wf_outlook_run_location_as_of", Columns: []string{"location_id", "as_of"}},
{Name: "idx_wf_outlook_run_as_of", Columns: []string{"as_of"}},
},
},
{
Name: tableOutlooks,
Columns: []fksinks.PostgresColumn{
{Name: "run_event_id", Type: "TEXT REFERENCES outlook_runs(event_id) ON DELETE CASCADE", Nullable: false},
{Name: "outlook_index", Type: "INTEGER", Nullable: false},
{Name: "as_of", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "product", Type: "TEXT", Nullable: false},
{Name: "day", Type: "INTEGER", Nullable: false},
{Name: "outlook_type", Type: "TEXT", Nullable: false},
{Name: "label", Type: "TEXT", Nullable: false},
{Name: "label_text", Type: "TEXT", Nullable: true},
{Name: "severity_rank", Type: "INTEGER", Nullable: true},
{Name: "valid_from", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "valid_to", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "issued_at", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "expires_at", Type: "TIMESTAMPTZ", Nullable: false},
{Name: "forecaster", Type: "TEXT", Nullable: true},
{Name: "headline", Type: "TEXT", Nullable: true},
{Name: "summary", Type: "TEXT", Nullable: true},
{Name: "discussion", Type: "TEXT", Nullable: true},
{Name: "source_url", Type: "TEXT", Nullable: true},
{Name: "image_url", Type: "TEXT", Nullable: true},
{Name: "contains_location", Type: "BOOLEAN", Nullable: false},
{Name: "geometry_json", Type: "TEXT", Nullable: false},
},
PrimaryKey: []string{"run_event_id", "outlook_index"},
PruneColumn: "as_of",
Indexes: []fksinks.PostgresIndex{
{Name: "idx_wf_outlooks_contains_valid", Columns: []string{"contains_location", "valid_from", "valid_to"}},
{Name: "idx_wf_outlooks_day_type_label", Columns: []string{"day", "outlook_type", "label"}},
{Name: "idx_wf_outlooks_valid", Columns: []string{"valid_from", "valid_to"}},
},
},
},
MapEvent: mapPostgresEvent,
}

View File

@@ -1,6 +1,9 @@
package postgres
import "testing"
import (
"strings"
"testing"
)
func TestWeatherPostgresSchemaShape(t *testing.T) {
s := PostgresSchema()
@@ -20,6 +23,8 @@ func TestWeatherPostgresSchemaShape(t *testing.T) {
tableAlertRuns: true,
tableAlerts: true,
tableAlertReferences: true,
tableOutlookRuns: true,
tableOutlooks: true,
}
if len(s.Tables) != len(wantTables) {
@@ -43,6 +48,29 @@ func TestWeatherPostgresSchemaShape(t *testing.T) {
}
}
func TestWeatherPostgresSchemaIncludesOutlookTables(t *testing.T) {
runColumns := columnsForTable(t, tableOutlookRuns)
for _, col := range []string{"event_id", "event_kind", "event_source", "event_schema", "event_emitted_at", "event_effective_at", "location_id", "location_name", "latitude", "longitude", "as_of", "issued_at", "outlook_count"} {
if !runColumns[col] {
t.Fatalf("%s missing %s column", tableOutlookRuns, col)
}
}
assertTablePrimaryKey(t, tableOutlookRuns, []string{"event_id"})
assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_location_as_of", []string{"location_id", "as_of"})
assertTableIndex(t, tableOutlookRuns, "idx_wf_outlook_run_as_of", []string{"as_of"})
outlookColumns := columnsForTable(t, tableOutlooks)
for _, col := range []string{"run_event_id", "outlook_index", "as_of", "product", "day", "outlook_type", "label", "label_text", "severity_rank", "valid_from", "valid_to", "issued_at", "expires_at", "forecaster", "headline", "summary", "discussion", "source_url", "image_url", "contains_location", "geometry_json"} {
if !outlookColumns[col] {
t.Fatalf("%s missing %s column", tableOutlooks, col)
}
}
assertTablePrimaryKey(t, tableOutlooks, []string{"run_event_id", "outlook_index"})
assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_contains_valid", []string{"contains_location", "valid_from", "valid_to"})
assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_day_type_label", []string{"day", "outlook_type", "label"})
assertTableIndex(t, tableOutlooks, "idx_wf_outlooks_valid", []string{"valid_from", "valid_to"})
}
func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) {
runColumns := columnsForTable(t, tableWeatherStoryRuns)
if !runColumns["as_of"] {
@@ -60,6 +88,39 @@ func TestWeatherPostgresSchemaIncludesWeatherStoryColumns(t *testing.T) {
}
}
func assertTablePrimaryKey(t *testing.T, table string, want []string) {
t.Helper()
for _, tbl := range PostgresSchema().Tables {
if tbl.Name != table {
continue
}
if strings.Join(tbl.PrimaryKey, ",") != strings.Join(want, ",") {
t.Fatalf("%s primary key = %#v, want %#v", table, tbl.PrimaryKey, want)
}
return
}
t.Fatalf("missing table %q", table)
}
func assertTableIndex(t *testing.T, table string, name string, want []string) {
t.Helper()
for _, tbl := range PostgresSchema().Tables {
if tbl.Name != table {
continue
}
for _, idx := range tbl.Indexes {
if idx.Name == name {
if strings.Join(idx.Columns, ",") != strings.Join(want, ",") {
t.Fatalf("%s index %s columns = %#v, want %#v", table, name, idx.Columns, want)
}
return
}
}
t.Fatalf("%s missing index %s", table, name)
}
t.Fatalf("missing table %q", table)
}
func columnsForTable(t *testing.T, table string) map[string]bool {
t.Helper()

View File

@@ -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.

View File

@@ -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,
}
}

View File

@@ -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})
}

View File

@@ -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 = `<?xml version="1.0"?>
<rss version="2.0">
<channel>
<title>SPC AC RSS</title>
<lastBuildDate>Thu, 11 Jun 2026 21:00:00 +0000</lastBuildDate>
</channel>
</rss>`

42
model/outlook.go Normal file
View File

@@ -0,0 +1,42 @@
package model
import (
"encoding/json"
"time"
)
// WeatherOutlookRun is a snapshot of convective outlook polygons for a
// configured location as-of a provider issue time.
type WeatherOutlookRun struct {
LocationID string `json:"locationId,omitempty"`
LocationName string `json:"locationName,omitempty"`
Latitude *float64 `json:"latitude,omitempty"`
Longitude *float64 `json:"longitude,omitempty"`
AsOf time.Time `json:"asOf"`
IssuedAt *time.Time `json:"issuedAt,omitempty"`
Outlooks []WeatherOutlook `json:"outlooks"`
}
// WeatherOutlook is a canonical representation of one outlook polygon.
type WeatherOutlook struct {
ID string `json:"id"`
Provider string `json:"provider"`
Product string `json:"product"`
Day int `json:"day"`
OutlookType string `json:"outlookType"`
Label string `json:"label"`
LabelText string `json:"labelText,omitempty"`
SeverityRank *int `json:"severityRank,omitempty"`
ValidFrom time.Time `json:"validFrom"`
ValidTo time.Time `json:"validTo"`
IssuedAt time.Time `json:"issuedAt"`
ExpiresAt time.Time `json:"expiresAt"`
Forecaster string `json:"forecaster,omitempty"`
Headline string `json:"headline,omitempty"`
Summary string `json:"summary,omitempty"`
Discussion string `json:"discussion,omitempty"`
SourceURL string `json:"sourceUrl,omitempty"`
ImageURL string `json:"imageUrl,omitempty"`
ContainsLocation bool `json:"containsLocation"`
Geometry json.RawMessage `json:"geometry"`
}

View File

@@ -24,10 +24,13 @@ 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"
SchemaWeatherForecastDiscussionV1 = "weather.forecast_discussion.v1"
SchemaWeatherStoryV1 = "weather.weather_story.v1"
SchemaWeatherAlertV1 = "weather.alert.v1"
SchemaWeatherOutlookV1 = "weather.outlook.v1"
)