diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index a4382fe..b353afa 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -28,116 +28,6 @@ import ( db "github.com/openclimatefix/data-platform/internal/server/postgres/gen" ) -// --- Reuseable Functions for Route Logic ------------------------------------------------------- - -// timeWindowToPgWindow converts a TimeWindow protobuf message to a pair of pgtype.Timestamp values. -func timeWindowToPgWindow( - window *pb.TimeWindow, -) (start pgtype.Timestamp, end pgtype.Timestamp, err error) { - currentTime := time.Now().UTC() - if window == nil || (window.StartTimestampUtc == nil && window.EndTimestampUtc == nil) { - start = pgtype.Timestamp{Time: currentTime.Add(-48 * time.Hour), Valid: true} - end = pgtype.Timestamp{Time: currentTime.Add(36 * time.Hour), Valid: true} - } else if window.StartTimestampUtc != nil && window.EndTimestampUtc != nil { - start = pgtype.Timestamp{Time: window.StartTimestampUtc.AsTime(), Valid: true} - end = pgtype.Timestamp{Time: window.EndTimestampUtc.AsTime(), Valid: true} - } else { - err = errors.New( - "invalid time window: both start and end timestamps must be provided or neither", - ) - } - - return start, end, err -} - -// timeptrToPgTimestamp converts a protobuf Timestamp pointer to a pgtype.Timestamp. -// If the pointer is nil, it returns the current time truncated to the nearest minute. -func timeptrToPgTimestamp(t *timestamppb.Timestamp) pgtype.Timestamp { - if t == nil { - return pgtype.Timestamp{ - Time: time.Now().UTC().Truncate(time.Minute), - Valid: true, - } - } - - return pgtype.Timestamp{Time: t.AsTime().UTC(), Valid: true} -} - -// extractSIPStatPtrFromMap gets a key's value from a map as a pointer, and converts it -// to a smallint percentage. If it doesn't exist, it returns nil. -func extractSIPStatPtrFromMap(m map[string]float32, key string) *int16 { - val, exists := m[key] - if !exists { - return nil - } - - sip_val := int16(val * 30000.0) - - return &sip_val -} - -// prepareForecastParams generates the database parameters for a single forecast from a gRPC request. -func prepareForecastParams( - req *pb.CreateForecastRequest, - geometryUuid uuid.UUID, - sourceTypeId int16, - forecasterId int32, -) (db.CreateForecastsParams, error) { - initTime := req.InitTimeUtc.AsTime().Truncate(time.Minute) - - fUuid, err := uuid.NewV7() - if err != nil { - return db.CreateForecastsParams{}, fmt.Errorf("failed to generate uuidv7: %w", err) - } - - // Manually overwrite the 48-bit timestamp with the initTime milliseconds - ms := uint64(initTime.UnixMilli()) - fUuid[0] = byte(ms >> 40) - fUuid[1] = byte(ms >> 32) - fUuid[2] = byte(ms >> 24) - fUuid[3] = byte(ms >> 16) - fUuid[4] = byte(ms >> 8) - fUuid[5] = byte(ms) - - firstHorizon := int32(req.Values[0].HorizonMins) - lastHorizon := int32(req.Values[len(req.Values)-1].HorizonMins) - - periodStart := initTime.Add(time.Duration(firstHorizon) * time.Minute) - periodEnd := initTime.Add(time.Duration(lastHorizon) * time.Minute) - - targetPeriod := pgtype.Range[pgtype.Timestamp]{ - Lower: pgtype.Timestamp{Time: periodStart, Valid: true}, - Upper: pgtype.Timestamp{Time: periodEnd, Valid: true}, - LowerType: pgtype.Inclusive, - UpperType: pgtype.Inclusive, - Valid: true, - } - - var createdTime pgtype.Timestamp - if req.CreatedTimestampUtc != nil { - createdTime = pgtype.Timestamp{Time: req.CreatedTimestampUtc.AsTime(), Valid: true} - } else { - createdTime = pgtype.Timestamp{ - Time: time.Now().UTC().Truncate(time.Minute), - Valid: true, - } - } - - return db.CreateForecastsParams{ - ForecastUuid: fUuid, - GeometryUuid: geometryUuid, - SourceTypeID: sourceTypeId, - ForecasterID: forecasterId, - InitTimeUtc: pgtype.Timestamp{Time: initTime, Valid: true}, - ValueResolutionMins: int16(req.Values[1].HorizonMins - req.Values[0].HorizonMins), - TargetPeriod: targetPeriod, - Metadata: req.Metadata, - CreatedAtUtc: createdTime, - }, nil -} - -// --- Server Implementation ---------------------------------------------------------------------- - func NewDataPlatformDataServiceServerImpl() *DataPlatformDataServiceServerImpl { return &DataPlatformDataServiceServerImpl{} } @@ -146,8 +36,6 @@ func NewDataPlatformDataServiceServerImpl() *DataPlatformDataServiceServerImpl { // It requires the database transaction for the request to be set in the context. type DataPlatformDataServiceServerImpl struct{} -// --- Server Method Implementations -------------------------------------------------------------- - // CreateForecast implements dp.DataPlatformDataServiceServer. func (s *DataPlatformDataServiceServerImpl) CreateForecast( ctx context.Context, @@ -207,14 +95,14 @@ func (s *DataPlatformDataServiceServerImpl) CreateForecast( Msg("found forecaster") // Create a new forecast - fParams, err := prepareForecastParams( + fParams, err := mapCreateForecast( req, uuid.MustParse(req.LocationUuid), dbSource.SourceTypeID, dbForecaster.ForecasterID, ) if err != nil { - return nil, fmt.Errorf("failed to prepare forecast params: %w", err) + return nil, fmt.Errorf("failed to map forecast params: %w", err) } countF, err := querier.CreateForecasts(ctx, []db.CreateForecastsParams{fParams}) @@ -327,19 +215,7 @@ func (s *DataPlatformDataServiceServerImpl) GetLatestForecasts( Int("dp.forecasts.count", len(dbListForecasts)). Msg("fetched latest forecasts") - forecasts := make([]*pb.GetLatestForecastsResponse_Forecast, len(dbListForecasts)) - for i, fc := range dbListForecasts { - forecasts[i] = &pb.GetLatestForecastsResponse_Forecast{ - InitializationTimestampUtc: timestamppb.New(fc.InitTimeUtc.Time), - Forecaster: &pb.Forecaster{ - ForecasterName: fc.ForecasterName, - ForecasterVersion: fc.ForecasterVersion, - }, - LocationUuid: fc.GeometryUuid.String(), - Metadata: fc.Metadata, - CreatedTimestampUtc: timestamppb.New(fc.CreatedAtUtc.Time), - } - } + forecasts := MapSlice(dbListForecasts, mapLatestForecast) return &pb.GetLatestForecastsResponse{ Forecasts: forecasts, @@ -444,13 +320,7 @@ func (s *DataPlatformDataServiceServerImpl) ListForecasters( return nil, fmt.Errorf("no forecasters found with the specified filters: %w", err) } - forecasters := make([]*pb.Forecaster, len(dbListForecasters)) - for i, fc := range dbListForecasters { - forecasters[i] = &pb.Forecaster{ - ForecasterName: fc.ForecasterName, - ForecasterVersion: fc.ForecasterVersion, - } - } + forecasters := MapSlice(dbListForecasters, mapForecaster) return &pb.ListForecastersResponse{ Forecasters: forecasters, @@ -493,10 +363,7 @@ func (s *DataPlatformDataServiceServerImpl) StreamForecastData( l.Debug().Str("loc", locStr).Msg("STARTING database query") - locationUuid, err := uuid.Parse(locStr) - if err != nil { - return status.Errorf(codes.InvalidArgument, "Invalid location UUID: %v", err) - } + locationUuid := uuid.MustParse(locStr) // Query with the pool directly so each concurrent request gets a fresh connection. // This is to avoid very large data requests choking the memory of the API. @@ -553,53 +420,10 @@ func (s *DataPlatformDataServiceServerImpl) StreamForecastData( ) } - otherStatistics := make(map[string]float32) - if row.P02Sip != nil { - otherStatistics["p02"] = float32(*row.P02Sip) / 30000.0 - } - - if row.P10Sip != nil { - otherStatistics["p10"] = float32(*row.P10Sip) / 30000.0 - } - - if row.P25Sip != nil { - otherStatistics["p25"] = float32(*row.P25Sip) / 30000.0 - } - - if row.P75Sip != nil { - otherStatistics["p75"] = float32(*row.P75Sip) / 30000.0 - } - - if row.P90Sip != nil { - otherStatistics["p90"] = float32(*row.P90Sip) / 30000.0 - } - - if row.P98Sip != nil { - otherStatistics["p98"] = float32(*row.P98Sip) / 30000.0 - } - - metadata := make(map[string]string) - if req.IncludeMetadata && row.Metadata != nil { - for k, v := range row.Metadata.AsMap() { - metadata[k] = v.(string) - } - } - - batch = append(batch, &pb.ForecastDatum{ - InitTimestamp: timestamppb.New(row.InitTimeUtc.Time), - LocationUuid: locationUuid.String(), - ForecasterFullname: fmt.Sprintf( - "%s:%s", - row.ForecasterName, - row.ForecasterVersion, - ), - HorizonMins: uint32(row.HorizonMins), - P50Fraction: float32(row.P50Sip) / 30000.0, - OtherStatisticsFractions: otherStatistics, - CreatedTimestampUtc: timestamppb.New(row.CreatedAtUtc.Time), - EffectiveCapacityWatts: uint64(row.CapacityWatts), - Metadata: metadata, - }) + batch = append( + batch, + mapStreamedForecastDatum(row, locationUuid, req.IncludeMetadata), + ) if len(batch) == batchSize { select { case resChan <- &pb.StreamForecastDataResponse{Values: batch}: @@ -667,10 +491,7 @@ func (s *DataPlatformDataServiceServerImpl) GetWeekAverageDeltas( querier := db.New(ix.GetTxFromContext(ctx)) // Get the location and source - locationUuid, err := uuid.Parse(req.LocationUuid) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "Invalid location UUID: %v", err) - } + locationUuid := uuid.MustParse(req.LocationUuid) gstprms := db.GetSourceAtTimestampParams{ GeometryUuid: locationUuid, @@ -737,16 +558,12 @@ func (s *DataPlatformDataServiceServerImpl) GetWeekAverageDeltas( } // Convert the deltas to the response format - deltas := make([]*pb.GetWeekAverageDeltasResponse_AverageDelta, len(dbDeltas)) - for i, delta := range dbDeltas { - deltas[i] = &pb.GetWeekAverageDeltasResponse_AverageDelta{ - DeltaFraction: float32(delta.AvgDeltaSip) / 30000.0, - HorizonMins: uint32(delta.HorizonMins), - EffectiveCapacityWatts: uint64( - dbSource.CapacityWatts, - ), // Should this be done over time? - } - } + deltas := MapSlice( + dbDeltas, + func(row db.GetWeekAverageDeltasForLocationsRow) *pb.GetWeekAverageDeltasResponse_AverageDelta { + return mapWeekAverageDelta(row, dbSource.CapacityWatts) + }, + ) return &pb.GetWeekAverageDeltasResponse{ Deltas: deltas, @@ -773,10 +590,7 @@ func (s *DataPlatformDataServiceServerImpl) GetObservationsAsTimeseries( ) } - start, end, err := timeWindowToPgWindow(req.TimeWindow) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "Invalid time window: %v", err) - } + start, end := timeWindowToPgWindow(req.TimeWindow) goprms := db.GetObservationsBetweenParams{ GeometryUuid: locationUuid, @@ -795,14 +609,7 @@ func (s *DataPlatformDataServiceServerImpl) GetObservationsAsTimeseries( ) } - values := make([]*pb.GetObservationsAsTimeseriesResponse_Value, len(dbObs)) - for i, obs := range dbObs { - values[i] = &pb.GetObservationsAsTimeseriesResponse_Value{ - ValueFraction: float32(obs.ValueSip) / 30000.0, - TimestampUtc: timestamppb.New(obs.ObservationTimestampUtc.Time), - EffectiveCapacityWatts: uint64(obs.CapacityWatts), - } - } + values := MapSlice(dbObs, mapObservationAsTimeseries) return &pb.GetObservationsAsTimeseriesResponse{ LocationUuid: locationUuid.String(), @@ -819,10 +626,7 @@ func (s *DataPlatformDataServiceServerImpl) CreateObservations( querier := db.New(ix.GetTxFromContext(ctx)) // Get the location and source - locationUuid, err := uuid.Parse(req.LocationUuid) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "Invalid location UUID: %v", err) - } + locationUuid := uuid.MustParse(req.LocationUuid) cfprms := db.GetSourceAtTimestampParams{ GeometryUuid: locationUuid, @@ -918,15 +722,7 @@ func (s *DataPlatformDataServiceServerImpl) GetLatestObservations( return nil, fmt.Errorf("backend communication error: %w", err) } - observations := make([]*pb.GetLatestObservationsResponse_Observation, len(dbObs)) - for i, obs := range dbObs { - observations[i] = &pb.GetLatestObservationsResponse_Observation{ - LocationUuid: obs.GeometryUuid.String(), - TimestampUtc: timestamppb.New(obs.ObservationTimestampUtc.Time), - ValueFraction: float32(obs.ValueSip) / 30000.0, - EffectiveCapacityWatts: uint64(obs.CapacityWatts), - } - } + observations := MapSlice(dbObs, mapLatestObservation) l.Debug(). Int16("dp.source.type_id", goprms.SourceTypeID). @@ -978,13 +774,7 @@ func (s *DataPlatformDataServiceServerImpl) ListObservers( Int("dp.observers.count", len(dbListObservers)). Msg("found observers") - observers := make([]*pb.ListObserversResponse_ObserverSummary, len(dbListObservers)) - for i, ob := range dbListObservers { - observers[i] = &pb.ListObserversResponse_ObserverSummary{ - ObserverUuid: ob.ObserverUuid.String(), - ObserverName: ob.ObserverName, - } - } + observers := MapSlice(dbListObservers, mapObserver) return &pb.ListObserversResponse{ Observers: observers, @@ -1047,48 +837,7 @@ func (s *DataPlatformDataServiceServerImpl) GetForecastAtTimestamp( ) } - values := make([]*pb.GetForecastAtTimestampResponse_Value, len(dbPredictions)) - for i, value := range dbPredictions { - otherStats := make(map[string]float32) - if value.P02Sip != nil { - otherStats["p02"] = float32(*value.P02Sip) / 30000.0 - } - - if value.P10Sip != nil { - otherStats["p10"] = float32(*value.P10Sip) / 30000.0 - } - - if value.P25Sip != nil { - otherStats["p25"] = float32(*value.P25Sip) / 30000.0 - } - - if value.P75Sip != nil { - otherStats["p75"] = float32(*value.P75Sip) / 30000.0 - } - - if value.P90Sip != nil { - otherStats["p90"] = float32(*value.P90Sip) / 30000.0 - } - - if value.P98Sip != nil { - otherStats["p98"] = float32(*value.P98Sip) / 30000.0 - } - - values[i] = &pb.GetForecastAtTimestampResponse_Value{ - ValueFraction: float32(value.P50Sip) / 30000.0, - EffectiveCapacityWatts: uint64(value.CapacityWatts), - LocationUuid: value.GeometryUuid.String(), - LocationName: value.GeometryName, - Latlng: &pb.LatLng{ - Latitude: value.Latitude, - Longitude: value.Longitude, - }, - Metadata: value.Metadata, - InitializationTimestampUtc: timestamppb.New(value.InitTimeUtc.Time), - CreatedTimestampUtc: timestamppb.New(value.CreatedAtUtc.Time), - OtherStatisticsFractions: otherStats, - } - } + values := MapSlice(dbPredictions, mapPredictionAtTime) return &pb.GetForecastAtTimestampResponse{ TimestampUtc: req.TimestampUtc, @@ -1139,18 +888,7 @@ func (s *DataPlatformDataServiceServerImpl) GetObservationsAtTimestamp( ) } - observations := make([]*pb.GetObservationsAtTimestampResponse_Value, len(dbObs)) - for i, obs := range dbObs { - observations[i] = &pb.GetObservationsAtTimestampResponse_Value{ - ValueFraction: float32(obs.ValueSip) / 30000.0, - EffectiveCapacityWatts: uint64(obs.CapacityWatts), - LocationUuid: obs.GeometryUuid.String(), - Latlng: &pb.LatLng{ - Latitude: obs.Latitude, - Longitude: obs.Longitude, - }, - } - } + observations := MapSlice(dbObs, mapObservationAtTimestamp) return &pb.GetObservationsAtTimestampResponse{ TimestampUtc: req.TimestampUtc, @@ -1245,14 +983,7 @@ func (s *DataPlatformDataServiceServerImpl) GetLocationAsTimeseries( ) } - values := make([]*pb.GetLocationAsTimeseriesResponse_LocationSnapshot, len(dbValues)) - for i, v := range dbValues { - values[i] = &pb.GetLocationAsTimeseriesResponse_LocationSnapshot{ - EffectiveCapacityWatts: uint64(v.CapacityWatts), - TimestampUtc: timestamppb.New(v.ValidFromUtc.Time), - Metadata: v.Metadata, - } - } + values := MapSlice(dbValues, mapLocationSnapshot) return &pb.GetLocationAsTimeseriesResponse{ Values: values, @@ -1344,15 +1075,6 @@ func (s *DataPlatformDataServiceServerImpl) UpdateLocation( l := zerolog.Ctx(ctx) querier := db.New(ix.GetTxFromContext(ctx)) - if req.NewEffectiveCapacityWatts == nil && - req.NewLocationName == nil && - req.NewMetadata == nil { - return nil, status.Error( - codes.InvalidArgument, - "At least one of new effective capacity, new location name, or new metadata must be provided.", - ) - } - // Set the valid from time to now if not provided validFrom := time.Now().UTC().Truncate(time.Minute) if req.ValidFromUtc != nil { @@ -1505,10 +1227,7 @@ func (s *DataPlatformDataServiceServerImpl) GetLocationsAsGeoJSON( locationUuids := make([]uuid.UUID, len(req.LocationUuids)) for i, id := range req.LocationUuids { - locationUuids[i], err = uuid.Parse(id) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "Invalid location UUID: %v", err) - } + locationUuids[i] = uuid.MustParse(id) } ggprms := db.GetGeometryGeoJSONParams{ @@ -1568,45 +1287,7 @@ func (s *DataPlatformDataServiceServerImpl) GetForecastAsTimeseries( return nil, fmt.Errorf("no forecasts found for the given parameters: %w", err) } - out := make([]*pb.GetForecastAsTimeseriesResponse_Value, len(dbPreds)) - for i, pred := range dbPreds { - otherStats := make(map[string]float32) - if pred.P02Sip != nil { - otherStats["p02"] = float32(*pred.P02Sip) / 30000.0 - } - - if pred.P10Sip != nil { - otherStats["p10"] = float32(*pred.P10Sip) / 30000.0 - } - - if pred.P25Sip != nil { - otherStats["p25"] = float32(*pred.P25Sip) / 30000.0 - } - - if pred.P75Sip != nil { - otherStats["p75"] = float32(*pred.P75Sip) / 30000.0 - } - - if pred.P90Sip != nil { - otherStats["p90"] = float32(*pred.P90Sip) / 30000.0 - } - - if pred.P98Sip != nil { - otherStats["p98"] = float32(*pred.P98Sip) / 30000.0 - } - - out[i] = &pb.GetForecastAsTimeseriesResponse_Value{ - TargetTimestampUtc: timestamppb.New( - pred.InitTimeUtc.Time.Add(time.Duration(pred.HorizonMins) * time.Minute), - ), - P50ValueFraction: float32(pred.P50Sip) / 30000.0, - EffectiveCapacityWatts: uint64(pred.CapacityWatts), - InitializationTimestampUtc: timestamppb.New(pred.InitTimeUtc.Time), - CreatedTimestampUtc: timestamppb.New(pred.CreatedAtUtc.Time), - OtherStatisticsFractions: otherStats, - Metadata: pred.Metadata, - } - } + out := MapSlice(dbPreds, mapForecastAsTimeseriesFromForecastValue) return &pb.GetForecastAsTimeseriesResponse{ LocationUuid: req.LocationUuid, @@ -1628,10 +1309,7 @@ func (s *DataPlatformDataServiceServerImpl) GetForecastAsTimeseries( } // Get the predictions for the given location source - start, end, err := timeWindowToPgWindow(req.TimeWindow) - if err != nil { - return nil, status.Errorf(codes.InvalidArgument, "Invalid time window: %v", err) - } + start, end := timeWindowToPgWindow(req.TimeWindow) pivotTime := pgtype.Timestamp{Valid: false} if req.PivotTimestampUtc != nil { @@ -1674,43 +1352,7 @@ func (s *DataPlatformDataServiceServerImpl) GetForecastAsTimeseries( Msg(fmt.Sprintf("found %d predictions", len(dbValues))) } - values := make([]*pb.GetForecastAsTimeseriesResponse_Value, len(dbValues)) - for i, value := range dbValues { - otherStats := make(map[string]float32) - if value.P02Sip != nil { - otherStats["p02"] = float32(*value.P02Sip) / 30000.0 - } - - if value.P10Sip != nil { - otherStats["p10"] = float32(*value.P10Sip) / 30000.0 - } - - if value.P25Sip != nil { - otherStats["p25"] = float32(*value.P25Sip) / 30000.0 - } - - if value.P75Sip != nil { - otherStats["p75"] = float32(*value.P75Sip) / 30000.0 - } - - if value.P90Sip != nil { - otherStats["p90"] = float32(*value.P90Sip) / 30000.0 - } - - if value.P98Sip != nil { - otherStats["p98"] = float32(*value.P98Sip) / 30000.0 - } - - values[i] = &pb.GetForecastAsTimeseriesResponse_Value{ - TargetTimestampUtc: timestamppb.New(value.TargetTimeUtc.Time), - P50ValueFraction: float32(value.P50Sip) / 30000.0, - OtherStatisticsFractions: otherStats, - EffectiveCapacityWatts: uint64(value.CapacityWatts), - InitializationTimestampUtc: timestamppb.New(value.InitTimeUtc.Time), - CreatedTimestampUtc: timestamppb.New(value.CreatedAtUtc.Time), - Metadata: value.Metadata, - } - } + values := MapSlice(dbValues, mapForecastAsTimeseriesFromLocationValue) return &pb.GetForecastAsTimeseriesResponse{ LocationUuid: dbSource.GeometryUuid.String(), @@ -1762,18 +1404,10 @@ func (s *DataPlatformDataServiceServerImpl) ListLocations( } for _, loc := range glResp { - locations = append(locations, &pb.ListLocationsResponse_LocationSummary{ - LocationUuid: loc.GeometryUuid.String(), - LocationName: loc.GeometryName, - Latlng: &pb.LatLng{ - Latitude: loc.Latitude, - Longitude: loc.Longitude, - }, - EffectiveCapacityWatts: uint64(loc.CapacityWatts), - EnergySource: pb.EnergySource(loc.SourceTypeID), - LocationType: pb.LocationType(loc.GeometryTypeID), - Metadata: loc.MetadataJsonb, - }) + locations = append(locations, mapLocationSummary( + loc.GeometryUuid, loc.GeometryName, loc.Latitude, loc.Longitude, + loc.CapacityWatts, loc.SourceTypeID, loc.GeometryTypeID, loc.MetadataJsonb, + )) } } else if req.EnclosedLocationUuidFilter != nil { llprms := db.ListSourcesAtTimestampWithoutParams{ @@ -1791,18 +1425,10 @@ func (s *DataPlatformDataServiceServerImpl) ListLocations( } for _, loc := range glResp { - locations = append(locations, &pb.ListLocationsResponse_LocationSummary{ - LocationUuid: loc.GeometryUuid.String(), - LocationName: loc.GeometryName, - Latlng: &pb.LatLng{ - Latitude: loc.Latitude, - Longitude: loc.Longitude, - }, - EffectiveCapacityWatts: uint64(loc.CapacityWatts), - EnergySource: pb.EnergySource(loc.SourceTypeID), - LocationType: pb.LocationType(loc.GeometryTypeID), - Metadata: loc.MetadataJsonb, - }) + locations = append(locations, mapLocationSummary( + loc.GeometryUuid, loc.GeometryName, loc.Latitude, loc.Longitude, + loc.CapacityWatts, loc.SourceTypeID, loc.GeometryTypeID, loc.MetadataJsonb, + )) } } else { lsprms := db.ListSourcesAtTimestampParams{ @@ -1820,18 +1446,10 @@ func (s *DataPlatformDataServiceServerImpl) ListLocations( } for _, loc := range glResp { - locations = append(locations, &pb.ListLocationsResponse_LocationSummary{ - LocationUuid: loc.GeometryUuid.String(), - LocationName: loc.GeometryName, - Latlng: &pb.LatLng{ - Latitude: loc.Latitude, - Longitude: loc.Longitude, - }, - EffectiveCapacityWatts: uint64(loc.CapacityWatts), - EnergySource: pb.EnergySource(loc.SourceTypeID), - LocationType: pb.LocationType(loc.GeometryTypeID), - Metadata: loc.MetadataJsonb, - }) + locations = append(locations, mapLocationSummary( + loc.GeometryUuid, loc.GeometryName, loc.Latitude, loc.Longitude, + loc.CapacityWatts, loc.SourceTypeID, loc.GeometryTypeID, loc.MetadataJsonb, + )) } } @@ -2002,7 +1620,7 @@ func (s *DataPlatformDataServiceServerImpl) StreamCreateForecasts( sourceCache[sKey] = sInfo } - fParams, err := prepareForecastParams( + fParams, err := mapCreateForecast( req, sInfo.geometryUuid, sKey.sourceTypeId, diff --git a/internal/server/postgres/dataserverimpl_test.go b/internal/server/postgres/dataserverimpl_test.go index 1d4ddbc..efa7070 100644 --- a/internal/server/postgres/dataserverimpl_test.go +++ b/internal/server/postgres/dataserverimpl_test.go @@ -2691,7 +2691,7 @@ func TestPrepareForecastParams(t *testing.T) { for _, tc := range testcases { t.Run(tc.name, func(t *testing.T) { - params, err := prepareForecastParams(tc.req, geomID, sourceID, forecasterID) + params, err := mapCreateForecast(tc.req, geomID, sourceID, forecasterID) if tc.shouldErr { require.Error(t, err) return diff --git a/internal/server/postgres/mappers.go b/internal/server/postgres/mappers.go new file mode 100644 index 0000000..bcf935a --- /dev/null +++ b/internal/server/postgres/mappers.go @@ -0,0 +1,380 @@ +package postgres + +import ( + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" + + pb "github.com/openclimatefix/data-platform/internal/gen/ocf/dp" + db "github.com/openclimatefix/data-platform/internal/server/postgres/gen" +) + +// MapSlice transforms a slice of type T into a slice of type U using a mapping function. +func MapSlice[T, U any](input []T, mapper func(T) U) []U { + if input == nil { + return nil + } + + out := make([]U, len(input)) + for i, v := range input { + out[i] = mapper(v) + } + + return out +} + +// timeWindowToPgWindow converts a TimeWindow protobuf message to a pair of pgtype.Timestamp values. +// If the TimeWindow is nil or its StartTimestampUtc is nil, it defaults to a window from 48 hours ago to 36 hours in the future. Protovalidate ensures at the boundary that the start is always before the end, so we don't need to check that here. +func timeWindowToPgWindow( + window *pb.TimeWindow, +) (start pgtype.Timestamp, end pgtype.Timestamp) { + currentTime := time.Now().UTC() + if window == nil || window.StartTimestampUtc == nil { + start = pgtype.Timestamp{Time: currentTime.Add(-48 * time.Hour), Valid: true} + end = pgtype.Timestamp{Time: currentTime.Add(36 * time.Hour), Valid: true} + } else { + start = pgtype.Timestamp{Time: window.StartTimestampUtc.AsTime(), Valid: true} + end = pgtype.Timestamp{Time: window.EndTimestampUtc.AsTime(), Valid: true} + } + + return start, end +} + +// timeptrToPgTimestamp converts a protobuf Timestamp pointer to a pgtype.Timestamp. +// If the pointer is nil, it returns the current time truncated to the nearest minute. +func timeptrToPgTimestamp(t *timestamppb.Timestamp) pgtype.Timestamp { + if t == nil { + return pgtype.Timestamp{ + Time: time.Now().UTC().Truncate(time.Minute), + Valid: true, + } + } + + return pgtype.Timestamp{Time: t.AsTime().UTC(), Valid: true} +} + +// extractSIPStatPtrFromMap gets a key's value from a map as a pointer, and converts it +// to a smallint percentage. If it doesn't exist, it returns nil. +func extractSIPStatPtrFromMap(m map[string]float32, key string) *int16 { + val, exists := m[key] + if !exists { + return nil + } + + sip_val := int16(val * 30000.0) + + return &sip_val +} + +// sipToFraction converts a SIP value to a fraction. +func sipToFraction(sip int16) float32 { + return float32(sip) / 30000.0 +} + +// buildOtherStatsMap constructs a map of other statistics from optional SIP pointers. +// Only keys that are not nil will be included in the returned map. +func buildOtherStatsMap(p02, p10, p25, p75, p90, p98 *int16) map[string]float32 { + otherStats := make(map[string]float32) + if p02 != nil { + otherStats["p02"] = sipToFraction(*p02) + } + + if p10 != nil { + otherStats["p10"] = sipToFraction(*p10) + } + + if p25 != nil { + otherStats["p25"] = sipToFraction(*p25) + } + + if p75 != nil { + otherStats["p75"] = sipToFraction(*p75) + } + + if p90 != nil { + otherStats["p90"] = sipToFraction(*p90) + } + + if p98 != nil { + otherStats["p98"] = sipToFraction(*p98) + } + + return otherStats +} + +// mapCreateForecast generates the database parameters for a single forecast from a gRPC request. +func mapCreateForecast( + req *pb.CreateForecastRequest, + geometryUuid uuid.UUID, + sourceTypeId int16, + forecasterId int32, +) (db.CreateForecastsParams, error) { + initTime := req.InitTimeUtc.AsTime().Truncate(time.Minute) + + fUuid, err := uuid.NewV7() + if err != nil { + return db.CreateForecastsParams{}, fmt.Errorf("failed to generate uuidv7: %w", err) + } + + // Manually overwrite the 48-bit timestamp with the initTime milliseconds + ms := uint64(initTime.UnixMilli()) + fUuid[0] = byte(ms >> 40) + fUuid[1] = byte(ms >> 32) + fUuid[2] = byte(ms >> 24) + fUuid[3] = byte(ms >> 16) + fUuid[4] = byte(ms >> 8) + fUuid[5] = byte(ms) + + firstHorizon := int32(req.Values[0].HorizonMins) + lastHorizon := int32(req.Values[len(req.Values)-1].HorizonMins) + + periodStart := initTime.Add(time.Duration(firstHorizon) * time.Minute) + periodEnd := initTime.Add(time.Duration(lastHorizon) * time.Minute) + + targetPeriod := pgtype.Range[pgtype.Timestamp]{ + Lower: pgtype.Timestamp{Time: periodStart, Valid: true}, + Upper: pgtype.Timestamp{Time: periodEnd, Valid: true}, + LowerType: pgtype.Inclusive, + UpperType: pgtype.Inclusive, + Valid: true, + } + + var createdTime pgtype.Timestamp + if req.CreatedTimestampUtc != nil { + createdTime = pgtype.Timestamp{Time: req.CreatedTimestampUtc.AsTime(), Valid: true} + } else { + createdTime = pgtype.Timestamp{ + Time: time.Now().UTC().Truncate(time.Minute), + Valid: true, + } + } + + return db.CreateForecastsParams{ + ForecastUuid: fUuid, + GeometryUuid: geometryUuid, + SourceTypeID: sourceTypeId, + ForecasterID: forecasterId, + InitTimeUtc: pgtype.Timestamp{Time: initTime, Valid: true}, + ValueResolutionMins: int16(req.Values[1].HorizonMins - req.Values[0].HorizonMins), + TargetPeriod: targetPeriod, + Metadata: req.Metadata, + CreatedAtUtc: createdTime, + }, nil +} + +func mapLatestForecast( + fc db.GetLatestForecastsAtHorizonSincePivotRow, +) *pb.GetLatestForecastsResponse_Forecast { + return &pb.GetLatestForecastsResponse_Forecast{ + InitializationTimestampUtc: timestamppb.New(fc.InitTimeUtc.Time), + Forecaster: &pb.Forecaster{ + ForecasterName: fc.ForecasterName, + ForecasterVersion: fc.ForecasterVersion, + }, + LocationUuid: fc.GeometryUuid.String(), + Metadata: fc.Metadata, + CreatedTimestampUtc: timestamppb.New(fc.CreatedAtUtc.Time), + } +} + +func mapForecaster(fc db.GetForecastersByFiltersRow) *pb.Forecaster { + return &pb.Forecaster{ + ForecasterName: fc.ForecasterName, + ForecasterVersion: fc.ForecasterVersion, + } +} + +func mapWeekAverageDelta( + delta db.GetWeekAverageDeltasForLocationsRow, + capacityWatts int64, +) *pb.GetWeekAverageDeltasResponse_AverageDelta { + return &pb.GetWeekAverageDeltasResponse_AverageDelta{ + DeltaFraction: float32(delta.AvgDeltaSip) / 30000.0, + HorizonMins: uint32(delta.HorizonMins), + EffectiveCapacityWatts: uint64(capacityWatts), + } +} + +func mapObservationAsTimeseries( + obs db.GetObservationsBetweenRow, +) *pb.GetObservationsAsTimeseriesResponse_Value { + return &pb.GetObservationsAsTimeseriesResponse_Value{ + ValueFraction: sipToFraction(obs.ValueSip), + TimestampUtc: timestamppb.New(obs.ObservationTimestampUtc.Time), + EffectiveCapacityWatts: uint64(obs.CapacityWatts), + } +} + +func mapLatestObservation( + obs db.GetLatestObservationsRow, +) *pb.GetLatestObservationsResponse_Observation { + return &pb.GetLatestObservationsResponse_Observation{ + LocationUuid: obs.GeometryUuid.String(), + TimestampUtc: timestamppb.New(obs.ObservationTimestampUtc.Time), + ValueFraction: sipToFraction(obs.ValueSip), + EffectiveCapacityWatts: uint64(obs.CapacityWatts), + } +} + +func mapObserver(ob db.ObsObserver) *pb.ListObserversResponse_ObserverSummary { + return &pb.ListObserversResponse_ObserverSummary{ + ObserverUuid: ob.ObserverUuid.String(), + ObserverName: ob.ObserverName, + } +} + +func mapPredictionAtTime( + value db.ListPredictionsAtTimeForLocationsRow, +) *pb.GetForecastAtTimestampResponse_Value { + return &pb.GetForecastAtTimestampResponse_Value{ + ValueFraction: sipToFraction(value.P50Sip), + EffectiveCapacityWatts: uint64(value.CapacityWatts), + LocationUuid: value.GeometryUuid.String(), + LocationName: value.GeometryName, + Latlng: &pb.LatLng{ + Latitude: value.Latitude, + Longitude: value.Longitude, + }, + Metadata: value.Metadata, + InitializationTimestampUtc: timestamppb.New(value.InitTimeUtc.Time), + CreatedTimestampUtc: timestamppb.New(value.CreatedAtUtc.Time), + OtherStatisticsFractions: buildOtherStatsMap( + value.P02Sip, + value.P10Sip, + value.P25Sip, + value.P75Sip, + value.P90Sip, + value.P98Sip, + ), + } +} + +func mapObservationAtTimestamp( + obs db.ListObservationsAtTimeForLocationsRow, +) *pb.GetObservationsAtTimestampResponse_Value { + return &pb.GetObservationsAtTimestampResponse_Value{ + ValueFraction: sipToFraction(obs.ValueSip), + EffectiveCapacityWatts: uint64(obs.CapacityWatts), + LocationUuid: obs.GeometryUuid.String(), + Latlng: &pb.LatLng{ + Latitude: obs.Latitude, + Longitude: obs.Longitude, + }, + } +} + +func mapLocationSnapshot( + v db.GetSourceHistoryRow, +) *pb.GetLocationAsTimeseriesResponse_LocationSnapshot { + return &pb.GetLocationAsTimeseriesResponse_LocationSnapshot{ + EffectiveCapacityWatts: uint64(v.CapacityWatts), + TimestampUtc: timestamppb.New(v.ValidFromUtc.Time), + Metadata: v.Metadata, + } +} + +func mapForecastAsTimeseriesFromForecastValue( + pred db.ListPredictionsForForecastsRow, +) *pb.GetForecastAsTimeseriesResponse_Value { + return &pb.GetForecastAsTimeseriesResponse_Value{ + TargetTimestampUtc: timestamppb.New( + pred.InitTimeUtc.Time.Add(time.Duration(pred.HorizonMins) * time.Minute), + ), + P50ValueFraction: sipToFraction(pred.P50Sip), + EffectiveCapacityWatts: uint64(pred.CapacityWatts), + InitializationTimestampUtc: timestamppb.New(pred.InitTimeUtc.Time), + CreatedTimestampUtc: timestamppb.New(pred.CreatedAtUtc.Time), + OtherStatisticsFractions: buildOtherStatsMap( + pred.P02Sip, + pred.P10Sip, + pred.P25Sip, + pred.P75Sip, + pred.P90Sip, + pred.P98Sip, + ), + Metadata: pred.Metadata, + } +} + +func mapForecastAsTimeseriesFromLocationValue( + value db.ListPredictionsForLocationRow, +) *pb.GetForecastAsTimeseriesResponse_Value { + return &pb.GetForecastAsTimeseriesResponse_Value{ + TargetTimestampUtc: timestamppb.New(value.TargetTimeUtc.Time), + P50ValueFraction: sipToFraction(value.P50Sip), + OtherStatisticsFractions: buildOtherStatsMap( + value.P02Sip, + value.P10Sip, + value.P25Sip, + value.P75Sip, + value.P90Sip, + value.P98Sip, + ), + EffectiveCapacityWatts: uint64(value.CapacityWatts), + InitializationTimestampUtc: timestamppb.New(value.InitTimeUtc.Time), + CreatedTimestampUtc: timestamppb.New(value.CreatedAtUtc.Time), + Metadata: value.Metadata, + } +} + +func mapLocationSummary( + geomUuid uuid.UUID, + geomName string, + lat, lon float32, + cap int64, + srcType, geomType int16, + meta *structpb.Struct, +) *pb.ListLocationsResponse_LocationSummary { + return &pb.ListLocationsResponse_LocationSummary{ + LocationUuid: geomUuid.String(), + LocationName: geomName, + Latlng: &pb.LatLng{ + Latitude: lat, + Longitude: lon, + }, + EffectiveCapacityWatts: uint64(cap), + EnergySource: pb.EnergySource(srcType), + LocationType: pb.LocationType(geomType), + Metadata: meta, + } +} + +func mapStreamedForecastDatum( + row db.ListPredictionsForForecastsRow, + locUuid uuid.UUID, + includeMetadata bool, +) *pb.ForecastDatum { + metadata := make(map[string]string) + if includeMetadata && row.Metadata != nil { + for k, v := range row.Metadata.AsMap() { + metadata[k] = v.(string) + } + } + + return &pb.ForecastDatum{ + InitTimestamp: timestamppb.New(row.InitTimeUtc.Time), + LocationUuid: locUuid.String(), + ForecasterFullname: fmt.Sprintf( + "%s:%s", + row.ForecasterName, + row.ForecasterVersion, + ), + HorizonMins: uint32(row.HorizonMins), + P50Fraction: sipToFraction(row.P50Sip), + OtherStatisticsFractions: buildOtherStatsMap( + row.P02Sip, + row.P10Sip, + row.P25Sip, + row.P75Sip, + row.P90Sip, + row.P98Sip, + ), + CreatedTimestampUtc: timestamppb.New(row.CreatedAtUtc.Time), + EffectiveCapacityWatts: uint64(row.CapacityWatts), + Metadata: metadata, + } +} diff --git a/internal/server/postgres/mappers_test.go b/internal/server/postgres/mappers_test.go new file mode 100644 index 0000000..ec38127 --- /dev/null +++ b/internal/server/postgres/mappers_test.go @@ -0,0 +1,452 @@ +package postgres + +import ( + "strconv" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" + + pb "github.com/openclimatefix/data-platform/internal/gen/ocf/dp" + db "github.com/openclimatefix/data-platform/internal/server/postgres/gen" +) + +func Test_MapSlice(t *testing.T) { + tests := []struct { + name string + input []int + expected []string + }{ + { + name: "nil input returns nil", + input: nil, + expected: nil, + }, + { + name: "empty slice returns empty slice", + input: []int{}, + expected: []string{}, + }, + { + name: "populated slice maps correctly", + input: []int{1, 2, 3}, + expected: []string{"1", "2", "3"}, + }, + } + + mapper := func(i int) string { return strconv.Itoa(i) } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := MapSlice(tt.input, mapper) + require.Equal(t, tt.expected, res) + }) + } +} + +func Test_timeptrToPgTimestamp(t *testing.T) { + now := time.Now().UTC() + tests := []struct { + name string + input *timestamppb.Timestamp + validateResult func(*testing.T, pgtype.Timestamp) + }{ + { + name: "nil input returns truncated current time", + input: nil, + validateResult: func(t *testing.T, res pgtype.Timestamp) { + require.True(t, res.Valid) + // It should be within a couple of seconds of Now().Truncate(time.Minute) + expected := time.Now().UTC().Truncate(time.Minute) + require.WithinDuration(t, expected, res.Time, 2*time.Second) + }, + }, + { + name: "valid input maps exactly", + input: timestamppb.New(now), + validateResult: func(t *testing.T, res pgtype.Timestamp) { + require.True(t, res.Valid) + require.Equal(t, now, res.Time) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := timeptrToPgTimestamp(tt.input) + tt.validateResult(t, res) + }) + } +} + +func Test_extractSIPStatPtrFromMap(t *testing.T) { + m := map[string]float32{"p10": 0.1, "p90": 0.9} + + tests := []struct { + name string + inputMap map[string]float32 + key string + expected *int16 + }{ + { + name: "nil map returns nil", + inputMap: nil, + key: "p10", + expected: nil, + }, + { + name: "missing key returns nil", + inputMap: m, + key: "p50", + expected: nil, + }, + { + name: "existing key returns sip value", + inputMap: m, + key: "p10", + expected: func() *int16 { v := int16(3000); return &v }(), + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := extractSIPStatPtrFromMap(tt.inputMap, tt.key) + if tt.expected == nil { + require.Nil(t, res) + } else { + require.NotNil(t, res) + require.Equal(t, *tt.expected, *res) + } + }) + } +} + +func Test_sipToFraction(t *testing.T) { + tests := []struct { + name string + input int16 + expected float32 + }{ + { + name: "zero", + input: 0, + expected: 0.0, + }, + { + name: "max positive", + input: 30000, + expected: 1.0, + }, + { + name: "max negative", + input: -30000, + expected: -1.0, + }, + { + name: "half", + input: 15000, + expected: 0.5, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := sipToFraction(tt.input) + require.InDelta(t, tt.expected, res, 0.0001) + }) + } +} + +func Test_buildOtherStatsMap(t *testing.T) { + v10 := int16(3000) + v90 := int16(27000) + v25 := int16(7500) + + tests := []struct { + name string + p02 *int16 + p10 *int16 + p25 *int16 + p75 *int16 + p90 *int16 + p98 *int16 + expected map[string]float32 + }{ + { + name: "all nil returns empty map", + expected: map[string]float32{}, + }, + { + name: "partially populated", + p10: &v10, + p90: &v90, + expected: map[string]float32{ + "p10": 0.1, + "p90": 0.9, + }, + }, + { + name: "fully populated", + p02: &v10, // Just using v10 for convenience + p10: &v10, + p25: &v25, + p75: &v25, + p90: &v90, + p98: &v90, + expected: map[string]float32{ + "p02": 0.1, + "p10": 0.1, + "p25": 0.25, + "p75": 0.25, + "p90": 0.9, + "p98": 0.9, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := buildOtherStatsMap(tt.p02, tt.p10, tt.p25, tt.p75, tt.p90, tt.p98) + require.Equal(t, tt.expected, res) + }) + } +} + +func Test_timeWindowToPgWindow(t *testing.T) { + now := time.Now().UTC() + startTs := timestamppb.New(now.Add(-2 * time.Hour)) + endTs := timestamppb.New(now.Add(2 * time.Hour)) + + tests := []struct { + name string + input *pb.TimeWindow + validateResult func(*testing.T, pgtype.Timestamp, pgtype.Timestamp) + }{ + { + name: "nil window applies defaults", + input: nil, + validateResult: func(t *testing.T, start pgtype.Timestamp, end pgtype.Timestamp) { + require.True(t, start.Valid) + require.True(t, end.Valid) + require.WithinDuration(t, now.Add(-48*time.Hour), start.Time, 5*time.Second) + require.WithinDuration(t, now.Add(36*time.Hour), end.Time, 5*time.Second) + }, + }, + { + name: "start timestamp nil applies defaults", + input: &pb.TimeWindow{ + EndTimestampUtc: endTs, + }, // Assuming protovalidate let this slip somehow + validateResult: func(t *testing.T, start pgtype.Timestamp, end pgtype.Timestamp) { + require.True(t, start.Valid) + require.True(t, end.Valid) + require.WithinDuration(t, now.Add(-48*time.Hour), start.Time, 5*time.Second) + require.WithinDuration(t, now.Add(36*time.Hour), end.Time, 5*time.Second) + }, + }, + { + name: "perfectly populated window", + input: &pb.TimeWindow{ + StartTimestampUtc: startTs, + EndTimestampUtc: endTs, + }, + validateResult: func(t *testing.T, start pgtype.Timestamp, end pgtype.Timestamp) { + require.True(t, start.Valid) + require.True(t, end.Valid) + require.Equal(t, startTs.AsTime(), start.Time) + require.Equal(t, endTs.AsTime(), end.Time) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + start, end := timeWindowToPgWindow(tt.input) + tt.validateResult(t, start, end) + }) + } +} + +func Test_mapLocationSummary(t *testing.T) { + id := uuid.New() + meta, _ := structpb.NewStruct(map[string]interface{}{"key": "value"}) + + tests := []struct { + name string + metadata *structpb.Struct + validateResult func(*testing.T, *pb.ListLocationsResponse_LocationSummary) + }{ + { + name: "valid metadata", + metadata: meta, + validateResult: func(t *testing.T, res *pb.ListLocationsResponse_LocationSummary) { + require.NotNil(t, res.Metadata) + require.Equal(t, "value", res.Metadata.Fields["key"].GetStringValue()) + }, + }, + { + name: "nil metadata", + metadata: nil, + validateResult: func(t *testing.T, res *pb.ListLocationsResponse_LocationSummary) { + require.Nil(t, res.Metadata) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := mapLocationSummary( + id, + "Test Location", + 51.5, + -0.1, + 1000, + 1, + 1, + tt.metadata, + ) + + require.Equal(t, id.String(), res.LocationUuid) + require.Equal(t, "Test Location", res.LocationName) + require.Equal(t, float32(51.5), res.Latlng.Latitude) + require.Equal(t, float32(-0.1), res.Latlng.Longitude) + require.Equal(t, uint64(1000), res.EffectiveCapacityWatts) + require.Equal(t, pb.EnergySource(1), res.EnergySource) + require.Equal(t, pb.LocationType(1), res.LocationType) + + tt.validateResult(t, res) + }) + } +} + +func Test_mapStreamedForecastDatum(t *testing.T) { + id := uuid.New() + meta, _ := structpb.NewStruct(map[string]interface{}{"key": "value"}) + + baseRow := db.ListPredictionsForForecastsRow{ + ForecasterName: "test", + ForecasterVersion: "1.0", + HorizonMins: 60, + P50Sip: 15000, + CapacityWatts: 1000, + InitTimeUtc: pgtype.Timestamp{Time: time.Now().UTC(), Valid: true}, + CreatedAtUtc: pgtype.Timestamp{Time: time.Now().UTC(), Valid: true}, + Metadata: meta, + } + + tests := []struct { + name string + includeMetadata bool + row db.ListPredictionsForForecastsRow + validateResult func(*testing.T, *pb.ForecastDatum) + }{ + { + name: "includeMetadata false, DB has metadata", + includeMetadata: false, + row: baseRow, + validateResult: func(t *testing.T, res *pb.ForecastDatum) { + require.NotNil(t, res.Metadata) + require.Empty(t, res.Metadata) + }, + }, + { + name: "includeMetadata true, DB metadata is nil", + includeMetadata: true, + row: func() db.ListPredictionsForForecastsRow { + r := baseRow + r.Metadata = nil + return r + }(), + validateResult: func(t *testing.T, res *pb.ForecastDatum) { + require.NotNil(t, res.Metadata) + require.Empty(t, res.Metadata) + }, + }, + { + name: "includeMetadata true, DB has metadata", + includeMetadata: true, + row: baseRow, + validateResult: func(t *testing.T, res *pb.ForecastDatum) { + require.NotNil(t, res.Metadata) + require.Len(t, res.Metadata, 1) + require.Equal(t, "value", res.Metadata["key"]) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := mapStreamedForecastDatum(tt.row, id, tt.includeMetadata) + + require.Equal(t, id.String(), res.LocationUuid) + require.Equal(t, "test:1.0", res.ForecasterFullname) + require.Equal(t, uint32(60), res.HorizonMins) + require.InDelta(t, 0.5, res.P50Fraction, 0.0001) + + tt.validateResult(t, res) + }) + } +} + +func Test_mapForecastAsTimeseriesFromLocationValue(t *testing.T) { + p50 := int16(15000) + p10 := int16(3000) + p90 := int16(27000) + + initTime := time.Now().UTC() + targetTime := initTime.Add(time.Hour) + + baseRow := db.ListPredictionsForLocationRow{ + P50Sip: p50, + CapacityWatts: 10000, + TargetTimeUtc: pgtype.Timestamp{Time: targetTime, Valid: true}, + InitTimeUtc: pgtype.Timestamp{Time: initTime, Valid: true}, + CreatedAtUtc: pgtype.Timestamp{Time: initTime, Valid: true}, + } + + tests := []struct { + name string + row db.ListPredictionsForLocationRow + validateResult func(*testing.T, *pb.GetForecastAsTimeseriesResponse_Value) + }{ + { + name: "sparse row only p50", + row: baseRow, + validateResult: func(t *testing.T, v *pb.GetForecastAsTimeseriesResponse_Value) { + require.InDelta(t, 0.5, v.P50ValueFraction, 0.0001) + require.Empty(t, v.OtherStatisticsFractions) + }, + }, + { + name: "fully populated row", + row: func() db.ListPredictionsForLocationRow { + r := baseRow + r.P10Sip = &p10 + r.P90Sip = &p90 + return r + }(), + validateResult: func(t *testing.T, v *pb.GetForecastAsTimeseriesResponse_Value) { + require.InDelta(t, 0.5, v.P50ValueFraction, 0.0001) + require.Len(t, v.OtherStatisticsFractions, 2) + require.InDelta(t, 0.1, v.OtherStatisticsFractions["p10"], 0.0001) + require.InDelta(t, 0.9, v.OtherStatisticsFractions["p90"], 0.0001) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + res := mapForecastAsTimeseriesFromLocationValue(tt.row) + + require.Equal(t, uint64(10000), res.EffectiveCapacityWatts) + require.Equal(t, targetTime, res.TargetTimestampUtc.AsTime()) + require.Equal(t, initTime, res.InitializationTimestampUtc.AsTime()) + + tt.validateResult(t, res) + }) + } +} diff --git a/internal/server/postgres/sql/migrations/00010_7_plevels.sql b/internal/server/postgres/sql/migrations/00010_7_plevels.sql index 3e22968..fce4033 100644 --- a/internal/server/postgres/sql/migrations/00010_7_plevels.sql +++ b/internal/server/postgres/sql/migrations/00010_7_plevels.sql @@ -18,5 +18,5 @@ ALTER TABLE pred.predicted_generation_values ALTER TABLE pred.predicted_generation_values DROP COLUMN p02_sip, DROP COLUMN p25_sip, - DROP COLUMN p75_sip; - DROP COLUMN p98_sip, + DROP COLUMN p75_sip, + DROP COLUMN p98_sip; diff --git a/proto/ocf/dp/dp-data.messages.proto b/proto/ocf/dp/dp-data.messages.proto index bb9bb04..66671b6 100644 --- a/proto/ocf/dp/dp-data.messages.proto +++ b/proto/ocf/dp/dp-data.messages.proto @@ -44,6 +44,12 @@ message TimeWindow { message: "start_timestamp_utc must be before end_timestamp_utc" expression: "this.start_timestamp_utc <= this.end_timestamp_utc" }; + + option (buf.validate.message).cel = { + id: "both_or_neither_timestamps" + message: "Both start and end timestamps must be provided or neither" + expression: "(has(this.start_timestamp_utc) && has(this.end_timestamp_utc)) || (!has(this.start_timestamp_utc) && !has(this.end_timestamp_utc))" + }; } @@ -547,12 +553,12 @@ message UpdateLocationRequest { // Ensure one of the updatable fields is set. option (buf.validate.message).cel = { - id: "at_least_one_field" + id: "has_update_field" message: "at least one updatable field must be set" expression: "has(this.new_location_name) " "|| has(this.new_effective_capacity_watts) " - "|| has(this.new_metadata) ? true : false" + "|| has(this.new_metadata)" }; } @@ -584,8 +590,8 @@ message GetLocationsAsGeoJSONRequest { (buf.validate.field).repeated.max_items = 1000, (buf.validate.field).repeated.unique = true, (buf.validate.field).repeated.items = { - string: {uuid: true} - } + string: {uuid: true} + } ]; /* If true, the GeoJSON will not be simplified. * Defaults to false if not set to reduce response size.