Use one Core Load calculation for app, web and HA

.changeset/unified-live-household-load.md

@@ -0,0 +1,5 @@
+---
+"ftw": patch
+---
+
+Use one qualified Core household power balance for the box page, phone sessions and FTW Cloud. Keep missing or invalid load unavailable instead of recalculating it from smoothed readings or showing zero.

go/cmd/ftw/app_link.go

@@ -55,7 +55,15 @@ type appSite struct {
 	// siteMeterStale is how long a site-meter reading stays current. It is
 	// the watchdog timeout, so the app's freshness claim and the box's own
 	// dispatch gate are answering from the same number.
-	siteMeterStale time.Duration
+	siteMeterStale     time.Duration
+	measurementOptions func() telemetry.ForecastOptions
+}
+
+func newAppSite(tel *telemetry.Store, ctrl *control.State, ctrlMu *sync.Mutex,
+	revision *control.Revision, siteMeterStale time.Duration, options func() telemetry.ForecastOptions,
+) *appSite {
+	return &appSite{tel: tel, ctrl: ctrl, ctrlMu: ctrlMu, revision: revision,
+		started: processStarted, siteMeterStale: siteMeterStale, measurementOptions: options}
 }
 
 func (a *appSite) Snapshot() appproto.Snapshot {
@@ -66,7 +74,14 @@ func (a *appSite) Snapshot() appproto.Snapshot {
 	fuseA, fusePhases := a.ctrl.SiteFuseAmps, a.ctrl.SiteFusePhases
 	a.ctrlMu.Unlock()
 
-	snap := appproto.Snapshot{Mode: mode, ControlRev: rev, FuseA: fuseA}
+	var opts telemetry.ForecastOptions
+	if a.measurementOptions != nil {
+		opts = a.measurementOptions()
+	}
+	live := a.tel.LiveMeasurementNow(siteMeterDriver, opts)
+	snap := appproto.Snapshot{Mode: mode, ControlRev: rev, FuseA: fuseA,
+		GridW: live.GridW, PVW: live.PVW, BatteryW: live.BatteryW, EVW: live.EVW,
+		LoadW: live.HouseholdW, LoadWKnown: live.Valid}
 
 	uptimeMs := time.Since(a.started).Milliseconds()
 	seen := map[string]bool{}
@@ -101,7 +116,6 @@ func (a *appSite) Snapshot() appproto.Snapshot {
 		addSource(siteMeterDriver)
 		if health := a.tel.DriverHealth(siteMeterDriver); health != nil && health.IsOnline() {
 			if reading := a.tel.Get(siteMeterDriver, telemetry.DerMeter); reading != nil {
-				snap.GridW = reading.SmoothedW
 				snap.GridPhaseA = gridPhaseAmps(reading, fusePhases)
 			}
 		} else {
@@ -114,13 +128,6 @@ func (a *appSite) Snapshot() appproto.Snapshot {
 
 	for _, reading := range a.tel.ReadingsByType(telemetry.DerPV) {
 		addSource(reading.Driver)
-		if health := a.tel.DriverHealth(reading.Driver); health != nil && health.IsOnline() {
-			// Site convention: positive watts flow into the site, so PV is
-			// negative while generating. No sign conversion happens here —
-			// the driver boundary already did it, and doing it twice is how
-			// a solar panel starts consuming power.
-			snap.PVW += reading.SmoothedW
-		}
 	}
 
 	var socTotal float64
@@ -131,7 +138,6 @@ func (a *appSite) Snapshot() appproto.Snapshot {
 		if health == nil || !health.IsOnline() {
 			continue
 		}
-		snap.BatteryW += reading.SmoothedW
 		if reading.SoC != nil {
 			socTotal += *reading.SoC
 			socCount++
@@ -145,28 +151,15 @@ func (a *appSite) Snapshot() appproto.Snapshot {
 		snap.BatterySoC = socTotal / float64(socCount)
 	}
 
-	// EV chargers. Known means the site has one at all — an idle charger
-	// is a real 0 W reading, a site without one sends no field 10 and the
-	// app draws no EV node. Positive while charging: a charger consumes
-	// like any other load. The watts come from SumOnlineEVW so a charger
-	// that cannot take a command but is still drawing is counted here the
-	// same way the LAN dashboard counts it — IsOnline() would hide that
-	// draw in the house node.
+	// Keep charger presence and source metadata separate from qualified watts.
+	// An idle charger is a real zero; a missing configured flow still makes
+	// the household balance unavailable.
 	for _, reading := range a.tel.ReadingsByType(telemetry.DerEV) {
 		addSource(reading.Driver)
 		snap.EVWKnown = true
 	}
-	snap.EVW = a.tel.SumOnlineEVW()
-
-	// grid = load + battery + pv + ev + v2x, all site-signed. Rearranged,
-	// not re-derived: a second formula here would be a second thing to
-	// keep in step with docs/site-convention.md. House load is never
-	// negative; metering noise that would push it below zero is the car
-	// or the battery, not the house consuming in reverse.
-	snap.LoadW = snap.GridW - snap.BatteryW - snap.PVW -
-		snap.EVW - a.tel.SumOnlineV2XW()
-	if snap.LoadW < 0 {
-		snap.LoadW = 0
+	for _, source := range live.Sources {
+		addSource(source.Flow.Driver)
 	}
 
 	return snap
@@ -670,6 +663,7 @@ func startAppLink(
 	prefs *config.PlannerPrefs,
 	revision *control.Revision,
 	siteMeterStale time.Duration,
+	measurementOptions func() telemetry.ForecastOptions,
 	gateway *lateAPI,
 	cloud *cloudSession,
 	agents *agentPorts,
@@ -690,10 +684,7 @@ func startAppLink(
 		return enroll, nil, false, nil
 	}
 
-	site := &appSite{
-		tel: tel, ctrl: ctrl, ctrlMu: ctrlMu, revision: revision,
-		started: processStarted, siteMeterStale: siteMeterStale,
-	}
+	site := newAppSite(tel, ctrl, ctrlMu, revision, siteMeterStale, measurementOptions)
 	info := newAppBoxInfo(boxID, build)
 	modes := &appModes{ctrl: ctrl, ctrlMu: ctrlMu, state: st, mpc: planner, prefs: prefs}
 	plans := &appPlans{planner: planner, ctrl: ctrl, ctrlMu: ctrlMu}

go/cmd/ftw/cloud_link.go

@@ -388,16 +388,14 @@ func (n *cloudNotices) notify(kind string) {
 func startCloudLink(
 	ctx context.Context, statePath, build string, st *state.Store, enroll *appenroll.Identity,
 	tel *telemetry.Store, ctrl *control.State, ctrlMu *sync.Mutex, revision *control.Revision,
-	siteMeterStale time.Duration, priceSvc *prices.Service, gateway *lateAPI, notices *cloudNotices,
+	siteMeterStale time.Duration, measurementOptions func() telemetry.ForecastOptions, priceSvc *prices.Service, gateway *lateAPI, notices *cloudNotices,
 	tunnel *appuplink.Tunnel, carsRT *cars.Runtime, webPush *notifications.WebPush,
 ) *cloudlink.Link {
 	if st == nil || enroll == nil {
 		return nil
 	}
-	site := &cloudSite{api: gateway, cars: carsRT, site: &appSite{
-		tel: tel, ctrl: ctrl, ctrlMu: ctrlMu, revision: revision,
-		started: processStarted, siteMeterStale: siteMeterStale,
-	}}
+	site := &cloudSite{api: gateway, cars: carsRT,
+		site: newAppSite(tel, ctrl, ctrlMu, revision, siteMeterStale, measurementOptions)}
 	if priceSvc != nil && priceSvc.Zone != "" && priceSvc.Store != nil {
 		site.prices = &appPrices{svc: priceSvc}
 	}

go/cmd/ftw/ha_bridge_owner_test.go

@@ -87,7 +87,7 @@ func newRealHAOwner(modes chan<- string) *haBridgeOwner {
 		return nil
 	}}
 	o := newHABridgeOwner(func(c *config.HomeAssistant, names []string) (haBridgeHandle, error) {
-		b, err := ha.Start(c, tel, ctrl, ctrlMu, names, cb, nil, nil)
+		b, err := ha.Start(c, tel, ctrl, ctrlMu, names, cb, nil, nil, nil)
 		if err != nil {
 			return nil, err
 		}

go/cmd/ftw/live_load_test.go

@@ -0,0 +1,169 @@
+package main
+
+import (
+	"encoding/json"
+	"math"
+	"net/http/httptest"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/srcfl/ftw/go/internal/api"
+	"github.com/srcfl/ftw/go/internal/appproto"
+	"github.com/srcfl/ftw/go/internal/config"
+	"github.com/srcfl/ftw/go/internal/control"
+	"github.com/srcfl/ftw/go/internal/telemetry"
+)
+
+func liveLoadStatus(t *testing.T, tel *telemetry.Store, ctrl *control.State, options func() telemetry.ForecastOptions) map[string]any {
+	t.Helper()
+	srv := api.New(&api.Deps{Tel: tel, Ctrl: ctrl, CtrlMu: &sync.Mutex{},
+		Cfg: &config.Config{}, CfgMu: &sync.RWMutex{}, CapMu: &sync.RWMutex{},
+		HistoryMeasurementSources: options})
+	rr := httptest.NewRecorder()
+	srv.ServeHTTP(rr, httptest.NewRequest("GET", "/api/status", nil))
+	if rr.Code != 200 {
+		t.Fatal("status request failed")
+	}
+	var out map[string]any
+	if err := json.Unmarshal(rr.Body.Bytes(), &out); err != nil {
+		t.Fatal("invalid status response")
+	}
+	return out
+}
+
+func liveLoadData(t *testing.T, fields map[string]any) json.RawMessage {
+	t.Helper()
+	data, err := json.Marshal(fields)
+	if err != nil {
+		t.Fatal("invalid power evidence")
+	}
+	return data
+}
+
+func TestLiveLoadAppAndLANShareSourcePowerAndValidity(t *testing.T) {
+	for _, scenario := range []string{"zero", "raw_not_smoothed", "site_side_ac", "missing_meter", "stale_power", "skew", "negative", "offline_ev", "idle_ev", "paused_ev", "faulted_ev"} {
+		t.Run(scenario, func(t *testing.T) {
+			tel := telemetry.NewStore()
+			for _, name := range []string{"meter", "solar", "battery", "charger"} {
+				tel.EnsureDriverHealth(name)
+			}
+			ctrl := &control.State{SiteMeterDriver: "meter", Mode: control.ModeSelfConsumption}
+			tel.Update("meter", telemetry.DerMeter, 1000, nil, nil)
+			want, valid := 1000.0, true
+			switch scenario {
+			case "zero":
+				tel.Update("meter", telemetry.DerMeter, 0, nil, nil)
+				want = 0
+			case "raw_not_smoothed":
+				tel.Update("meter", telemetry.DerMeter, 5000, nil, nil)
+				want = 5000
+				if tel.Get("meter", telemetry.DerMeter).SmoothedW == want {
+					t.Fatal("test did not distinguish smoothed power")
+				}
+			case "site_side_ac":
+				tel.Update("battery", telemetry.DerBattery, -2500, nil, liveLoadData(t, map[string]any{"control_power_w": -2000, "control_power_available": true}))
+				want = 3000
+			case "missing_meter":
+				tel.Remove("meter")
+				valid = false
+			case "stale_power":
+				tel.Update("meter", telemetry.DerMeter, 1000, nil, liveLoadData(t, map[string]any{"control_power_observed_at": time.Now().Add(-2 * time.Minute).Format(time.RFC3339Nano)}))
+				valid = false
+			case "skew":
+				tel.Update("solar", telemetry.DerPV, -500, nil, liveLoadData(t, map[string]any{"control_power_observed_at": time.Now().Add(-45 * time.Second).Format(time.RFC3339Nano)}))
+				valid = false
+			case "negative":
+				tel.Update("battery", telemetry.DerBattery, 2000, nil, nil)
+				valid = false
+			case "offline_ev":
+				tel.Update("charger", telemetry.DerEV, 500, nil, nil)
+				tel.DriverHealthMut("charger").SetOffline()
+				valid = false
+			case "idle_ev", "paused_ev":
+				fields := map[string]any{"connected": scenario == "paused_ev", "charging": false,
+					"control_power_observed_at": time.Now().Add(-time.Hour).Format(time.RFC3339Nano),
+					"control_power_available":   true, "control_power_confirmed": true}
+				tel.Update("charger", telemetry.DerEV, 0, nil, liveLoadData(t, fields))
+			case "faulted_ev":
+				tel.Update("charger", telemetry.DerEV, 500, nil, nil)
+				tel.SetDriverDeviceFault("charger", true, "command rejected")
+				want = 500
+			}
+			site := &appSite{tel: tel, ctrl: ctrl, ctrlMu: &sync.Mutex{}, revision: &control.Revision{}, started: time.Now(), siteMeterStale: time.Minute}
+			snap := site.Snapshot()
+			status := liveLoadStatus(t, tel, ctrl, nil)
+			if snap.LoadWKnown != valid || (status["load_w"] != nil) != valid {
+				t.Fatal("live adapters disagree on load validity")
+			}
+			if valid && (snap.LoadW != want || status["load_w"] != want) {
+				t.Fatal("live adapters did not use the same source watts")
+			}
+			cloud := (&cloudSite{site: site}).Live()
+			value, present := cloud.Fields[strconv.Itoa(int(appproto.FidLoadW))]
+			if present != valid || valid && value != int64(math.Round(want)) {
+				t.Fatal("Cloud changed load watts or validity")
+			}
+			if scenario == "site_side_ac" && (snap.BatteryW != -2000 || status["bat_w"] != float64(-2000)) {
+				t.Fatal("live battery power differs from the AC balance")
+			}
+		})
+	}
+}
+
+func TestLiveLoadOptionsFollowConfigurationForAppLANAndCloud(t *testing.T) {
+	tel := telemetry.NewStore()
+	tel.EnsureDriverHealth("meter")
+	tel.Update("meter", telemetry.DerMeter, 1000, nil, nil)
+	ctrl := &control.State{SiteMeterDriver: "meter"}
+	settings := &forecastSiteConfig{value: forecastSite{
+		Options:        telemetry.ForecastOptions{HouseholdInvalidReason: "unconfirmed_battery_identity"},
+		HistoryOptions: telemetry.ForecastOptions{ExpectedFlows: []telemetry.ForecastFlow{{Driver: "battery", DerType: telemetry.DerBattery}}},
+	}}
+	options := func() telemetry.ForecastOptions { return settings.Snapshot().HistoryOptions }
+	// Both link constructors use this constructor; Cloud uses the same appSite
+	// adapter for its readable live copy as the encrypted phone session.
+	site := newAppSite(tel, ctrl, &sync.Mutex{}, &control.Revision{}, time.Minute, options)
+	for _, batteryDeclared := range []bool{true, false, true} {
+		settings.mu.Lock()
+		if batteryDeclared {
+			settings.value.HistoryOptions.ExpectedFlows = []telemetry.ForecastFlow{{Driver: "battery", DerType: telemetry.DerBattery}}
+		} else {
+			settings.value.HistoryOptions.ExpectedFlows = nil
+		}
+		settings.mu.Unlock()
+		snap := site.Snapshot()
+		status := liveLoadStatus(t, tel, ctrl, options)
+		_, cloudKnown := (&cloudSite{site: site}).Live().Fields[strconv.Itoa(int(appproto.FidLoadW))]
+		known := !batteryDeclared
+		if snap.LoadWKnown != known || (status["load_w"] != nil) != known || cloudKnown != known {
+			t.Fatal("configured never-emitted flow differs between clients")
+		}
+		if known && (snap.LoadW != 1000 || status["load_w"] != float64(1000)) {
+			t.Fatal("learning-only identity gate affected live load")
+		}
+		if batteryDeclared {
+			found := false
+			for _, source := range snap.Sources {
+				found = found || source.ID == "battery"
+			}
+			if !found {
+				t.Fatal("never-emitted source missing from app metadata")
+			}
+		}
+	}
+}
+
+func TestLiveLoadWithoutConfiguredMeterIsUnavailable(t *testing.T) {
+	tel := telemetry.NewStore()
+	ctrl := &control.State{}
+	site := newAppSite(tel, ctrl, &sync.Mutex{}, &control.Revision{}, time.Minute, nil)
+	status := liveLoadStatus(t, tel, ctrl, nil)
+	if _, present := status["load_w"]; !present || status["load_w"] != nil || status["grid_w"] != nil || site.Snapshot().LoadWKnown {
+		t.Fatal("missing configuration became a measured zero")
+	}
+	if _, present := (&cloudSite{site: site}).Live().Fields[strconv.Itoa(int(appproto.FidLoadW))]; present {
+		t.Fatal("Cloud invented load without a meter")
+	}
+}

go/cmd/ftw/main.go

@@ -827,6 +827,9 @@ func main() {
 		return state.Device{}
 	}
 	forecastSettings := newForecastSiteConfig(st)
+	liveMeasurementOptions := func() telemetry.ForecastOptions {
+		return forecastSettings.Snapshot().HistoryOptions
+	}
 	forecastSettings.identity = func(name string) (string, bool) {
 		if id, ok := runningDeviceID(reg, name); ok {
 			return id, true
@@ -1037,7 +1040,7 @@ func main() {
 	// the broker did not accept. Its start closure reads the registry and
 	// control wiring only when it runs.
 	haOwner := newHABridgeOwner(func(c *config.HomeAssistant, names []string) (haBridgeHandle, error) {
-		b, err := ha.Start(c, tel, ctrl, ctrlMu, names, haCallbacks(ctrl, ctrlMu, st, mpcSvc, plannerPrefs), mpcPlanSource(mpcSvc), haEnergySource(st))
+		b, err := ha.Start(c, tel, ctrl, ctrlMu, names, haCallbacks(ctrl, ctrlMu, st, mpcSvc, plannerPrefs), mpcPlanSource(mpcSvc), haEnergySource(st), liveMeasurementOptions)
 		if err != nil {
 			return nil, err
 		}
@@ -2403,7 +2406,7 @@ func main() {
 	appEnroll, appUplink, appLinkEnabled, appLinkErr := startAppLink(
 		ctx, cfg, identityKeyPath, boxID, Version,
 		st, tel, mpcSvc, lpMgr, lpController, carsRT, priceSvc, ctrl, ctrlMu, plannerPrefs,
-		controlRev, appLinkWatchdog, appAPI, appCloud, agents,
+		controlRev, appLinkWatchdog, liveMeasurementOptions, appAPI, appCloud, agents,
 	)
 	switch {
 	case appLinkErr != nil:
@@ -2424,7 +2427,7 @@ func main() {
 		appTunnel = appUplink.Tunnel()
 	}
 	cloudLink := startCloudLink(ctx, statePath, Version, st, appEnroll,
-		tel, ctrl, ctrlMu, controlRev, appLinkWatchdog, priceSvc, appAPI, cloudNotify, appTunnel, carsRT, webPush)
+		tel, ctrl, ctrlMu, controlRev, appLinkWatchdog, liveMeasurementOptions, priceSvc, appAPI, cloudNotify, appTunnel, carsRT, webPush)
 	appCloud.bind(cloudLink)
 	if cloudLink != nil {
 		agents.home = cloudLink.Home
@@ -2501,7 +2504,7 @@ func main() {
 		SiteDispatchBlocked: func() string {
 			return siteDispatchNow(tel, cfg, cfgMu, ctrl, ctrlMu, time.Now()).Reason
 		},
-		HistoryMeasurementSources: func() telemetry.ForecastOptions { return forecastSettings.Snapshot().HistoryOptions },
+		HistoryMeasurementSources: liveMeasurementOptions,
 		SiteMeasurementSources: func() telemetry.ForecastOptions {
 			return forecastSettings.Snapshot().Options
 		},
@@ -3062,6 +3065,7 @@ func main() {
 				batW := troubleshootingSumOnlineW(tel, telemetry.DerBattery)
 				evW := tel.SumOnlineEVW()
 				v2xW := tel.SumOnlineV2XW()
+				live := tel.LiveMeasurementNow(siteMeterDriver, liveMeasurementOptions())
 				attrs := []any{
 					"mode", mode,
 					"plan_stale", planMissingNow,
@@ -3072,14 +3076,17 @@ func main() {
 					"bat_w", batW,
 					"ev_w", evW,
 					"v2x_w", v2xW,
+					"load_valid", live.Valid,
 					"self_tune_active", selfTuneActive,
 					"self_tune_driver", selfTuneName,
 					"self_tune_command_w", selfTuneCmd,
 					"targets", targets,
 					"final_targets", finalTargets,
 				}
-				if haveGrid {
-					attrs = append(attrs, "load_w", gridW-batW-pvW-evW-v2xW)
+				if live.Valid {
+					attrs = append(attrs, "load_w", live.HouseholdW)
+				} else {
+					attrs = append(attrs, "load_reason", live.Reason)
 				}
 				slog.Info("troubleshooting: dispatch decision", attrs...)
 			}