Add SPC outlook normalization
This commit is contained in:
314
internal/normalizers/spc/convective_outlook.go
Normal file
314
internal/normalizers/spc/convective_outlook.go
Normal 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
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user