fix(history): keep P1 household and PV energy observations

.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.

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) {

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 {

go/cmd/ftw/history_disabled_battery_test.go

@@ -0,0 +1,180 @@
+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},
+		{"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) {
+			path := filepath.Join(t.TempDir(), "state.db")
+			st, err := state.Open(path)
+			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 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 {
+				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)
+			}
+		})
+	}
+}

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
 			}