From ad16957f341217985c3e1645bde490f520e1ca24 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sun, 18 Oct 2026 17:15:57 +0200 Subject: [PATCH 1/2] fix(history): honor disabled battery reads and retain PV energy --- .changeset/p1-disabled-battery-pv-energy.md | 5 + go/cmd/ftw/energy_history.go | 11 +- go/cmd/ftw/forecast_measurements.go | 6 +- go/cmd/ftw/history_disabled_battery_test.go | 165 ++++++++++++++++++++ go/internal/state/energy_ledger.go | 8 + 5 files changed, 191 insertions(+), 4 deletions(-) create mode 100644 .changeset/p1-disabled-battery-pv-energy.md create mode 100644 go/cmd/ftw/history_disabled_battery_test.go diff --git a/.changeset/p1-disabled-battery-pv-energy.md b/.changeset/p1-disabled-battery-pv-energy.md new file mode 100644 index 000000000..5edcb2d74 --- /dev/null +++ b/.changeset/p1-disabled-battery-pv-energy.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Keep household history when a P1 meter has battery reading turned off. Use observed PV power when a zero energy counter cannot account for reported generation, and retain the final power interval when a usable counter arrives. diff --git a/go/cmd/ftw/energy_history.go b/go/cmd/ftw/energy_history.go index ad5453aac..69fb45d3e 100644 --- a/go/cmd/ftw/energy_history.go +++ b/go/cmd/ftw/energy_history.go @@ -118,9 +118,16 @@ func buildEnergyObservations(st *state.Store, tel *telemetry.Store, ctrl tickPer continue } data := energyNumbers(r.Data) + generationW := math.Max(-r.RawW, 0) + counter := firstEnergyNumber(data, "generation_wh", "total_generation_wh", "lifetime_wh") + // Some PV meters publish zero for an unavailable cumulative register + // while reporting generation. Keep the measured power available for + // bounded integration; a returning counter re-baselines in the ledger. + if counter != nil && *counter == 0 && generationW > 0 { + counter = nil + } appendDirection(id, deviceID, state.AssetPV, r.Driver, - state.FlowPVGeneration, r.UpdatedAt.UnixMilli(), math.Max(-r.RawW, 0), - firstEnergyNumber(data, "generation_wh", "total_generation_wh", "lifetime_wh"), false) + state.FlowPVGeneration, r.UpdatedAt.UnixMilli(), generationW, counter, false) } for _, r := range tel.ReadingsByType(telemetry.DerEV) { diff --git a/go/cmd/ftw/forecast_measurements.go b/go/cmd/ftw/forecast_measurements.go index 064a22a8a..f55b82638 100644 --- a/go/cmd/ftw/forecast_measurements.go +++ b/go/cmd/ftw/forecast_measurements.go @@ -82,12 +82,14 @@ func forecastMeasurementTopology(cfg *config.Config, catalog []drivers.CatalogEn } for _, item := range active { d, e := item.driver, item.entry + readBattery, hasReadBattery := d.Config["read_battery"].(bool) + batteryExpected := (readBattery || d.BatteryCapacityWh > 0 || d.BatteryTelemetryOnly) && (!hasReadBattery || readBattery) if d.IsSiteMeter { add(d, telemetry.DerMeter) } if !item.found { unknown = append(unknown, d.Name+":unknown_driver_capabilities") - if d.BatteryCapacityWh > 0 || d.BatteryTelemetryOnly { + if batteryExpected { add(d, telemetry.DerBattery) } continue @@ -98,7 +100,7 @@ func forecastMeasurementTopology(cfg *config.Config, catalog []drivers.CatalogEn } else if ev { add(d, telemetry.DerEV) } - if bat && (d.BatteryCapacityWh > 0 || d.BatteryTelemetryOnly) { + if bat && batteryExpected { add(d, telemetry.DerBattery) } if pv { diff --git a/go/cmd/ftw/history_disabled_battery_test.go b/go/cmd/ftw/history_disabled_battery_test.go new file mode 100644 index 000000000..c4a027963 --- /dev/null +++ b/go/cmd/ftw/history_disabled_battery_test.go @@ -0,0 +1,165 @@ +package main + +import ( + "context" + "encoding/json" + "math" + "path/filepath" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +func TestHistoryDisabledBatteryKeepsP1Load(t *testing.T) { + for _, knownCatalog := range []bool{true, false} { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + if err := st.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + cfg := &config.Config{Drivers: []config.Driver{ + {Name: "p1", Lua: "drivers/hybrid.lua", IsSiteMeter: true, BatteryCapacityWh: 10000, BatteryTelemetryOnly: true, Config: map[string]any{"read_pv": false, "read_battery": false}}, + {Name: "solar", Lua: "drivers/solar.lua"}, + }} + catalog := measurementCatalog() + if !knownCatalog { + catalog = nil + } + tel := telemetry.NewStore() + tel.Update("p1", telemetry.DerMeter, 1000, nil, nil) + tel.RecordDriverSuccess("p1") + tel.Update("solar", telemetry.DerPV, -2000, nil, nil) + tel.RecordDriverSuccess("solar") + opts := historyMeasurementOptions(cfg, catalog) + at := time.Now().Add(-2 * time.Second).UnixMilli() + for i := int64(0); i < 2; i++ { + for _, kind := range []telemetry.DerType{telemetry.DerMeter, telemetry.DerPV} { + name := map[telemetry.DerType]string{telemetry.DerMeter: "p1", telemetry.DerPV: "solar"}[kind] + tel.Get(name, kind).UpdatedAt = time.UnixMilli(at + i*1000) + } + if _, err := persistTelemetryTick(st, tel, tickPersistControl{SiteMeterDriver: "p1"}, at+i*1000, time.Minute, nil, opts); err != nil { + t.Fatal(err) + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + points, err := st.LoadHistory(at, at+1000, 0) + if err != nil { + t.Fatal(err) + } + if len(points) == 0 { + t.Fatal("history empty") + } + for _, p := range points { + if p.Missing&(state.MissingBattery|state.MissingLoad) != 0 || p.LoadW != 3000 { + t.Fatalf("catalog=%v: disabled battery blocked saved P1 load: %+v", knownCatalog, points) + } + } + native, err := (&appHistory{st: st}).Window(context.Background(), []string{"load_w"}, state.EnergyLedgerBucketMS, at/state.EnergyLedgerBucketMS*state.EnergyLedgerBucketMS, at+state.EnergyLedgerBucketMS) + if err != nil { + t.Fatal(err) + } + if len(native["load_w"]) == 0 || native["load_w"][0].W <= 0 { + t.Fatal("saved household ledger did not reach native app history") + } + // A previously observed battery remains significant even if config no + // longer requires its first reading. Never replace a stale flow by zero. + tel.Update("p1", telemetry.DerBattery, 500, nil, nil) + tel.Get("p1", telemetry.DerBattery).UpdatedAt = time.Now().Add(-2 * time.Minute) + p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "p1"}, at, time.Minute, opts) + if p.Missing&(state.MissingBattery|state.MissingLoad) != state.MissingBattery|state.MissingLoad { + t.Fatal("disabled config hid an observed stale battery") + } + } +} + +func TestBatteryReadingConfigDeclaresRequiredFlow(t *testing.T) { + for _, tc := range []struct { + name string + read any + capacity float64 + want bool + }{ + {"disabled despite stored capacity", false, 10000, false}, + {"existing battery", nil, 10000, true}, + {"explicit telemetry reading", true, 0, true}, + } { + t.Run(tc.name, func(t *testing.T) { + d := config.Driver{Name: "hybrid", Lua: "drivers/hybrid.lua", BatteryCapacityWh: tc.capacity} + if tc.read != nil { + d.Config = map[string]any{"read_battery": tc.read, "read_pv": false} + } + opts := forecastMeasurementOptions(&config.Config{Drivers: []config.Driver{d}}, measurementCatalog()) + found := false + for _, f := range opts.ExpectedFlows { + found = found || f.DerType == telemetry.DerBattery + } + if found != tc.want { + t.Fatalf("battery required=%v, want %v", found, tc.want) + } + }) + } +} + +func TestPVZeroCounterKeepsObservedEnergy(t *testing.T) { + for _, tc := range []struct { + name string + counters []float64 + watts float64 + step time.Duration + wantWh float64 + }{ + {"unavailable zero", []float64{0, 0, 0}, 3000, time.Minute, 100}, + {"positive counter returns", []float64{0, 0, 100, 150}, 3000, time.Minute, 150}, + {"counter resets to zero", []float64{100, 0, 0, 50}, 3000, time.Minute, 150}, + {"real zero generation", []float64{0, 0, 0}, 0, time.Minute, 0}, + {"positive coarse counter stays authoritative", []float64{100, 100, 100}, 3000, time.Minute, 0}, + {"unobserved gap stays missing", []float64{0, 0}, 3000, 5 * time.Minute, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + id, err := st.RegisterDevice(state.Device{DriverName: "solar", Make: "PV meter", Serial: "test-pv"}) + if err != nil { + t.Fatal(err) + } + tel := telemetry.NewStore() + base := time.Now().Add(-time.Hour).Truncate(5 * time.Minute).UnixMilli() + for i, counter := range tc.counters { + at := base + int64(i)*tc.step.Milliseconds() + raw, _ := json.Marshal(map[string]float64{"generation_wh": counter}) + tel.Update("solar", telemetry.DerPV, -tc.watts, nil, raw) + tel.RecordDriverSuccess("solar") + tel.Get("solar", telemetry.DerPV).UpdatedAt = time.UnixMilli(at) + obs := buildEnergyObservations(st, tel, tickPersistControl{}, state.HistoryPoint{}, testEnergyIdentity(st)) + if err := st.RecordTickWithEnergy(state.HistoryPoint{TsMs: at, JSON: "{}"}, nil, obs); err != nil { + t.Fatal(err) + } + } + points, _, err := st.LoadEnergyHistory(state.EnergyHistoryQuery{AssetID: state.HardwareEnergyAssetID(id, state.AssetPV), SinceMS: base, UntilMS: base + int64(len(tc.counters))*tc.step.Milliseconds(), BucketMS: state.EnergyLedgerBucketMS, Limit: 100}) + if err != nil { + t.Fatal(err) + } + var wh float64 + for _, p := range points { + wh += p.EnergyWh + if tc.name == "unavailable zero" && p.EnergyWh > 0 && (p.Source != "power_telemetry" || p.Quality != "integrated") { + t.Fatal("power-derived energy claimed a measured counter") + } + } + if math.Abs(wh-tc.wantWh) > 0.0001 { + t.Fatalf("saved PV = %g Wh, want %g Wh", wh, tc.wantWh) + } + }) + } +} diff --git a/go/internal/state/energy_ledger.go b/go/internal/state/energy_ledger.go index 7aa1427ef..aab2916ac 100644 --- a/go/internal/state/energy_ledger.go +++ b/go/internal/state/energy_ledger.go @@ -221,6 +221,14 @@ func recordEnergyObservationTx(tx *sql.Tx, o EnergyObservation) error { // the fallback interval twice. switch { case !haveCounter: + // Power may already cover the preceding intervals before the + // first usable counter. Close only its final observed interval; + // the new cumulative value is a baseline, not additional energy. + if o.PowerW != nil && havePower { + if err := integratePower(tx, o, power, *o.PowerW, "counter_resume_fallback"); err != nil { + return err + } + } if err := addLedgerMarker(tx, o, "hardware_counter", "gap", "counter_baseline"); err != nil { return err } From 9a396c4fcaef6f0d98cd420000914147ad625ea9 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Sun, 18 Oct 2026 17:19:28 +0200 Subject: [PATCH 2/2] test(history): retain PV cursor behavior across reopen --- go/cmd/ftw/history_disabled_battery_test.go | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/go/cmd/ftw/history_disabled_battery_test.go b/go/cmd/ftw/history_disabled_battery_test.go index c4a027963..51a54fcc5 100644 --- a/go/cmd/ftw/history_disabled_battery_test.go +++ b/go/cmd/ftw/history_disabled_battery_test.go @@ -118,13 +118,15 @@ func TestPVZeroCounterKeepsObservedEnergy(t *testing.T) { }{ {"unavailable zero", []float64{0, 0, 0}, 3000, time.Minute, 100}, {"positive counter returns", []float64{0, 0, 100, 150}, 3000, time.Minute, 150}, + {"previous Core zero cursor", []float64{0, 0, 100, 150}, 3000, time.Minute, 150}, {"counter resets to zero", []float64{100, 0, 0, 50}, 3000, time.Minute, 150}, {"real zero generation", []float64{0, 0, 0}, 0, time.Minute, 0}, {"positive coarse counter stays authoritative", []float64{100, 100, 100}, 3000, time.Minute, 0}, {"unobserved gap stays missing", []float64{0, 0}, 3000, 5 * time.Minute, 0}, } { t.Run(tc.name, func(t *testing.T) { - st, err := state.Open(filepath.Join(t.TempDir(), "state.db")) + path := filepath.Join(t.TempDir(), "state.db") + st, err := state.Open(path) if err != nil { t.Fatal(err) } @@ -142,9 +144,22 @@ func TestPVZeroCounterKeepsObservedEnergy(t *testing.T) { tel.RecordDriverSuccess("solar") tel.Get("solar", telemetry.DerPV).UpdatedAt = time.UnixMilli(at) obs := buildEnergyObservations(st, tel, tickPersistControl{}, state.HistoryPoint{}, testEnergyIdentity(st)) + if i == 0 && tc.name == "previous Core zero cursor" { + // Seed exactly the zero-counter cursor the older Core kept. + obs[0].CounterWh = &counter + } if err := st.RecordTickWithEnergy(state.HistoryPoint{TsMs: at, JSON: "{}"}, nil, obs); err != nil { t.Fatal(err) } + if i == 1 && (tc.name == "positive counter returns" || tc.name == "previous Core zero cursor") { + if err := st.Close(); err != nil { + t.Fatal(err) + } + st, err = state.Open(path) + if err != nil { + t.Fatal(err) + } + } } points, _, err := st.LoadEnergyHistory(state.EnergyHistoryQuery{AssetID: state.HardwareEnergyAssetID(id, state.AssetPV), SinceMS: base, UntilMS: base + int64(len(tc.counters))*tc.step.Milliseconds(), BucketMS: state.EnergyLedgerBucketMS, Limit: 100}) if err != nil {