From 1809df79f10aa74d1acc0990c6ae187ddcf8ca73 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Wed, 29 Jul 2026 02:57:43 +0530 Subject: [PATCH 1/3] feat(go): Add mappers --- internal/server/postgres/dataserverimpl.go | 346 ++---------------- internal/server/postgres/mappers.go | 256 +++++++++++++ internal/server/postgres/mappers_test.go | 128 +++++++ .../sql/migrations/00010_7_plevels.sql | 4 +- 4 files changed, 412 insertions(+), 322 deletions(-) create mode 100644 internal/server/postgres/mappers.go create mode 100644 internal/server/postgres/mappers_test.go diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index a4382fe..cc72bdc 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -31,51 +31,6 @@ import ( // --- 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, @@ -327,19 +282,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 +387,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, @@ -553,53 +490,7 @@ 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}: @@ -737,16 +628,9 @@ 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, @@ -795,14 +679,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(), @@ -918,15 +795,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 +847,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 +910,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 +961,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 +1056,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, @@ -1568,45 +1372,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, @@ -1674,43 +1440,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 +1492,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 +1513,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 +1534,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, + )) } } diff --git a/internal/server/postgres/mappers.go b/internal/server/postgres/mappers.go new file mode 100644 index 0000000..c745d19 --- /dev/null +++ b/internal/server/postgres/mappers.go @@ -0,0 +1,256 @@ +package postgres + +import ( + "errors" + "fmt" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + pb "github.com/openclimatefix/data-platform/internal/gen/ocf/dp" + db "github.com/openclimatefix/data-platform/internal/server/postgres/gen" + "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +// 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. +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 +} + +// 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. +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 +} + +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..9f56115 --- /dev/null +++ b/internal/server/postgres/mappers_test.go @@ -0,0 +1,128 @@ +package postgres + +import ( + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + pb "github.com/openclimatefix/data-platform/internal/gen/ocf/dp" + db "github.com/openclimatefix/data-platform/internal/server/postgres/gen" + "github.com/stretchr/testify/assert" + "google.golang.org/protobuf/types/known/structpb" + "google.golang.org/protobuf/types/known/timestamppb" +) + +func Test_sipToFraction(t *testing.T) { + assert.InDelta(t, 0.5, sipToFraction(15000), 0.0001) + assert.InDelta(t, 1.0, sipToFraction(30000), 0.0001) + assert.InDelta(t, 0.0, sipToFraction(0), 0.0001) + assert.InDelta(t, -0.5, sipToFraction(-15000), 0.0001) +} + +func Test_buildOtherStatsMap(t *testing.T) { + v10 := int16(3000) + v90 := int16(27000) + + // Partial array of pointers + m := buildOtherStatsMap(nil, &v10, nil, nil, &v90, nil) + + assert.Len(t, m, 2) + assert.InDelta(t, 0.1, m["p10"], 0.0001) + assert.InDelta(t, 0.9, m["p90"], 0.0001) + + // Empty + mEmpty := buildOtherStatsMap(nil, nil, nil, nil, nil, nil) + assert.Len(t, mEmpty, 0) +} + +func Test_mapLocationSummary(t *testing.T) { + id := uuid.New() + meta, _ := structpb.NewStruct(map[string]interface{}{"key": "value"}) + + res := mapLocationSummary( + id, + "Test Location", + 51.5, + -0.1, + 1000, + 1, + 1, + meta, + ) + + assert.Equal(t, id.String(), res.LocationUuid) + assert.Equal(t, "Test Location", res.LocationName) + assert.Equal(t, float32(51.5), res.Latlng.Latitude) + assert.Equal(t, float32(-0.1), res.Latlng.Longitude) + assert.Equal(t, uint64(1000), res.EffectiveCapacityWatts) + assert.Equal(t, pb.EnergySource(1), res.EnergySource) + assert.Equal(t, pb.LocationType(1), res.LocationType) + assert.Equal(t, "value", res.Metadata.Fields["key"].GetStringValue()) +} + +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)) + + window := &pb.TimeWindow{ + StartTimestampUtc: startTs, + EndTimestampUtc: endTs, + } + + start, end, err := timeWindowToPgWindow(window) + assert.NoError(t, err) + assert.True(t, start.Valid) + assert.True(t, end.Valid) + assert.Equal(t, startTs.AsTime(), start.Time) + assert.Equal(t, endTs.AsTime(), end.Time) + + // Nil window + start, end, err = timeWindowToPgWindow(nil) + assert.NoError(t, err) + assert.True(t, start.Valid) + assert.True(t, end.Valid) + // Should be -48h and +36h roughly + assert.WithinDuration(t, now.Add(-48*time.Hour), start.Time, 5*time.Second) + assert.WithinDuration(t, now.Add(36*time.Hour), end.Time, 5*time.Second) + + // Invalid window + invalidWindow := &pb.TimeWindow{ + StartTimestampUtc: startTs, + } + _, _, err = timeWindowToPgWindow(invalidWindow) + assert.Error(t, err) +} + +func Test_mapForecastAsTimeseriesFromLocation(t *testing.T) { + p50 := int16(15000) + p10 := int16(3000) + p90 := int16(27000) + + initTime := time.Now().UTC() + targetTime := initTime.Add(time.Hour) + + rows := []db.ListPredictionsForLocationRow{ + { + P50Sip: p50, + P10Sip: &p10, + P90Sip: &p90, + CapacityWatts: 10000, + TargetTimeUtc: pgtype.Timestamp{Time: targetTime, Valid: true}, + InitTimeUtc: pgtype.Timestamp{Time: initTime, Valid: true}, + CreatedAtUtc: pgtype.Timestamp{Time: initTime, Valid: true}, + }, + } + + res := MapSlice(rows, mapForecastAsTimeseriesFromLocationValue) + assert.Len(t, res, 1) + + v := res[0] + assert.InDelta(t, 0.5, v.P50ValueFraction, 0.0001) + assert.Equal(t, uint64(10000), v.EffectiveCapacityWatts) + assert.InDelta(t, 0.1, v.OtherStatisticsFractions["p10"], 0.0001) + assert.InDelta(t, 0.9, v.OtherStatisticsFractions["p90"], 0.0001) + assert.Equal(t, targetTime, v.TargetTimestampUtc.AsTime()) + assert.Equal(t, initTime, v.InitializationTimestampUtc.AsTime()) +} 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; From 9c5b7764d8d4384c4a2c2d033a84bc8f32771135 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Wed, 29 Jul 2026 12:11:40 +0530 Subject: [PATCH 2/3] fix(proto): Timewindow, uuid validation to edge Moves the time window validation, and location uuid validation, to protovalidate. This prevents the need for some checks within the go code. --- internal/server/postgres/dataserverimpl.go | 55 +-- internal/server/postgres/mappers.go | 140 ++++-- internal/server/postgres/mappers_test.go | 490 +++++++++++++++++---- proto/ocf/dp/dp-data.messages.proto | 14 +- 4 files changed, 536 insertions(+), 163 deletions(-) diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index cc72bdc..108c10f 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -31,7 +31,7 @@ import ( // --- Reuseable Functions for Route Logic ------------------------------------------------------- // timeWindowToPgWindow converts a TimeWindow protobuf message to a pair of pgtype.Timestamp values. -// prepareForecastParams generates the database parameters for a single forecast from a gRPC request. +// PrepareForecastParams generates the database parameters for a single forecast from a gRPC request. func prepareForecastParams( req *pb.CreateForecastRequest, geometryUuid uuid.UUID, @@ -430,10 +430,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. @@ -490,7 +487,10 @@ func (s *DataPlatformDataServiceServerImpl) StreamForecastData( ) } - batch = append(batch, mapStreamedForecastDatum(row, locationUuid, req.IncludeMetadata)) + batch = append( + batch, + mapStreamedForecastDatum(row, locationUuid, req.IncludeMetadata), + ) if len(batch) == batchSize { select { case resChan <- &pb.StreamForecastDataResponse{Values: batch}: @@ -558,10 +558,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, @@ -628,9 +625,12 @@ func (s *DataPlatformDataServiceServerImpl) GetWeekAverageDeltas( } // Convert the deltas to the response format - deltas := MapSlice(dbDeltas, func(row db.GetWeekAverageDeltasForLocationsRow) *pb.GetWeekAverageDeltasResponse_AverageDelta { - return mapWeekAverageDelta(row, dbSource.CapacityWatts) - }) + deltas := MapSlice( + dbDeltas, + func(row db.GetWeekAverageDeltasForLocationsRow) *pb.GetWeekAverageDeltasResponse_AverageDelta { + return mapWeekAverageDelta(row, dbSource.CapacityWatts) + }, + ) return &pb.GetWeekAverageDeltasResponse{ Deltas: deltas, @@ -657,10 +657,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, @@ -696,10 +693,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, @@ -1148,15 +1142,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 { @@ -1309,10 +1294,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{ @@ -1394,10 +1376,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 { diff --git a/internal/server/postgres/mappers.go b/internal/server/postgres/mappers.go index c745d19..a949d14 100644 --- a/internal/server/postgres/mappers.go +++ b/internal/server/postgres/mappers.go @@ -1,16 +1,16 @@ package postgres import ( - "errors" "fmt" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" - pb "github.com/openclimatefix/data-platform/internal/gen/ocf/dp" - db "github.com/openclimatefix/data-platform/internal/server/postgres/gen" "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. @@ -18,31 +18,30 @@ 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, err error) { +) (start pgtype.Timestamp, end pgtype.Timestamp) { currentTime := time.Now().UTC() - if window == nil || (window.StartTimestampUtc == nil && window.EndTimestampUtc == nil) { + 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 if window.StartTimestampUtc != nil && window.EndTimestampUtc != nil { + } else { 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 + return start, end } // timeptrToPgTimestamp converts a protobuf Timestamp pointer to a pgtype.Timestamp. @@ -77,30 +76,39 @@ func sipToFraction(sip int16) float32 { } // 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 } -func mapLatestForecast(fc db.GetLatestForecastsAtHorizonSincePivotRow) *pb.GetLatestForecastsResponse_Forecast { +func mapLatestForecast( + fc db.GetLatestForecastsAtHorizonSincePivotRow, +) *pb.GetLatestForecastsResponse_Forecast { return &pb.GetLatestForecastsResponse_Forecast{ InitializationTimestampUtc: timestamppb.New(fc.InitTimeUtc.Time), Forecaster: &pb.Forecaster{ @@ -120,7 +128,10 @@ func mapForecaster(fc db.GetForecastersByFiltersRow) *pb.Forecaster { } } -func mapWeekAverageDelta(delta db.GetWeekAverageDeltasForLocationsRow, capacityWatts int64) *pb.GetWeekAverageDeltasResponse_AverageDelta { +func mapWeekAverageDelta( + delta db.GetWeekAverageDeltasForLocationsRow, + capacityWatts int64, +) *pb.GetWeekAverageDeltasResponse_AverageDelta { return &pb.GetWeekAverageDeltasResponse_AverageDelta{ DeltaFraction: float32(delta.AvgDeltaSip) / 30000.0, HorizonMins: uint32(delta.HorizonMins), @@ -128,7 +139,9 @@ func mapWeekAverageDelta(delta db.GetWeekAverageDeltasForLocationsRow, capacityW } } -func mapObservationAsTimeseries(obs db.GetObservationsBetweenRow) *pb.GetObservationsAsTimeseriesResponse_Value { +func mapObservationAsTimeseries( + obs db.GetObservationsBetweenRow, +) *pb.GetObservationsAsTimeseriesResponse_Value { return &pb.GetObservationsAsTimeseriesResponse_Value{ ValueFraction: sipToFraction(obs.ValueSip), TimestampUtc: timestamppb.New(obs.ObservationTimestampUtc.Time), @@ -136,7 +149,9 @@ func mapObservationAsTimeseries(obs db.GetObservationsBetweenRow) *pb.GetObserva } } -func mapLatestObservation(obs db.GetLatestObservationsRow) *pb.GetLatestObservationsResponse_Observation { +func mapLatestObservation( + obs db.GetLatestObservationsRow, +) *pb.GetLatestObservationsResponse_Observation { return &pb.GetLatestObservationsResponse_Observation{ LocationUuid: obs.GeometryUuid.String(), TimestampUtc: timestamppb.New(obs.ObservationTimestampUtc.Time), @@ -152,12 +167,14 @@ func mapObserver(ob db.ObsObserver) *pb.ListObserversResponse_ObserverSummary { } } -func mapPredictionAtTime(value db.ListPredictionsAtTimeForLocationsRow) *pb.GetForecastAtTimestampResponse_Value { +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, + ValueFraction: sipToFraction(value.P50Sip), + EffectiveCapacityWatts: uint64(value.CapacityWatts), + LocationUuid: value.GeometryUuid.String(), + LocationName: value.GeometryName, Latlng: &pb.LatLng{ Latitude: value.Latitude, Longitude: value.Longitude, @@ -165,11 +182,20 @@ func mapPredictionAtTime(value db.ListPredictionsAtTimeForLocationsRow) *pb.GetF 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), + OtherStatisticsFractions: buildOtherStatsMap( + value.P02Sip, + value.P10Sip, + value.P25Sip, + value.P75Sip, + value.P90Sip, + value.P98Sip, + ), } } -func mapObservationAtTimestamp(obs db.ListObservationsAtTimeForLocationsRow) *pb.GetObservationsAtTimestampResponse_Value { +func mapObservationAtTimestamp( + obs db.ListObservationsAtTimeForLocationsRow, +) *pb.GetObservationsAtTimestampResponse_Value { return &pb.GetObservationsAtTimestampResponse_Value{ ValueFraction: sipToFraction(obs.ValueSip), EffectiveCapacityWatts: uint64(obs.CapacityWatts), @@ -181,7 +207,9 @@ func mapObservationAtTimestamp(obs db.ListObservationsAtTimeForLocationsRow) *pb } } -func mapLocationSnapshot(v db.GetSourceHistoryRow) *pb.GetLocationAsTimeseriesResponse_LocationSnapshot { +func mapLocationSnapshot( + v db.GetSourceHistoryRow, +) *pb.GetLocationAsTimeseriesResponse_LocationSnapshot { return &pb.GetLocationAsTimeseriesResponse_LocationSnapshot{ EffectiveCapacityWatts: uint64(v.CapacityWatts), TimestampUtc: timestamppb.New(v.ValidFromUtc.Time), @@ -189,7 +217,9 @@ func mapLocationSnapshot(v db.GetSourceHistoryRow) *pb.GetLocationAsTimeseriesRe } } -func mapForecastAsTimeseriesFromForecastValue(pred db.ListPredictionsForForecastsRow) *pb.GetForecastAsTimeseriesResponse_Value { +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), @@ -198,16 +228,32 @@ func mapForecastAsTimeseriesFromForecastValue(pred db.ListPredictionsForForecast 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, + 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 { +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), + 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), @@ -215,7 +261,14 @@ func mapForecastAsTimeseriesFromLocationValue(value db.ListPredictionsForLocatio } } -func mapLocationSummary(geomUuid uuid.UUID, geomName string, lat, lon float32, cap int64, srcType, geomType int16, meta *structpb.Struct) *pb.ListLocationsResponse_LocationSummary { +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, @@ -230,7 +283,11 @@ func mapLocationSummary(geomUuid uuid.UUID, geomName string, lat, lon float32, c } } -func mapStreamedForecastDatum(row db.ListPredictionsForForecastsRow, locUuid uuid.UUID, includeMetadata bool) *pb.ForecastDatum { +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() { @@ -246,11 +303,18 @@ func mapStreamedForecastDatum(row db.ListPredictionsForForecastsRow, locUuid uui 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, + 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 index 9f56115..ec38127 100644 --- a/internal/server/postgres/mappers_test.go +++ b/internal/server/postgres/mappers_test.go @@ -1,101 +1,398 @@ package postgres import ( + "strconv" "testing" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" - pb "github.com/openclimatefix/data-platform/internal/gen/ocf/dp" - db "github.com/openclimatefix/data-platform/internal/server/postgres/gen" - "github.com/stretchr/testify/assert" + "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) { - assert.InDelta(t, 0.5, sipToFraction(15000), 0.0001) - assert.InDelta(t, 1.0, sipToFraction(30000), 0.0001) - assert.InDelta(t, 0.0, sipToFraction(0), 0.0001) - assert.InDelta(t, -0.5, sipToFraction(-15000), 0.0001) + 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) - // Partial array of pointers - m := buildOtherStatsMap(nil, &v10, nil, nil, &v90, nil) + 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, + }, + }, + } - assert.Len(t, m, 2) - assert.InDelta(t, 0.1, m["p10"], 0.0001) - assert.InDelta(t, 0.9, m["p90"], 0.0001) + 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)) - // Empty - mEmpty := buildOtherStatsMap(nil, nil, nil, nil, nil, nil) - assert.Len(t, mEmpty, 0) + 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"}) - res := mapLocationSummary( - id, - "Test Location", - 51.5, - -0.1, - 1000, - 1, - 1, - meta, - ) - - assert.Equal(t, id.String(), res.LocationUuid) - assert.Equal(t, "Test Location", res.LocationName) - assert.Equal(t, float32(51.5), res.Latlng.Latitude) - assert.Equal(t, float32(-0.1), res.Latlng.Longitude) - assert.Equal(t, uint64(1000), res.EffectiveCapacityWatts) - assert.Equal(t, pb.EnergySource(1), res.EnergySource) - assert.Equal(t, pb.LocationType(1), res.LocationType) - assert.Equal(t, "value", res.Metadata.Fields["key"].GetStringValue()) + 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_timeWindowToPgWindow(t *testing.T) { - now := time.Now().UTC() - startTs := timestamppb.New(now.Add(-2 * time.Hour)) - endTs := timestamppb.New(now.Add(2 * time.Hour)) +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"]) + }, + }, + } - window := &pb.TimeWindow{ - StartTimestampUtc: startTs, - EndTimestampUtc: endTs, - } - - start, end, err := timeWindowToPgWindow(window) - assert.NoError(t, err) - assert.True(t, start.Valid) - assert.True(t, end.Valid) - assert.Equal(t, startTs.AsTime(), start.Time) - assert.Equal(t, endTs.AsTime(), end.Time) - - // Nil window - start, end, err = timeWindowToPgWindow(nil) - assert.NoError(t, err) - assert.True(t, start.Valid) - assert.True(t, end.Valid) - // Should be -48h and +36h roughly - assert.WithinDuration(t, now.Add(-48*time.Hour), start.Time, 5*time.Second) - assert.WithinDuration(t, now.Add(36*time.Hour), end.Time, 5*time.Second) - - // Invalid window - invalidWindow := &pb.TimeWindow{ - StartTimestampUtc: startTs, - } - _, _, err = timeWindowToPgWindow(invalidWindow) - assert.Error(t, err) + 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_mapForecastAsTimeseriesFromLocation(t *testing.T) { +func Test_mapForecastAsTimeseriesFromLocationValue(t *testing.T) { p50 := int16(15000) p10 := int16(3000) p90 := int16(27000) @@ -103,26 +400,53 @@ func Test_mapForecastAsTimeseriesFromLocation(t *testing.T) { initTime := time.Now().UTC() targetTime := initTime.Add(time.Hour) - rows := []db.ListPredictionsForLocationRow{ + 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) + }, + }, { - P50Sip: p50, - P10Sip: &p10, - P90Sip: &p90, - CapacityWatts: 10000, - TargetTimeUtc: pgtype.Timestamp{Time: targetTime, Valid: true}, - InitTimeUtc: pgtype.Timestamp{Time: initTime, Valid: true}, - CreatedAtUtc: pgtype.Timestamp{Time: initTime, Valid: true}, + 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) + }, }, } - res := MapSlice(rows, mapForecastAsTimeseriesFromLocationValue) - assert.Len(t, res, 1) + 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()) - v := res[0] - assert.InDelta(t, 0.5, v.P50ValueFraction, 0.0001) - assert.Equal(t, uint64(10000), v.EffectiveCapacityWatts) - assert.InDelta(t, 0.1, v.OtherStatisticsFractions["p10"], 0.0001) - assert.InDelta(t, 0.9, v.OtherStatisticsFractions["p90"], 0.0001) - assert.Equal(t, targetTime, v.TargetTimestampUtc.AsTime()) - assert.Equal(t, initTime, v.InitializationTimestampUtc.AsTime()) + tt.validateResult(t, res) + }) + } } 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. From 9a8e4e47f6ab12eb5986e7b1f77c0c7113228d29 Mon Sep 17 00:00:00 2001 From: devsjc <47188100+devsjc@users.noreply.github.com> Date: Wed, 29 Jul 2026 15:10:26 +0530 Subject: [PATCH 3/3] chore(repo): Move last helper --- internal/server/postgres/dataserverimpl.go | 73 +------------------ .../server/postgres/dataserverimpl_test.go | 2 +- internal/server/postgres/mappers.go | 60 +++++++++++++++ 3 files changed, 64 insertions(+), 71 deletions(-) diff --git a/internal/server/postgres/dataserverimpl.go b/internal/server/postgres/dataserverimpl.go index 108c10f..b353afa 100644 --- a/internal/server/postgres/dataserverimpl.go +++ b/internal/server/postgres/dataserverimpl.go @@ -28,71 +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. -// 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{} } @@ -101,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, @@ -162,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}) @@ -1687,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 index a949d14..bcf935a 100644 --- a/internal/server/postgres/mappers.go +++ b/internal/server/postgres/mappers.go @@ -106,6 +106,66 @@ func buildOtherStatsMap(p02, p10, p25, p75, p90, p98 *int16) map[string]float32 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 {