gsn/predictor
1
Fork 0
Code Issues Pull requests 1 Projects Releases Activity

Decart Interp #9

Open
a.antonov wants to merge 2 commits from decart into main
pull from: decart
merge into: gsn:main
Conversation 0 Commits 2 Files changed 32 +1197 -137
20 changed files with 1197 additions and 137 deletions
Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 84d1664b29 - Show all commits

1
.gitignore vendored Normal file
View file

@ -0,0 +1 @@
bin

7
api/rest/predictor.swagger.yml
View file

@ -375,6 +375,13 @@ components:
profile, hand off to `fallback_index`, or clip to the boundary.
properties:
launch: { $ref: "#/components/schemas/Launch" }
dataset:
type: string
format: date-time
description: |
Forecast run to predict from, given as its epoch. Defaults to the
active dataset. The named run must be stored; a request for one that
is not is rejected rather than answered from a different run.
direction:
type: string
enum: [forward, reverse]

220
cmd/compare-tawhiri/main.go
View file

@ -24,6 +24,7 @@ import (
"net/http"
"net/url"
"os"
"strings"
"text/tabwriter"
"time"
)
@ -75,15 +76,22 @@ func main() {
var worst float64
compared := 0
for _, s := range sites {
p := params{lat: s.lat, lng: s.lng, alt: *alt, launch: launchTime,
// The site suite compares trajectories, not notations, so fold western
// longitudes into [0, 360) here: that is the only range Tawhiri accepts.
// The longitude suite below sends raw values precisely to test that edge.
lng := s.lng
if lng < 0 {
lng += 360
}
p := params{lat: s.lat, lng: lng, alt: *alt, launch: launchTime,
ascent: *ascent, burst: *burst, descent: *descent}
ours, err := predict(*server+"/api/v1/prediction", p, "")
ours, _, err := predict(*server+"/api/v1/prediction", p, "")
if err != nil {
fmt.Fprintf(tw, "%s\tlocal error: %v\n", s.name, err)
continue
}
theirs, err := predict(*tawhiri, p, datasetParam)
theirs, _, err := predict(*tawhiri, p, datasetParam)
if err != nil {
fmt.Fprintf(tw, "%s\ttawhiri error: %v\n", s.name, err)
continue
@ -106,10 +114,17 @@ func main() {
}
tw.Flush()
breaks := comparePairs(*server, *tawhiri, longitudeParams(launchTime, *alt, *ascent, *burst, *descent), datasetParam)
breaks += compareLongitudes(*server, *tawhiri, longitudeParams(launchTime, *alt, *ascent, *burst, *descent), datasetParam)
if compared == 0 {
fmt.Println("\nVERDICT: NO COMPARISONS (every site errored — see rows above)")
os.Exit(1)
}
if breaks > 0 {
fmt.Printf("\nVERDICT: INCOMPATIBLE (%d longitude(s) Tawhiri accepts and we reject)\n", breaks)
os.Exit(3)
}
fmt.Printf("\ncompared %d/%d sites; worst landing distance: %.2f km\n", compared, len(sites), worst/1000)
switch {
case worst < 1000:
@ -153,17 +168,20 @@ type result struct {
dataset string
}
func predict(endpoint string, p params, dataset string) (result, error) {
// Tawhiri requires longitude in [0, 360); normalize so both endpoints get
// the same request. Returned trajectory longitudes are [-180, 180] on both
// sides, so the comparison stays consistent.
lng := p.lng
if lng < 0 {
lng += 360
}
// predict sends p verbatim and returns the parsed result plus the HTTP status.
//
// Nothing is normalised here on purpose. This function used to fold negative
// longitudes into [0, 360) before sending, which made every longitude notation
// look identical to both endpoints — so the tool could not see a disagreement
// about notation even in principle, and ours drifted to [-180, 360) with an
// asymmetry nobody had chosen. Callers that want a fold do it themselves.
//
// The status is returned separately from the error so a caller can tell a
// deliberate rejection from a transport failure.
func predict(endpoint string, p params, dataset string) (result, int, error) {
q := url.Values{}
q.Set("launch_latitude", fmt.Sprintf("%.4f", p.lat))
q.Set("launch_longitude", fmt.Sprintf("%.4f", lng))
q.Set("launch_longitude", fmt.Sprintf("%.4f", p.lng))
q.Set("launch_altitude", fmt.Sprintf("%.0f", p.alt))
q.Set("launch_datetime", p.launch.Format(time.RFC3339))
q.Set("ascent_rate", fmt.Sprintf("%.2f", p.ascent))
@ -191,10 +209,10 @@ func predict(endpoint string, p params, dataset string) (result, error) {
break
}
if lastErr != nil {
return result{}, lastErr
return result{}, 0, lastErr
}
if status != 200 {
return result{}, fmt.Errorf("HTTP %d: %s", status, truncate(string(body), 160))
return result{}, status, fmt.Errorf("HTTP %d: %s", status, truncate(string(body), 160))
}
var doc struct {
@ -211,7 +229,7 @@ func predict(endpoint string, p params, dataset string) (result, error) {
} `json:"request"`
}
if err := json.Unmarshal(body, &doc); err != nil {
return result{}, err
return result{}, status, err
}
var r result
@ -230,7 +248,7 @@ func predict(endpoint string, p params, dataset string) (result, error) {
r.landLat, r.landLng, r.landAlt = last.Latitude, last.Longitude, last.Altitude
}
}
return r, nil
return r, status, nil
}
type readinessResp struct {
@ -274,3 +292,173 @@ func truncate(s string, n int) string {
}
return s[:n] + "…"
}
// longitudeCases are the notations a client can put in launch_longitude, sent
// verbatim to both endpoints.
//
// Longitude is on a circle, so every one of these names a real meridian and no
// value here is geometrically wrong; what differs is which notations each service
// agrees to read. Tawhiri implements a half-open [0, 360) — one clean fundamental
// domain, every meridian spelled exactly once. We accept a superset deliberately,
// because the frontend holds longitudes in [-180, 180] and a western launch
// therefore arrives negative.
//
// The pairs below name the same meridian in the two notations, so an accepted
// pair must produce the same trajectory: (-51.7, 308.3), (-180, 180), (0, 360),
// (-90, 270). That is what makes this a comparison rather than a status table.
func longitudeCases() []float64 {
return []float64{
0, 180, 270, 308.3, 359.999, // inside Tawhiri's domain
-0.0001, -51.7, -90, -180, -200, // signed convention: our extension
360, // 0 spelled redundantly; Tawhiri refuses it
5170, // a plain typo, refused by both
}
}
func longitudeParams(launch time.Time, alt, ascent, burst, descent float64) params {
// 64.1 N is Nuuk's latitude: high enough to be a realistic Arctic site, far
// enough from the pole that the comparison is about longitude alone.
return params{lat: 64.1, alt: alt, launch: launch, ascent: ascent, burst: burst, descent: descent}
}
// compareLongitudes reports how the two services read each notation and returns
// the number of compatibility breaks — longitudes Tawhiri accepts and we refuse.
// Those are the only rows that are a defect: refusing what a Tawhiri client is
// entitled to send is what a drop-in replacement must never do. The reverse, us
// accepting what Tawhiri refuses, is the extension the product depends on.
func compareLongitudes(server, tawhiri string, base params, datasetParam string) int {
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
fmt.Fprintln(tw, "\nlongitude\tours\ttawhiri\tlanding Δ\tverdict")
fmt.Fprintln(tw, "---------\t----\t-------\t---------\t-------")
breaks := 0
for _, lng := range longitudeCases() {
p := base
p.lng = lng
ours, oursStatus, oursErr := predict(server+"/api/v1/prediction", p, "")
theirs, theirsStatus, theirsErr := predict(tawhiri, p, datasetParam)
code := func(status int, err error) string {
if err == nil {
return "200"
}
if status == 0 {
return "transport"
}
return fmt.Sprintf("%d", status)
}
delta := ""
verdict := ""
switch {
case oursErr == nil && theirsErr == nil:
d := haversine(ours.landLat, ours.landLng, theirs.landLat, theirs.landLng)
delta = fmt.Sprintf("%.2f km", d/1000)
verdict = "agree"
if d > 50000 {
verdict = "ACCEPTED BY BOTH, TRAJECTORIES DIVERGE"
}
// v1 must publish the convention Tawhiri publishes, not merely the same
// place in a different notation. It used to answer -51.2115 where
// Tawhiri answers 308.7884; haversine cannot see that, because a 360
// degree difference vanishes from sin(dlam/2).
if (ours.landLng < 0) != (theirs.landLng < 0) {
verdict = fmt.Sprintf("OUTPUT FORMAT DIFFERS (ours %.4f, theirs %.4f)", ours.landLng, theirs.landLng)
breaks++
}
case !aboutLongitude(oursErr) || !aboutLongitude(theirsErr):
// A rejection for some other reason says nothing about notation. Before
// this check, an upstream "No matching dataset found" — which happens as
// soon as sondehub rotates past our GFS run — was reported as "Tawhiri
// refuses this longitude", i.e. the tool inventing compatibility news.
verdict = "inconclusive: " + firstReason(oursErr, theirsErr)
case oursErr != nil && theirsErr != nil:
verdict = "agree (both refuse)"
case oursErr != nil:
verdict = "BREAK: Tawhiri accepts, we refuse"
breaks++
default:
verdict = "extension: we accept, Tawhiri refuses"
}
fmt.Fprintf(tw, "%g\t%s\t%s\t%s\t%s\n",
lng, code(oursStatus, oursErr), code(theirsStatus, theirsErr), delta, verdict)
}
tw.Flush()
return breaks
}
// aboutLongitude reports whether err is a rejection of the longitude itself
// rather than of something else in the request. A nil error passes: a successful
// call is always informative.
func aboutLongitude(err error) bool {
if err == nil {
return true
}
return strings.Contains(strings.ToLower(err.Error()), "longitude")
}
// firstReason summarises whichever side failed for an unrelated reason.
func firstReason(errs ...error) string {
for _, err := range errs {
if err != nil && !aboutLongitude(err) {
return truncate(err.Error(), 90)
}
}
return "unknown"
}
// longitudePairs are two notations naming one meridian: signed, then unsigned.
//
// Tawhiri reads only the unsigned one, so this is the only way the signed form —
// the extension the product actually depends on — can be checked against the
// reference at all. Our answer for the signed value must match Tawhiri's answer
// for its unsigned twin. Without this the "extension" rows in the table above are
// merely unrefuted, not verified.
//
// (0, 360) is absent on purpose: Tawhiri refuses 360, so that pair has no
// reference side.
func longitudePairs() [][2]float64 {
return [][2]float64{{-51.7, 308.3}, {-180, 180}, {-90, 270}, {-0.0001, 359.9999}}
}
// comparePairs checks each signed notation we accept against Tawhiri's answer for
// the same meridian written the way Tawhiri accepts it. Returns the number of
// pairs that disagree by more than the site suite's own tolerance.
func comparePairs(server, tawhiri string, base params, datasetParam string) int {
tw := tabwriter.NewWriter(os.Stdout, 0, 0, 2, ' ', 0)
fmt.Fprintln(tw, "\nours (signed)\tvs tawhiri (unsigned)\tlanding Δ\tverdict")
fmt.Fprintln(tw, "-------------\t---------------------\t---------\t-------")
bad := 0
for _, pair := range longitudePairs() {
signed, unsigned := pair[0], pair[1]
p := base
p.lng = signed
ours, _, oursErr := predict(server+"/api/v1/prediction", p, "")
q := base
q.lng = unsigned
theirs, _, theirsErr := predict(tawhiri, q, datasetParam)
switch {
case oursErr != nil:
fmt.Fprintf(tw, "%g\t%g\t\tours refused: %v\n", signed, unsigned, oursErr)
bad++
case theirsErr != nil:
fmt.Fprintf(tw, "%g\t%g\t\ttawhiri refused the unsigned twin: %v\n", signed, unsigned, theirsErr)
bad++
default:
d := haversine(ours.landLat, ours.landLng, theirs.landLat, theirs.landLng)
verdict := "agree — the extension reads the same meridian"
if d > 50000 {
verdict = "DIVERGENT — signed notation is not the same meridian"
bad++
}
fmt.Fprintf(tw, "%g\t%g\t%.2f km\t%s\n", signed, unsigned, d/1000, verdict)
}
}
tw.Flush()
return bad
}

62
internal/api/mapping.go
View file

@ -2,22 +2,72 @@ package api
import (
"fmt"
"math"
"net/http"
"time"
"predictor-refactored/internal/api/async"
"predictor-refactored/internal/engine"
"predictor-refactored/internal/numerics"
apirest "predictor-refactored/pkg/rest"
)
// normalizeLng folds a longitude into [0, 360) for internal use.
func normalizeLng(lng float64) float64 {
if lng < 0 {
return lng + 360
// longitudeLimit bounds what either API version accepts, in degrees.
//
// Symmetric on purpose. Upstream Tawhiri implements a half-open [0, 360) — one
// fundamental domain, every meridian spelled exactly once (measured). We accept a
// superset because the frontend holds longitudes in [-180, 180], so a western
// launch arrives negative, and we tolerate one full turn either way. Symmetry is
// the whole point: the previous bound, [-180, 360), had its two ends justified by
// unrelated things — the lower end was the signed convention's edge, the upper end
// was the domain on which the old single-fold normalizeLng happened to be correct.
//
// Beyond one turn a value is a typo rather than a notation. 5170 is arithmetically
// a perfectly good way to write 50 E, and upstream refuses it too.
const longitudeLimit = 360.0
// validateLat refuses latitudes outside [-90, 90].
//
// Shared by both API versions. Unlike longitude this is a real geometric bound:
// the poles are the ends of the axis, so no notation makes 100 meaningful. v2
// checked it and v1 did not.
func validateLat(lat float64) error {
if math.IsNaN(lat) || lat < -90 || lat > 90 {
return apiError(http.StatusBadRequest, fmt.Sprintf("launch latitude must be in [-90, 90], got %g", lat))
}
return lng
return nil
}
// signedLng converts an internal [0, 360) longitude back to [-180, 180).
// validateLng refuses longitudes outside [-longitudeLimit, longitudeLimit].
//
// Shared by both API versions. v1 used to apply no check at all while v2 enforced
// [-180, 360), so the same launch could be accepted by one and refused by the
// other — a difference in behaviour where only format may differ.
func validateLng(lng float64) error {
if math.IsNaN(lng) || lng < -longitudeLimit || lng > longitudeLimit {
return apiError(http.StatusBadRequest,
fmt.Sprintf("launch longitude must be in [-%g, %g], got %g", longitudeLimit, longitudeLimit, lng))
}
return nil
}
// normalizeLng folds a longitude into the [0, 360) the wind grid is indexed on.
//
// A true modulo, via the same helper the integrator applies on every step
// (numerics.PyMod in toGeo), so the request path and the engine agree by
// construction rather than by coincidence. It used to be a single
// `if lng < 0 { lng += 360 }`, correct only on [-360, 360): -400 became -40,
// reached the grid, and came back as "lng=-40 out of range" — a complaint about
// wind data for what was an input problem.
func normalizeLng(lng float64) float64 {
return numerics.PyMod(lng, 360)
}
// signedLng converts an internal [0, 360) longitude to [-180, 180).
//
// This is v2's published format. v1 publishes the internal [0, 360) unchanged,
// because that is the convention upstream Tawhiri publishes and v1 exists to be a
// drop-in for it.
func signedLng(lng float64) float64 {
if lng > 180 {
return lng - 360

88
internal/api/mapping_test.go Normal file
View file

@ -0,0 +1,88 @@
package api
import (
"math"
"testing"
)
// Longitude handling is shared by both API versions on purpose: the two may
// differ in format, names and features, never in what the maths does. v1 used to
// apply no range check at all while v2 enforced [-180, 360), so the same launch
// could be accepted by one and refused by the other.
func TestValidateLngAcceptsBothConventionsAndOneTurnEitherWay(t *testing.T) {
// Upstream Tawhiri implements a half-open [0, 360) — measured: 0, 180, 270 and
// 308.3 accepted, everything negative and 360 refused. We accept a superset
// because the frontend holds longitudes in [-180, 180], so a western launch
// arrives negative. The bound is symmetric so there is nothing to explain
// about one end that does not hold for the other.
for _, lng := range []float64{0, 0.1, 180, 270, 308.3, 359.999, 360, -0.0001, -51.7, -180, -200, -360} {
if err := validateLng(lng); err != nil {
t.Errorf("validateLng(%g) = %v, want accepted", lng, err)
}
}
}
func TestValidateLngRefusesTypos(t *testing.T) {
// Arithmetically every one of these names a real meridian — 5170 is 50 E. They
// are refused because they are far likelier to be a unit slip or a swapped
// field than an intended notation, and upstream refuses them too.
for _, lng := range []float64{360.0001, -360.0001, 400, -400, 5170, math.Inf(1), math.Inf(-1)} {
if err := validateLng(lng); err == nil {
t.Errorf("validateLng(%g) was accepted, want refused", lng)
}
}
}
func TestNormalizeLngFoldsAnyInputIntoTheGridDomain(t *testing.T) {
// The wind grid is indexed from LonStart = 0 over 360 degrees, so every value
// handed to it must land in [0, 360).
//
// This was a single `if lng < 0 { lng += 360 }`, correct only on [-360, 360):
// -400 became -40, reached the grid, and came back as "lng=-40 out of range" —
// a complaint about wind data for what was an input problem.
cases := map[float64]float64{
0: 0, 0.5: 0.5, 180: 180, 359.5: 359.5,
360: 0, 720: 0, -360: 0,
-0.5: 359.5, -51.7: 308.3, -180: 180, -200: 160, -400: 320, 5170: 130,
}
for in, want := range cases {
if got := normalizeLng(in); math.Abs(got-want) > 1e-9 {
t.Errorf("normalizeLng(%g) = %g, want %g", in, got, want)
}
if got := normalizeLng(in); got < 0 || got >= 360 {
t.Errorf("normalizeLng(%g) = %g, outside [0, 360)", in, got)
}
}
}
func TestSignedLngIsTheV2Format(t *testing.T) {
// v2 publishes [-180, 180): its own convenient format. v1 publishes the
// internal [0, 360) unchanged, because that is what upstream Tawhiri publishes
// and v1 exists to be a drop-in for it.
cases := map[float64]float64{0: 0, 90: 90, 180: 180, 180.5: -179.5, 308.3: -51.7, 359.5: -0.5}
for in, want := range cases {
if got := signedLng(in); math.Abs(got-want) > 1e-9 {
t.Errorf("signedLng(%g) = %g, want %g", in, got, want)
}
}
}
func TestValidateLatIsSharedByBothVersions(t *testing.T) {
// Unlike longitude, latitude is a genuine geometric bound: +-90 are the poles
// and there is nothing beyond them, so no notation makes 100 meaningful.
//
// v2 checked this and v1 did not, which is a difference in behaviour rather
// than format. On v1 an out-of-range latitude reached the wind grid and came
// back as "lat=... out of range" — a complaint about data for an input problem.
for _, lat := range []float64{-90, -89.999, 0, 52.2, 89.999, 90} {
if err := validateLat(lat); err != nil {
t.Errorf("validateLat(%g) = %v, want accepted", lat, err)
}
}
for _, lat := range []float64{-90.0001, 90.0001, 100, -100, math.NaN(), math.Inf(1)} {
if err := validateLat(lat); err == nil {
t.Errorf("validateLat(%g) was accepted, want refused", lat)
}
}
}

86
internal/api/prediction.go
View file

@ -2,6 +2,7 @@ package api
import (
"context"
"fmt"
"net/http"
"time"
@ -23,9 +24,45 @@ func (h *Handler) ReadinessCheck(_ context.Context) (*apirest.ReadinessResponse,
return resp, nil
}
// fieldFor resolves the dataset a request asked for, or the active one.
//
// A request naming a dataset that is not stored is refused. The parameter used
// to be declared in the spec and never read, so a client asking for a specific
// forecast run — an archival date, a reproducible comparison — was silently
// served from whatever happened to be loaded instead.
func (h *Handler) fieldFor(ctx context.Context, want apirest.OptDateTime) (weather.WindField, error) {
if epoch, ok := want.Get(); ok {
field, err := h.mgr.FieldFor(ctx, epoch.UTC())
if err != nil {
return nil, apiError(http.StatusBadRequest, err.Error())
}
return field, nil
}
field := h.mgr.Active()
if field == nil {
return nil, apiError(http.StatusServiceUnavailable, "no dataset loaded, service is starting up")
}
return field, nil
}
// runFailure turns an integration failure into a 400.
//
// The sampler reporting an error means it had no data — most often a launch time
// past the dataset's horizon — so the trajectory is not a forecast. It used to be
// discarded, and the run continued with zero wind: the balloon took off and
// landed on the same spot, returned as a successful prediction.
func runFailure(events *engine.EventSink, field weather.WindField) error {
err := events.Err()
if err == nil {
return nil
}
return apiError(http.StatusBadRequest,
fmt.Sprintf("%s (dataset %s)", err, field.Epoch().UTC().Format(time.RFC3339)))
}
// PerformPredictionV2 implements POST /api/v2/prediction.
func (h *Handler) PerformPredictionV2(_ context.Context, req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) {
resp, err := h.runPredictionV2(req)
func (h *Handler) PerformPredictionV2(ctx context.Context, req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) {
resp, err := h.runPredictionV2(ctx, req)
if err == nil {
h.metrics.Prediction("v2", resp.CompletedAt.Sub(resp.StartedAt), nil)
}
@ -60,23 +97,22 @@ func (h *Handler) CancelPredictionJob(_ context.Context, params apirest.CancelPr
// runPredictionV2 is the synchronous prediction core, shared by the v2
// endpoint and the async worker pool.
func (h *Handler) runPredictionV2(req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) {
func (h *Handler) runPredictionV2(ctx context.Context, req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) {
// Validate the request shape before checking dataset availability, so a
// malformed request is a 400 regardless of startup state.
lat := req.Launch.Latitude
rawLng := req.Launch.Longitude
alt := req.Launch.Altitude.Or(0)
if lat < -90 || lat > 90 {
return nil, apiError(http.StatusBadRequest, "launch.latitude must be in [-90, 90]")
if err := validateLat(lat); err != nil {
return nil, err
}
if rawLng < -180 || rawLng >= 360 {
return nil, apiError(http.StatusBadRequest, "launch.longitude must be in [-180, 360)")
if err := validateLng(req.Launch.Longitude); err != nil {
return nil, err
}
lng := normalizeLng(rawLng)
lng := normalizeLng(req.Launch.Longitude)
field := h.mgr.Active()
if field == nil {
return nil, apiError(http.StatusServiceUnavailable, "no dataset loaded, service is starting up")
field, err := h.fieldFor(ctx, req.Dataset)
if err != nil {
return nil, err
}
events := engine.NewEventSink()
@ -90,6 +126,9 @@ func (h *Handler) runPredictionV2(req *apirest.PredictionV2Request) (*apirest.Pr
started := time.Now().UTC()
results := prof.Run(float64(req.Launch.Time.Unix()), engine.State{Lat: lat, Lng: lng, Altitude: alt}, events)
completed := time.Now().UTC()
if err := runFailure(events, field); err != nil {
return nil, err
}
resp := &apirest.PredictionV2Response{
Stages: make([]apirest.StageResult, 0, len(results)),
@ -105,10 +144,10 @@ func (h *Handler) runPredictionV2(req *apirest.PredictionV2Request) (*apirest.Pr
}
// PerformPrediction implements GET /api/v1/prediction (Tawhiri-compatible).
func (h *Handler) PerformPrediction(_ context.Context, params apirest.PerformPredictionParams) (*apirest.PredictionResponse, error) {
field := h.mgr.Active()
if field == nil {
return nil, apiError(http.StatusServiceUnavailable, "no dataset loaded, service is starting up")
func (h *Handler) PerformPrediction(ctx context.Context, params apirest.PerformPredictionParams) (*apirest.PredictionResponse, error) {
field, err := h.fieldFor(ctx, params.Dataset)
if err != nil {
return nil, err
}
profileKind := "standard_profile"
@ -118,6 +157,12 @@ func (h *Handler) PerformPrediction(_ context.Context, params apirest.PerformPre
ascentRate := params.AscentRate.Or(5)
descentRate := params.DescentRate.Or(5)
launchAlt := params.LaunchAltitude.Or(0)
if err := validateLat(params.LaunchLatitude); err != nil {
return nil, err
}
if err := validateLng(params.LaunchLongitude); err != nil {
return nil, err
}
lng := normalizeLng(params.LaunchLongitude)
launchTime := float64(params.LaunchDatetime.Unix())
@ -142,6 +187,9 @@ func (h *Handler) PerformPrediction(_ context.Context, params apirest.PerformPre
started := time.Now().UTC()
results := prof.Run(launchTime, engine.State{Lat: params.LaunchLatitude, Lng: lng, Altitude: launchAlt}, events)
completed := time.Now().UTC()
if err := runFailure(events, field); err != nil {
return nil, err
}
h.metrics.Prediction(profileKind, completed.Sub(started), nil)
resp := &apirest.PredictionResponse{
@ -231,7 +279,11 @@ func tawhiriItem(name string, r engine.Result) apirest.PredictionResponsePredict
traj = append(traj, apirest.TawhiriPoint{
Datetime: time.Unix(int64(t), 0).UTC(),
Latitude: p.Lat,
Longitude: signedLng(p.Lng),
// v1 publishes [0, 360), the convention upstream Tawhiri publishes —
// measured: it answers 308.7884 where we used to answer -51.2115 for
// the same meridian. v1 exists to be a drop-in, so it matches. v2 keeps
// the signed format (mapping.go).
Longitude: p.Lng,
Altitude: p.Altitude,
})
}

6
internal/api/transport.go
View file

@ -74,7 +74,11 @@ func New(port int, d Deps) (*Server, error) {
Workers: d.AsyncWorkers,
QueueSize: d.AsyncQueueSize,
ResultTTL: d.AsyncResultTTL,
}, h.runPredictionV2, d.Metrics, d.Log)
// The async queue outlives the HTTP request that enqueued the job, so a
// worker cannot inherit its context.
}, func(req *apirest.PredictionV2Request) (*apirest.PredictionV2Response, error) {
return h.runPredictionV2(context.Background(), req)
}, d.Metrics, d.Log)
ogenSrv, err := apirest.NewServer(h, apirest.WithMiddleware(middleware.OgenLogging(d.Log)))
if err != nil {

95
internal/datasets/manager.go
View file

@ -161,6 +161,57 @@ func (m *Manager) SelectFor(t time.Time, lat, lng float64) weather.WindField {
return nil
}
// FieldFor returns the global dataset whose epoch is epoch, loading it from
// storage on demand when it is stored but not currently active.
//
// Being stored is enough. A caller asking for a specific run — an archival
// forecast, a reproducible comparison — must not have to make it the service's
// active dataset first, and asking for it must not repoint the service: Load
// appends, and Active keeps returning the first global it finds.
//
// Errors when no stored dataset carries that epoch, so a request naming a
// dataset the service cannot serve is refused rather than quietly answered from
// a different one.
//
// ponytail: loaded datasets are never evicted, so a long-lived service asked for
// many distinct epochs accumulates one mmap and fd each. Add LRU eviction here
// if that becomes real; the files are mmap-backed, so the cost is address space
// and descriptors, not resident memory.
func (m *Manager) FieldFor(ctx context.Context, epoch time.Time) (weather.WindField, error) {
if f := m.activeFieldAt(epoch); f != nil {
return f, nil
}
stored, err := m.store.List()
if err != nil {
return nil, fmt.Errorf("list stored datasets: %w", err)
}
for _, id := range stored {
if !id.Subset.IsGlobal() || !id.Epoch.Equal(epoch) {
continue
}
if err := m.Load(ctx, id); err != nil {
return nil, fmt.Errorf("load %s: %w", id.Filename(), err)
}
if f := m.activeFieldAt(epoch); f != nil {
return f, nil
}
}
return nil, fmt.Errorf("dataset %s is not stored", epoch.UTC().Format(time.RFC3339))
}
// activeFieldAt returns the loaded global field with exactly this epoch, or nil.
func (m *Manager) activeFieldAt(epoch time.Time) weather.WindField {
m.activeMu.RLock()
defer m.activeMu.RUnlock()
for _, d := range m.active {
if d.ID.Subset.IsGlobal() && d.ID.Epoch.Equal(epoch) {
return d.Field
}
}
return nil
}
// LoadedDatasets returns snapshots of every currently-loaded dataset.
func (m *Manager) LoadedDatasets() []LoadedDatasetInfo {
m.activeMu.RLock()
@ -309,36 +360,48 @@ func (m *Manager) Load(ctx context.Context, id DatasetID) error {
//
// Returns the JobID started, or empty string when nothing was scheduled.
func (m *Manager) Refresh(ctx context.Context, freshnessTTL time.Duration) (string, error) {
if a := m.activeGlobal(); a != nil && time.Since(a.ID.Epoch) < freshnessTTL {
return "", nil
}
if datasets, err := m.store.List(); err == nil {
for _, id := range datasets {
// Get something usable loaded first, whatever its age.
//
// freshnessTTL answers one question only: go fetch something newer? It must
// never decide whether data already on disk may be read. While one check
// served both, a stored run older than the TTL was skipped even when it was
// the only dataset present — active stayed empty, every prediction answered
// "no dataset loaded" with gigabytes of usable wind on disk, and with no
// reachable origin that state was permanent.
if m.activeGlobal() == nil {
if stored, err := m.store.List(); err == nil {
for _, id := range stored { // newest first
if !id.Subset.IsGlobal() {
continue
}
if time.Since(id.Epoch) > freshnessTTL {
continue
}
if a := m.activeGlobal(); a != nil && a.ID.Equals(id) {
return "", nil
}
if err := m.Load(ctx, id); err == nil {
break
}
}
}
}
active := m.activeGlobal()
if active != nil && time.Since(active.ID.Epoch) < freshnessTTL {
return "", nil
}
}
}
latest, err := m.src.LatestEpoch(ctx)
if err != nil {
return "", fmt.Errorf("latest epoch: %w", err)
}
id := DatasetID{Epoch: latest}
if a := m.activeGlobal(); a != nil && !latest.After(a.ID.Epoch) {
if active != nil && !latest.After(active.ID.Epoch) {
return "", nil
}
// Another replica sharing this volume may have committed it already.
id := DatasetID{Epoch: latest}
if m.store.Exists(id) {
if err := m.Load(ctx, id); err == nil {
return "", nil
}
}
jobID := m.Download(id)
go m.loadAfterCompletion(jobID, id)
return jobID, nil

166
internal/datasets/manager_test.go Normal file
View file

@ -0,0 +1,166 @@
package datasets
import (
"context"
"errors"
"testing"
"time"
"predictor-refactored/internal/weather"
)
// stubField is a WindField that carries only its epoch.
type stubField struct{ epoch time.Time }
func (f stubField) Wind(_, _, _, _ float64) (weather.Sample, error) {
return weather.Sample{}, nil
}
func (f stubField) Epoch() time.Time { return f.epoch }
func (f stubField) Source() string { return "fake" }
// fakeSource records what it was asked to open and can simulate an unreachable
// origin via latestErr.
type fakeSource struct {
latest time.Time
latestErr error
opened []DatasetID
}
func (s *fakeSource) ID() string { return "fake" }
func (s *fakeSource) LatestEpoch(context.Context) (time.Time, error) {
if s.latestErr != nil {
return time.Time{}, s.latestErr
}
return s.latest, nil
}
func (s *fakeSource) Download(context.Context, DatasetID, Storage, ProgressSink, Throttle) error {
return errors.New("fakeSource does not download")
}
func (s *fakeSource) Open(_ context.Context, id DatasetID, _ Storage) (weather.WindField, error) {
s.opened = append(s.opened, id)
return stubField{epoch: id.Epoch}, nil
}
func (s *fakeSource) Coverage(id DatasetID) Coverage {
return Coverage{
Region: Region{MinLat: -90, MaxLat: 90, MinLng: 0, MaxLng: 360},
StartTime: id.Epoch,
EndTime: id.Epoch.Add(192 * time.Hour),
}
}
// fakeStore is an in-memory Storage holding a fixed set of committed datasets,
// newest first, as LocalStore.List promises.
type fakeStore struct{ ids []DatasetID }
func (s *fakeStore) SourceID() string { return "fake" }
func (s *fakeStore) Path(id DatasetID) string { return "/dev/null/" + id.Filename() }
func (s *fakeStore) Exists(id DatasetID) bool {
for _, have := range s.ids {
if have.Equals(id) {
return true
}
}
return false
}
func (s *fakeStore) List() ([]DatasetID, error) { return s.ids, nil }
func (s *fakeStore) Remove(DatasetID) error { return nil }
func (s *fakeStore) BeginWrite(DatasetID) (TempHandle, error) { return nil, errors.New("not used") }
func (s *fakeStore) Lock(context.Context) (func(), error) { return func() {}, nil }
// TestRefreshLoadsAStoredDatasetOlderThanTheFreshnessTTL pins the distinction
// the freshness TTL is allowed to make.
//
// A dataset on disk is usable wind data whatever its age. The TTL answers "go
// fetch something newer?", not "may I read what is already here?". While both
// questions shared one check, an archival run that was the only dataset present
// got skipped, active stayed empty, and every prediction answered "no dataset
// loaded" with gigabytes of usable data on disk — and, with no reachable origin,
// forever.
func TestRefreshLoadsAStoredDatasetOlderThanTheFreshnessTTL(t *testing.T) {
old := time.Now().UTC().Add(-30 * 24 * time.Hour).Truncate(time.Hour)
src := &fakeSource{latestErr: errors.New("origin unreachable")}
store := &fakeStore{ids: []DatasetID{{Epoch: old}}}
m := New(src, store, nil, nil)
// The origin probe is expected to fail; loading what is on disk is not.
_, _ = m.Refresh(context.Background(), 48*time.Hour)
if m.Active() == nil {
t.Fatal("Active() is nil: the stored dataset was never loaded")
}
if got := m.Active().Epoch(); !got.Equal(old) {
t.Errorf("loaded epoch = %s, want %s", got, old)
}
}
// TestRefreshPrefersTheNewestStoredDataset guards the ordering the fix relies
// on: falling back to disk must not mean falling back to the oldest file there.
func TestRefreshPrefersTheNewestStoredDataset(t *testing.T) {
newer := time.Now().UTC().Add(-10 * 24 * time.Hour).Truncate(time.Hour)
older := newer.Add(-20 * 24 * time.Hour)
src := &fakeSource{latestErr: errors.New("origin unreachable")}
store := &fakeStore{ids: []DatasetID{{Epoch: newer}, {Epoch: older}}}
m := New(src, store, nil, nil)
_, _ = m.Refresh(context.Background(), 48*time.Hour)
if m.Active() == nil {
t.Fatal("Active() is nil")
}
if got := m.Active().Epoch(); !got.Equal(newer) {
t.Errorf("loaded epoch = %s, want the newest stored %s", got, newer)
}
}
// TestFieldForLoadsAStoredDatasetThatIsNotActive covers asking for a specific
// run by epoch — the archival-forecast case. Being stored is enough; the caller
// should not have to make it the active dataset first.
func TestFieldForLoadsAStoredDatasetThatIsNotActive(t *testing.T) {
active := time.Now().UTC().Truncate(time.Hour)
archive := active.Add(-30 * 24 * time.Hour)
src := &fakeSource{latest: active}
store := &fakeStore{ids: []DatasetID{{Epoch: active}, {Epoch: archive}}}
m := New(src, store, nil, nil)
if err := m.Load(context.Background(), DatasetID{Epoch: active}); err != nil {
t.Fatalf("Load active: %v", err)
}
field, err := m.FieldFor(context.Background(), archive)
if err != nil {
t.Fatalf("FieldFor(%s): %v", archive, err)
}
if got := field.Epoch(); !got.Equal(archive) {
t.Errorf("epoch = %s, want %s", got, archive)
}
// The active dataset must be unaffected: asking for an archival run for one
// prediction should not repoint the service at it.
if got := m.Active().Epoch(); !got.Equal(active) {
t.Errorf("Active() moved to %s, want %s", got, active)
}
}
// TestFieldForRejectsAnEpochThatIsNotStored is the half that matters most: a
// requested dataset that cannot be served must say so rather than quietly
// substituting a different one, which is what ignoring the parameter did.
func TestFieldForRejectsAnEpochThatIsNotStored(t *testing.T) {
active := time.Now().UTC().Truncate(time.Hour)
src := &fakeSource{latest: active}
store := &fakeStore{ids: []DatasetID{{Epoch: active}}}
m := New(src, store, nil, nil)
if err := m.Load(context.Background(), DatasetID{Epoch: active}); err != nil {
t.Fatalf("Load: %v", err)
}
missing := active.Add(-365 * 24 * time.Hour)
field, err := m.FieldFor(context.Background(), missing)
if err == nil {
t.Fatalf("FieldFor(%s) returned field with epoch %s, want an error", missing, field.Epoch())
}
}

156
internal/engine/engine_test.go
View file

@ -1,7 +1,9 @@
package engine
import (
"fmt"
"math"
"strings"
"testing"
"time"
@ -261,3 +263,157 @@ func TestPolygonOutsideAntimeridian(t *testing.T) {
t.Errorf("(0, 0) should be outside")
}
}
// horizonWind has data up to horizon seconds and fails past it, the way a real
// dataset's time axis does (numerics.AxisError from Axis.Locate).
type horizonWind struct{ horizon float64 }
func (w horizonWind) Wind(t, _, _, _ float64) (weather.Sample, error) {
if t > w.horizon {
return weather.Sample{}, fmt.Errorf("hour=%v out of range", t/3600)
}
return weather.Sample{U: 10, V: 0}, nil
}
func (w horizonWind) Epoch() time.Time { return time.Unix(0, 0).UTC() }
func (w horizonWind) Source() string { return "test" }
// TestWindSamplingFailureIsRecordedNotSwallowed pins the difference between an
// event and a failure.
//
// WindTransport used to discard the sampler's error and return a zero rate. A
// zero rate is indistinguishable from calm air, so a launch past the dataset's
// horizon integrated with no wind at all and produced a balloon that took off
// and landed on the same spot — returned as a successful forecast. An error from
// the sampler means there was no data, so the trajectory past that point is not
// a forecast and must not be presented as one.
func TestWindSamplingFailureIsRecordedNotSwallowed(t *testing.T) {
sink := NewEventSink()
ascend := &Propagator{
Name: "ascent",
Step: 60,
Model: Sum(ConstantRate(5), WindTransport(horizonWind{horizon: 600}, sink)),
Constraints: []Constraint{Altitude{Op: OpGreaterEqual, Limit: 30000, On: ActionStop}},
}
prof := Profile{Stages: []*Propagator{ascend}, Direction: Forward}
prof.Run(0, State{Lat: 0, Lng: 0, Altitude: 0}, sink)
err := sink.Err()
if err == nil {
t.Fatal("sink reports no failure; the sampling error was swallowed")
}
if !strings.Contains(err.Error(), "out of range") {
t.Errorf("error %q does not carry the sampler's reason", err)
}
}
// TestWindSamplingSuccessLeavesNoFailure keeps the check above from firing on
// healthy runs, which would turn every prediction into a 400.
func TestWindSamplingSuccessLeavesNoFailure(t *testing.T) {
sink := NewEventSink()
ascend := &Propagator{
Name: "ascent",
Step: 60,
Model: Sum(ConstantRate(5), WindTransport(horizonWind{horizon: 1e9}, sink)),
Constraints: []Constraint{Altitude{Op: OpGreaterEqual, Limit: 30000, On: ActionStop}},
}
prof := Profile{Stages: []*Propagator{ascend}, Direction: Forward}
prof.Run(0, State{Lat: 0, Lng: 0, Altitude: 0}, sink)
if err := sink.Err(); err != nil {
t.Fatalf("healthy run reported a failure: %v", err)
}
}
// TestAboveModelStaysNonFatal separates the two paths explicitly: extrapolating
// above the highest pressure level is a warning the caller may ignore, and must
// not become a hard failure.
func TestAboveModelStaysNonFatal(t *testing.T) {
sink := NewEventSink()
ascend := &Propagator{
Name: "ascent",
Step: 60,
Model: Sum(ConstantRate(5), WindTransport(aboveModelWind{}, sink)),
Constraints: []Constraint{Altitude{Op: OpGreaterEqual, Limit: 30000, On: ActionStop}},
}
prof := Profile{Stages: []*Propagator{ascend}, Direction: Forward}
prof.Run(0, State{Lat: 0, Lng: 0, Altitude: 0}, sink)
if err := sink.Err(); err != nil {
t.Fatalf("above_model became a failure: %v", err)
}
if len(sink.Snapshot()) == 0 {
t.Error("above_model event was not emitted")
}
}
// TestEveryModelHonoursIncludeWind pins the promise ModelSpec.IncludeWind makes.
//
// buildConstantRate and buildParachuteDescent both took BuildDeps as `_` and
// returned a bare vertical model, so include_wind was accepted and discarded.
// Only buildPiecewise ever called maybeAddWind. POST /api/v2/prediction with the
// spec's own documented example therefore answered 200 with a trajectory that
// rose and fell on the spot — a wind-free forecast presented as a forecast.
//
// The loop over modelFactories is the part that matters going forward: a newly
// registered model type cannot be added without deciding here what include_wind
// does to it, which is exactly the step that was skipped before.
func TestEveryModelHonoursIncludeWind(t *testing.T) {
cases := map[string]struct {
spec ModelSpec
wantError bool
}{
"constant_rate": {spec: ModelSpec{Type: "constant_rate", Rate: 5, IncludeWind: true}},
"parachute_descent": {spec: ModelSpec{Type: "parachute_descent", SeaLevelRate: 5, IncludeWind: true}},
"piecewise": {spec: ModelSpec{Type: "piecewise", IncludeWind: true,
Segments: []PiecewiseSegmentSpec{{Until: math.Inf(1), Rate: 5}}}},
// include_wind on the wind model itself would sum the wind into itself.
// Refused rather than ignored: a silently doubled wind is a wrong forecast
// that looks right.
"wind": {spec: ModelSpec{Type: "wind", IncludeWind: true}, wantError: true},
}
for name := range modelFactories {
if _, ok := cases[name]; !ok {
t.Errorf("model %q is registered but has no include_wind case here", name)
}
}
deps := BuildDeps{Wind: fixedWind{u: 10, v: -4}}
for name, tc := range cases {
t.Run(name, func(t *testing.T) {
built, err := BuildModel(tc.spec, deps)
if tc.wantError {
if err == nil {
t.Fatal("expected include_wind to be refused, got no error")
}
return
}
if err != nil {
t.Fatalf("BuildModel: %v", err)
}
m := built.Model
if built.Build != nil {
m = built.Build(StageContext{})
}
rate := m(0, State{Lat: 0, Lng: 0, Altitude: 1000})
if rate.East != 10 || rate.North != -4 {
t.Errorf("include_wind ignored: east/north = %v/%v, want 10/-4", rate.East, rate.North)
}
})
}
}
// TestIncludeWindWithoutAFieldIsRefused covers the other half of maybeAddWind,
// which returned the wind-free model when deps.Wind was nil. A request that asks
// for wind and cannot get it must fail, not quietly become a vertical drop.
func TestIncludeWindWithoutAFieldIsRefused(t *testing.T) {
for _, typ := range []string{"constant_rate", "parachute_descent", "piecewise"} {
t.Run(typ, func(t *testing.T) {
spec := ModelSpec{Type: typ, Rate: 5, SeaLevelRate: 5, IncludeWind: true,
Segments: []PiecewiseSegmentSpec{{Until: math.Inf(1), Rate: 5}}}
if _, err := BuildModel(spec, BuildDeps{}); err == nil {
t.Error("include_wind with no wind field was accepted")
}
})
}
}

41
internal/engine/events.go
View file

@ -26,10 +26,13 @@ type EventSummary struct {
}
// EventSink collects events from models and the integrator, aggregating
// duplicate types into a single EventSummary. Safe for concurrent use.
// duplicate types into a single EventSummary. It also carries the run's first
// unrecoverable failure, which is a different thing from an event — see Fail.
// Safe for concurrent use.
type EventSink struct {
mu sync.Mutex
summaries map[string]*EventSummary
err error
}
// NewEventSink returns an empty sink.
@ -61,6 +64,42 @@ func (s *EventSink) Emit(typ string, t float64, state State, message string) {
}
}
// Fail records the run's first unrecoverable error. Nil-safe, like Emit.
//
// Deliberately not an event. An event is an observation the caller may choose to
// ignore: "above_model" means samples above the highest pressure level were
// extrapolated, which degrades the answer without invalidating it. A failure
// means the sampler had no data at all, so the trajectory from that point on is
// not a forecast and must not be returned as one.
//
// Routing both through Emit is how the second came to be treated like the first.
// WindTransport discarded the sampler's error and returned a zero rate, which is
// indistinguishable from calm air, so any launch past the dataset's horizon
// integrated with no wind and produced a balloon that took off and landed on the
// same spot — reported as a successful prediction.
//
// Only the first failure is kept: it is the one that explains the run.
func (s *EventSink) Fail(err error) {
if s == nil || err == nil {
return
}
s.mu.Lock()
defer s.mu.Unlock()
if s.err == nil {
s.err = err
}
}
// Err returns the first failure passed to Fail, or nil if the run was sound.
func (s *EventSink) Err() error {
if s == nil {
return nil
}
s.mu.Lock()
defer s.mu.Unlock()
return s.err
}
// Snapshot returns a stable copy of every summary in deterministic order
// (sorted by Type).
func (s *EventSink) Snapshot() []EventSummary {

7
internal/engine/models.go
View file

@ -1,7 +1,9 @@
package engine
import (
"fmt"
"sort"
"time"
"predictor-refactored/internal/numerics"
"predictor-refactored/internal/weather"
@ -84,6 +86,11 @@ func WindTransport(field weather.WindField, events *EventSink) Model {
return func(t float64, s State) numerics.Rate {
sample, err := field.Wind(t, s.Lat, s.Lng, s.Altitude)
if err != nil {
// Recorded, not swallowed. A zero rate here is indistinguishable
// from calm air, so returning one turns "we have no data for this
// time" into "the balloon did not move" — see EventSink.Fail.
events.Fail(fmt.Errorf("no wind data at %s: %w",
time.Unix(int64(t), 0).UTC().Format(time.RFC3339), err))
return numerics.Rate{}
}
if sample.AboveModel && events != nil {

52
internal/engine/registry.go
View file

@ -205,21 +205,32 @@ func buildPolygon(spec ConstraintSpec, _ BuildDeps) (Constraint, error) {
return NewPolygon(spec.Vertices, mode, act, spec.Label), nil
}
func buildConstantRate(spec ModelSpec, _ BuildDeps) (BuiltModel, error) {
return BuiltModel{Model: ConstantRate(spec.Rate)}, nil
func buildConstantRate(spec ModelSpec, deps BuildDeps) (BuiltModel, error) {
if err := checkWind(spec.IncludeWind, deps); err != nil {
return BuiltModel{}, err
}
return BuiltModel{Model: addWind(ConstantRate(spec.Rate), spec.IncludeWind, deps)}, nil
}
func buildParachuteDescent(spec ModelSpec, _ BuildDeps) (BuiltModel, error) {
func buildParachuteDescent(spec ModelSpec, deps BuildDeps) (BuiltModel, error) {
if spec.SeaLevelRate <= 0 {
return BuiltModel{}, fmt.Errorf("parachute_descent requires positive sea_level_rate")
}
return BuiltModel{Model: ParachuteDescent(spec.SeaLevelRate)}, nil
if err := checkWind(spec.IncludeWind, deps); err != nil {
return BuiltModel{}, err
}
return BuiltModel{Model: addWind(ParachuteDescent(spec.SeaLevelRate), spec.IncludeWind, deps)}, nil
}
func buildWind(_ ModelSpec, deps BuildDeps) (BuiltModel, error) {
func buildWind(spec ModelSpec, deps BuildDeps) (BuiltModel, error) {
if deps.Wind == nil {
return BuiltModel{}, fmt.Errorf("wind model requires a loaded wind field")
}
// Refused rather than ignored: summing the wind into itself would double the
// drift, and a doubled wind is a wrong forecast that looks like a right one.
if spec.IncludeWind {
return BuiltModel{}, fmt.Errorf("wind model does not take include_wind: it would sum the wind into itself")
}
return BuiltModel{Model: WindTransport(deps.Wind, deps.Events)}, nil
}
@ -231,12 +242,17 @@ func buildPiecewise(spec ModelSpec, deps BuildDeps) (BuiltModel, error) {
return BuiltModel{}, fmt.Errorf("piecewise: unknown segment reference %q", s.Reference)
}
}
// Checked here, not in the closure below: the profile runner calls Build once
// per stage and has nowhere to return an error to.
if err := checkWind(spec.IncludeWind, deps); err != nil {
return BuiltModel{}, err
}
// Always build lazily: the profile runner supplies a StageContext before
// each stage, which is what resolves absolute / profile-relative /
// propagator-relative segment times uniformly.
return BuiltModel{
Build: func(ctx StageContext) Model {
return maybeAddWind(Piecewise(resolveSegments(spec.Segments, ctx)), spec.IncludeWind, deps)
return addWind(Piecewise(resolveSegments(spec.Segments, ctx)), spec.IncludeWind, deps)
},
}, nil
}
@ -266,12 +282,26 @@ func segmentBase(reference string, ctx StageContext) float64 {
}
}
// maybeAddWind sums a WindTransport model into base when the spec asks for it.
func maybeAddWind(base Model, includeWind bool, deps BuildDeps) Model {
if !includeWind {
return base
// checkWind reports whether include_wind can be satisfied at all.
//
// A request that asks for wind and cannot have it is an error. This used to
// return the wind-free model instead, which turned a missing wind field into a
// balloon that rose and fell on the spot, reported as a successful forecast.
func checkWind(includeWind bool, deps BuildDeps) error {
if includeWind && deps.Wind == nil {
return fmt.Errorf("include_wind requires a loaded wind field")
}
if deps.Wind == nil {
return nil
}
// addWind sums a WindTransport model into base when the spec asks for it.
//
// Callers must have called checkWind first. The split exists because the
// piecewise builder combines inside a closure the profile runner invokes per
// stage, with no way to surface an error — so validation happens at build time
// and combination stays infallible, rather than an error being swallowed there.
func addWind(base Model, includeWind bool, deps BuildDeps) Model {
if !includeWind {
return base
}
return Sum(base, WindTransport(deps.Wind, deps.Events))

29
internal/windviz/windviz.go
View file

@ -68,9 +68,20 @@ const (
// Rasterize samples field over req and returns the U/V grid payload.
//
// Data is laid out in wind-js scan order: row 0 is the northernmost
// latitude (la1), each row runs west→east, longitudes increasing. Per-cell
// sampling errors (e.g. altitude outside the model) are written as 0 rather
// than failing the whole request; a time outside coverage is a hard error.
// latitude (la1), each row runs west→east, longitudes increasing.
//
// Cells that cannot be sampled individually — a regional dataset queried outside
// its region — are written as 0 and the request still succeeds, because a partly
// covered grid is worth drawing. A grid in which nothing could be sampled is an
// error instead. Time is one value for the whole request, so a time past the
// dataset's horizon fails every cell, and returning that as zeros made a velocity
// layer draw an atmosphere at perfect rest where the honest answer is "we have no
// data for this time".
//
// Known gap: an individual zero-filled cell is indistinguishable from genuine
// calm in the payload. The wind-js format expresses missing data as null, which
// would mean typing Data as []*float64; there is no consumer to validate that
// against since the browser wind layer was removed in the Cesium migration.
func Rasterize(field weather.WindField, req Request) (Field, error) {
step := req.Step
if step <= 0 {
@ -125,6 +136,8 @@ func Rasterize(field weather.WindField, req Request) (Field, error) {
v := make([]float64, nx*ny)
// Row 0 = north (la1); rows descend in latitude.
var failed int
var firstErr error
for j := range ny {
lat := maxLat - float64(j)*step
for i := range nx {
@ -132,12 +145,20 @@ func Rasterize(field weather.WindField, req Request) (Field, error) {
s, err := field.Wind(req.Time, lat, normLng(lng), req.Altitude)
idx := j*nx + i
if err != nil {
continue // leave as 0
failed++
if firstErr == nil {
firstErr = err
}
continue // leave as 0; see the doc comment on this distinction
}
u[idx] = s.U
v[idx] = s.V
}
}
if failed == nx*ny {
return nil, fmt.Errorf("no wind data anywhere in the requested grid at %s: %w",
time.Unix(int64(req.Time), 0).UTC().Format(time.RFC3339), firstErr)
}
refTime := time.Unix(int64(req.Time), 0).UTC().Format("2006-01-02T15:04:05.000Z")
mk := func(num int, name string, data []float64) Component {

85
internal/windviz/windviz_test.go
View file

@ -1,6 +1,8 @@
package windviz
import (
"fmt"
"strings"
"testing"
"time"
@ -94,3 +96,86 @@ func TestCacheRoundTrip(t *testing.T) {
t.Errorf("cache should hit after put")
}
}
// horizonWind has data up to horizon seconds and fails past it, the way a real
// dataset's time axis does.
type horizonWind struct{ horizon float64 }
func (w horizonWind) Wind(t, _, _, _ float64) (weather.Sample, error) {
if t > w.horizon {
return weather.Sample{}, fmt.Errorf("hour=%v out of range", t/3600)
}
return weather.Sample{U: 7, V: -3}, nil
}
func (w horizonWind) Epoch() time.Time { return time.Unix(0, 0).UTC() }
func (w horizonWind) Source() string { return "test" }
// northOnlyWind has data only in the northern hemisphere, the way a regional
// subset does. Its failures are genuinely per-cell.
type northOnlyWind struct{}
func (northOnlyWind) Wind(_, lat, _, _ float64) (weather.Sample, error) {
if lat < 0 {
return weather.Sample{}, fmt.Errorf("lat=%v outside region", lat)
}
return weather.Sample{U: 5, V: 1}, nil
}
func (northOnlyWind) Epoch() time.Time { return time.Unix(0, 0).UTC() }
func (northOnlyWind) Source() string { return "test" }
// TestRasterizeRefusesWhenNoCellHasData is the case the package comment already
// promised and the code never implemented ("a time outside coverage is a hard
// error").
//
// Time is one value for the whole request, so a time past the dataset's horizon
// fails every cell. Each failure was written as a zero, so the response was a
// complete grid of zero wind — a velocity layer draws that as an atmosphere at
// perfect rest. Missing data must not be rendered as calm air.
func TestRasterizeRefusesWhenNoCellHasData(t *testing.T) {
f := horizonWind{horizon: 600}
out, err := Rasterize(f, Request{Time: 700, MinLng: 0, MaxLng: 360, Step: 90})
if err == nil {
t.Fatalf("Rasterize returned %d components instead of an error", len(out))
}
if !strings.Contains(err.Error(), "out of range") {
t.Errorf("error %q does not carry the sampler's reason", err)
}
}
// TestRasterizeStillSucceedsWithinCoverage keeps the check above from rejecting
// healthy requests.
func TestRasterizeStillSucceedsWithinCoverage(t *testing.T) {
f := horizonWind{horizon: 600}
out, err := Rasterize(f, Request{Time: 300, MinLng: 0, MaxLng: 360, Step: 90})
if err != nil {
t.Fatalf("Rasterize within coverage: %v", err)
}
if got := out[0].Data[0]; got != 7 {
t.Errorf("u = %v, want 7", got)
}
}
// TestRasterizeZeroFillsIndividualGaps preserves the documented per-cell
// behaviour: a partly-covered grid is still worth drawing, so gaps stay zero and
// the request succeeds. Only a grid with nothing in it is refused.
func TestRasterizeZeroFillsIndividualGaps(t *testing.T) {
out, err := Rasterize(northOnlyWind{}, Request{
MinLat: -30, MaxLat: 30, MinLng: 0, MaxLng: 30, Step: 30,
})
if err != nil {
t.Fatalf("Rasterize partly-covered grid: %v", err)
}
// Row 0 is the northernmost latitude (+30), the last row is -30.
nx := out[0].Header.Nx
ny := out[0].Header.Ny
if got := out[0].Data[0]; got != 5 {
t.Errorf("northern cell u = %v, want 5", got)
}
if got := out[0].Data[(ny-1)*nx]; got != 0 {
t.Errorf("southern gap u = %v, want 0", got)
}
}

121
pkg/rest/oas_client_gen.go
View file

@ -4,6 +4,7 @@ package rest
import (
"context"
"io"
"net/url"
"strings"
"time"
@ -239,7 +240,13 @@ func (c *Client) sendCancelDatasetJob(ctx context.Context, params CancelDatasetJ
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeCancelDatasetJobResponse(resp)
@ -331,7 +338,13 @@ func (c *Client) sendCancelPredictionJob(ctx context.Context, params CancelPredi
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeCancelPredictionJobResponse(resp)
@ -408,7 +421,13 @@ func (c *Client) sendCreatePredictionJob(ctx context.Context, request *Predictio
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeCreatePredictionJobResponse(resp)
@ -500,7 +519,13 @@ func (c *Client) sendDeleteDataset(ctx context.Context, params DeleteDatasetPara
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeDeleteDatasetResponse(resp)
@ -592,7 +617,13 @@ func (c *Client) sendGetDatasetJob(ctx context.Context, params GetDatasetJobPara
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeGetDatasetJobResponse(resp)
@ -684,7 +715,13 @@ func (c *Client) sendGetPredictionJob(ctx context.Context, params GetPredictionJ
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeGetPredictionJobResponse(resp)
@ -758,7 +795,13 @@ func (c *Client) sendGetServiceStatus(ctx context.Context) (res *StatusResponse,
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeGetServiceStatusResponse(resp)
@ -955,7 +998,13 @@ func (c *Client) sendGetWindField(ctx context.Context, params GetWindFieldParams
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeGetWindFieldResponse(resp)
@ -1029,7 +1078,13 @@ func (c *Client) sendGetWindMeta(ctx context.Context) (res *WindMeta, err error)
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeGetWindMetaResponse(resp)
@ -1103,7 +1158,13 @@ func (c *Client) sendListDatasetJobs(ctx context.Context) (res []DownloadJob, er
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeListDatasetJobsResponse(resp)
@ -1177,7 +1238,13 @@ func (c *Client) sendListDatasets(ctx context.Context) (res *DatasetList, err er
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeListDatasetsResponse(resp)
@ -1433,7 +1500,13 @@ func (c *Client) sendPerformPrediction(ctx context.Context, params PerformPredic
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodePerformPredictionResponse(resp)
@ -1510,7 +1583,13 @@ func (c *Client) sendPerformPredictionV2(ctx context.Context, request *Predictio
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodePerformPredictionV2Response(resp)
@ -1584,7 +1663,13 @@ func (c *Client) sendReadinessCheck(ctx context.Context) (res *ReadinessResponse
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeReadinessCheckResponse(resp)
@ -1661,7 +1746,13 @@ func (c *Client) sendTriggerDatasetDownload(ctx context.Context, request *Downlo
return res, errors.Wrap(err, "do request")
}
body := resp.Body
defer body.Close()
defer func() {
// Drain the body to EOF before closing, so the underlying
// connection can be reused by the Transport regardless of the
// response status code. See https://github.com/ogen-go/ogen/issues/1670.
_, _ = io.Copy(io.Discard, body)
_ = body.Close()
}()
stage = "DecodeResponse"
result, err := decodeTriggerDatasetDownloadResponse(resp)

30
pkg/rest/oas_handlers_gen.go
View file

@ -71,7 +71,7 @@ func (s *Server) handleCancelDatasetJobRequest(args [1]string, argsEscaped bool,
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -225,7 +225,7 @@ func (s *Server) handleCancelPredictionJobRequest(args [1]string, argsEscaped bo
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -379,7 +379,7 @@ func (s *Server) handleCreatePredictionJobRequest(args [0]string, argsEscaped bo
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -533,7 +533,7 @@ func (s *Server) handleDeleteDatasetRequest(args [1]string, argsEscaped bool, w
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -687,7 +687,7 @@ func (s *Server) handleGetDatasetJobRequest(args [1]string, argsEscaped bool, w
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -841,7 +841,7 @@ func (s *Server) handleGetPredictionJobRequest(args [1]string, argsEscaped bool,
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -995,7 +995,7 @@ func (s *Server) handleGetServiceStatusRequest(args [0]string, argsEscaped bool,
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -1130,7 +1130,7 @@ func (s *Server) handleGetWindFieldRequest(args [0]string, argsEscaped bool, w h
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -1308,7 +1308,7 @@ func (s *Server) handleGetWindMetaRequest(args [0]string, argsEscaped bool, w ht
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -1443,7 +1443,7 @@ func (s *Server) handleListDatasetJobsRequest(args [0]string, argsEscaped bool,
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -1578,7 +1578,7 @@ func (s *Server) handleListDatasetsRequest(args [0]string, argsEscaped bool, w h
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -1713,7 +1713,7 @@ func (s *Server) handlePerformPredictionRequest(args [0]string, argsEscaped bool
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -1907,7 +1907,7 @@ func (s *Server) handlePerformPredictionV2Request(args [0]string, argsEscaped bo
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -2061,7 +2061,7 @@ func (s *Server) handleReadinessCheckRequest(args [0]string, argsEscaped bool, w
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)
@ -2196,7 +2196,7 @@ func (s *Server) handleTriggerDatasetDownloadRequest(args [0]string, argsEscaped
if code != 0 {
codeAttr := semconv.HTTPResponseStatusCode(code)
attrs = append(attrs, codeAttr)
span.SetAttributes(codeAttr)
span.SetAttributes(attrs...)
}
attrOpt := metric.WithAttributes(attrs...)

31
pkg/rest/oas_json_gen.go
View file

@ -4417,6 +4417,12 @@ func (s *PredictionV2Request) encodeFields(e *jx.Encoder) {
e.FieldStart("launch")
s.Launch.Encode(e)
}
{
if s.Dataset.Set {
e.FieldStart("dataset")
s.Dataset.Encode(e, json.EncodeDateTime)
}
}
{
if s.Direction.Set {
e.FieldStart("direction")
@ -4449,12 +4455,13 @@ func (s *PredictionV2Request) encodeFields(e *jx.Encoder) {
}
}
var jsonFieldsNameOfPredictionV2Request = [5]string{
var jsonFieldsNameOfPredictionV2Request = [6]string{
0: "launch",
1: "direction",
2: "profile",
3: "globals",
4: "options",
1: "dataset",
2: "direction",
3: "profile",
4: "globals",
5: "options",
}
// Decode decodes PredictionV2Request from json.
@ -4477,6 +4484,16 @@ func (s *PredictionV2Request) Decode(d *jx.Decoder) error {
}(); err != nil {
return errors.Wrap(err, "decode field \"launch\"")
}
case "dataset":
if err := func() error {
s.Dataset.Reset()
if err := s.Dataset.Decode(d, json.DecodeDateTime); err != nil {
return err
}
return nil
}(); err != nil {
return errors.Wrap(err, "decode field \"dataset\"")
}
case "direction":
if err := func() error {
s.Direction.Reset()
@ -4488,7 +4505,7 @@ func (s *PredictionV2Request) Decode(d *jx.Decoder) error {
return errors.Wrap(err, "decode field \"direction\"")
}
case "profile":
requiredBitSet[0] |= 1 << 2
requiredBitSet[0] |= 1 << 3
if err := func() error {
s.Profile = make([]StageSpec, 0)
if err := d.Arr(func(d *jx.Decoder) error {
@ -4542,7 +4559,7 @@ func (s *PredictionV2Request) Decode(d *jx.Decoder) error {
// Validate required fields.
var failures []validate.FieldError
for i, mask := range [1]uint8{
0b00000101,
0b00001001,
} {
if result := (requiredBitSet[i] & mask) ^ mask; result != 0 {
// Mask only required fields and check equality to mask using XOR.

21
pkg/rest/oas_response_encoders_gen.go
View file

@ -14,14 +14,12 @@ import (
func encodeCancelDatasetJobResponse(response *CancelDatasetJobNoContent, w http.ResponseWriter, span trace.Span) error {
w.WriteHeader(204)
span.SetStatus(codes.Ok, http.StatusText(204))
return nil
}
func encodeCancelPredictionJobResponse(response *CancelPredictionJobNoContent, w http.ResponseWriter, span trace.Span) error {
w.WriteHeader(204)
span.SetStatus(codes.Ok, http.StatusText(204))
return nil
}
@ -29,7 +27,6 @@ func encodeCancelPredictionJobResponse(response *CancelPredictionJobNoContent, w
func encodeCreatePredictionJobResponse(response *PredictionJob, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(202)
span.SetStatus(codes.Ok, http.StatusText(202))
e := new(jx.Encoder)
response.Encode(e)
@ -42,7 +39,6 @@ func encodeCreatePredictionJobResponse(response *PredictionJob, w http.ResponseW
func encodeDeleteDatasetResponse(response *DeleteDatasetNoContent, w http.ResponseWriter, span trace.Span) error {
w.WriteHeader(204)
span.SetStatus(codes.Ok, http.StatusText(204))
return nil
}
@ -50,7 +46,6 @@ func encodeDeleteDatasetResponse(response *DeleteDatasetNoContent, w http.Respon
func encodeGetDatasetJobResponse(response *DownloadJob, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -64,7 +59,6 @@ func encodeGetDatasetJobResponse(response *DownloadJob, w http.ResponseWriter, s
func encodeGetPredictionJobResponse(response *PredictionJob, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -78,7 +72,6 @@ func encodeGetPredictionJobResponse(response *PredictionJob, w http.ResponseWrit
func encodeGetServiceStatusResponse(response *StatusResponse, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -92,7 +85,6 @@ func encodeGetServiceStatusResponse(response *StatusResponse, w http.ResponseWri
func encodeGetWindFieldResponse(response []WindComponent, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
e.ArrStart()
@ -110,7 +102,6 @@ func encodeGetWindFieldResponse(response []WindComponent, w http.ResponseWriter,
func encodeGetWindMetaResponse(response *WindMeta, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -124,7 +115,6 @@ func encodeGetWindMetaResponse(response *WindMeta, w http.ResponseWriter, span t
func encodeListDatasetJobsResponse(response []DownloadJob, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
e.ArrStart()
@ -142,7 +132,6 @@ func encodeListDatasetJobsResponse(response []DownloadJob, w http.ResponseWriter
func encodeListDatasetsResponse(response *DatasetList, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -156,7 +145,6 @@ func encodeListDatasetsResponse(response *DatasetList, w http.ResponseWriter, sp
func encodePerformPredictionResponse(response *PredictionResponse, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -170,7 +158,6 @@ func encodePerformPredictionResponse(response *PredictionResponse, w http.Respon
func encodePerformPredictionV2Response(response *PredictionV2Response, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -184,7 +171,6 @@ func encodePerformPredictionV2Response(response *PredictionV2Response, w http.Re
func encodeReadinessCheckResponse(response *ReadinessResponse, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(200)
span.SetStatus(codes.Ok, http.StatusText(200))
e := new(jx.Encoder)
response.Encode(e)
@ -198,7 +184,6 @@ func encodeReadinessCheckResponse(response *ReadinessResponse, w http.ResponseWr
func encodeTriggerDatasetDownloadResponse(response *DownloadAccepted, w http.ResponseWriter, span trace.Span) error {
w.Header().Set("Content-Type", "application/json; charset=utf-8")
w.WriteHeader(202)
span.SetStatus(codes.Ok, http.StatusText(202))
e := new(jx.Encoder)
response.Encode(e)
@ -217,10 +202,8 @@ func encodeErrorResponse(response *DefaultErrorStatusCode, w http.ResponseWriter
code = http.StatusOK
}
w.WriteHeader(code)
if st := http.StatusText(code); code >= http.StatusBadRequest {
span.SetStatus(codes.Error, st)
} else {
span.SetStatus(codes.Ok, st)
if code >= http.StatusInternalServerError {
span.SetStatus(codes.Error, http.StatusText(code))
}
e := new(jx.Encoder)

20
pkg/rest/oas_schemas_gen.go
View file

@ -2615,13 +2615,15 @@ func (s *PredictionResponseWarnings) init() PredictionResponseWarnings {
return m
}
// A profile-driven prediction. `profile` is an ordered chain of
// propagators; each integrates from where the previous ended. A stage's
// `constraints` decide when it ends and what happens next: stop the
// profile, hand off to `fallback_index`, or clip to the boundary.
// A profile-driven prediction. `profile` is an ordered chain of propagators; each integrates from
// where the previous ended. A stage's `constraints` decide when it ends and what happens next: stop
// the profile, hand off to `fallback_index`, or clip to the boundary.
// Ref: #/components/schemas/PredictionV2Request
type PredictionV2Request struct {
Launch Launch `json:"launch"`
// Forecast run to predict from, given as its epoch. Defaults to the active dataset. The named run must
// be stored; a request for one that is not is rejected rather than answered from a different run.
Dataset OptDateTime `json:"dataset"`
// Forward integrates launch→landing; reverse integrates backward in time.
Direction OptPredictionV2RequestDirection `json:"direction"`
Profile []StageSpec `json:"profile"`
@ -2635,6 +2637,11 @@ func (s *PredictionV2Request) GetLaunch() Launch {
return s.Launch
}
// GetDataset returns the value of Dataset.
func (s *PredictionV2Request) GetDataset() OptDateTime {
return s.Dataset
}
// GetDirection returns the value of Direction.
func (s *PredictionV2Request) GetDirection() OptPredictionV2RequestDirection {
return s.Direction
@ -2660,6 +2667,11 @@ func (s *PredictionV2Request) SetLaunch(val Launch) {
s.Launch = val
}
// SetDataset sets the value of Dataset.
func (s *PredictionV2Request) SetDataset(val OptDateTime) {
s.Dataset = val
}
// SetDirection sets the value of Direction.
func (s *PredictionV2Request) SetDirection(val OptPredictionV2RequestDirection) {
s.Direction = val