86 lines
2.5 KiB
Go
86 lines
2.5 KiB
Go
package data
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"time"
|
|
)
|
|
|
|
type PredictPoint struct {
|
|
ForecastTime time.Time
|
|
RainMMx1000 int32
|
|
}
|
|
|
|
// ForecastRainAtIssued returns forecast hourly rows for a given station/provider at an exact issued_at time.
|
|
func ForecastRainAtIssued(ctx context.Context, stationID, provider string, issuedAt time.Time) ([]PredictPoint, error) {
|
|
const q = `
|
|
SELECT forecast_time, rain_mm_x1000
|
|
FROM forecast_hourly
|
|
WHERE station_id=$1 AND provider=$2 AND issued_at=$3
|
|
ORDER BY forecast_time ASC`
|
|
rows, err := DB().QueryContext(ctx, q, stationID, provider, issuedAt)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []PredictPoint
|
|
for rows.Next() {
|
|
var p PredictPoint
|
|
var rain sql.NullInt32
|
|
if err := rows.Scan(&p.ForecastTime, &rain); err != nil {
|
|
continue
|
|
}
|
|
if rain.Valid {
|
|
p.RainMMx1000 = rain.Int32
|
|
}
|
|
out = append(out, p)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ResolveIssuedAtInBucket finds the latest issued_at in [bucket, bucket+1h) for station/provider.
|
|
func ResolveIssuedAtInBucket(ctx context.Context, stationID, provider string, bucketHour time.Time) (time.Time, bool, error) {
|
|
const q = `SELECT issued_at FROM forecast_hourly WHERE station_id=$1 AND provider=$2 AND issued_at >= $3 AND issued_at < $3 + interval '1 hour' ORDER BY issued_at DESC LIMIT 1`
|
|
var t time.Time
|
|
err := DB().QueryRowContext(ctx, q, stationID, provider, bucketHour).Scan(&t)
|
|
if err == sql.ErrNoRows {
|
|
return time.Time{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return time.Time{}, false, err
|
|
}
|
|
return t, true, nil
|
|
}
|
|
|
|
// UpsertForecastRain writes rain-only rows for a provider at issued_at, upserting by key.
|
|
// Only the rain_mm_x1000 column is set/updated; other columns remain NULL or unchanged.
|
|
type UpsertRainItem struct {
|
|
ForecastTime time.Time
|
|
RainMMx1000 int32
|
|
}
|
|
|
|
func UpsertForecastRain(ctx context.Context, stationID, provider string, issuedAt time.Time, items []UpsertRainItem) error {
|
|
if len(items) == 0 {
|
|
return nil
|
|
}
|
|
const q = `
|
|
INSERT INTO forecast_hourly (
|
|
station_id, provider, issued_at, forecast_time, rain_mm_x1000
|
|
) VALUES ($1,$2,$3,$4,$5)
|
|
ON CONFLICT (station_id, provider, issued_at, forecast_time) DO UPDATE SET
|
|
rain_mm_x1000 = EXCLUDED.rain_mm_x1000`
|
|
tx, err := DB().BeginTx(ctx, &sql.TxOptions{})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() {
|
|
_ = tx.Rollback()
|
|
}()
|
|
for _, it := range items {
|
|
if _, err := tx.ExecContext(ctx, q, stationID, provider, issuedAt, it.ForecastTime, it.RainMMx1000); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return tx.Commit()
|
|
}
|