From 90817e4b73579d7bf29741b45f7b56c3864d25e1 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 12 Oct 2026 19:59:52 +0200 Subject: [PATCH 1/3] fix(history): retain working measurements when devices are offline Signed-off-by: Fredrik Ahlgren --- .changeset/history-offline-devices.md | 5 + docs/architecture.md | 5 + go/cmd/ftw/energy_history_test.go | 22 ++- go/cmd/ftw/energy_identity_test.go | 3 + go/cmd/ftw/history_options_test.go | 15 +- go/cmd/ftw/history_partial_test.go | 22 +++ go/cmd/ftw/main.go | 140 ++++++++--------- go/internal/api/api.go | 56 ++++--- go/internal/api/api_history_quality.go | 74 +++++++++ go/internal/api/api_history_quality_test.go | 94 ++++++++++++ go/internal/state/history_dashboard.go | 41 ++++- go/internal/state/history_dashboard_cost.go | 6 + .../state/history_dashboard_retention.go | 27 +++- go/internal/state/history_hot.go | 1 + go/internal/state/history_plain.go | 2 +- go/internal/state/history_quality.go | 69 +++++++++ go/internal/state/history_quality_test.go | 141 ++++++++++++++++++ go/internal/state/history_schema.go | 2 +- go/internal/state/history_sqlite.go | 6 +- go/internal/state/history_writer.go | 3 + go/internal/state/store.go | 47 ++++-- go/internal/state/store_ts.go | 3 + go/internal/telemetry/forecast.go | 36 ++++- state-schema.json | 2 +- web/app.js | 130 +++++++++------- web/components/ftw-bar-chart.js | 2 + web/components/ftw-energy-cake.js | 1 + web/components/ftw-history-card.js | 10 +- web/energy-history.js | 24 ++- web/history-quality.test.mjs | 33 ++++ web/index.html | 1 + web/style.css | 2 + 32 files changed, 812 insertions(+), 213 deletions(-) create mode 100644 .changeset/history-offline-devices.md create mode 100644 go/cmd/ftw/history_partial_test.go create mode 100644 go/internal/api/api_history_quality.go create mode 100644 go/internal/api/api_history_quality_test.go create mode 100644 go/internal/state/history_quality.go create mode 100644 go/internal/state/history_quality_test.go create mode 100644 web/history-quality.test.mjs diff --git a/.changeset/history-offline-devices.md b/.changeset/history-offline-devices.md new file mode 100644 index 000000000..923ea6680 --- /dev/null +++ b/.changeset/history-offline-devices.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Keep history for working devices when another device is offline. Show missing power, state of charge and energy as unknown, preserve gaps through retention and backup, and flag incomplete measurements or an invalid household balance. State schema 8 preserves this evidence; return to an older Core with its matching full backup. diff --git a/docs/architecture.md b/docs/architecture.md index c4c415343..66db7245f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -37,6 +37,11 @@ A separate module requires a concrete benefit and: A percentage goal and a goal to reach the car's own limit are distinct. State schema 7 blocks rollback to a Core that would ignore the saved goal mode. +State schema 8 adds durable missing-value evidence for chart and energy buckets. +An offline flow leaves other devices recording. Missing inputs make only their +values and the derived household balance unknown. Retention, downsampling and +full backups keep that evidence; a recovered sample cannot fill an earlier gap. +Use a matching full backup when returning to an older Core. `GET /api/loadpoints` advertises `vehicle_limit_goal_supported`; clients must require that flag before saving `schedule.finish_at_vehicle_limit`. Existing percentage goals keep their meaning. In vehicle-limit mode, the planner uses diff --git a/go/cmd/ftw/energy_history_test.go b/go/cmd/ftw/energy_history_test.go index e4a8d1867..e52123cc0 100644 --- a/go/cmd/ftw/energy_history_test.go +++ b/go/cmd/ftw/energy_history_test.go @@ -82,16 +82,16 @@ func TestBuildHistoryPointExcludesUnavailableTelemetry(t *testing.T) { point, available := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "site-meter"}, now.UnixMilli(), time.Minute) - if available { - t.Fatalf("offline site meter produced history: %+v", point) + if !available || point.Missing&state.MissingGrid == 0 { + t.Fatalf("offline site meter became known or lost live devices: %+v", point) } tel.Update("site-meter", telemetry.DerMeter, 800, nil, nil) tel.RecordDriverSuccess("site-meter") point, available = buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "site-meter"}, now.UnixMilli(), time.Minute) - if available { - t.Fatalf("missing PV/battery treated as zero in household history: %+v", point) + if !available || point.Missing&(state.MissingPV|state.MissingBattery|state.MissingLoad) == 0 { + t.Fatalf("missing PV/battery became known or lost live meter: %+v", point) } // The aggregate becomes known only after every significant flow recovers. tel.Update("stale-pv", telemetry.DerPV, -800, nil, nil) @@ -140,8 +140,8 @@ func TestBuildHistoryPointRequiresFreshEVAndV2X(t *testing.T) { point, available := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "site-meter"}, now.UnixMilli(), time.Minute) - if available { - t.Fatalf("aged EV/V2X became zero household demand: %+v", point) + if !available || point.Missing&(state.MissingEV|state.MissingV2X|state.MissingLoad) == 0 { + t.Fatalf("aged EV/V2X became known or lost live meter: %+v", point) } } @@ -162,6 +162,9 @@ func TestPersistTelemetryTickUsesPersistenceFreshness(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { _ = st.Close() }) + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } tickMS := time.Now().Add(-5 * time.Second).UnixMilli() tel := telemetry.NewStore() tel.EnsureDriverHealth("meter") @@ -247,6 +250,9 @@ func TestStaleMeterTickKeepsSamplesAndIndependentLedgerWithoutDispatch(t *testin if err != nil { t.Fatal(err) } + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } t.Cleanup(func() { _ = st.Close() }) pvDeviceID, err := st.RegisterDevice(state.Device{ DriverName: "solar", Make: "PV Maker", Serial: "PV-42", @@ -293,8 +299,8 @@ func TestStaleMeterTickKeepsSamplesAndIndependentLedgerWithoutDispatch(t *testin if err != nil { t.Fatal(err) } - if len(history) != 0 { - t.Fatalf("stale meter wrote legacy history: %+v", history) + if len(history) == 0 || history[0].Missing&state.MissingGrid == 0 { + t.Fatalf("stale meter lost independent history or became known: %+v", history) } samples, err := st.LoadSeries("solar", "pv_w", 0, time.Now().Add(time.Second).UnixMilli(), 0) if err != nil { diff --git a/go/cmd/ftw/energy_identity_test.go b/go/cmd/ftw/energy_identity_test.go index 30baf1033..2586b666e 100644 --- a/go/cmd/ftw/energy_identity_test.go +++ b/go/cmd/ftw/energy_identity_test.go @@ -52,6 +52,9 @@ func TestStartupAliasCannotReplayCountersWhileRawHistoryKeepsWriting(t *testing. t.Fatal(err) } t.Cleanup(func() { _ = st.Close() }) + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } serial := state.Device{DriverName: "meter", Make: "inverter", Serial: "one", MAC: "aa:bb:cc:dd:ee:ff"} if _, err := st.RegisterDevice(serial); err != nil { t.Fatal(err) diff --git a/go/cmd/ftw/history_options_test.go b/go/cmd/ftw/history_options_test.go index 2c511e503..ace120995 100644 --- a/go/cmd/ftw/history_options_test.go +++ b/go/cmd/ftw/history_options_test.go @@ -54,20 +54,19 @@ func TestHistoryKeepsRunningWithOptionalPVAndUnconfirmedIdentity(t *testing.T) { t.Fatalf("history stopped: point=%+v available=%v", point, ok) } - // A known source that goes silent still stops history; only optional, - // never-reported PV is treated as absent. + // Missing flows remain unknown while independent live history continues. tel.DriverHealthMut("pixii").SetOffline() - if _, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"}, - time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); ok { - t.Fatal("offline battery treated as zero in history") + if p, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"}, + time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); !ok || p.Missing&(state.MissingBattery|state.MissingLoad) == 0 { + t.Fatal("offline battery lost independent history or became known") } tel.Update("pixii", telemetry.DerBattery, 500, nil, nil) tel.RecordDriverSuccess("pixii") // Once the Zap does report PV, a stale PV reading stops history again. tel.Update("sourceful-zap", telemetry.DerPV, -100, nil, nil) tel.Get("sourceful-zap", telemetry.DerPV).UpdatedAt = time.Now().Add(-2 * time.Minute) - if _, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"}, - time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); ok { - t.Fatal("stale reported PV treated as zero in history") + if p, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "sourceful-zap"}, + time.Now().UnixMilli(), time.Minute, snap.HistoryOptions); !ok || p.Missing&state.MissingPV == 0 { + t.Fatal("stale PV lost independent history or became known") } } diff --git a/go/cmd/ftw/history_partial_test.go b/go/cmd/ftw/history_partial_test.go new file mode 100644 index 000000000..fcb2032d8 --- /dev/null +++ b/go/cmd/ftw/history_partial_test.go @@ -0,0 +1,22 @@ +package main + +import ( + "github.com/srcfl/ftw/go/internal/telemetry" + "testing" + "time" +) + +func TestHistoryRetainsLiveMeterWhenBatteryOffline(t *testing.T) { + tel := telemetry.NewStore() + for _, name := range []string{"meter", "battery"} { + tel.EnsureDriverHealth(name) + } + tel.Update("meter", telemetry.DerMeter, 1200, nil, nil) + tel.RecordDriverSuccess("meter") + tel.DriverHealthMut("battery").SetOffline() + opts := telemetry.ForecastOptions{ExpectedFlows: []telemetry.ForecastFlow{{Driver: "battery", DerType: telemetry.DerBattery}}} + p, available := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "meter"}, time.Now().UnixMilli(), time.Minute, opts) + if !available || p.GridW != 1200 { + t.Fatalf("live meter lost when battery offline: available=%v point=%+v", available, p) + } +} diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index cffc17663..60f44120c 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -2505,6 +2505,7 @@ func main() { SiteDispatchBlocked: func() string { return siteDispatchNow(tel, cfg, cfgMu, ctrl, ctrlMu, time.Now()).Reason }, + HistoryMeasurementSources: func() telemetry.ForecastOptions { return forecastSettings.Snapshot().HistoryOptions }, SiteMeasurementSources: func() telemetry.ForecastOptions { return forecastSettings.Snapshot().Options }, @@ -3894,7 +3895,7 @@ func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersis energyObservations := buildEnergyObservations(st, tel, ctrl, hp, identity) filtered := energyObservations[:0] for _, observation := range energyObservations { - if !historyAvailable && observation.AssetKind == state.AssetObservedConsumer { + if (!historyAvailable || hp.Missing&state.MissingLoad != 0) && observation.AssetKind == state.AssetObservedConsumer { continue } if observation.AssetKind != state.AssetObservedConsumer { @@ -3921,102 +3922,81 @@ func persistTelemetryTick(st *state.Store, tel *telemetry.Store, ctrl tickPersis } func buildHistoryPoint(tel *telemetry.Store, ctrl tickPersistControl, nowMs int64, historyMaxAge time.Duration, options ...telemetry.ForecastOptions) (state.HistoryPoint, bool) { - unavailable := state.HistoryPoint{TsMs: nowMs} - if tel == nil || ctrl.SiteMeterDriver == "" { - return unavailable, false + if tel == nil { + return state.HistoryPoint{TsMs: nowMs}, false } opts := telemetry.ForecastOptions{MaxAge: historyMaxAge} if len(options) > 0 { opts = options[0] opts.MaxAge = historyMaxAge } - // Polls can update telemetry while dispatch runs. Keep the tick's history - // timestamp, but judge reading freshness when this snapshot is captured. + opts.UseWatchdogTimeout = true balance := tel.ForecastMeasurementNow(ctrl.SiteMeterDriver, opts) - if !balance.Valid { - return unavailable, false - } - now := balance.At - gridW, pvW, batW := balance.GridW, balance.PVW, balance.BatteryW - evW, v2xW, loadW := balance.EVW, balance.V2XW, balance.HouseholdW - readingUsable := func(driver string, updatedAt time.Time) bool { - h := tel.DriverHealth(driver) - if !h.TelemetryLive() || updatedAt.After(now) { - return false + p := state.HistoryPoint{TsMs: nowMs, GridW: balance.GridW, PVW: balance.PVW, BatW: balance.BatteryW, LoadW: balance.HouseholdW} + for kind, flag := range map[telemetry.DerType]state.HistoryMissing{telemetry.DerMeter: state.MissingGrid, telemetry.DerPV: state.MissingPV, telemetry.DerBattery: state.MissingBattery, telemetry.DerEV: state.MissingEV, telemetry.DerV2X: state.MissingV2X} { + if !balance.Known[kind] { + p.Missing |= flag } - maxAge := historyMaxAge - if h.WatchdogTimeoutOverride > 0 { - maxAge = h.WatchdogTimeoutOverride - } - return maxAge <= 0 || now.Sub(updatedAt) <= maxAge } - var avgSoC, sumSoC float64 - var socCount int - for _, r := range tel.ReadingsByType(telemetry.DerBattery) { - if readingUsable(r.Driver, r.UpdatedAt) && r.SoC != nil { - sumSoC += *r.SoC - socCount++ - } + if opts.PVInvalidReason != "" { + p.Missing |= state.MissingPV } - if socCount > 0 { - avgSoC = sumSoC / float64(socCount) - } - - // Per-driver detail packed into the JSON column. The schema is - // schema-less by design — UI code reads what it understands and - // ignores the rest, so drivers can add fields without a migration. - perDriver := make(map[string]map[string]float64) - for name := range tel.AllHealth() { - row := map[string]float64{} - if r := tel.Get(name, telemetry.DerBattery); r != nil && - readingUsable(name, r.UpdatedAt) { - row["bat_w"] = r.SmoothedW - if r.SoC != nil { - row["soc"] = *r.SoC - } + if !balance.Valid { + p.Missing |= state.MissingLoad + } + if balance.Reason == "invalid_balance" || strings.HasPrefix(balance.Reason, "duplicate_") || strings.HasPrefix(balance.Reason, "invalid_sign:") || balance.Reason == "time_skew" { + p.Missing |= state.BalanceMismatch + } + perDriver := map[string]map[string]any{} + var socSum float64 + var batteries, socCount int + var hasPower bool + for _, source := range balance.Sources { + hasPower = hasPower || source.PowerW != nil + name := source.Flow.Driver + if name == "" { + continue } - if r := tel.Get(name, telemetry.DerPV); r != nil && - readingUsable(name, r.UpdatedAt) { - row["pv_w"] = r.SmoothedW - } - if r := tel.Get(name, telemetry.DerMeter); r != nil && - readingUsable(name, r.UpdatedAt) { - row["meter_w"] = r.SmoothedW - } - // EV charge power: required for the live chart's EV series - // (web/app.js reads `d.ev_w` per driver from /api/history). - // Without it the chart's EV trace is always zero — the in-memory - // /api/status DOES carry ev_w, but history rows never did until - // this row was added. - if r := tel.Get(name, telemetry.DerEV); r != nil && - readingUsable(name, r.UpdatedAt) { - row["ev_w"] = r.SmoothedW - } - if r := tel.Get(name, telemetry.DerV2X); r != nil && - readingUsable(name, r.UpdatedAt) { - row["v2x_w"] = r.SmoothedW - if r.SoC != nil { - row["v2x_vehicle_soc"] = *r.SoC + row := perDriver[name] + if row == nil { + row = map[string]any{} + perDriver[name] = row + } + field := map[telemetry.DerType]string{telemetry.DerMeter: "meter_w", telemetry.DerPV: "pv_w", telemetry.DerBattery: "bat_w", telemetry.DerEV: "ev_w", telemetry.DerV2X: "v2x_w"}[source.Flow.DerType] + if field != "" { + row[field] = source.PowerW + } + if source.Flow.DerType == telemetry.DerBattery { + batteries++ + row["soc"] = source.SoC + if source.SoC != nil { + socSum += *source.SoC + socCount++ } } - perDriver[name] = row + if source.Flow.DerType == telemetry.DerV2X { + row["v2x_vehicle_soc"] = source.SoC + } } - targets := make(map[string]float64) + if socCount > 0 && socCount == batteries { + p.BatSoC = socSum / float64(socCount) + } else { + p.Missing |= state.MissingSoC + } + // Numeric storage remains finite. The durable mask, never these placeholders, + // decides whether a chart or energy total is known. + p.ClearMissingValues() + targets := map[string]float64{} for _, t := range ctrl.LastTargets { targets[t.Driver] = t.TargetW } - jsonBlob, _ := json.Marshal(map[string]any{ - "drivers": perDriver, - "targets": targets, - "ev_w": evW, - "v2x_w": v2xW, - "load_house_w": loadW, - "forecast_measurement_quality": "complete_instantaneous_balance_v1", - }) - return state.HistoryPoint{ - TsMs: nowMs, GridW: gridW, PVW: pvW, BatW: batW, LoadW: loadW, BatSoC: avgSoC, - JSON: string(jsonBlob), - }, true + quality := "incomplete_measurement" + if balance.Valid { + quality = "complete_instantaneous_balance_v1" + } + blob, _ := json.Marshal(map[string]any{"drivers": perDriver, "targets": targets, "ev_w": p.Missing.Value(state.MissingEV, balance.EVW), "v2x_w": p.Missing.Value(state.MissingV2X, balance.V2XW), "load_house_w": p.LoadW, "forecast_measurement_quality": quality, "history_reason": balance.Reason}) + p.JSON = string(blob) + return p, hasPower } func restoreLatestMPCDiagnostic(st *state.Store, svc *mpc.Service, now time.Time) { diff --git a/go/internal/api/api.go b/go/internal/api/api.go index 9466dbb73..d289fb96b 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -74,6 +74,7 @@ const ( // One instance is shared across all handlers; mutations use the contained // mutexes from each package. type Deps struct { + HistoryMeasurementSources func() telemetry.ForecastOptions // SiteMeasurementSources reuses the configured physical-flow inventory. // These are source declarations, never forecast values or derived load. SiteMeasurementSources func() telemetry.ForecastOptions @@ -1140,12 +1141,12 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { // instead of "no data yet". Mirror the old `len(pts) > 1` guard. if d.Intervals > 0 { energyToday = map[string]any{ - "import_wh": d.ImportWh, - "export_wh": d.ExportWh, - "pv_wh": d.PVWh, - "bat_charged_wh": d.BatChargedWh, - "bat_discharged_wh": d.BatDischargedWh, - "load_wh": d.LoadWh, + "import_wh": d.Missing.Value(state.MissingGrid, d.ImportWh), + "export_wh": d.Missing.Value(state.MissingGrid, d.ExportWh), + "pv_wh": d.Missing.Value(state.MissingPV, d.PVWh), + "bat_charged_wh": d.Missing.Value(state.MissingBattery, d.BatChargedWh), + "bat_discharged_wh": d.Missing.Value(state.MissingBattery, d.BatDischargedWh), + "load_wh": d.Missing.Value(state.MissingLoad, d.LoadWh), } } } else { @@ -1253,6 +1254,7 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { } resp["energy"] = energy } + s.qualifyStatusHistory(resp, ctrl.SiteMeterDriver) writeJSON(w, 200, resp) } @@ -1272,14 +1274,17 @@ func currentGridEnergySlot(st *state.Store, now time.Time) (map[string]any, erro if err != nil { return nil, err } + if d.Intervals == 0 { + d.Missing |= state.MissingGrid + } return map[string]any{ "slot_start_ms": slotStart.UnixMilli(), "slot_end_ms": slotEnd.UnixMilli(), "elapsed_s": now.Sub(slotStart).Seconds(), - "import_wh": d.ImportWh, - "export_wh": d.ExportWh, + "import_wh": d.Missing.Value(state.MissingGrid, d.ImportWh), + "export_wh": d.Missing.Value(state.MissingGrid, d.ExportWh), // Observational only — see comment above. Do not bill against this. - "net_wh": d.ImportWh - d.ExportWh, + "net_wh": d.Missing.Value(state.MissingGrid, d.ImportWh-d.ExportWh), }, nil } @@ -2274,11 +2279,18 @@ func (s *Server) handleHistory(w http.ResponseWriter, r *http.Request) { } inner["ts"] = row.TsMs // Fill from flat columns for charting - inner["grid_w"] = row.GridW - inner["pv_w"] = row.PVW - inner["bat_w"] = row.BatW - inner["load_w"] = row.LoadW - inner["bat_soc"] = row.BatSoC + inner["grid_w"] = row.Missing.Value(state.MissingGrid, row.GridW) + inner["pv_w"] = row.Missing.Value(state.MissingPV, row.PVW) + inner["bat_w"] = row.Missing.Value(state.MissingBattery, row.BatW) + inner["load_w"] = row.Missing.Value(state.MissingLoad, row.LoadW) + inner["bat_soc"] = row.Missing.Value(state.MissingSoC, row.BatSoC) + inner["load_house_w"] = inner["load_w"] + for field, flag := range map[string]state.HistoryMissing{"ev_w": state.MissingEV, "v2x_w": state.MissingV2X} { + if row.Missing&flag != 0 { + inner[field] = nil + } + } + inner["history_quality"] = row.Missing.Quality() inner["n"] = row.N inner["resolution_ms"] = row.ResolutionMS inner["first_ms"] = row.FirstMS @@ -2426,14 +2438,18 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { } } + if de.Intervals == 0 { + de.Missing |= state.MissingPower + } out = append(out, map[string]any{ + "history_quality": de.Missing.Quality(), "day": dayKey, - "import_wh": de.ImportWh, - "export_wh": de.ExportWh, - "pv_wh": de.PVWh, - "bat_charged_wh": de.BatChargedWh, - "bat_discharged_wh": de.BatDischargedWh, - "load_wh": de.LoadWh, + "import_wh": de.Missing.Value(state.MissingGrid, de.ImportWh), + "export_wh": de.Missing.Value(state.MissingGrid, de.ExportWh), + "pv_wh": de.Missing.Value(state.MissingPV, de.PVWh), + "bat_charged_wh": de.Missing.Value(state.MissingBattery, de.BatChargedWh), + "bat_discharged_wh": de.Missing.Value(state.MissingBattery, de.BatDischargedWh), + "load_wh": de.Missing.Value(state.MissingLoad, de.LoadWh), }) } writeJSON(w, 200, map[string]any{"days": out, "tz": localzone.Name()}) diff --git a/go/internal/api/api_history_quality.go b/go/internal/api/api_history_quality.go new file mode 100644 index 000000000..12a5b76c5 --- /dev/null +++ b/go/internal/api/api_history_quality.go @@ -0,0 +1,74 @@ +package api + +import ( + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" + "strings" +) + +func (s *Server) qualifyStatusHistory(resp map[string]any, siteMeter string) { + if s.deps.HistoryMeasurementSources == nil { + return + } + opts := s.deps.HistoryMeasurementSources() + opts.UseWatchdogTimeout = true + r := s.deps.Tel.ForecastMeasurementNow(siteMeter, opts) + var missing state.HistoryMissing + for kind, flag := range map[telemetry.DerType]state.HistoryMissing{telemetry.DerMeter: state.MissingGrid, telemetry.DerPV: state.MissingPV, telemetry.DerBattery: state.MissingBattery, telemetry.DerEV: state.MissingEV, telemetry.DerV2X: state.MissingV2X} { + if !r.Known[kind] { + missing |= flag + } + } + if !r.Valid { + missing |= state.MissingLoad + } + if opts.PVInvalidReason != "" { + missing |= state.MissingPV + } + if r.Reason == "invalid_balance" || strings.HasPrefix(r.Reason, "duplicate_") || strings.HasPrefix(r.Reason, "invalid_sign:") || r.Reason == "time_skew" { + missing |= state.BalanceMismatch + } + resp["grid_w"] = missing.Value(state.MissingGrid, r.GridW) + resp["pv_w"] = missing.Value(state.MissingPV, r.PVW) + resp["bat_w"] = missing.Value(state.MissingBattery, r.BatteryW) + resp["ev_w"] = missing.Value(state.MissingEV, r.EVW) + if missing&state.MissingEV != 0 { + resp["ev_charging_w"] = nil + } + resp["v2x_w"] = missing.Value(state.MissingV2X, r.V2XW) + resp["load_w"] = missing.Value(state.MissingLoad, r.HouseholdW) + resp["history_quality"] = missing.Quality() + resp["history_reason"] = r.Reason + drivers, _ := resp["drivers"].(map[string]any) + for _, source := range r.Sources { + if d, ok := drivers[source.Flow.Driver].(map[string]any); ok { + field := map[telemetry.DerType]string{telemetry.DerMeter: "meter_w", telemetry.DerPV: "pv_w", telemetry.DerBattery: "bat_w", telemetry.DerEV: "ev_w", telemetry.DerV2X: "v2x_w"}[source.Flow.DerType] + if field != "" { + d[field] = source.PowerW + } + if source.Flow.DerType == telemetry.DerBattery { + d["soc"] = source.SoC + } + } + } + var count int + var sum float64 + for _, source := range r.Sources { + if source.Flow.DerType == telemetry.DerBattery { + if source.SoC == nil { + missing |= state.MissingSoC + } + count++ + if source.SoC != nil { + sum += *source.SoC + } + } + } + if count == 0 { + missing |= state.MissingSoC + } + if count > 0 { + sum /= float64(count) + } + resp["bat_soc"] = missing.Value(state.MissingSoC, sum) +} diff --git a/go/internal/api/api_history_quality_test.go b/go/internal/api/api_history_quality_test.go new file mode 100644 index 000000000..7520d22f4 --- /dev/null +++ b/go/internal/api/api_history_quality_test.go @@ -0,0 +1,94 @@ +package api + +import ( + "context" + "encoding/json" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" + "net/http/httptest" + "path/filepath" + "testing" + "time" +) + +func TestPartialHistoryAPIKeepsKnownFlowsAndNullsMissingValues(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + defer st.Close() + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + now := time.Now().Add(-time.Minute).UnixMilli() + for i := 0; i < 4; i++ { + p := state.HistoryPoint{TsMs: now + int64(i*10000), GridW: 1200, PVW: -300, Missing: state.MissingBattery | state.MissingLoad | state.MissingSoC, JSON: `{"drivers":{"meter":{"meter_w":1200},"battery":{"bat_w":null}}}`} + if err := st.EnqueueTelemetryTick(&p, nil, nil); err != nil { + t.Fatal(err) + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + srv := New(&Deps{State: st}) + rr := httptest.NewRecorder() + srv.handleHistory(rr, httptest.NewRequest("GET", "/api/history?range=5m&points=1", nil)) + var h struct { + Items []map[string]any `json:"items"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &h); err != nil { + t.Fatal(err) + } + if len(h.Items) != 1 { + t.Fatal(rr.Body.String()) + } + p := h.Items[0] + if p["grid_w"] != float64(1200) || p["pv_w"] != float64(-300) || p["bat_w"] != nil || p["bat_soc"] != nil || p["load_w"] != nil || p["history_quality"] != "incomplete" { + t.Fatal(p) + } + rr = httptest.NewRecorder() + srv.handleEnergyDaily(rr, httptest.NewRequest("GET", "/api/energy/daily?days=7", nil)) + var d struct { + Days []map[string]any `json:"days"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &d); err != nil { + t.Fatal(err) + } + today := d.Days[len(d.Days)-1] + if today["import_wh"] != float64(10) || today["pv_wh"] != float64(2.5) || today["load_wh"] != nil || today["bat_charged_wh"] != nil || today["history_quality"] != "incomplete" { + t.Fatal(today) + } + if d.Days[0]["import_wh"] != nil { + t.Fatal("empty day shown as measured zero", d.Days[0]) + } +} + +func TestLiveHistoryQualitySeparatesOfflineFromInvalidBalance(t *testing.T) { + tel := telemetry.NewStore() + for _, name := range []string{"meter", "battery"} { + tel.EnsureDriverHealth(name) + } + tel.Update("meter", telemetry.DerMeter, 1200, nil, nil) + tel.RecordDriverSuccess("meter") + tel.DriverHealthMut("battery").SetOffline() + srv := New(&Deps{Tel: tel, HistoryMeasurementSources: func() telemetry.ForecastOptions { + return telemetry.ForecastOptions{ExpectedFlows: []telemetry.ForecastFlow{{Driver: "battery", DerType: telemetry.DerBattery}}} + }}) + resp := map[string]any{} + srv.qualifyStatusHistory(resp, "meter") + if resp["grid_w"] != float64(1200) || resp["bat_w"] != nil || resp["load_w"] != nil || resp["history_quality"] != "incomplete" { + t.Fatal(resp) + } + tel.Update("battery", telemetry.DerBattery, 2000, nil, nil) + tel.RecordDriverSuccess("battery") + srv.qualifyStatusHistory(resp, "meter") + if resp["bat_w"] != float64(2000) || resp["load_w"] != nil || resp["history_quality"] != "balance_mismatch" { + t.Fatal(resp) + } + tel.Update("battery", telemetry.DerBattery, 0, nil, nil) + tel.RecordDriverSuccess("battery") + srv.qualifyStatusHistory(resp, "meter") + if resp["bat_w"] != float64(0) || resp["load_w"] != float64(1200) || resp["history_quality"] != "complete" { + t.Fatal(resp) + } +} diff --git a/go/internal/state/history_dashboard.go b/go/internal/state/history_dashboard.go index d884d992a..daac8b7ff 100644 --- a/go/internal/state/history_dashboard.go +++ b/go/internal/state/history_dashboard.go @@ -59,7 +59,12 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi } if p.TsMs > previous { if previous > 0 && p.TsMs-previous <= maxCostIntegrationGap.Milliseconds() { - if err := addSiteIntervalTx(ctx, tx, previous, p); err != nil { + var previousMissing HistoryMissing + err := tx.QueryRowContext(ctx, `SELECT value FROM history_site_cursor WHERE key='missing'`).Scan(&previousMissing) + if err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + if err := addSiteIntervalTx(ctx, tx, previous, p, previousMissing); err != nil { return err } } @@ -67,6 +72,11 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi return err } } + if p.TsMs > previous { + if _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('missing',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, p.Missing); err != nil { + return err + } + } start := bucketStart(p.TsMs, HistoryResolutionMS) var n int64 var seen []byte @@ -83,10 +93,17 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi json=CASE WHEN excluded.last_ms>=last_ms THEN excluded.json ELSE json END,last_ms=MAX(last_ms,excluded.last_ms), grid_w=(grid_w*n+excluded.grid_w)/ (n+1),pv_w=(pv_w*n+excluded.pv_w)/(n+1),bat_w=(bat_w*n+excluded.bat_w)/(n+1),load_w=(load_w*n+excluded.load_w)/(n+1),bat_soc=(bat_soc*n+excluded.bat_soc)/(n+1),n=n+1,seen_ms=excluded.seen_ms`, start, HistoryResolutionMS, p.TsMs, p.TsMs, 1, p.GridW, p.PVW, p.BatW, p.LoadW, p.BatSoC, p.JSON, seen) - return err + if err != nil { + return err + } + return putHistoryQuality(ctx, tx, "dashboard", start, HistoryResolutionMS, p.Missing) } -func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *HistoryPoint) error { +func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *HistoryPoint, previousMissing HistoryMissing) error { + point := *p + point.Missing |= previousMissing + point.ClearMissingValues() + p = &point for from := previous; from < p.TsMs; { start := bucketStart(from, ArchiveResolutionMS) to := min(p.TsMs, start+ArchiveResolutionMS) @@ -101,10 +118,13 @@ func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *Histo _, err := tx.ExecContext(ctx, `INSERT INTO history_site_energy VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(start_ms,resolution_ms) DO UPDATE SET first_ms=MIN(first_ms,excluded.first_ms),last_ms=MAX(last_ms,excluded.last_ms),covered_ms=covered_ms+excluded.covered_ms,n=n+excluded.n, import_wh=import_wh+excluded.import_wh,export_wh=export_wh+excluded.export_wh,pv_wh=pv_wh+excluded.pv_wh,bat_charge_wh=bat_charge_wh+excluded.bat_charge_wh,bat_discharge_wh=bat_discharge_wh+excluded.bat_discharge_wh,load_wh=load_wh+excluded.load_wh,ev_wh=ev_wh+excluded.ev_wh`, - start, ArchiveResolutionMS, from, to, ms, intervals, math.Max(0, p.GridW)*hours, math.Max(0, -p.GridW)*hours, -p.PVW*hours, math.Max(0, p.BatW)*hours, math.Max(0, -p.BatW)*hours, p.LoadW*hours, math.Max(0, p.GridW-p.BatW-p.PVW-p.LoadW)*hours) + start, ArchiveResolutionMS, from, to, ms, intervals, math.Max(0, p.GridW)*hours, math.Max(0, -p.GridW)*hours, -p.PVW*hours, math.Max(0, p.BatW)*hours, math.Max(0, -p.BatW)*hours, p.LoadW*hours, siteEVWh(p, hours)) if err != nil { return err } + if err := putHistoryQuality(ctx, tx, "energy", start, ArchiveResolutionMS, p.Missing); err != nil { + return err + } from = to } return nil @@ -119,7 +139,15 @@ func (s *Store) siteEnergyFrom(ctx context.Context) (int64, bool, error) { return from, err == nil, err } +func siteEVWh(p *HistoryPoint, hours float64) float64 { + if p.Missing&(MissingGrid|MissingPV|MissingBattery|MissingLoad|MissingEV|MissingV2X) != 0 { + return 0 + } + return math.Max(0, p.GridW-p.BatW-p.PVW-p.LoadW) * hours +} + type siteEnergyRow struct { + missing HistoryMissing start, width, first, last, covered, n int64 imp, exp, pv, charge, discharge, load, ev float64 } @@ -130,14 +158,14 @@ type siteEnergyRow struct { func (s *Store) walkSiteEnergy(ctx context.Context, from, to int64, visit func(siteEnergyRow) error) error { ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - rows, err := s.history.QueryContext(ctx, `SELECT * FROM history_site_energy WHERE start_ms>=? AND start_ms=? AND last_ms<=? ORDER BY start_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to) + rows, err := s.history.QueryContext(ctx, `SELECT e.*,COALESCE(q.missing,0) FROM history_site_energy e LEFT JOIN history_quality q ON q.kind='energy' AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms WHERE e.start_ms>=? AND e.start_ms=? AND last_ms<=? ORDER BY e.start_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to) if err != nil { return err } defer rows.Close() for rows.Next() { var r siteEnergyRow - if err := rows.Scan(&r.start, &r.width, &r.first, &r.last, &r.covered, &r.n, &r.imp, &r.exp, &r.pv, &r.charge, &r.discharge, &r.load, &r.ev); err != nil { + if err := rows.Scan(&r.start, &r.width, &r.first, &r.last, &r.covered, &r.n, &r.imp, &r.exp, &r.pv, &r.charge, &r.discharge, &r.load, &r.ev, &r.missing); err != nil { return err } if err := visit(r); err != nil { @@ -157,6 +185,7 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner out.BatDischargedWh += r.discharge out.LoadWh += r.load out.Intervals += r.n + out.Missing |= r.missing return nil }) return out, err diff --git a/go/internal/state/history_dashboard_cost.go b/go/internal/state/history_dashboard_cost.go index 1d7411e2f..926a84218 100644 --- a/go/internal/state/history_dashboard_cost.go +++ b/go/internal/state/history_dashboard_cost.go @@ -39,6 +39,9 @@ func (s *Store) integrateHistoryRange(ctx context.Context, since, until int64, s out.ImportWh += r.imp out.ExportWh += r.exp out.LoadWh += r.load + if r.missing&(MissingGrid|MissingLoad) != 0 { + return nil + } out.HistoryCoveredMs += r.covered for idx < len(slots) && slots[idx].EndMs <= r.first { idx++ @@ -90,6 +93,9 @@ func (s *Store) ImportWhIntervals(ctx context.Context, intervals [][2]int64) ([] idx++ } if idx < len(intervals) && intervals[idx][0] <= r.first && intervals[idx][1] >= r.last { + if r.missing&MissingGrid != 0 { + return nil + } wh[idx] += r.imp covered[idx] += r.covered } diff --git a/go/internal/state/history_dashboard_retention.go b/go/internal/state/history_dashboard_retention.go index 2c32f43e9..bcf213cd0 100644 --- a/go/internal/state/history_dashboard_retention.go +++ b/go/internal/state/history_dashboard_retention.go @@ -22,10 +22,14 @@ func (s *Store) maintainDashboard(ctx context.Context, now time.Time) error { for { var n int64 err := s.writeArchiveBatch(ctx, func(ctx context.Context, tx *sql.Tx) error { + if _, err := tx.ExecContext(ctx, `DELETE FROM history_quality WHERE kind='dashboard' AND (start_ms,resolution_ms) IN (SELECT start_ms,resolution_ms FROM history_dashboard WHERE last_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND d.start_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms= 1 && i <= 3 { + p.Missing = MissingBattery | MissingLoad | MissingSoC + p.ClearMissingValues() + } + dashboardTick(t, s, fmt.Sprint(i), &p) + } + check := func(st *Store) { + t.Helper() + points, err := st.LoadHistory(ms, ms+120000, 1) + if err != nil || len(points) != 1 { + t.Fatal(points, err) + } + p := points[0] + if p.GridW != 1200 || p.PVW != -300 || p.Missing != MissingBattery|MissingLoad|MissingSoC { + t.Fatal(p) + } + d, err := st.DailyEnergy(ms, ms+120000) + if err != nil { + t.Fatal(err) + } + near(t, "independent grid import", d.ImportWh, 20) + near(t, "independent PV generation", d.PVWh, 5) + if d.Missing&MissingLoad == 0 || d.Missing&MissingGrid != 0 { + t.Fatal(d) + } + // The recovery endpoint must not fill the preceding offline interval. + near(t, "known battery charge", d.BatChargedWh, float64(200*(21000+20000))/3600000) + if err := st.SaveDailyEnergy("2026-10-12", d); err != nil { + t.Fatal(err) + } + cached, ok, err := st.LoadDailyEnergy("2026-10-12") + if err != nil || !ok || cached.Missing != d.Missing || cached.Intervals != d.Intervals { + t.Fatal(cached, ok, err) + } + } + check(s) + for _, age := range []time.Duration{48 * time.Hour, 40 * 24 * time.Hour, 732 * 24 * time.Hour} { + if err := s.maintainDashboard(context.Background(), base.Add(age)); err != nil { + t.Fatal(err) + } + // Chart retention ends at two years, energy quality keeps its quarter-hour evidence. + if age < AggregateRetention { + check(s) + } else { + d, err := s.DailyEnergy(ms, ms+120000) + if err != nil || d.Missing&MissingLoad == 0 || d.ImportWh != 20 { + t.Fatal(d, err) + } + } + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + s, err = Open(path) + if err != nil { + t.Fatal(err) + } + defer s.Close() + d, err := s.DailyEnergy(ms, ms+120000) + if err != nil || d.Missing&MissingLoad == 0 { + t.Fatal(d, err) + } + dst := filepath.Join(t.TempDir(), "backup.gz") + if err := s.BackupToCompressed(dst); err != nil { + t.Fatal(err) + } + in, err := os.Open(dst) + if err != nil { + t.Fatal(err) + } + defer in.Close() + z, err := gzip.NewReader(in) + if err != nil { + t.Fatal(err) + } + defer z.Close() + restoredPath := filepath.Join(t.TempDir(), "state.db") + out, err := os.Create(restoredPath) + if err != nil { + t.Fatal(err) + } + if _, err := io.Copy(out, z); err != nil { + t.Fatal(err) + } + if err := out.Close(); err != nil { + t.Fatal(err) + } + restored, err := Open(restoredPath) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + got, err := restored.DailyEnergy(ms, ms+120000) + if err != nil || got.Missing != d.Missing || got.ImportWh != d.ImportWh { + t.Fatal(got, err) + } +} diff --git a/go/internal/state/history_schema.go b/go/internal/state/history_schema.go index 71779be6a..17e86785e 100644 --- a/go/internal/state/history_schema.go +++ b/go/internal/state/history_schema.go @@ -2,7 +2,7 @@ package state var historySchema = append(append([]string{}, sqliteLegacyHistoryStmts...), `CREATE TABLE IF NOT EXISTS ts_latest (driver_id INTEGER NOT NULL,metric_id INTEGER NOT NULL,ts_ms INTEGER NOT NULL,value REAL NOT NULL,PRIMARY KEY(driver_id,metric_id)) WITHOUT ROWID`, - dashboardSchema, siteEnergySchema, siteEnergyCursorSchema, + dashboardSchema, siteEnergySchema, siteEnergyCursorSchema, historyQualitySchema, historyDailyQualitySchema, `CREATE INDEX IF NOT EXISTS idx_dashboard_last ON history_dashboard(last_ms)`, `CREATE INDEX IF NOT EXISTS idx_dashboard_age ON history_dashboard(resolution_ms,start_ms)`, `CREATE INDEX IF NOT EXISTS idx_site_energy_age ON history_site_energy(resolution_ms,start_ms)`, diff --git a/go/internal/state/history_sqlite.go b/go/internal/state/history_sqlite.go index b5eac2949..8d850a986 100644 --- a/go/internal/state/history_sqlite.go +++ b/go/internal/state/history_sqlite.go @@ -24,7 +24,7 @@ var historyTables = []string{ "history_hot", "history_warm", "history_cold", "ts_drivers", "ts_metrics", "ts_samples", "energy_daily", "energy_ledger_meta", "energy_assets", "energy_ledger_entries", "energy_ledger_cursors", } -var sqliteHistoryTables = append(append([]string{}, historyTables...), "ts_series_hour", "ts_archive_days", "ts_buckets", "ts_bucket_days", "ts_legacy_bucket_days", "ts_aggregate_hours", "ts_latest", "history_dashboard", "history_site_energy", "history_site_cursor") +var sqliteHistoryTables = append(append([]string{}, historyTables...), "ts_series_hour", "ts_archive_days", "ts_buckets", "ts_bucket_days", "ts_legacy_bucket_days", "ts_aggregate_hours", "ts_latest", "history_dashboard", "history_site_energy", "history_site_cursor", "history_quality", "history_daily_quality") func ensureHistorySchema(exec func(string) error) error { for _, stmt := range historySchema { @@ -374,7 +374,9 @@ func historyOrder(table string) string { return " ORDER BY driver_id, metric_id, hour_ms" case "ts_samples": return " ORDER BY driver_id, metric_id, ts_ms" - case "energy_daily": + case "history_quality": + return " ORDER BY kind, start_ms, resolution_ms" + case "energy_daily", "history_daily_quality": return " ORDER BY day" case "energy_assets": return " ORDER BY asset_id" diff --git a/go/internal/state/history_writer.go b/go/internal/state/history_writer.go index f9b35ff93..b3bcc5f60 100644 --- a/go/internal/state/history_writer.go +++ b/go/internal/state/history_writer.go @@ -150,6 +150,9 @@ func (s *Store) EnqueueTelemetryTick(p *HistoryPoint, samples []Sample, observat if w == nil { return errors.New("history writer is unavailable") } + if p != nil && p.Missing != 0 && !s.aggregateHistory.Load() { + return w.reject("partial history requires aggregate storage") + } if err := validateHistorySamples(samples); err != nil { return w.reject(err.Error()) } diff --git a/go/internal/state/store.go b/go/internal/state/store.go index 0555ec114..e8d95e0c0 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -25,7 +25,7 @@ const ( // SchemaVersion identifies the on-disk state format for update rollback. // Increase it before a release that cannot safely reopen the same state.db // with the prior Core version. - SchemaVersion = 7 + SchemaVersion = 8 // LegacyReleaseMarker is the value release notes publish in the legacy // `` marker. Only Cores before v3.6.0-beta.1 // (schema 4 and older) read it. Those Cores copied their whole history @@ -1376,6 +1376,7 @@ func (s *Store) batteryModelKey(driverName string) string { // HistoryPoint is one row of the history table. type HistoryPoint struct { + Missing HistoryMissing N int64 `json:"n,omitempty"` ResolutionMS int64 `json:"resolution_ms,omitempty"` FirstMS int64 `json:"first_ms,omitempty"` @@ -1390,6 +1391,9 @@ type HistoryPoint struct { // RecordHistory inserts a live hot-tier entry into SQLite. func (s *Store) RecordHistory(p HistoryPoint) error { + if p.Missing != 0 { + return errors.New("partial history must use the aggregate tick writer") + } var normalizeErr error p, normalizeErr = normalizeHistoryPoint(p) if normalizeErr != nil { @@ -1412,6 +1416,9 @@ func (s *Store) RecordHistory(p HistoryPoint) error { // A retry is safe: the last history point for each timestamp wins. func (s *Store) BulkRecordHistory(pts []HistoryPoint) error { for _, p := range pts { + if p.Missing != 0 { + return errors.New("partial history must use the aggregate tick writer") + } if _, err := normalizeHistoryPoint(p); err != nil { return err } @@ -1477,15 +1484,15 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, // COALESCE to 0 so NULL columns (from partial aggregations) scan cleanly. const tierUnion = ` WITH all_rows AS ( - SELECT ts_ms, grid_w, pv_w, bat_w, load_w, bat_soc, 0 AS tier,1 AS n,1 AS resolution_ms,ts_ms AS first_ms FROM history_hot + SELECT ts_ms, grid_w, pv_w, bat_w, load_w, bat_soc, 0 AS tier,1 AS n,1 AS resolution_ms,ts_ms AS first_ms,0 AS missing FROM history_hot WHERE ts_ms BETWEEN ? AND ? UNION ALL - SELECT ts_ms, grid_w, pv_w, bat_w, load_w, bat_soc, 1,1,900000,ts_ms FROM history_warm + SELECT ts_ms, grid_w, pv_w, bat_w, load_w, bat_soc, 1,1,900000,ts_ms,0 FROM history_warm WHERE ts_ms BETWEEN ? AND ? UNION ALL - SELECT ts_ms, grid_w, pv_w, bat_w, load_w, bat_soc, 2,1,86400000,ts_ms FROM history_cold + SELECT ts_ms, grid_w, pv_w, bat_w, load_w, bat_soc, 2,1,86400000,ts_ms,0 FROM history_cold WHERE ts_ms BETWEEN ? AND ? - UNION ALL SELECT last_ms,grid_w,pv_w,bat_w,load_w,bat_soc,-1,n,resolution_ms,first_ms FROM history_dashboard WHERE last_ms BETWEEN ? AND ? + UNION ALL SELECT last_ms,grid_w,pv_w,bat_w,load_w,bat_soc,-1,n,d.resolution_ms,first_ms,COALESCE(q.missing,0) FROM history_dashboard d LEFT JOIN history_quality q ON q.kind='dashboard' AND q.start_ms=d.start_ms AND q.resolution_ms=d.resolution_ms WHERE last_ms BETWEEN ? AND ? ), deduped AS ( SELECT * FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY ts_ms ORDER BY tier) AS rn FROM all_rows) WHERE rn=1 @@ -1505,14 +1512,14 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, SELECT MAX(ts_ms) AS ts_ms, SUM(COALESCE(grid_w, 0)*n)/SUM(n) AS grid_w, SUM(COALESCE(pv_w, 0)*n)/SUM(n) AS pv_w, SUM(COALESCE(bat_w, 0)*n)/SUM(n) AS bat_w, SUM(COALESCE(load_w, 0)*n)/SUM(n) AS load_w, - SUM(COALESCE(bat_soc, 0)*n)/SUM(n) AS bat_soc,SUM(n) AS n,MAX(resolution_ms) AS resolution_ms,MIN(first_ms) AS first_ms + SUM(COALESCE(bat_soc, 0)*n)/SUM(n) AS bat_soc,SUM(n) AS n,MAX(resolution_ms) AS resolution_ms,MIN(first_ms) AS first_ms,`+historyMissingAggregate+` AS missing FROM deduped GROUP BY (ts_ms - ?) / ? ) SELECT ts_ms,grid_w,pv_w,bat_w,load_w,bat_soc, COALESCE((SELECT json FROM history_dashboard WHERE last_ms=b.ts_ms), (SELECT json FROM history_hot WHERE ts_ms=b.ts_ms), (SELECT json FROM history_warm WHERE ts_ms=b.ts_ms), - (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms + (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms,missing FROM bucketed b ORDER BY ts_ms`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, bucketMs) } else { rows, err = s.history.QueryContext(ctx, tierUnion+` @@ -1520,7 +1527,7 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, COALESCE((SELECT json FROM history_dashboard WHERE last_ms=b.ts_ms), (SELECT json FROM history_hot WHERE ts_ms=b.ts_ms), (SELECT json FROM history_warm WHERE ts_ms=b.ts_ms), - (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms + (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms,missing FROM deduped b ORDER BY ts_ms LIMIT ?`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, maxRawSeriesPoints+1) } if err != nil { @@ -1532,7 +1539,7 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, var bytes int for rows.Next() { var p HistoryPoint - if err := rows.Scan(&p.TsMs, &p.GridW, &p.PVW, &p.BatW, &p.LoadW, &p.BatSoC, &p.JSON, &p.N, &p.ResolutionMS, &p.FirstMS); err != nil { + if err := rows.Scan(&p.TsMs, &p.GridW, &p.PVW, &p.BatW, &p.LoadW, &p.BatSoC, &p.JSON, &p.N, &p.ResolutionMS, &p.FirstMS, &p.Missing); err != nil { return all, err } if p.ResolutionMS > 1 { @@ -1558,6 +1565,7 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, // Import/Export are grid-boundary W split on sign; PV and LoadWh are always // positive accumulations; BatCharged/BatDischarged split bat_w on sign. type DayEnergy struct { + Missing HistoryMissing ImportWh float64 ExportWh float64 PVWh float64 @@ -1713,13 +1721,13 @@ func (s *Store) dailyEnergyFromArchive(sinceMs, untilMs int64) (DayEnergy, error // authoritative — no TTL, no staleness check needed. func (s *Store) LoadDailyEnergy(day string) (DayEnergy, bool, error) { const q = ` - SELECT import_wh, export_wh, pv_wh, bat_charged_wh, bat_discharged_wh, load_wh - FROM energy_daily WHERE day = ? + SELECT import_wh, export_wh, pv_wh, bat_charged_wh, bat_discharged_wh, load_wh,q.missing,q.n + FROM energy_daily d JOIN history_daily_quality q ON q.day=d.day WHERE d.day = ? ` var d DayEnergy err := s.history.QueryRow(q, day).Scan( &d.ImportWh, &d.ExportWh, &d.PVWh, - &d.BatChargedWh, &d.BatDischargedWh, &d.LoadWh, + &d.BatChargedWh, &d.BatDischargedWh, &d.LoadWh, &d.Missing, &d.Intervals, ) if err == sql.ErrNoRows { return DayEnergy{}, false, nil @@ -1750,12 +1758,23 @@ func (s *Store) SaveDailyEnergy(day string, de DayEnergy) error { load_wh = excluded.load_wh, computed_at_ms = excluded.computed_at_ms ` - _, err := s.history.Exec(q, day, + tx, err := s.history.Begin() + if err != nil { + return err + } + defer tx.Rollback() + _, err = tx.Exec(q, day, de.ImportWh, de.ExportWh, de.PVWh, de.BatChargedWh, de.BatDischargedWh, de.LoadWh, time.Now().UnixMilli(), ) - return err + if err != nil { + return err + } + if _, err := tx.Exec(`INSERT INTO history_daily_quality VALUES(?,?,?) ON CONFLICT(day) DO UPDATE SET missing=excluded.missing,n=excluded.n`, day, de.Missing, de.Intervals); err != nil { + return err + } + return tx.Commit() } // CountHistoryWithoutMarker returns the number of history rows across all diff --git a/go/internal/state/store_ts.go b/go/internal/state/store_ts.go index 961a8aa95..738202e11 100644 --- a/go/internal/state/store_ts.go +++ b/go/internal/state/store_ts.go @@ -370,6 +370,9 @@ func (s *Store) RecordTickWithOptionalHistory(p *HistoryPoint, samples []Sample, // commit succeed and will never retry it again. Its current, possibly uncertain // commit retains its receipt until a later batch succeeds. func (s *Store) recordHistoryBatch(ctx context.Context, batchID, payloadHash string, p *HistoryPoint, samples []Sample, observations []EnergyObservation, acknowledgedSequence int64) (int64, error) { + if p != nil && p.Missing != 0 && !s.aggregateHistory.Load() { + return 0, errors.New("partial history requires aggregate storage") + } if err := validateHistorySamples(samples); err != nil { return 0, err } diff --git a/go/internal/telemetry/forecast.go b/go/internal/telemetry/forecast.go index 8ed431ad5..c9c1aa43d 100644 --- a/go/internal/telemetry/forecast.go +++ b/go/internal/telemetry/forecast.go @@ -33,12 +33,21 @@ type ForecastOptions struct { ExpectedFlows []ForecastFlow MaxAge time.Duration MaxSkew time.Duration + UseWatchdogTimeout bool `json:",omitempty"` } // ForecastReading is one coherent snapshot, in site signs and raw watts. // Valid describes the household balance. PVValid only describes the PV sources. // Neither flag claims that the instantaneous sample covers a time interval. +type ForecastSource struct { + Flow ForecastFlow + PowerW *float64 + SoC *float64 +} + type ForecastReading struct { + Known map[DerType]bool + Sources map[string]ForecastSource At, Earliest, Latest, PVEarliest, PVLatest time.Time GridW, PVW, BatteryW, EVW, V2XW, HouseholdW float64 Valid, PVValid bool @@ -68,8 +77,11 @@ func (s *Store) ForecastMeasurementNow(siteMeter string, opts ForecastOptions) F } func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts ForecastOptions) ForecastReading { - out := ForecastReading{At: now, Valid: true, PVValid: true} + out := ForecastReading{At: now, Valid: true, PVValid: true, Known: map[DerType]bool{DerMeter: true, DerPV: true, DerBattery: true, DerEV: true, DerV2X: true}, Sources: map[string]ForecastSource{}} if s == nil { + for kind := range out.Known { + out.Known[kind] = false + } out.Valid = false out.PVValid = false out.Reason = "no_telemetry" @@ -116,6 +128,7 @@ func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts if f.FlowID != "" { if old, ok := ids[f.FlowID]; ok { fail("duplicate_flow:"+f.FlowID, f.DerType == DerPV || old.DerType == DerPV) + out.Known[f.DerType], out.Known[old.DerType] = false, false } ids[f.FlowID] = f } @@ -126,6 +139,7 @@ func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts if f.DerType == DerEV { if _, ok := flows[f.Driver+":"+DerV2X.String()]; ok { fail("duplicate_charger:"+f.Driver, false) + out.Known[DerEV], out.Known[DerV2X] = false, false } } } @@ -143,6 +157,12 @@ func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts pvCount++ } r := s.readings[k] + source := ForecastSource{Flow: f} + out.Sources[k] = source + maxAge := opts.MaxAge + if opts.UseWatchdogTimeout && s.health[f.Driver] != nil && s.health[f.Driver].WatchdogTimeoutOverride > 0 { + maxAge = s.health[f.Driver].WatchdogTimeoutOverride + } reason := "" var powerW float64 var powerAt time.Time @@ -167,7 +187,7 @@ func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts reason = "unknown_power:" case math.IsNaN(powerW) || math.IsInf(powerW, 0): reason = "nonfinite:" - case powerAt.IsZero() || now.Sub(powerAt) > opts.MaxAge: + case powerAt.IsZero() || now.Sub(powerAt) > maxAge: reason = "stale:" case powerAt.After(now): reason = "future:" @@ -176,8 +196,15 @@ func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts } if reason != "" { fail(reason+k, pv) + out.Known[f.DerType] = false continue } + source.PowerW = &powerW + if r.SoC != nil && !math.IsNaN(*r.SoC) && !math.IsInf(*r.SoC, 0) && *r.SoC >= 0 && *r.SoC <= 1 { + soc := *r.SoC + source.SoC = &soc + } + out.Sources[k] = source if out.Earliest.IsZero() || powerAt.Before(out.Earliest) { out.Earliest = powerAt } @@ -219,6 +246,11 @@ func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts if math.IsNaN(out.PVW) || math.IsInf(out.PVW, 0) { fail("nonfinite_pv_sum", true) } + for kind, w := range map[DerType]float64{DerMeter: out.GridW, DerPV: out.PVW, DerBattery: out.BatteryW, DerEV: out.EVW, DerV2X: out.V2XW} { + if math.IsNaN(w) || math.IsInf(w, 0) { + out.Known[kind] = false + } + } out.HouseholdW = out.GridW - out.PVW - out.BatteryW - out.EVW - out.V2XW if math.IsNaN(out.HouseholdW) || math.IsInf(out.HouseholdW, 0) || out.HouseholdW < 0 { fail("invalid_balance", false) diff --git a/state-schema.json b/state-schema.json index 4625a3260..778427295 100644 --- a/state-schema.json +++ b/state-schema.json @@ -1,4 +1,4 @@ { - "version": 7, + "version": 8, "legacy_marker": 4 } diff --git a/web/app.js b/web/app.js index ffb2e8e1f..b0fb9ed64 100644 --- a/web/app.js +++ b/web/app.js @@ -95,6 +95,7 @@ load: [], timestamps: [], // Energy counters (cumulative Wh, today-scoped) + quality: [], e_import: [], e_export: [], e_pv: [], @@ -155,12 +156,12 @@ } // Ensure a battery-driver slot exists. Backfills bat/target arrays - // with zeros up to current timestamps length so row indices align + // with unknowns up to the preceding timestamp so row indices align // with the other chartHistory series. function ensureBatteryDriver(name) { if (chartBatteries[name]) return chartBatteries[name]; - var pad = chartHistory.timestamps.length; - var slot = { bat: new Array(pad).fill(0), target: new Array(pad).fill(0) }; + var pad = Math.max(0, chartHistory.timestamps.length - 1); + var slot = { bat: new Array(pad).fill(null), target: new Array(pad).fill(null) }; chartBatteries[name] = slot; syncBatteryLegend(); return slot; @@ -212,8 +213,8 @@ } function ensureEVDriver(name) { if (chartEVs[name]) return chartEVs[name]; - var pad = chartHistory.timestamps.length; - var slot = { ev: new Array(pad).fill(0) }; + var pad = Math.max(0, chartHistory.timestamps.length - 1); + var slot = { ev: new Array(pad).fill(null) }; chartEVs[name] = slot; syncEVLegend(); return slot; @@ -374,12 +375,14 @@ } function formatKwh(wh) { - var kwh = (wh || 0) / 1000; + if (wh == null) return "Unknown"; + var kwh = wh / 1000; if (kwh >= 100) return kwh.toFixed(0) + " kWh"; if (kwh >= 10) return kwh.toFixed(1) + " kWh"; return kwh.toFixed(2) + " kWh"; } function formatSignedWh(wh) { + if (wh == null) return "Unknown"; var v = Number(wh) || 0; var abs = Math.abs(v); var sign = v > 0.5 ? "+" : v < -0.5 ? "-" : "±"; @@ -387,6 +390,7 @@ return sign + Math.round(abs) + " Wh"; } function formatWhMagnitude(wh) { + if (wh == null) return "Unknown"; var abs = Math.abs(Number(wh) || 0); if (abs >= 1000) return (abs / 1000).toFixed(2) + " kWh"; return Math.round(abs) + " Wh"; @@ -472,13 +476,13 @@ drivers[name] = d; }); out.drivers = drivers; - if (have.pv_w) out.pv_w = sums.pv_w; - if (have.bat_w) out.bat_w = sums.bat_w; - if (have.ev_w) { + if (have.pv_w && data.pv_w != null) out.pv_w = sums.pv_w; + if (have.bat_w && data.bat_w != null) out.bat_w = sums.bat_w; + if (have.ev_w && data.ev_w != null) { out.ev_w = sums.ev_w; out.ev_charging_w = sums.ev_w; } - if (out.grid_w != null) { + if (out.grid_w != null && data.load_w != null) { out.load_w = Math.max(0, (out.grid_w || 0) - (out.bat_w || 0) - (out.pv_w || 0) - (out.ev_w || 0)); } else { // No live meter: do not keep a previous load or treat JSON null as 0 W. @@ -491,13 +495,14 @@ if (!windowMs || windowMs <= 0 || values.length < 3) return values; var out = new Array(values.length); for (var i = 0; i < values.length; i++) { + if (values[i] == null) { out[i] = null; continue; } var t = timestamps[i]; var sum = 0; var n = 0; for (var j = i; j >= 0; j--) { if (t - timestamps[j] > windowMs) break; var v = values[j]; - if (v == null || !isFinite(v)) continue; + if (v == null || !isFinite(v)) break; sum += v; n++; } @@ -682,8 +687,8 @@ // SEPARATELY within the slot — the bill is import × import_price // plus export × export_price, never their net. Render both // directions so the operator sees the slot's true exposure. - var impWh = Number(slot.import_wh) || 0; - var expWh = Number(slot.export_wh) || 0; + var impWh = slot.import_wh; + var expWh = slot.export_wh; slotBadge.textContent = "15m ↑" + formatWhMagnitude(impWh) + " ↓" + formatWhMagnitude(expWh); slotBadge.className = "grid-slot-badge" + (impWh > expWh + 5 ? " slot-import" : expWh > impWh + 5 ? " slot-export" : ""); @@ -706,7 +711,7 @@ return online && d.pv_w != null; }); var pvConfigured = Object.keys(data.drivers || {}).some(function (name) { - return (data.drivers[name] || {}).pv_w != null; + return Object.prototype.hasOwnProperty.call(data.drivers[name] || {}, "pv_w"); }); var pvDir = document.getElementById("pv-dir"); if (pvConfigured && !pvLive) { @@ -733,10 +738,10 @@ var cardEvWEl = document.getElementById("card-ev-w"); var cardEvSubEl = document.getElementById("card-ev-sub"); if (cardEvWEl) { - var evWNow = data.ev_charging_w || 0; + var evWNow = data.ev_charging_w; cardEvWEl.textContent = formatW(evWNow); cardEvWEl.className = evWNow > 1 ? "card-value val-load" : "card-value val-neutral"; - if (cardEvSubEl) cardEvSubEl.textContent = evWNow > 1 ? "charging" : "charger"; + if (cardEvSubEl) cardEvSubEl.textContent = evWNow == null ? "no data" : evWNow > 1 ? "charging" : "charger"; } // Battery — positive=charge, negative=discharge. Same honesty as solar: @@ -749,7 +754,7 @@ return online && d.bat_w != null; }); var batConfigured = Object.keys(data.drivers || {}).some(function (name) { - return (data.drivers[name] || {}).bat_w != null; + return Object.prototype.hasOwnProperty.call(data.drivers[name] || {}, "bat_w"); }); if (batConfigured && !batLive) { batW.textContent = "—"; @@ -1045,12 +1050,13 @@ chartHistory.pv.push(data.pv_w); chartHistory.load.push(data.load_w); chartHistory.timestamps.push(now); - chartHistory.e_import.push(t.import_wh || 0); - chartHistory.e_export.push(t.export_wh || 0); - chartHistory.e_pv.push(t.pv_wh || 0); - chartHistory.e_charged.push(t.bat_charged_wh || 0); - chartHistory.e_discharged.push(t.bat_discharged_wh || 0); - chartHistory.e_load.push(t.load_wh || 0); + chartHistory.quality.push(data.history_quality || "complete"); + chartHistory.e_import.push(t.import_wh == null ? null : t.import_wh); + chartHistory.e_export.push(t.export_wh == null ? null : t.export_wh); + chartHistory.e_pv.push(t.pv_wh == null ? null : t.pv_wh); + chartHistory.e_charged.push(t.bat_charged_wh == null ? null : t.bat_charged_wh); + chartHistory.e_discharged.push(t.bat_discharged_wh == null ? null : t.bat_discharged_wh); + chartHistory.e_load.push(t.load_wh == null ? null : t.load_wh); // Per-battery-driver push. Any driver in data.drivers that exposes // bat_w is considered battery-capable and gets its own chart series. @@ -1059,31 +1065,28 @@ var seenEVs = {}; Object.keys(drivers).forEach(function (name) { var d = drivers[name] || {}; - if (d.bat_w != null) { + if (Object.prototype.hasOwnProperty.call(d, "bat_w")) { seenBatteries[name] = true; var bslot = ensureBatteryDriver(name); - bslot.bat.push(d.bat_w || 0); + bslot.bat.push(d.status === "offline" || d.bat_w == null ? null : d.bat_w); bslot.target.push((targetsByDriver && targetsByDriver[name]) || 0); } - if (d.ev_w != null) { + if (Object.prototype.hasOwnProperty.call(d, "ev_w")) { seenEVs[name] = true; var eslot = ensureEVDriver(name); - eslot.ev.push(d.ev_w || 0); + eslot.ev.push(d.status === "offline" || d.ev_w == null ? null : d.ev_w); } }); - // Drivers that have history but didn't report this cycle: push a - // 0 so index alignment with chartHistory.timestamps stays intact. - // (Keeps the line continuous; the driver offline/gap is already - // visible through the driver card's status indicator.) + // Preserve alignment while leaving a gap for an unavailable device. Object.keys(chartBatteries).forEach(function (name) { if (seenBatteries[name]) return; var slot = chartBatteries[name]; - slot.bat.push(0); - slot.target.push(0); + slot.bat.push(null); + slot.target.push(null); }); Object.keys(chartEVs).forEach(function (name) { if (seenEVs[name]) return; - chartEVs[name].ev.push(0); + chartEVs[name].ev.push(null); }); if (chartHistory.grid.length > CHART_POINTS) { @@ -1139,6 +1142,14 @@ var windowMs = CHART_RANGE_MS[chartRange] || CHART_RANGE_MS["5m"]; var now = Date.now(); var windowStart = now - windowMs; + var qualityNote = document.getElementById("chart-history-quality"); + if (qualityNote) { + var qualities = chartHistory.quality.filter(function (_,i) { return chartHistory.timestamps[i] >= windowStart; }); + var mismatch = qualities.indexOf("balance_mismatch") >= 0; + var partial = mismatch || qualities.indexOf("incomplete") >= 0; + qualityNote.hidden = !partial; + qualityNote.textContent = mismatch ? "Measurements do not balance. Home use is unknown for those samples." : "Some measurements are missing. Working devices keep recording; gaps are unknown."; + } // Forward-looking forecast: at least 15 minutes so the transition // from actual→predicted is smooth. For longer ranges, use half the // past window. Keeps the live trace dominant on wide views while @@ -1153,7 +1164,7 @@ var smoothedPVSeries = null; var smoothedLoadSeries = null; if (chartView === "energy") { - function toKwh(arr) { return arr.map(function(x){ return x / 1000; }); } + function toKwh(arr) { return arr.map(function(x){ return x == null ? null : x / 1000; }); } series = [ { data: toKwh(chartHistory.e_import), color: "#ef4444", width: 2, dash: [], name: "Import", fill: true }, { data: toKwh(chartHistory.e_export), color: "#22c55e", width: 2, dash: [], name: "Export", fill: true }, @@ -1302,12 +1313,14 @@ series.forEach(function (sr) { if (sr.data.length < 2 || chartHistory.timestamps.length < 2) return; - var pts = []; + var segments = [], pts = []; for (var j = 0; j < sr.data.length; j++) { var ts = chartHistory.timestamps[j]; - if (sr.data[j] == null) continue; + if (sr.data[j] == null) { if (pts.length) segments.push(pts); pts = []; continue; } pts.push({ x: tsToX(ts), y: valToY(sr.data[j]) }); } + if (pts.length) segments.push(pts); + segments.forEach(function (pts) { if (pts.length < 2) return; if (sr.fill) { @@ -1340,6 +1353,7 @@ if (sr.width >= 2) { liveTips.push({ x: pts[pts.length - 1].x, y: pts[pts.length - 1].y, color: sr.color }); } + }); }); // ---- Forward-looking forecast (dashed lines from plan) ---- @@ -4909,16 +4923,17 @@ chartEVs = {}; data.items.forEach(function (it) { var et = it.energy_today || {}; - chartHistory.grid.push(it.grid_w || 0); - chartHistory.pv.push(it.pv_w || 0); - chartHistory.load.push(it.load_w || 0); + chartHistory.grid.push(it.grid_w == null ? null : it.grid_w); + chartHistory.pv.push(it.pv_w == null ? null : it.pv_w); + chartHistory.load.push(it.load_w == null ? null : it.load_w); chartHistory.timestamps.push(it.ts || 0); - chartHistory.e_import.push(et.import_wh || 0); - chartHistory.e_export.push(et.export_wh || 0); - chartHistory.e_pv.push(et.pv_wh || 0); - chartHistory.e_charged.push(et.bat_charged_wh || 0); - chartHistory.e_discharged.push(et.bat_discharged_wh || 0); - chartHistory.e_load.push(et.load_wh || 0); + chartHistory.quality.push(it.history_quality || "complete"); + chartHistory.e_import.push(et.import_wh == null ? null : et.import_wh); + chartHistory.e_export.push(et.export_wh == null ? null : et.export_wh); + chartHistory.e_pv.push(et.pv_wh == null ? null : et.pv_wh); + chartHistory.e_charged.push(et.bat_charged_wh == null ? null : et.bat_charged_wh); + chartHistory.e_discharged.push(et.bat_discharged_wh == null ? null : et.bat_discharged_wh); + chartHistory.e_load.push(et.load_wh == null ? null : et.load_wh); // Per-driver discovery from this item's drivers + targets maps. // A driver can expose bat_w, ev_w, or both. @@ -4927,26 +4942,26 @@ var seenBat = {}, seenEV = {}; Object.keys(itDrivers).forEach(function (name) { var d = itDrivers[name] || {}; - if (d.bat_w != null) { + if (Object.prototype.hasOwnProperty.call(d, "bat_w")) { seenBat[name] = true; var bslot = ensureBatteryDriver(name); - bslot.bat.push(d.bat_w || 0); - bslot.target.push(itTargets[name] || 0); + bslot.bat.push(d.status === "offline" || d.bat_w == null ? null : d.bat_w); + bslot.target.push(itTargets[name] == null ? null : itTargets[name]); } - if (d.ev_w != null) { + if (Object.prototype.hasOwnProperty.call(d, "ev_w")) { seenEV[name] = true; - ensureEVDriver(name).ev.push(d.ev_w || 0); + ensureEVDriver(name).ev.push(d.ev_w == null ? null : d.ev_w); } }); Object.keys(chartBatteries).forEach(function (name) { if (seenBat[name]) return; var slot = chartBatteries[name]; - slot.bat.push(0); - slot.target.push(0); + slot.bat.push(null); + slot.target.push(null); }); Object.keys(chartEVs).forEach(function (name) { if (seenEV[name]) return; - chartEVs[name].ev.push(0); + chartEVs[name].ev.push(null); }); }); syncBatteryLegend(); @@ -5029,6 +5044,7 @@ totals.export_wh += arr[i].export_wh || 0; totals.pv_wh += arr[i].pv_wh || 0; } + if (arr.some(function (day) { return ["import_wh", "load_wh", "export_wh", "pv_wh"].some(function (key) { return day[key] == null; }); })) { totals.incomplete = true; } historyCakeEl.setTotals(totals); }) .catch(function () { /* network blip — leave the previous render */ }) @@ -5091,11 +5107,10 @@ // the row blob. Skip rows with no battery sample so the chart // doesn't draw spurious zero bars. var points = items - .filter(function (it) { return it.bat_w != null || it.bat_soc != null; }) .map(function (it) { return { ts_ms: it.ts || it.ts_ms, - bat_w: it.bat_w || 0, + bat_w: it.bat_w == null ? null : it.bat_w, soc_pct: (it.bat_soc != null) ? it.bat_soc * 100 : null, }; }); @@ -5114,6 +5129,7 @@ ctx.clearRect(0, 0, cssW, cssH); var C = chartColors(); + if (!points.some(function(p) { return p.bat_w != null || p.soc_pct != null; })) { ctx.fillStyle = C.muted; ctx.font = "13px system-ui"; ctx.fillText("Battery measurements unknown", 16, 28); return; } var pad = { left: 36, right: 36, top: 8, bottom: 18 }; var plotW = cssW - pad.left - pad.right; var plotH = cssH - pad.top - pad.bottom; @@ -5176,7 +5192,7 @@ var started = false; for (var k = 0; k < points.length; k++) { var sp = points[k]; - if (sp.soc_pct == null) continue; + if (sp.soc_pct == null) { started = false; continue; } var sx = xOf(sp.ts_ms); var sy = socOf(sp.soc_pct); if (!started) { ctx.moveTo(sx, sy); started = true; } diff --git a/web/components/ftw-bar-chart.js b/web/components/ftw-bar-chart.js index d7465a6ba..e3d85947f 100644 --- a/web/components/ftw-bar-chart.js +++ b/web/components/ftw-bar-chart.js @@ -255,6 +255,7 @@ class FtwBarChart extends FtwElement { let max = 0, sum = 0, count = 0; for (const d of this._data) { + if (d.value == null) continue; const v = Number(d.value) || 0; if (v > max) max = v; sum += v; @@ -263,6 +264,7 @@ class FtwBarChart extends FtwElement { const avg = count > 0 ? sum / count : 0; const colsSvg = this._data.map((d) => { + if (d.value == null) return `
?
`; const v = Number(d.value) || 0; // 2% floor keeps tiny-but-nonzero values visible; gate on v>0 so // truly-zero buckets render as empty columns, not 2% slivers. diff --git a/web/components/ftw-energy-cake.js b/web/components/ftw-energy-cake.js index 6c4911151..79bf78eda 100644 --- a/web/components/ftw-energy-cake.js +++ b/web/components/ftw-energy-cake.js @@ -161,6 +161,7 @@ class FtwEnergyCake extends FtwElement { if (!this._totals) { return `
No data yet.
`; } + if (this._totals.incomplete) return `
Incomplete measurements. The energy split is unknown.
`; const importWh = Math.max(0, Number(this._totals.import_wh) || 0); const loadWh = Math.max(0, Number(this._totals.load_wh) || 0); const exportWh = Math.max(0, Number(this._totals.export_wh) || 0); diff --git a/web/components/ftw-history-card.js b/web/components/ftw-history-card.js index 2cb46d355..662db8719 100644 --- a/web/components/ftw-history-card.js +++ b/web/components/ftw-history-card.js @@ -359,11 +359,13 @@ class FtwHistoryCard extends FtwElement { .then((resp) => { if (seq !== this._reqSeq) return; const buckets = (resp && resp.days) || []; - let sum = 0; + let sum = 0, known = 0; // Pass kWh (not Wh) as `value` so the bar-chart's avg label // matches the per-bar displayValue units — otherwise the chart // would show bars as "10.5" and the avg line as "10500". const data = buckets.map((b) => { + if (b[field] == null) return {label:fmtDayShort(b.day), value:null, displayValue:"?", title:fmtDayShort(b.day)+": Unknown"}; + known++; const wh = Number(b[field]) || 0; sum += wh; const kwh = wh / 1000; @@ -383,14 +385,16 @@ class FtwHistoryCard extends FtwElement { // via a trailing so the avg reads // as secondary text (smaller, muted) instead of a second // prominent number. - if (data.length) { + if (known > 0 && known < data.length) { + this._totalEl.textContent = fmtKwh(sum) + " · incomplete period"; + } else if (known > 0) { const avgWh = sum / data.length; this._totalEl.innerHTML = escapeHtml(fmtKwh(sum) + " total") + ' (' + escapeHtml(fmtKwh(avgWh) + " avg") + ")"; } else { - this._totalEl.textContent = "— kWh"; + this._totalEl.textContent = "Unknown"; } this._chart.data = data; }; diff --git a/web/energy-history.js b/web/energy-history.js index a9cbb0add..d6b2e775a 100644 --- a/web/energy-history.js +++ b/web/energy-history.js @@ -46,13 +46,13 @@ } function sum(days, key) { - return days.reduce(function (total, day) { - var value = Number(day[key] || 0); - return total + (Number.isFinite(value) && value > 0 ? value : 0); - }, 0); + var known = days.filter(function (day) { return day[key] != null && Number.isFinite(Number(day[key])); }); + if (!known.length) return null; + return known.reduce(function (total, day) { return total + Math.max(0, Number(day[key])); }, 0); } function formatKWh(wh) { + if (wh == null) return "Unknown"; var kwh = Math.max(0, Number(wh) || 0) / 1000; var digits = kwh < 10 ? 1 : 0; return kwh.toLocaleString(undefined, { @@ -93,7 +93,8 @@ ]; summary.innerHTML = cards.map(function (card) { return '
' + card.label + '' + - formatKWh(sum(days, card.key)) + '' + card.note + '
'; + formatKWh(sum(days, card.key)) + '' + + (days.some(function (day) { return day[card.key] == null; }) ? 'Incomplete period · known days only' : card.note) + ''; }).join(''); } @@ -104,6 +105,10 @@ target.innerHTML = '

What stands out

No energy history is available for this period.

'; return; } + if (days.some(function (day) { return dailySeries.some(function (series) { return day[series.key] == null; }) || day.history_quality === 'balance_mismatch'; })) { + target.innerHTML = '

Incomplete measurements

Working devices keep recording. Missing values are unknown; totals cover known days only. Home use needs all measured flows to balance.

'; + return; + } var totalLoad = sum(days, 'load_wh'); var totalPV = sum(days, 'pv_wh'); var charged = sum(days, 'bat_charged_wh'); @@ -142,7 +147,7 @@ if (legend) { legend.innerHTML = dailySeries.map(function (series) { return '' + series.label + ''; - }).join(''); + }).join('') + (days.some(function (day) { return dailySeries.some(function (series) { return day[series.key] == null; }); }) ? '? Unknown' : ''); } if (!days.length) { ctx.fillStyle = C.muted; @@ -181,6 +186,11 @@ var totalBarsW = dailySeries.length * barW + (dailySeries.length - 1) * barGap; var startX = pad.left + groupW * dayIndex + (groupW - totalBarsW) / 2; dailySeries.forEach(function (series, seriesIndex) { + if (day[series.key] == null) { + ctx.fillStyle = C.muted; ctx.textAlign = 'center'; + ctx.fillText('?', startX + seriesIndex * (barW + barGap) + barW / 2, pad.top + plotH - 4); + return; + } var value = Math.max(0, Number(day[series.key]) || 0); var barH = plotH * value / maxWh; ctx.fillStyle = series.color; @@ -284,7 +294,7 @@ } else if (uncertain) { note.className = 'energy-history-quality-note'; note.textContent = uncertain + ' gap or counter reset marker' + (uncertain === 1 ? ' is' : 's are') + - ' recorded in this period. Trusted daily totals stay above.'; + ' recorded in this period. Daily totals and missing measurements stay above.'; } else { note.className = 'energy-history-quality-note good'; note.textContent = 'No counter jumps or data gaps were found in this selection.'; diff --git a/web/history-quality.test.mjs b/web/history-quality.test.mjs new file mode 100644 index 000000000..67b8776a4 --- /dev/null +++ b/web/history-quality.test.mjs @@ -0,0 +1,33 @@ +import assert from 'node:assert/strict'; +import {readFileSync} from 'node:fs'; +import test from 'node:test'; +import vm from 'node:vm'; +const app=readFileSync(new URL('./app.js',import.meta.url),'utf8'); +const daily=readFileSync(new URL('./energy-history.js',import.meta.url),'utf8'); +function fn(source,name,next) {return source.slice(source.indexOf(' function '+name+'('),source.indexOf(' function '+next+'('));} + +test('chart smoothing leaves missing samples and does not cross a gap',()=>{ + const c={};vm.createContext(c);vm.runInContext(fn(app,'smoothSeriesForChart','statusClass'),c); + const result=c.smoothSeriesForChart([100,200,null,800,1000],[0,1000,2000,3000,4000],10000); + assert.deepEqual(Array.from(result),[100,150,null,800,900]); +}); + +test('daily totals distinguish no measurements, measured zero and partial coverage',()=>{ + const c={};vm.createContext(c);vm.runInContext(fn(daily,'sum','formatDay'),c); + assert.equal(c.sum([{import_wh:null}],'import_wh'),null); + assert.equal(c.formatKWh(null),'Unknown'); + assert.equal(c.sum([{import_wh:0}],'import_wh'),0); + assert.equal(c.sum([{import_wh:null},{import_wh:2500}],'import_wh'),2500); +}); + +test('daily view renders unknown home use and a measured grid total',()=>{ + const els={};for(const id of ['energy-history-summary','energy-history-insight']) els[id]={innerHTML:''}; + const c={document:{getElementById:id=>els[id]||null},dailySeries:[{key:'load_wh'},{key:'pv_wh'},{key:'import_wh'},{key:'export_wh'}]}; + vm.createContext(c);vm.runInContext(fn(daily,'sum','formatDay')+fn(daily,'renderSummary','drawDailyChart'),c); + const days=[{day:'2026-10-12',load_wh:null,pv_wh:3000,import_wh:1200,export_wh:0,bat_charged_wh:null,bat_discharged_wh:null,history_quality:'incomplete'}]; + c.renderSummary(days);c.renderInsights(days); + assert.match(els['energy-history-summary'].innerHTML,/Unknown/); + assert.match(els['energy-history-summary'].innerHTML,/1.2 kWh/); + assert.match(els['energy-history-insight'].innerHTML,/Incomplete measurements/); + assert.doesNotMatch(els['energy-history-insight'].innerHTML,/%/); +}); diff --git a/web/index.html b/web/index.html index 3cd2177f3..9233aff44 100644 --- a/web/index.html +++ b/web/index.html @@ -435,6 +435,7 @@

Live

+
Grid diff --git a/web/style.css b/web/style.css index 004f9ef95..5f6d2f814 100644 --- a/web/style.css +++ b/web/style.css @@ -2344,3 +2344,5 @@ body.ftw-app .forecast-card .twin-profile-toggle button { min-height: 40px; font .cs-row[data-tone="warning"] { border-color: var(--amber); } .cs-row[data-tone="alarm"] { border-color: var(--red-e); } .cs-device { color: var(--fg-dim); } + +.history-data-warning { color: var(--accent-e, #f5b942); font-size: 13px; margin: 8px 0; } From 9ba70a6ad6c18a81bb334a649245eff23d9ae2f2 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 12 Oct 2026 20:10:02 +0200 Subject: [PATCH 2/3] fix(history): keep partial rows safe for schema-7 native rollback Signed-off-by: Fredrik Ahlgren --- .changeset/history-offline-devices.md | 2 +- docs/architecture.md | 16 ++-- go/internal/state/history_dashboard.go | 68 +++++++++++++---- .../state/history_dashboard_retention.go | 74 ++++++++++++++----- go/internal/state/history_plain.go | 2 +- go/internal/state/history_quality_test.go | 49 ++++++++++++ go/internal/state/history_schema.go | 7 ++ go/internal/state/history_sqlite.go | 6 +- go/internal/state/store.go | 15 ++-- state-schema.json | 2 +- 10 files changed, 192 insertions(+), 49 deletions(-) diff --git a/.changeset/history-offline-devices.md b/.changeset/history-offline-devices.md index 923ea6680..7bb6e22b9 100644 --- a/.changeset/history-offline-devices.md +++ b/.changeset/history-offline-devices.md @@ -2,4 +2,4 @@ "ftw": patch --- -Keep history for working devices when another device is offline. Show missing power, state of charge and energy as unknown, preserve gaps through retention and backup, and flag incomplete measurements or an invalid household balance. State schema 8 preserves this evidence; return to an older Core with its matching full backup. +Keep history for working devices when another device is offline. Show missing power, state of charge and energy as unknown, preserve gaps through retention and backup, and flag incomplete measurements or an invalid household balance. Separate additive schema-7 tables keep partial rows out of older readers and preserve them across a native rollback. diff --git a/docs/architecture.md b/docs/architecture.md index 66db7245f..2859f474f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -37,11 +37,6 @@ A separate module requires a concrete benefit and: A percentage goal and a goal to reach the car's own limit are distinct. State schema 7 blocks rollback to a Core that would ignore the saved goal mode. -State schema 8 adds durable missing-value evidence for chart and energy buckets. -An offline flow leaves other devices recording. Missing inputs make only their -values and the derived household balance unknown. Retention, downsampling and -full backups keep that evidence; a recovered sample cannot fill an earlier gap. -Use a matching full backup when returning to an older Core. `GET /api/loadpoints` advertises `vehicle_limit_goal_supported`; clients must require that flag before saving `schedule.finish_at_vehicle_limit`. Existing percentage goals keep their meaning. In vehicle-limit mode, the planner uses @@ -63,6 +58,17 @@ a reading after the observed connection and no measured charging. An ambiguous match cannot finish the goal. A charger declining current is reported as a refusal, never as an invented battery level or proof that the target was reached. +### History measurement quality + +An offline flow leaves working devices recording. Missing inputs make their +values and the derived household balance unknown. Separate additive schema-7 +tables preserve partial chart and energy rows, including missing-value evidence, +through retention, downsampling and full backup. A recovered sample cannot fill +an earlier gap. Older Cores see only complete rows in the original tables; +their integration cursor cannot bridge a partial interval after rollback. +Partial rows remain stored and become visible again on the newer Core. + + ### Control state and disk writes Core loads EV restart records before starting control. During operation, a diff --git a/go/internal/state/history_dashboard.go b/go/internal/state/history_dashboard.go index daac8b7ff..0c995d2c7 100644 --- a/go/internal/state/history_dashboard.go +++ b/go/internal/state/history_dashboard.go @@ -24,18 +24,38 @@ const siteEnergyCursorSchema = `CREATE TABLE IF NOT EXISTS history_site_cursor ( // A missing usable site snapshot breaks integration even if it recovers before // the maximum gap. The cursor and chart/energy rows share the tick receipt. func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoint) error { + // Older schema-7 Cores read only the original tables. Keep partial rows + // separate, and clear their cursor so rollback cannot bridge a gap. + if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO history_partial_cursor SELECT key,value FROM history_site_cursor`); err != nil { + return err + } + var legacyLast, partialLast int64 + if err := tx.QueryRowContext(ctx, `SELECT COALESCE((SELECT value FROM history_site_cursor WHERE key='last'),0),COALESCE((SELECT value FROM history_partial_cursor WHERE key='last'),0)`).Scan(&legacyLast, &partialLast); err != nil { + return err + } + if legacyLast > partialLast { + if _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, legacyLast); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('missing',0) ON CONFLICT(key) DO UPDATE SET value=0`); err != nil { + return err + } + } if p == nil { - _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('last',0) ON CONFLICT(key) DO UPDATE SET value=0`) + if _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('last',0) ON CONFLICT(key) DO UPDATE SET value=0`); err != nil { + return err + } + _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('last',0) ON CONFLICT(key) DO UPDATE SET value=0`) return err } var previous int64 - err := tx.QueryRowContext(ctx, `SELECT value FROM history_site_cursor WHERE key='last'`).Scan(&previous) + err := tx.QueryRowContext(ctx, `SELECT value FROM history_partial_cursor WHERE key='last'`).Scan(&previous) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } hadCursor := err == nil var from int64 - err = tx.QueryRowContext(ctx, `SELECT value FROM history_site_cursor WHERE key='from'`).Scan(&from) + err = tx.QueryRowContext(ctx, `SELECT value FROM history_partial_cursor WHERE key='from'`).Scan(&from) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } @@ -53,14 +73,14 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi } else { previous = 0 } - if _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('from',?)`, from); err != nil { + if _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('from',?)`, from); err != nil { return err } } if p.TsMs > previous { if previous > 0 && p.TsMs-previous <= maxCostIntegrationGap.Milliseconds() { var previousMissing HistoryMissing - err := tx.QueryRowContext(ctx, `SELECT value FROM history_site_cursor WHERE key='missing'`).Scan(&previousMissing) + err := tx.QueryRowContext(ctx, `SELECT value FROM history_partial_cursor WHERE key='missing'`).Scan(&previousMissing) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } @@ -68,19 +88,35 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi return err } } - if _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, p.TsMs); err != nil { + if _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, p.TsMs); err != nil { + return err + } + } + if p.TsMs > previous { + if _, err := tx.ExecContext(ctx, `INSERT INTO history_partial_cursor VALUES('missing',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, p.Missing); err != nil { return err } } if p.TsMs > previous { - if _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('missing',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, p.Missing); err != nil { + legacyLast := p.TsMs + if p.Missing != 0 { + legacyLast = 0 + } + if _, err := tx.ExecContext(ctx, `INSERT INTO history_site_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, legacyLast); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO history_site_cursor SELECT key,value FROM history_partial_cursor WHERE key='from'`); err != nil { return err } } + table, kind := "history_dashboard", "dashboard" + if p.Missing != 0 { + table, kind = "history_dashboard_partial", "partial_dashboard" + } start := bucketStart(p.TsMs, HistoryResolutionMS) var n int64 var seen []byte - err = tx.QueryRowContext(ctx, `SELECT n,seen_ms FROM history_dashboard WHERE start_ms=? AND resolution_ms=?`, start, HistoryResolutionMS).Scan(&n, &seen) + err = tx.QueryRowContext(ctx, `SELECT n,seen_ms FROM `+table+` WHERE start_ms=? AND resolution_ms=?`, start, HistoryResolutionMS).Scan(&n, &seen) if err != nil && !errors.Is(err, sql.ErrNoRows) { return err } @@ -88,7 +124,7 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi if !added { return nil } - _, err = tx.ExecContext(ctx, `INSERT INTO history_dashboard VALUES(?,?,?,?,?,?,?,?,?,?,?,?) + _, err = tx.ExecContext(ctx, `INSERT INTO `+table+` VALUES(?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(start_ms,resolution_ms) DO UPDATE SET first_ms=MIN(first_ms,excluded.first_ms), json=CASE WHEN excluded.last_ms>=last_ms THEN excluded.json ELSE json END,last_ms=MAX(last_ms,excluded.last_ms), grid_w=(grid_w*n+excluded.grid_w)/ (n+1),pv_w=(pv_w*n+excluded.pv_w)/(n+1),bat_w=(bat_w*n+excluded.bat_w)/(n+1),load_w=(load_w*n+excluded.load_w)/(n+1),bat_soc=(bat_soc*n+excluded.bat_soc)/(n+1),n=n+1,seen_ms=excluded.seen_ms`, @@ -96,7 +132,7 @@ func (s *Store) recordDashboardTx(ctx context.Context, tx *sql.Tx, p *HistoryPoi if err != nil { return err } - return putHistoryQuality(ctx, tx, "dashboard", start, HistoryResolutionMS, p.Missing) + return putHistoryQuality(ctx, tx, kind, start, HistoryResolutionMS, p.Missing) } func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *HistoryPoint, previousMissing HistoryMissing) error { @@ -104,6 +140,10 @@ func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *Histo point.Missing |= previousMissing point.ClearMissingValues() p = &point + table, kind := "history_site_energy", "energy" + if p.Missing != 0 { + table, kind = "history_site_energy_partial", "partial_energy" + } for from := previous; from < p.TsMs; { start := bucketStart(from, ArchiveResolutionMS) to := min(p.TsMs, start+ArchiveResolutionMS) @@ -115,14 +155,14 @@ func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *Histo intervals = 1 } hours := float64(ms) / 3600000 - _, err := tx.ExecContext(ctx, `INSERT INTO history_site_energy VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?) + _, err := tx.ExecContext(ctx, `INSERT INTO `+table+` VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(start_ms,resolution_ms) DO UPDATE SET first_ms=MIN(first_ms,excluded.first_ms),last_ms=MAX(last_ms,excluded.last_ms),covered_ms=covered_ms+excluded.covered_ms,n=n+excluded.n, import_wh=import_wh+excluded.import_wh,export_wh=export_wh+excluded.export_wh,pv_wh=pv_wh+excluded.pv_wh,bat_charge_wh=bat_charge_wh+excluded.bat_charge_wh,bat_discharge_wh=bat_discharge_wh+excluded.bat_discharge_wh,load_wh=load_wh+excluded.load_wh,ev_wh=ev_wh+excluded.ev_wh`, start, ArchiveResolutionMS, from, to, ms, intervals, math.Max(0, p.GridW)*hours, math.Max(0, -p.GridW)*hours, -p.PVW*hours, math.Max(0, p.BatW)*hours, math.Max(0, -p.BatW)*hours, p.LoadW*hours, siteEVWh(p, hours)) if err != nil { return err } - if err := putHistoryQuality(ctx, tx, "energy", start, ArchiveResolutionMS, p.Missing); err != nil { + if err := putHistoryQuality(ctx, tx, kind, start, ArchiveResolutionMS, p.Missing); err != nil { return err } from = to @@ -132,7 +172,7 @@ func addSiteIntervalTx(ctx context.Context, tx *sql.Tx, previous int64, p *Histo func (s *Store) siteEnergyFrom(ctx context.Context) (int64, bool, error) { var from int64 - err := s.history.QueryRowContext(ctx, `SELECT value FROM history_site_cursor WHERE key='from'`).Scan(&from) + err := s.history.QueryRowContext(ctx, `SELECT value FROM history_partial_cursor WHERE key='from' UNION ALL SELECT value FROM history_site_cursor WHERE key='from' LIMIT 1`).Scan(&from) if errors.Is(err, sql.ErrNoRows) { return 0, false, nil } @@ -158,7 +198,7 @@ type siteEnergyRow struct { func (s *Store) walkSiteEnergy(ctx context.Context, from, to int64, visit func(siteEnergyRow) error) error { ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - rows, err := s.history.QueryContext(ctx, `SELECT e.*,COALESCE(q.missing,0) FROM history_site_energy e LEFT JOIN history_quality q ON q.kind='energy' AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms WHERE e.start_ms>=? AND e.start_ms=? AND last_ms<=? ORDER BY e.start_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to) + rows, err := s.history.QueryContext(ctx, `WITH energy AS (SELECT *, 'energy' AS kind FROM history_site_energy UNION ALL SELECT *, 'partial_energy' AS kind FROM history_site_energy_partial) SELECT e.start_ms,e.resolution_ms,e.first_ms,e.last_ms,e.covered_ms,e.n,e.import_wh,e.export_wh,e.pv_wh,e.bat_charge_wh,e.bat_discharge_wh,e.load_wh,e.ev_wh,COALESCE(q.missing,0) FROM energy e LEFT JOIN history_quality q ON q.kind=e.kind AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms WHERE e.start_ms>=? AND e.start_ms=? AND last_ms<=? ORDER BY e.start_ms`, bucketStart(from, EnergyLedgerDailyBucketMS), to, from, to) if err != nil { return err } diff --git a/go/internal/state/history_dashboard_retention.go b/go/internal/state/history_dashboard_retention.go index bcf213cd0..34d7e71d2 100644 --- a/go/internal/state/history_dashboard_retention.go +++ b/go/internal/state/history_dashboard_retention.go @@ -19,13 +19,34 @@ func (s *Store) maintainDashboard(ctx context.Context, now time.Time) error { if err := s.rollSiteEnergy(ctx, now.Add(-AggregateRetention).UnixMilli()); err != nil { return err } + for _, partial := range []bool{false, true} { + if err := s.pruneDashboard(ctx, now, partial); err != nil { + return err + } + } + return nil +} +func dashboardStorage(partial bool) (string, string) { + if partial { + return "history_dashboard_partial", "partial_dashboard" + } + return "history_dashboard", "dashboard" +} +func energyStorage(partial bool) (string, string) { + if partial { + return "history_site_energy_partial", "partial_energy" + } + return "history_site_energy", "energy" +} +func (s *Store) pruneDashboard(ctx context.Context, now time.Time, partial bool) error { + table, kind := dashboardStorage(partial) for { var n int64 err := s.writeArchiveBatch(ctx, func(ctx context.Context, tx *sql.Tx) error { - if _, err := tx.ExecContext(ctx, `DELETE FROM history_quality WHERE kind='dashboard' AND (start_ms,resolution_ms) IN (SELECT start_ms,resolution_ms FROM history_dashboard WHERE last_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms 0 && partial[0]) rows, err := tx.QueryContext(ctx, `SELECT d.start_ms,first_ms,last_ms,n,grid_w,pv_w,bat_w,load_w,bat_soc,json,COALESCE(q.missing,0) - FROM history_dashboard d LEFT JOIN history_quality q ON q.kind='dashboard' AND q.start_ms=d.start_ms AND q.resolution_ms=d.resolution_ms WHERE d.resolution_ms=? AND d.start_ms>=? AND d.start_ms=? AND d.start_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms=? AND start_ms` marker. Only Cores before v3.6.0-beta.1 // (schema 4 and older) read it. Those Cores copied their whole history @@ -1493,6 +1493,7 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, SELECT ts_ms, grid_w, pv_w, bat_w, load_w, bat_soc, 2,1,86400000,ts_ms,0 FROM history_cold WHERE ts_ms BETWEEN ? AND ? UNION ALL SELECT last_ms,grid_w,pv_w,bat_w,load_w,bat_soc,-1,n,d.resolution_ms,first_ms,COALESCE(q.missing,0) FROM history_dashboard d LEFT JOIN history_quality q ON q.kind='dashboard' AND q.start_ms=d.start_ms AND q.resolution_ms=d.resolution_ms WHERE last_ms BETWEEN ? AND ? + UNION ALL SELECT last_ms,grid_w,pv_w,bat_w,load_w,bat_soc,-2,n,d.resolution_ms,first_ms,COALESCE(q.missing,0) FROM history_dashboard_partial d LEFT JOIN history_quality q ON q.kind='partial_dashboard' AND q.start_ms=d.start_ms AND q.resolution_ms=d.resolution_ms WHERE last_ms BETWEEN ? AND ? ), deduped AS ( SELECT * FROM (SELECT *, ROW_NUMBER() OVER (PARTITION BY ts_ms ORDER BY tier) AS rn FROM all_rows) WHERE rn=1 @@ -1516,19 +1517,19 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, FROM deduped GROUP BY (ts_ms - ?) / ? ) SELECT ts_ms,grid_w,pv_w,bat_w,load_w,bat_soc, - COALESCE((SELECT json FROM history_dashboard WHERE last_ms=b.ts_ms), + COALESCE((SELECT json FROM history_dashboard_partial WHERE last_ms=b.ts_ms),(SELECT json FROM history_dashboard WHERE last_ms=b.ts_ms), (SELECT json FROM history_hot WHERE ts_ms=b.ts_ms), (SELECT json FROM history_warm WHERE ts_ms=b.ts_ms), (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms,missing - FROM bucketed b ORDER BY ts_ms`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, bucketMs) + FROM bucketed b ORDER BY ts_ms`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, bucketMs) } else { rows, err = s.history.QueryContext(ctx, tierUnion+` SELECT ts_ms,COALESCE(grid_w,0),COALESCE(pv_w,0),COALESCE(bat_w,0),COALESCE(load_w,0),COALESCE(bat_soc,0), - COALESCE((SELECT json FROM history_dashboard WHERE last_ms=b.ts_ms), + COALESCE((SELECT json FROM history_dashboard_partial WHERE last_ms=b.ts_ms),(SELECT json FROM history_dashboard WHERE last_ms=b.ts_ms), (SELECT json FROM history_hot WHERE ts_ms=b.ts_ms), (SELECT json FROM history_warm WHERE ts_ms=b.ts_ms), (SELECT json FROM history_cold WHERE ts_ms=b.ts_ms),'{}'),n,resolution_ms,first_ms,missing - FROM deduped b ORDER BY ts_ms LIMIT ?`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, maxRawSeriesPoints+1) + FROM deduped b ORDER BY ts_ms LIMIT ?`, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, sinceMs, untilMs, maxRawSeriesPoints+1) } if err != nil { return nil, err @@ -1722,7 +1723,7 @@ func (s *Store) dailyEnergyFromArchive(sinceMs, untilMs int64) (DayEnergy, error func (s *Store) LoadDailyEnergy(day string) (DayEnergy, bool, error) { const q = ` SELECT import_wh, export_wh, pv_wh, bat_charged_wh, bat_discharged_wh, load_wh,q.missing,q.n - FROM energy_daily d JOIN history_daily_quality q ON q.day=d.day WHERE d.day = ? + FROM energy_daily_partial d JOIN history_daily_quality q ON q.day=d.day WHERE d.day = ? ` var d DayEnergy err := s.history.QueryRow(q, day).Scan( @@ -1746,7 +1747,7 @@ func (s *Store) SaveDailyEnergy(day string, de DayEnergy) error { s.historyWriteMu.Lock() defer s.historyWriteMu.Unlock() const q = ` - INSERT INTO energy_daily( + INSERT INTO energy_daily_partial( day, import_wh, export_wh, pv_wh, bat_charged_wh, bat_discharged_wh, load_wh, computed_at_ms ) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(day) DO UPDATE SET diff --git a/state-schema.json b/state-schema.json index 778427295..4625a3260 100644 --- a/state-schema.json +++ b/state-schema.json @@ -1,4 +1,4 @@ { - "version": 8, + "version": 7, "legacy_marker": 4 } From c0fdaa55215b002871386098a437279c62d5060a Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 12 Oct 2026 20:14:02 +0200 Subject: [PATCH 3/3] fix(history): qualify unmeasured PV and preserve known battery SoC Signed-off-by: Fredrik Ahlgren --- go/cmd/ftw/history_partial_test.go | 12 ++++++ go/internal/api/api_history_quality.go | 17 ++++---- go/internal/api/api_history_quality_test.go | 44 ++++++++++++++++++++- go/internal/telemetry/forecast.go | 3 ++ 4 files changed, 67 insertions(+), 9 deletions(-) diff --git a/go/cmd/ftw/history_partial_test.go b/go/cmd/ftw/history_partial_test.go index fcb2032d8..13728826c 100644 --- a/go/cmd/ftw/history_partial_test.go +++ b/go/cmd/ftw/history_partial_test.go @@ -1,6 +1,7 @@ package main import ( + "github.com/srcfl/ftw/go/internal/state" "github.com/srcfl/ftw/go/internal/telemetry" "testing" "time" @@ -20,3 +21,14 @@ func TestHistoryRetainsLiveMeterWhenBatteryOffline(t *testing.T) { t.Fatalf("live meter lost when battery offline: available=%v point=%+v", available, p) } } + +func TestHistoryConfiguredPVWithoutMeasurementIsUnknown(t *testing.T) { + tel := telemetry.NewStore() + tel.EnsureDriverHealth("meter") + tel.Update("meter", telemetry.DerMeter, 1200, nil, nil) + tel.RecordDriverSuccess("meter") + p, ok := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "meter"}, time.Now().UnixMilli(), time.Minute, telemetry.ForecastOptions{HouseholdInvalidReason: "configured_pv_without_measurement_source"}) + if !ok || p.Missing&state.MissingPV == 0 { + t.Fatalf("unmeasured configured PV reported as zero: %+v", p) + } +} diff --git a/go/internal/api/api_history_quality.go b/go/internal/api/api_history_quality.go index 12a5b76c5..08e6a6d18 100644 --- a/go/internal/api/api_history_quality.go +++ b/go/internal/api/api_history_quality.go @@ -47,28 +47,29 @@ func (s *Server) qualifyStatusHistory(resp map[string]any, siteMeter string) { d[field] = source.PowerW } if source.Flow.DerType == telemetry.DerBattery { - d["soc"] = source.SoC + if source.SoC == nil { + d["bat_soc"] = nil + } else { + d["bat_soc"] = *source.SoC + } } } } var count int - var sum float64 for _, source := range r.Sources { if source.Flow.DerType == telemetry.DerBattery { if source.SoC == nil { missing |= state.MissingSoC } count++ - if source.SoC != nil { - sum += *source.SoC - } } } if count == 0 { missing |= state.MissingSoC } - if count > 0 { - sum /= float64(count) + if missing&state.MissingSoC != 0 { + resp["bat_soc"] = nil } - resp["bat_soc"] = missing.Value(state.MissingSoC, sum) + // handleStatus already weights known battery SoCs by capacity. + // Qualification only clears that value when a required input is missing. } diff --git a/go/internal/api/api_history_quality_test.go b/go/internal/api/api_history_quality_test.go index 7520d22f4..92bfd4e49 100644 --- a/go/internal/api/api_history_quality_test.go +++ b/go/internal/api/api_history_quality_test.go @@ -5,6 +5,7 @@ import ( "encoding/json" "github.com/srcfl/ftw/go/internal/state" "github.com/srcfl/ftw/go/internal/telemetry" + "math" "net/http/httptest" "path/filepath" "testing" @@ -55,7 +56,7 @@ func TestPartialHistoryAPIKeepsKnownFlowsAndNullsMissingValues(t *testing.T) { t.Fatal(err) } today := d.Days[len(d.Days)-1] - if today["import_wh"] != float64(10) || today["pv_wh"] != float64(2.5) || today["load_wh"] != nil || today["bat_charged_wh"] != nil || today["history_quality"] != "incomplete" { + if math.Abs(today["import_wh"].(float64)-10) > 1e-9 || math.Abs(today["pv_wh"].(float64)-2.5) > 1e-9 || today["load_wh"] != nil || today["bat_charged_wh"] != nil || today["history_quality"] != "incomplete" { t.Fatal(today) } if d.Days[0]["import_wh"] != nil { @@ -92,3 +93,44 @@ func TestLiveHistoryQualitySeparatesOfflineFromInvalidBalance(t *testing.T) { t.Fatal(resp) } } + +func TestLiveHistoryQualityKeepsWeightedSoCAndClearsStaleDriverSoC(t *testing.T) { + tel := telemetry.NewStore() + for _, name := range []string{"meter", "large", "small"} { + tel.EnsureDriverHealth(name) + } + low, high := .2, .8 + tel.Update("meter", telemetry.DerMeter, 1200, nil, nil) + tel.RecordDriverSuccess("meter") + tel.Update("large", telemetry.DerBattery, 100, &low, nil) + tel.RecordDriverSuccess("large") + tel.Update("small", telemetry.DerBattery, 100, &high, nil) + tel.RecordDriverSuccess("small") + srv := New(&Deps{Tel: tel, HistoryMeasurementSources: func() telemetry.ForecastOptions { return telemetry.ForecastOptions{} }}) + drivers := map[string]any{"large": map[string]any{"bat_soc": low}, "small": map[string]any{"bat_soc": high}} + resp := map[string]any{"bat_soc": .3, "drivers": drivers} + srv.qualifyStatusHistory(resp, "meter") + if resp["bat_soc"] != .3 { + t.Error("capacity-weighted state of charge changed", resp) + } + tel.DriverHealthMut("small").SetOffline() + srv.qualifyStatusHistory(resp, "meter") + if resp["bat_soc"] != nil || drivers["small"].(map[string]any)["bat_soc"] != nil { + t.Fatal("offline battery keeps stale state of charge", resp) + } +} + +func TestLiveHistoryQualityConfiguredPVWithoutMeasurementIsUnknown(t *testing.T) { + tel := telemetry.NewStore() + tel.EnsureDriverHealth("meter") + tel.Update("meter", telemetry.DerMeter, 1200, nil, nil) + tel.RecordDriverSuccess("meter") + srv := New(&Deps{Tel: tel, HistoryMeasurementSources: func() telemetry.ForecastOptions { + return telemetry.ForecastOptions{HouseholdInvalidReason: "configured_pv_without_measurement_source"} + }}) + resp := map[string]any{} + srv.qualifyStatusHistory(resp, "meter") + if resp["pv_w"] != nil || resp["load_w"] != nil { + t.Fatal("unmeasured configured PV reported as zero", resp) + } +} diff --git a/go/internal/telemetry/forecast.go b/go/internal/telemetry/forecast.go index c9c1aa43d..27031690b 100644 --- a/go/internal/telemetry/forecast.go +++ b/go/internal/telemetry/forecast.go @@ -110,6 +110,9 @@ func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts out.PVValid = false out.PVReason = opts.PVInvalidReason } + if opts.PVInvalidReason != "" || opts.HouseholdInvalidReason == "configured_pv_without_measurement_source" { + out.Known[DerPV] = false + } if opts.HouseholdInvalidReason != "" { fail(opts.HouseholdInvalidReason, false) }