From 604ed96c954bccc06e3276e9da2a289c9dbccb16 Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 17:43:24 +0200 Subject: [PATCH 1/4] fix(history): bridge short source gaps instead of losing the day's Home use Since #1741 a Core restart (18-20 s), a source unknown for a few seconds or a writer pause over 15 s made the whole day's load_wh null, and an unknown PV also nulled pv_wh. The source writer now holds an episode that starts at the last fully known snapshot. If every source is known again with the same topology within LoadEstimateRecoveryWindow (2 min), the episode is written as an estimate marked EstimatedLoad: measured sources keep their snapshots, unknown ones hold the anchor's values. Otherwise it is written exactly as before. Pending time stays out of totals and expected coverage, and the day waits for it before closing. On open Core marks the cursor as stopped instead of deleting it, so a brief stop is bridged as an estimate and a longer one starts a new period as before. Older Cores still delete the cursor on open. Relates to #1750. Co-Authored-By: Claude Opus 5.5 --- .changeset/bridge-short-history-gaps.md | 5 + docs/site-convention.md | 7 +- .../state/history_source_bridge_test.go | 147 +++++++++++ go/internal/state/history_sources.go | 236 ++++++++++++++---- go/internal/state/history_sources_test.go | 31 ++- go/internal/state/history_sqlite.go | 5 +- 6 files changed, 360 insertions(+), 71 deletions(-) create mode 100644 .changeset/bridge-short-history-gaps.md create mode 100644 go/internal/state/history_source_bridge_test.go diff --git a/.changeset/bridge-short-history-gaps.md b/.changeset/bridge-short-history-gaps.md new file mode 100644 index 000000000..915d09f1c --- /dev/null +++ b/.changeset/bridge-short-history-gaps.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +Keep the day's Home use after a Core restart or a source that drops out for up to two minutes. The gap is bridged as a marked estimate; longer gaps, topology changes and stops over two minutes stay unknown. diff --git a/docs/site-convention.md b/docs/site-convention.md index 77b9d80d7..a699e7985 100644 --- a/docs/site-convention.md +++ b/docs/site-convention.md @@ -56,8 +56,11 @@ holds each qualified snapshot forward until the next safe snapshot, at most the sums when reading a chart interval, day or tariff period. A negative minute may contribute to a valid longer interval; it is never clipped first. Chart means need complete interval coverage. Missing sources, writer gaps, -process stops and topology changes cannot become measured zero. This is an -estimate from received measurements, not a correction of device clocks. +process stops and topology changes cannot become measured zero. A gap of at +most two minutes between fully known snapshots of one topology is bridged as a +marked estimate: measured sources keep their snapshots, and an unknown source +or a stopped Core holds the last known values. Longer gaps stay unknown. This +is an estimate from received measurements, not a correction of device clocks. In self-consumption mode, the controller drives the site meter toward the grid target (normally 0 W): it charges from live surplus and may discharge diff --git a/go/internal/state/history_source_bridge_test.go b/go/internal/state/history_source_bridge_test.go new file mode 100644 index 000000000..ea92d4d61 --- /dev/null +++ b/go/internal/state/history_source_bridge_test.go @@ -0,0 +1,147 @@ +package state + +import ( + "testing" + "time" +) + +// Two hours of a 1000 W house, sampled every 5 s, with one disturbance one +// hour in. Observed on v0.154.0-beta.1 (19 October 2026): each short +// disturbance made the whole day's Home use unknown. +func TestSourceLoadDaySurvivesShortGap(t *testing.T) { + const tick, hour = int64(5000), int64(3600000) + for _, c := range []struct { + name string + // stopMS is a process stop starting one hour in. + stopMS int64 + // receipt alters the snapshot off ms after the first hour. + receipt func(off int64, v *HistorySources) + known bool + // importWh is checked when set; the recorded grid must survive. + loadWh, importWh float64 + }{ + {name: "no disturbance", known: true, loadWh: 2000, importWh: 2000}, + {name: "20 s restart", stopMS: 20000, known: true, loadWh: 2000, importWh: 2000}, + {name: "20 s restart, EV and its device back after 30 s", stopMS: 20000, receipt: func(off int64, v *HistorySources) { + if off <= 50000 { + v.Missing, v.Topology = MissingEV, "grid" + } + }, known: true, loadWh: 2000, importWh: 2000}, + {name: "EV unknown 15 s", receipt: unknownFor(MissingEV, 15000), known: true, loadWh: 2000, importWh: 2000}, + // The unknown charger holds its last value; the meter keeps its own + // receipts. The household carries the 2 kW for 15 s, nothing else. + {name: "EV starts while unknown", receipt: func(off int64, v *HistorySources) { + if off > 0 { + v.GridW, v.EVW = 3000, 2000 + } + if off > 0 && off <= 15000 { + v.Missing, v.EVW = MissingEV, 0 + } + }, known: true, loadWh: 2000 + 2000*15.0/3600, importWh: 1000 + 1000*5.0/3600 + 3000*3595.0/3600}, + {name: "PV unknown 60 s", receipt: unknownFor(MissingPV, 60000), known: true, loadWh: 2000, importWh: 2000}, + {name: "PV unknown 145 s", receipt: unknownFor(MissingPV, 145000), importWh: 2000}, + {name: "EV unknown 5 min", receipt: unknownFor(MissingEV, 300000), importWh: 2000}, + {name: "3 min restart", stopMS: 180000}, + {name: "meter replaced", receipt: func(off int64, v *HistorySources) { + if off > 0 { + v.Topology = "new meter+charger" + } + }}, + } { + t.Run(c.name, func(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24*time.Hour).UnixMilli() - 24*hour + for at := base; at <= base+2*hour; at += tick { + off := at - base - hour + if c.stopMS > 0 && off > 0 && off < c.stopMS { + continue + } + v := HistorySources{GridW: 1000} + if c.receipt != nil && off >= 0 { + c.receipt(off, &v) + } + sourceTick(t, s, at, v) + if c.stopMS > 0 && off == 0 { + s = reopenSourceStore(t, s) + } + } + d, err := s.DailyEnergy(base, base+2*hour) + if err != nil { + t.Fatal(err) + } + if c.importWh != 0 { + near(t, "import Wh", d.ImportWh, c.importWh) + } + if got := d.Missing&MissingLoad == 0; got != c.known { + t.Fatalf("Home use known=%v, want %v (quality %s)", got, c.known, d.Missing.Quality()) + } + if !c.known { + return + } + if q := d.Missing.Quality(); q != "estimated" { + t.Fatalf("quality %s, want estimated", q) + } + if d.Missing&MissingPV != 0 { + t.Fatal("PV total became unknown") + } + if diff := d.LoadWh - c.loadWh; diff > 1e-6 || diff < -1e-6 { + t.Fatalf("Home use %v Wh, want %v", d.LoadWh, c.loadWh) + } + }) + } +} + +// A bridge that closes after midnight must still land in the closed day, so +// the day waits for it before it becomes final. +func TestSourceBridgeHoldsDayOpenUntilItCloses(t *testing.T) { + s := sourceStore(t) + midnight := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for at := midnight - 60000; at <= midnight+10000; at += 5000 { + v := HistorySources{GridW: 1000} + if at > midnight-10000 && at < midnight+10000 { + v.Missing = MissingEV + } + sourceTick(t, s, at, v) + through, _, err := s.HistorySourceProgress(t.Context()) + if err != nil { + t.Fatal(err) + } + if at < midnight+10000 && through > midnight-10000 { + t.Fatal("day closed before its bridge") + } + } + through, _, err := s.HistorySourceProgress(t.Context()) + if err != nil || through != midnight+10000 { + t.Fatal("bridge not committed", through, err) + } + d, err := s.DailyEnergy(midnight-60000, midnight) + if err != nil || d.Missing&MissingLoad != 0 { + t.Fatal("closed day lost its Home use", err) + } + near(t, "closed day Wh", d.LoadWh, 1000.0/60) +} + +func unknownFor(flag HistoryMissing, ms int64) func(int64, *HistorySources) { + return func(off int64, v *HistorySources) { + if off > 0 && off <= ms { + v.Missing = flag + } + } +} + +func reopenSourceStore(t *testing.T, s *Store) *Store { + t.Helper() + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + s, err := Open(path) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { s.Close() }) + if err := s.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + return s +} diff --git a/go/internal/state/history_sources.go b/go/internal/state/history_sources.go index 207442be0..933eccc2f 100644 --- a/go/internal/state/history_sources.go +++ b/go/internal/state/history_sources.go @@ -29,20 +29,31 @@ const sourceCursorSchema = `CREATE TABLE IF NOT EXISTS history_source_cursor(key const sourceEnergyTable = "history_source_energy" const maxSourceEnergyWidth int64 = 900000 +// maxSourceBridgeMS bounds an estimated bridge over a writer pause, a process +// restart or a briefly unknown source. It shares the history estimate budget +// of a recovered balance episode; longer gaps stay unknown. +const maxSourceBridgeMS = maxLoadEstimateEpisodeMS + +// sourceReceipt is one snapshot held in the writer cursor. Stopped marks the +// first receipt after a process stop. +type sourceReceipt struct { + HistorySources + Stopped bool `json:",omitempty"` +} + +type sourceCursor struct { + At int64 + Start int64 + Sources HistorySources + // Episode starts at the last fully known receipt. Its intervals wait for + // known sources of that topology: if they return within the bridge + // budget the intervals are an estimate, otherwise they are recorded as + // before. Pending time never enters totals or the period's coverage. + Episode []sourceReceipt `json:",omitempty"` +} + func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *HistoryPoint) error { - if p == nil || p.Sources == nil { - _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key='last'`) - return err - } - at := p.Sources.AtMS - if at <= 0 { - return errors.New("history source snapshot has no observation time") - } - var last struct { - At int64 - Start int64 - Sources HistorySources - } + var last sourceCursor var raw string err := tx.QueryRowContext(ctx, `SELECT value FROM history_source_cursor WHERE key='last'`).Scan(&raw) if err != nil && !errors.Is(err, sql.ErrNoRows) { @@ -53,15 +64,54 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History return err } } + // commit makes source energy final through end. Readers count the + // period's time as expected, and a closed day waits for it. + commit := func(end int64) error { + if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_periods VALUES(?,?) ON CONFLICT(start_ms) DO UPDATE SET last_ms=excluded.last_ms`, last.Start, end); err != nil { + return err + } + _, err := tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('through',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, end) + return err + } + // An abandoned episode keeps the original evidence and its unknowns, + // through its last receipt. + abandon := func() error { + n := len(last.Episode) + if n == 0 { + return nil + } + for i := 1; i < n; i++ { + if err := recordedSourceIntervalTx(ctx, tx, last.Episode[i-1].HistorySources, last.Episode[i]); err != nil { + return err + } + } + through := last.Episode[n-1].AtMS + last.Episode = nil + return commit(through) + } + drop := func() error { + if err := abandon(); err != nil { + return err + } + _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key IN ('last','stopped')`) + return err + } + if p == nil || p.Sources == nil { + return drop() + } + at := p.Sources.AtMS + if at <= 0 { + return errors.New("history source snapshot has no observation time") + } var through int64 if err := tx.QueryRowContext(ctx, `SELECT COALESCE((SELECT CAST(value AS INTEGER) FROM history_source_cursor WHERE key='through'),0)`).Scan(&through); err != nil { return err } - if at <= through { - if at < through { - _, err = tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key='last'`) + if latest := max(through, last.At); at <= latest { + if at == latest { + return nil } - return err + return drop() } if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO history_source_cursor VALUES('from',?)`, at); err != nil { return err @@ -72,60 +122,138 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History return err } } - if last.At == 0 { - last.Start = at + cur := sourceReceipt{HistorySources: *p.Sources} + if err := tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM history_source_cursor WHERE key='stopped')`).Scan(&cur.Stopped); err != nil { + return err } - if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_periods VALUES(?,?) ON CONFLICT(start_ms) DO UPDATE SET last_ms=excluded.last_ms`, last.Start, at); err != nil { + if _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key='stopped'`); err != nil { return err } - // Hold the previous receipt forward, bounded by the next safe tick. Do - // not assign a newly arrived charger value to time before its receipt. - if last.At > 0 && at-last.At <= maxLoadEstimateTickMS && last.Sources.Topology == p.Sources.Topology { - // A delayed writer may finish a just-closed local day. Invalidate only - // the nearby cached dates; source rows and old history stay intact. - low := time.UnixMilli(last.At).UTC().AddDate(0, 0, -1).Format("2006-01-02") - high := time.UnixMilli(at).UTC().AddDate(0, 0, 1).Format("2006-01-02") - for _, table := range []string{"energy_daily", "energy_daily_partial", "history_daily_quality"} { - if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE day>=? AND day<=?`, low, high); err != nil { - return err - } + known := func(v HistorySources) bool { return v.Missing&MissingPower == 0 } + switch { + case last.At == 0: + last.Start = at + case at-last.At > maxSourceBridgeMS: + if err := abandon(); err != nil { + return err } - v := last.Sources - missing := (v.Missing | p.Sources.Missing) & MissingPower - if missing != 0 { - missing |= MissingLoad - } - for from := last.At; from < at; { - start := bucketStart(from, ArchiveResolutionMS) - to := min(at, start+ArchiveResolutionMS) - ms := to - from - h := float64(ms) / 3600000 - if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_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,ev_wh=ev_wh+excluded.ev_wh,ev_charge_wh=ev_charge_wh+excluded.ev_charge_wh,ev_discharge_wh=ev_discharge_wh+excluded.ev_discharge_wh`, - start, ArchiveResolutionMS, from, to, ms, 1, - math.Max(0, v.GridW)*h, math.Max(0, -v.GridW)*h, -v.PVW*h, - math.Max(0, v.BatteryW)*h, math.Max(0, -v.BatteryW)*h, 0, (v.EVW+v.V2XW)*h, (math.Max(0, v.EVW)+math.Max(0, v.V2XW))*h, (math.Max(0, -v.EVW)+math.Max(0, -v.V2XW))*h); err != nil { + // A long stop starts a new period, as an older writer may have run. + if cur.Stopped { + last.Start = at + } + case len(last.Episode) > 0: + last.Episode = append(last.Episode, cur) + case !cur.Stopped && at-last.At <= maxLoadEstimateTickMS && last.Sources.Topology == cur.Topology && known(last.Sources) && known(cur.HistorySources): + // Hold the previous receipt forward, bounded by the next safe tick. Do + // not assign a newly arrived charger value to time before its receipt. + if err := writeSourceIntervalTx(ctx, tx, last.At, at, last.Sources, 0); err != nil { + return err + } + case known(last.Sources): + last.Episode = []sourceReceipt{{HistorySources: last.Sources}, cur} + default: + if err := recordedSourceIntervalTx(ctx, tx, last.Sources, cur); err != nil { + return err + } + } + if len(last.Episode) > 0 { + anchor := last.Episode[0].HistorySources + switch { + case known(cur.HistorySources) && cur.Topology == anchor.Topology && at-anchor.AtMS <= maxSourceBridgeMS: + if err := bridgeSourceEpisodeTx(ctx, tx, last.Episode); err != nil { return err } - if err := putHistoryQuality(ctx, tx, "source_energy", start, ArchiveResolutionMS, missing); err != nil { + last.Episode = nil + case at-anchor.AtMS > maxSourceBridgeMS: + if err := abandon(); err != nil { return err } - from = to } } - last.At, last.Sources = at, *p.Sources - b, err := json.Marshal(last) - if err != nil { + end := at + if len(last.Episode) > 0 { + end = last.Episode[0].AtMS + } + if err := commit(end); err != nil { return err } - if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, string(b)); err != nil { + last.At, last.Sources = at, cur.HistorySources + b, err := json.Marshal(last) + if err != nil { return err } - _, err = tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('through',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, at) + _, err = tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, string(b)) return err } +// recordedSourceIntervalTx keeps the original evidence: the previous receipt +// held to the next safe tick of the same topology. Either endpoint's unknown +// source leaves the interval, and so its Load, unknown. +func recordedSourceIntervalTx(ctx context.Context, tx *sql.Tx, prev HistorySources, cur sourceReceipt) error { + if cur.Stopped || cur.AtMS-prev.AtMS > maxLoadEstimateTickMS || prev.Topology != cur.Topology { + return nil + } + missing := (prev.Missing | cur.Missing) & MissingPower + if missing != 0 { + missing |= MissingLoad + } + return writeSourceIntervalTx(ctx, tx, prev.AtMS, cur.AtMS, prev, missing) +} + +// bridgeSourceEpisodeTx estimates the episode between two fully known +// receipts of one topology. A measured source keeps its own receipts; an +// unknown one, or a receipt of another topology, holds the first receipt's +// values. The marker keeps the estimate distinguishable from measurement. +func bridgeSourceEpisodeTx(ctx context.Context, tx *sql.Tx, episode []sourceReceipt) error { + anchor := episode[0].HistorySources + for i := 1; i < len(episode); i++ { + v := episode[i-1].HistorySources + if v.Topology != anchor.Topology || v.Missing&MissingLoad != 0 { + v = anchor + } + for flag, w := range map[HistoryMissing][2]*float64{MissingGrid: {&v.GridW, &anchor.GridW}, MissingPV: {&v.PVW, &anchor.PVW}, MissingBattery: {&v.BatteryW, &anchor.BatteryW}, MissingEV: {&v.EVW, &anchor.EVW}, MissingV2X: {&v.V2XW, &anchor.V2XW}} { + if v.Missing&flag != 0 { + *w[0] = *w[1] + } + } + if err := writeSourceIntervalTx(ctx, tx, episode[i-1].AtMS, episode[i].AtMS, v, EstimatedLoad); err != nil { + return err + } + } + return nil +} + +func writeSourceIntervalTx(ctx context.Context, tx *sql.Tx, from, until int64, v HistorySources, missing HistoryMissing) error { + // A delayed writer may finish a just-closed local day. Invalidate only + // the nearby cached dates; source rows and old history stay intact. + low := time.UnixMilli(from).UTC().AddDate(0, 0, -1).Format("2006-01-02") + high := time.UnixMilli(until).UTC().AddDate(0, 0, 1).Format("2006-01-02") + for _, table := range []string{"energy_daily", "energy_daily_partial", "history_daily_quality"} { + if _, err := tx.ExecContext(ctx, `DELETE FROM `+table+` WHERE day>=? AND day<=?`, low, high); err != nil { + return err + } + } + for from < until { + start := bucketStart(from, ArchiveResolutionMS) + to := min(until, start+ArchiveResolutionMS) + ms := to - from + h := float64(ms) / 3600000 + if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_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,ev_wh=ev_wh+excluded.ev_wh,ev_charge_wh=ev_charge_wh+excluded.ev_charge_wh,ev_discharge_wh=ev_discharge_wh+excluded.ev_discharge_wh`, + start, ArchiveResolutionMS, from, to, ms, 1, + math.Max(0, v.GridW)*h, math.Max(0, -v.GridW)*h, -v.PVW*h, + math.Max(0, v.BatteryW)*h, math.Max(0, -v.BatteryW)*h, 0, (v.EVW+v.V2XW)*h, (math.Max(0, v.EVW)+math.Max(0, v.V2XW))*h, (math.Max(0, -v.EVW)+math.Max(0, -v.V2XW))*h); err != nil { + return err + } + if err := putHistoryQuality(ctx, tx, "source_energy", start, ArchiveResolutionMS, missing); err != nil { + return err + } + from = to + } + return nil +} + func (s *Store) sourceEnergyFrom(ctx context.Context) (int64, bool, error) { var from int64 err := s.history.QueryRowContext(ctx, `SELECT CAST(value AS INTEGER) FROM history_source_cursor WHERE key='from'`).Scan(&from) diff --git a/go/internal/state/history_sources_test.go b/go/internal/state/history_sources_test.go index c5eb57448..b9ea6f1db 100644 --- a/go/internal/state/history_sources_test.go +++ b/go/internal/state/history_sources_test.go @@ -127,10 +127,16 @@ func TestSourceLoadKeepsUnknownDistinctFromZero(t *testing.T) { if err != nil || len(b) != 1 { t.Fatal(err) } + // A brief unknown between known receipts is bridged, but only as a + // marked estimate. Other missing evidence stays unknown. + bridged := kind == "missing EV" || kind == "write gap" known := b[0].missing&MissingLoad == 0 - if known != (kind == "zero" || kind == "duplicate") { + if known != (kind == "zero" || kind == "duplicate" || bridged) { t.Fatal("missing evidence classified as measured zero", kind) } + if (b[0].missing&EstimatedLoad != 0) != bridged { + t.Fatal("bridge marker wrong", kind) + } if known { near(t, "zero Wh", b[0].wh, 0) } @@ -138,7 +144,7 @@ func TestSourceLoadKeepsUnknownDistinctFromZero(t *testing.T) { } } -func TestSourceLoadUsesSnapshotTimeAndDoesNotBridgeRestart(t *testing.T) { +func TestSourceLoadUsesSnapshotTimeAndMarksRestartBridge(t *testing.T) { path := filepath.Join(t.TempDir(), "state.db") s, err := Open(path) if err != nil { @@ -159,23 +165,22 @@ func TestSourceLoadUsesSnapshotTimeAndDoesNotBridgeRestart(t *testing.T) { if first != base+20000 || covered != 20000 { t.Fatal("source power assigned before observation") } - if err := s.Close(); err != nil { - t.Fatal(err) - } - s, err = Open(path) - if err != nil { + s = reopenSourceStore(t, s) + sourceTick(t, s, base+50000, HistorySources{Topology: "grid", GridW: 1000}) + var missing HistoryMissing + if err := s.history.QueryRow(`SELECT SUM(covered_ms),(SELECT missing FROM history_quality WHERE kind='source_energy') FROM history_source_energy`).Scan(&covered, &missing); err != nil { t.Fatal(err) } - defer s.Close() - if err := s.EnableHistoryAggregation(); err != nil { - t.Fatal(err) + if covered != 30000 || missing != EstimatedLoad { + t.Fatal("brief stop not bridged as a marked estimate", covered, missing) } - sourceTick(t, s, base+50000, HistorySources{Topology: "grid", GridW: 1000}) + s = reopenSourceStore(t, s) + sourceTick(t, s, base+50000+maxSourceBridgeMS+1, HistorySources{Topology: "grid", GridW: 1000}) if err := s.history.QueryRow(`SELECT SUM(covered_ms) FROM history_source_energy`).Scan(&covered); err != nil { t.Fatal(err) } - if covered != 20000 { - t.Fatal("integrated across stopped process") + if covered != 30000 { + t.Fatal("integrated across a long stop") } } diff --git a/go/internal/state/history_sqlite.go b/go/internal/state/history_sqlite.go index 9fd37d286..2eb2be37d 100644 --- a/go/internal/state/history_sqlite.go +++ b/go/internal/state/history_sqlite.go @@ -154,9 +154,10 @@ func (s *Store) openHistory() error { } } // A process stop has no continuous sampling proof, even when it was brief. - // Keep completed energy and quality, but never bridge the stopped process. + // Keep completed energy and quality. Source history may bridge a brief + // stop only as a marked estimate; the older paths never bridge it. if _, err := s.history.Exec(`DELETE FROM history_load_estimate_pending; - DELETE FROM history_source_cursor WHERE key='last'; + INSERT INTO history_source_cursor VALUES('stopped','1') ON CONFLICT(key) DO UPDATE SET value=excluded.value; UPDATE history_partial_cursor SET value=0 WHERE key IN ('last','residual_eligible'); UPDATE history_site_cursor SET value=0 WHERE key='last'`); err != nil { return err From 17bdaf743397d5093494c112c2cff0496b664f0c Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 20:32:27 +0200 Subject: [PATCH 2/4] fix(history): forgiving Home use display, strict marker for training Follows Fredrik's two-layer direction (19 October). Display is forgiving: an unknown PV, battery or EV source holds its last known value; the site meter, a writer pause or a restart holds for at most two minutes; a topology change no longer stops integration. A day's flow is unknown only when it is unknown or unrecorded for more than a tenth of the range. Only two fully measured snapshots a safe tick apart, of one topology and one process, make a measured interval. Every other interval keeps EstimatedLoad, and quarter-hour retention now keeps that bit, so the training layer can skip it. Replaces the pending episodes of the first version. Closed source-era days cached under the old rule are recomputed once. Relates to #1750. Co-Authored-By: Claude Opus 5.5 --- .changeset/bridge-short-history-gaps.md | 2 +- docs/site-convention.md | 15 +- go/internal/state/history_dashboard.go | 60 ++++-- go/internal/state/history_quality.go | 2 +- .../state/history_source_bridge_test.go | 138 ++++++++---- go/internal/state/history_sources.go | 201 ++++++++---------- go/internal/state/history_sources_test.go | 2 +- go/internal/state/history_sqlite.go | 3 + 8 files changed, 237 insertions(+), 186 deletions(-) diff --git a/.changeset/bridge-short-history-gaps.md b/.changeset/bridge-short-history-gaps.md index 915d09f1c..276ba0f06 100644 --- a/.changeset/bridge-short-history-gaps.md +++ b/.changeset/bridge-short-history-gaps.md @@ -2,4 +2,4 @@ "ftw": patch --- -Keep the day's Home use after a Core restart or a source that drops out for up to two minutes. The gap is bridged as a marked estimate; longer gaps, topology changes and stops over two minutes stay unknown. +Keep the day's Home use when a device drops out or Core restarts. An unknown PV, battery or EV source holds its last known value, and the meter or a restart holds for up to two minutes, marked as estimated. A day's total is unknown only when more than a tenth of it is missing. Days cached under the old rule are recomputed once. diff --git a/docs/site-convention.md b/docs/site-convention.md index a699e7985..72872e5f8 100644 --- a/docs/site-convention.md +++ b/docs/site-convention.md @@ -56,11 +56,16 @@ holds each qualified snapshot forward until the next safe snapshot, at most the sums when reading a chart interval, day or tariff period. A negative minute may contribute to a valid longer interval; it is never clipped first. Chart means need complete interval coverage. Missing sources, writer gaps, -process stops and topology changes cannot become measured zero. A gap of at -most two minutes between fully known snapshots of one topology is bridged as a -marked estimate: measured sources keep their snapshots, and an unknown source -or a stopped Core holds the last known values. Longer gaps stay unknown. This -is an estimate from received measurements, not a correction of device clocks. +process stops and topology changes cannot become measured zero. + +History has two layers. Display is forgiving: an unknown PV, battery or EV +source holds its last known value, and the site meter, a writer pause or a +Core restart holds for at most two minutes. Every such minute is marked +`EstimatedLoad`, and readers never clear the mark. A day's flow is unknown only +when it is unknown or unrecorded for more than a tenth of the range; +unrecorded time counts as nothing. Training is strict: it uses only minutes +without the mark. This is an estimate from received measurements, not a +correction of device clocks. In self-consumption mode, the controller drives the site meter toward the grid target (normally 0 W): it charges from live surplus and may discharge diff --git a/go/internal/state/history_dashboard.go b/go/internal/state/history_dashboard.go index 01230129f..2ea1cc884 100644 --- a/go/internal/state/history_dashboard.go +++ b/go/internal/state/history_dashboard.go @@ -273,18 +273,23 @@ func (s *Store) walkSiteEnergy(ctx context.Context, from, to int64, visit func(s return rows.Err() } +// maxSourceDayUnknownShare is the share of a source-era range a flow may be +// unknown or unrecorded before its total is unknown. Below it the total is +// an estimate of what was recorded; unrecorded time counts as nothing. +const maxSourceDayUnknownShare = 10 // percent + func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEnergy, error) { + cut, hasSources, err := s.sourceEnergyFrom(ctx) + if err != nil { + return DayEnergy{}, err + } var out DayEnergy var originalLoad float64 var originalMissing HistoryMissing var estimatedMS, coveredMS int64 - var sourceMS int64 - var first, last int64 - err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { - if first == 0 || r.first < first { - first = r.first - } - last = max(last, r.last) + var sourceMS, sourceEraMS int64 + unknownMS := map[HistoryMissing]int64{} + err = s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { out.ImportWh += r.imp out.ExportWh += r.exp out.PVWh += r.pv @@ -293,8 +298,8 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner originalLoad += r.load originalMissing |= r.missing coveredMS += r.covered - if r.source { - sourceMS += r.covered + if hasSources && r.first >= cut { + sourceEraMS += r.covered } if r.missing&^MissingSoC == MissingLoad|BalanceMismatch || r.missing&^MissingSoC == MissingLoad|MissingEV { if r.estimateCovered > 0 && r.estimateCovered == r.covered { @@ -304,12 +309,25 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner } out.LoadWh += r.load out.Intervals += r.n - out.Missing |= r.missing if r.covered > 0 { out.observed |= MissingPower &^ r.missing } + if r.source { + // Source rows decide each flow by its unknown time below. + sourceMS += r.covered + for _, flag := range []HistoryMissing{MissingGrid, MissingPV, MissingBattery, MissingLoad, MissingEV, MissingV2X} { + if r.missing&flag != 0 { + unknownMS[flag] += r.covered + } + } + r.missing &^= MissingPower + } + out.Missing |= r.missing return nil }) + if err != nil { + return out, err + } if sourceMS == 0 && (out.LoadWh < 0 || (estimatedMS > 120000 && estimatedMS > coveredMS/20)) { out.LoadWh = originalLoad out.Missing = originalMissing @@ -318,21 +336,21 @@ func (s *Store) aggregateDayEnergy(ctx context.Context, from, to int64) (DayEner out.LoadWh = 0 out.Missing |= MissingLoad } - cut, hasSources, coverageErr := s.sourceEnergyFrom(ctx) - if coverageErr != nil { - return out, coverageErr - } if hasSources && to > cut { - expected, err := s.sourcePeriodCoverage(ctx, max(from, cut), to) + // The writer's last receipt ends the range, so a live tail is not a gap. + through, _, err := s.HistorySourceProgress(ctx) if err != nil { return out, err } - if sourceMS < expected { - out.Missing |= MissingLoad - } - if coveredMS < last-first { - out.Missing |= MissingLoad + span := max(0, min(to, through)-max(from, cut)) + // Unrecorded time leaves the other totals as recorded, as before; + // the household balance needs it covered. + unknownMS[MissingLoad] += max(0, span-sourceEraMS) + for flag, ms := range unknownMS { + if ms*100 > span*maxSourceDayUnknownShare { + out.Missing |= flag + } } } - return out, err + return out, nil } diff --git a/go/internal/state/history_quality.go b/go/internal/state/history_quality.go index f816643be..cd3604cc3 100644 --- a/go/internal/state/history_quality.go +++ b/go/internal/state/history_quality.go @@ -70,4 +70,4 @@ func putHistoryQuality(ctx context.Context, tx *sql.Tx, kind string, start, widt // SQL has no bitwise-OR aggregate. MAX per bit preserves unknown evidence // across every source row, including mixed complete/incomplete buckets. -const historyMissingAggregate = `MAX(missing&1)|MAX(missing&2)|MAX(missing&4)|MAX(missing&8)|MAX(missing&16)|MAX(missing&32)|MAX(missing&64)|MAX(missing&128)` +const historyMissingAggregate = `MAX(missing&1)|MAX(missing&2)|MAX(missing&4)|MAX(missing&8)|MAX(missing&16)|MAX(missing&32)|MAX(missing&64)|MAX(missing&128)|MAX(missing&256)` diff --git a/go/internal/state/history_source_bridge_test.go b/go/internal/state/history_source_bridge_test.go index ea92d4d61..104a1aada 100644 --- a/go/internal/state/history_source_bridge_test.go +++ b/go/internal/state/history_source_bridge_test.go @@ -1,6 +1,7 @@ package state import ( + "context" "testing" "time" ) @@ -8,28 +9,27 @@ import ( // Two hours of a 1000 W house, sampled every 5 s, with one disturbance one // hour in. Observed on v0.154.0-beta.1 (19 October 2026): each short // disturbance made the whole day's Home use unknown. -func TestSourceLoadDaySurvivesShortGap(t *testing.T) { - const tick, hour = int64(5000), int64(3600000) +func TestSourceLoadDayForgivesShortGaps(t *testing.T) { + const tick, hour, minute = int64(5000), int64(3600000), int64(60000) for _, c := range []struct { name string // stopMS is a process stop starting one hour in. stopMS int64 // receipt alters the snapshot off ms after the first hour. receipt func(off int64, v *HistorySources) - known bool - // importWh is checked when set; the recorded grid must survive. + // Zero Wh means the total must be unknown. loadWh, importWh float64 }{ - {name: "no disturbance", known: true, loadWh: 2000, importWh: 2000}, - {name: "20 s restart", stopMS: 20000, known: true, loadWh: 2000, importWh: 2000}, + {name: "no disturbance", loadWh: 2000, importWh: 2000}, + {name: "20 s restart", stopMS: 20000, loadWh: 2000, importWh: 2000}, {name: "20 s restart, EV and its device back after 30 s", stopMS: 20000, receipt: func(off int64, v *HistorySources) { if off <= 50000 { v.Missing, v.Topology = MissingEV, "grid" } - }, known: true, loadWh: 2000, importWh: 2000}, - {name: "EV unknown 15 s", receipt: unknownFor(MissingEV, 15000), known: true, loadWh: 2000, importWh: 2000}, + }, loadWh: 2000, importWh: 2000}, + {name: "EV unknown 15 s", receipt: unknownFor(MissingEV, 15000), loadWh: 2000, importWh: 2000}, // The unknown charger holds its last value; the meter keeps its own - // receipts. The household carries the 2 kW for 15 s, nothing else. + // readings. The household carries the 2 kW for 15 s, nothing else. {name: "EV starts while unknown", receipt: func(off int64, v *HistorySources) { if off > 0 { v.GridW, v.EVW = 3000, 2000 @@ -37,16 +37,24 @@ func TestSourceLoadDaySurvivesShortGap(t *testing.T) { if off > 0 && off <= 15000 { v.Missing, v.EVW = MissingEV, 0 } - }, known: true, loadWh: 2000 + 2000*15.0/3600, importWh: 1000 + 1000*5.0/3600 + 3000*3595.0/3600}, - {name: "PV unknown 60 s", receipt: unknownFor(MissingPV, 60000), known: true, loadWh: 2000, importWh: 2000}, - {name: "PV unknown 145 s", receipt: unknownFor(MissingPV, 145000), importWh: 2000}, - {name: "EV unknown 5 min", receipt: unknownFor(MissingEV, 300000), importWh: 2000}, - {name: "3 min restart", stopMS: 180000}, + }, loadWh: 2000 + 2000*15.0/3600, importWh: 1000 + 1000*5.0/3600 + 3000*3595.0/3600}, + {name: "PV unknown 145 s", receipt: unknownFor(MissingPV, 145000), loadWh: 2000, importWh: 2000}, + {name: "EV unknown 30 min", receipt: unknownFor(MissingEV, 30*minute), loadWh: 2000, importWh: 2000}, {name: "meter replaced", receipt: func(off int64, v *HistorySources) { if off > 0 { v.Topology = "new meter+charger" } - }}, + }, loadWh: 2000, importWh: 2000}, + // The meter holds for two minutes. The other 185 s count as nothing, + // well inside a tenth of the range; Home use loses the four whole + // minutes they touch. + {name: "meter unknown 5 min", receipt: unknownFor(MissingGrid, 5*minute), loadWh: 2000 - 1000*240.0/3600, importWh: 2000 - 1000*185.0/3600}, + {name: "3 min restart", stopMS: 3 * minute, loadWh: 1950, importWh: 1950}, + // Over a tenth of the range unrecorded: Home use is unknown, and + // the recorded import stays known, as before. + {name: "15 min restart", stopMS: 15 * minute, importWh: 1750}, + {name: "meter unknown 20 min", receipt: unknownFor(MissingGrid, 20*minute)}, + {name: "invalid configuration 30 min", receipt: unknownFor(MissingLoad, 30*minute), importWh: 2000}, } { t.Run(c.name, func(t *testing.T) { s := sourceStore(t) @@ -69,56 +77,100 @@ func TestSourceLoadDaySurvivesShortGap(t *testing.T) { if err != nil { t.Fatal(err) } - if c.importWh != 0 { - near(t, "import Wh", d.ImportWh, c.importWh) - } - if got := d.Missing&MissingLoad == 0; got != c.known { - t.Fatalf("Home use known=%v, want %v (quality %s)", got, c.known, d.Missing.Quality()) - } - if !c.known { - return + check := func(name string, flag HistoryMissing, got, want float64) { + t.Helper() + if known := d.Missing&flag == 0; known != (want != 0) { + t.Fatalf("%s known=%v, want %v", name, known, want != 0) + } + if want != 0 && (got-want > 1e-6 || want-got > 1e-6) { + t.Fatalf("%s %v Wh, want %v", name, got, want) + } } - if q := d.Missing.Quality(); q != "estimated" { + check("Home use", MissingLoad, d.LoadWh, c.loadWh) + check("import", MissingGrid, d.ImportWh, c.importWh) + if q := d.Missing.Quality(); c.loadWh != 0 && q != "estimated" { t.Fatalf("quality %s, want estimated", q) } if d.Missing&MissingPV != 0 { t.Fatal("PV total became unknown") } - if diff := d.LoadWh - c.loadWh; diff > 1e-6 || diff < -1e-6 { - t.Fatalf("Home use %v Wh, want %v", d.LoadWh, c.loadWh) - } }) } } -// A bridge that closes after midnight must still land in the closed day, so -// the day waits for it before it becomes final. -func TestSourceBridgeHoldsDayOpenUntilItCloses(t *testing.T) { +// Training reads only measured minutes, so every minute the display holds +// or bridges keeps its marker, also after retention merges minutes. +func TestSourceLoadMarksEveryEstimatedMinute(t *testing.T) { s := sourceStore(t) - midnight := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() - for at := midnight - 60000; at <= midnight+10000; at += 5000 { + base := time.Now().UTC().Truncate(24*time.Hour).UnixMilli() - 24*3600000 + for at := base; at <= base+30*60000; at += 5000 { v := HistorySources{GridW: 1000} - if at > midnight-10000 && at < midnight+10000 { + if at > base+10*60000 && at <= base+10*60000+15000 { v.Missing = MissingEV } sourceTick(t, s, at, v) - through, _, err := s.HistorySourceProgress(t.Context()) + } + marked := func(width int64) []int64 { + t.Helper() + rows, err := s.history.Query(`SELECT start_ms FROM history_quality WHERE kind='source_energy' AND resolution_ms=? AND missing&?<>0 ORDER BY start_ms`, width, EstimatedLoad) if err != nil { t.Fatal(err) } - if at < midnight+10000 && through > midnight-10000 { - t.Fatal("day closed before its bridge") + defer rows.Close() + var out []int64 + for rows.Next() { + var start int64 + if err := rows.Scan(&start); err != nil { + t.Fatal(err) + } + out = append(out, start-base) + } + return out + } + if got := marked(ArchiveResolutionMS); len(got) != 1 || got[0] != 10*60000 { + t.Fatal("estimated minutes", got) + } + if err := s.rollSiteEnergy(context.Background(), base+30*60000); err != nil { + t.Fatal(err) + } + if got := marked(15 * 60000); len(got) != 1 || got[0] != 0 { + t.Fatal("marker lost when minutes merged", got) + } +} + +// Closed days cached under the whole-day rule are recomputed once. +func TestSourceDayCacheRecomputedOnce(t *testing.T) { + s := sourceStore(t) + day := time.Now().UTC().Truncate(24*time.Hour).AddDate(0, 0, -1) + sourceTick(t, s, day.UnixMilli(), HistorySources{GridW: 1000}) + if _, err := s.history.Exec(`DELETE FROM history_migrations WHERE name=?`, sourceDayCoverageMigration); err != nil { + t.Fatal(err) + } + key := day.Format("2006-01-02") + save := func() { + t.Helper() + if err := s.SaveDailyEnergyForSourceMode(key, DayEnergy{Missing: MissingLoad, Intervals: 1}, true); err != nil { + t.Fatal(err) + } + } + cached := func() bool { + t.Helper() + _, ok, err := s.LoadDailyEnergy(key) + if err != nil { + t.Fatal(err) } + return ok } - through, _, err := s.HistorySourceProgress(t.Context()) - if err != nil || through != midnight+10000 { - t.Fatal("bridge not committed", through, err) + save() + s = reopenSourceStore(t, s) + if cached() { + t.Fatal("day cached under the old rule survived") } - d, err := s.DailyEnergy(midnight-60000, midnight) - if err != nil || d.Missing&MissingLoad != 0 { - t.Fatal("closed day lost its Home use", err) + save() + s = reopenSourceStore(t, s) + if !cached() { + t.Fatal("recomputed day dropped again") } - near(t, "closed day Wh", d.LoadWh, 1000.0/60) } func unknownFor(flag HistoryMissing, ms int64) func(int64, *HistorySources) { diff --git a/go/internal/state/history_sources.go b/go/internal/state/history_sources.go index 933eccc2f..d25605474 100644 --- a/go/internal/state/history_sources.go +++ b/go/internal/state/history_sources.go @@ -29,13 +29,14 @@ const sourceCursorSchema = `CREATE TABLE IF NOT EXISTS history_source_cursor(key const sourceEnergyTable = "history_source_energy" const maxSourceEnergyWidth int64 = 900000 -// maxSourceBridgeMS bounds an estimated bridge over a writer pause, a process -// restart or a briefly unknown source. It shares the history estimate budget -// of a recovered balance episode; longer gaps stay unknown. +// maxSourceBridgeMS bounds how long the site meter's last value, or a +// snapshot across a writer pause or process restart, may stand in for +// missing evidence. It shares the history estimate budget of a recovered +// balance episode. A longer gap stays uncovered. const maxSourceBridgeMS = maxLoadEstimateEpisodeMS -// sourceReceipt is one snapshot held in the writer cursor. Stopped marks the -// first receipt after a process stop. +// sourceReceipt is one snapshot. Stopped marks the first one after a +// process stop. type sourceReceipt struct { HistorySources Stopped bool `json:",omitempty"` @@ -45,11 +46,16 @@ type sourceCursor struct { At int64 Start int64 Sources HistorySources - // Episode starts at the last fully known receipt. Its intervals wait for - // known sources of that topology: if they return within the bridge - // budget the intervals are an estimate, otherwise they are recorded as - // before. Pending time never enters totals or the period's coverage. - Episode []sourceReceipt `json:",omitempty"` + // Held is each source's last known value. Its Missing names the sources + // not yet known since the cursor began. GridAt is when the site meter + // was last known. + Held *HistorySources `json:",omitempty"` + GridAt int64 `json:",omitempty"` +} + +// sourceFlows pairs each source's flag with its value. +func sourceFlows(v *HistorySources) map[HistoryMissing]*float64 { + return map[HistoryMissing]*float64{MissingGrid: &v.GridW, MissingPV: &v.PVW, MissingBattery: &v.BatteryW, MissingEV: &v.EVW, MissingV2X: &v.V2XW} } func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *HistoryPoint) error { @@ -64,41 +70,10 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History return err } } - // commit makes source energy final through end. Readers count the - // period's time as expected, and a closed day waits for it. - commit := func(end int64) error { - if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_periods VALUES(?,?) ON CONFLICT(start_ms) DO UPDATE SET last_ms=excluded.last_ms`, last.Start, end); err != nil { - return err - } - _, err := tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('through',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, end) - return err - } - // An abandoned episode keeps the original evidence and its unknowns, - // through its last receipt. - abandon := func() error { - n := len(last.Episode) - if n == 0 { - return nil - } - for i := 1; i < n; i++ { - if err := recordedSourceIntervalTx(ctx, tx, last.Episode[i-1].HistorySources, last.Episode[i]); err != nil { - return err - } - } - through := last.Episode[n-1].AtMS - last.Episode = nil - return commit(through) - } - drop := func() error { - if err := abandon(); err != nil { - return err - } + if p == nil || p.Sources == nil { _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key IN ('last','stopped')`) return err } - if p == nil || p.Sources == nil { - return drop() - } at := p.Sources.AtMS if at <= 0 { return errors.New("history source snapshot has no observation time") @@ -107,11 +82,12 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History if err := tx.QueryRowContext(ctx, `SELECT COALESCE((SELECT CAST(value AS INTEGER) FROM history_source_cursor WHERE key='through'),0)`).Scan(&through); err != nil { return err } - if latest := max(through, last.At); at <= latest { - if at == latest { + if at <= through { + if at == through { return nil } - return drop() + _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key IN ('last','stopped')`) + return err } if _, err := tx.ExecContext(ctx, `INSERT OR IGNORE INTO history_source_cursor VALUES('from',?)`, at); err != nil { return err @@ -129,52 +105,34 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History if _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key='stopped'`); err != nil { return err } - known := func(v HistorySources) bool { return v.Missing&MissingPower == 0 } switch { case last.At == 0: last.Start = at case at-last.At > maxSourceBridgeMS: - if err := abandon(); err != nil { - return err - } - // A long stop starts a new period, as an older writer may have run. + // The gap stays uncovered. A long stop starts a new period, as an + // older writer may have run. if cur.Stopped { last.Start = at } - case len(last.Episode) > 0: - last.Episode = append(last.Episode, cur) - case !cur.Stopped && at-last.At <= maxLoadEstimateTickMS && last.Sources.Topology == cur.Topology && known(last.Sources) && known(cur.HistorySources): - // Hold the previous receipt forward, bounded by the next safe tick. Do - // not assign a newly arrived charger value to time before its receipt. - if err := writeSourceIntervalTx(ctx, tx, last.At, at, last.Sources, 0); err != nil { - return err - } - case known(last.Sources): - last.Episode = []sourceReceipt{{HistorySources: last.Sources}, cur} default: - if err := recordedSourceIntervalTx(ctx, tx, last.Sources, cur); err != nil { + if err := holdSourceIntervalTx(ctx, tx, last, cur); err != nil { return err } } - if len(last.Episode) > 0 { - anchor := last.Episode[0].HistorySources - switch { - case known(cur.HistorySources) && cur.Topology == anchor.Topology && at-anchor.AtMS <= maxSourceBridgeMS: - if err := bridgeSourceEpisodeTx(ctx, tx, last.Episode); err != nil { - return err - } - last.Episode = nil - case at-anchor.AtMS > maxSourceBridgeMS: - if err := abandon(); err != nil { - return err - } + if last.Held == nil { + last.Held = &HistorySources{Missing: MissingPower &^ MissingLoad} + } + known := sourceFlows(last.Held) + for flag, w := range sourceFlows(&cur.HistorySources) { + if cur.Missing&flag == 0 { + *known[flag] = *w + last.Held.Missing &^= flag } } - end := at - if len(last.Episode) > 0 { - end = last.Episode[0].AtMS + if cur.Missing&MissingGrid == 0 { + last.GridAt = at } - if err := commit(end); err != nil { + if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_periods VALUES(?,?) ON CONFLICT(start_ms) DO UPDATE SET last_ms=excluded.last_ms`, last.Start, at); err != nil { return err } last.At, last.Sources = at, cur.HistorySources @@ -182,45 +140,39 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History if err != nil { return err } - _, err = tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, string(b)) + if _, err := tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('last',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, string(b)); err != nil { + return err + } + _, err = tx.ExecContext(ctx, `INSERT INTO history_source_cursor VALUES('through',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, at) return err } -// recordedSourceIntervalTx keeps the original evidence: the previous receipt -// held to the next safe tick of the same topology. Either endpoint's unknown -// source leaves the interval, and so its Load, unknown. -func recordedSourceIntervalTx(ctx context.Context, tx *sql.Tx, prev HistorySources, cur sourceReceipt) error { - if cur.Stopped || cur.AtMS-prev.AtMS > maxLoadEstimateTickMS || prev.Topology != cur.Topology { - return nil - } - missing := (prev.Missing | cur.Missing) & MissingPower - if missing != 0 { +// holdSourceIntervalTx holds the previous snapshot forward to this one. Do +// not assign a newly arrived charger value to time before its receipt. A +// source unknown in the snapshot keeps its last known value, the site meter +// for at most the bridge budget. Only two fully measured snapshots a safe +// tick apart, of one topology and one process, make a measured interval; +// every other interval carries EstimatedLoad, which readers never clear. +func holdSourceIntervalTx(ctx context.Context, tx *sql.Tx, last sourceCursor, cur sourceReceipt) error { + v := last.Sources + var missing HistoryMissing + for flag, w := range sourceFlows(&v) { + switch { + case v.Missing&flag == 0: + case last.Held == nil || last.Held.Missing&flag != 0 || flag == MissingGrid && cur.AtMS-last.GridAt > maxSourceBridgeMS: + missing, *w = missing|flag, 0 + default: + *w = *sourceFlows(last.Held)[flag] + } + } + // A snapshot with an invalid configuration has no household balance. + if missing != 0 || v.Missing&MissingLoad != 0 { missing |= MissingLoad } - return writeSourceIntervalTx(ctx, tx, prev.AtMS, cur.AtMS, prev, missing) -} - -// bridgeSourceEpisodeTx estimates the episode between two fully known -// receipts of one topology. A measured source keeps its own receipts; an -// unknown one, or a receipt of another topology, holds the first receipt's -// values. The marker keeps the estimate distinguishable from measurement. -func bridgeSourceEpisodeTx(ctx context.Context, tx *sql.Tx, episode []sourceReceipt) error { - anchor := episode[0].HistorySources - for i := 1; i < len(episode); i++ { - v := episode[i-1].HistorySources - if v.Topology != anchor.Topology || v.Missing&MissingLoad != 0 { - v = anchor - } - for flag, w := range map[HistoryMissing][2]*float64{MissingGrid: {&v.GridW, &anchor.GridW}, MissingPV: {&v.PVW, &anchor.PVW}, MissingBattery: {&v.BatteryW, &anchor.BatteryW}, MissingEV: {&v.EVW, &anchor.EVW}, MissingV2X: {&v.V2XW, &anchor.V2XW}} { - if v.Missing&flag != 0 { - *w[0] = *w[1] - } - } - if err := writeSourceIntervalTx(ctx, tx, episode[i-1].AtMS, episode[i].AtMS, v, EstimatedLoad); err != nil { - return err - } + if (v.Missing|cur.Missing)&(MissingPower|EstimatedLoad) != 0 || cur.Stopped || cur.AtMS-v.AtMS > maxLoadEstimateTickMS || v.Topology != cur.Topology { + missing |= EstimatedLoad } - return nil + return writeSourceIntervalTx(ctx, tx, v.AtMS, cur.AtMS, v, missing) } func writeSourceIntervalTx(ctx context.Context, tx *sql.Tx, from, until int64, v HistorySources, missing HistoryMissing) error { @@ -274,11 +226,32 @@ func (s *Store) HistorySourceProgress(ctx context.Context) (int64, bool, error) return through, err == nil, err } -func (s *Store) sourcePeriodCoverage(ctx context.Context, from, to int64) (int64, error) { - var ms int64 - err := s.history.QueryRowContext(ctx, `SELECT COALESCE(SUM(MIN(last_ms,?)-MAX(start_ms,?)),0) FROM history_source_periods WHERE last_ms>? AND start_ms=(SELECT date(CAST(value AS INTEGER)/1000,'unixepoch','-1 day') FROM history_source_cursor WHERE key='from')`); err != nil { + return err + } + } + if _, err := tx.ExecContext(ctx, `INSERT INTO history_migrations(name) VALUES(?)`, sourceDayCoverageMigration); err != nil { + return err + } + return tx.Commit() } + func (s *Store) sourcePeriods(ctx context.Context, from, to int64) ([][2]int64, error) { rows, err := s.history.QueryContext(ctx, `SELECT start_ms,last_ms FROM history_source_periods WHERE start_ms<=? AND last_ms>=? ORDER BY start_ms`, to, from) if err != nil { diff --git a/go/internal/state/history_sources_test.go b/go/internal/state/history_sources_test.go index b9ea6f1db..99120ff38 100644 --- a/go/internal/state/history_sources_test.go +++ b/go/internal/state/history_sources_test.go @@ -129,7 +129,7 @@ func TestSourceLoadKeepsUnknownDistinctFromZero(t *testing.T) { } // A brief unknown between known receipts is bridged, but only as a // marked estimate. Other missing evidence stays unknown. - bridged := kind == "missing EV" || kind == "write gap" + bridged := kind == "missing EV" || kind == "write gap" || kind == "topology" known := b[0].missing&MissingLoad == 0 if known != (kind == "zero" || kind == "duplicate" || bridged) { t.Fatal("missing evidence classified as measured zero", kind) diff --git a/go/internal/state/history_sqlite.go b/go/internal/state/history_sqlite.go index 2eb2be37d..23375702c 100644 --- a/go/internal/state/history_sqlite.go +++ b/go/internal/state/history_sqlite.go @@ -162,6 +162,9 @@ func (s *Store) openHistory() error { UPDATE history_site_cursor SET value=0 WHERE key='last'`); err != nil { return err } + if err := s.refreshSourceDayCache(context.Background()); err != nil { + return err + } ok = true return nil } From 2d1a434fc8ba92bc5cb1b485e3e6712d12ad5fca Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 20:38:30 +0200 Subject: [PATCH 3/4] fix(history): stop holding a source that would make the house negative A held value that turns the household residual negative cannot be right: the source changed while unknown, such as a charger that finished. That interval leaves the held flows unknown instead. Co-Authored-By: Claude Opus 5.5 --- go/internal/state/history_source_bridge_test.go | 11 +++++++++++ go/internal/state/history_sources.go | 15 +++++++++++++-- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/go/internal/state/history_source_bridge_test.go b/go/internal/state/history_source_bridge_test.go index 104a1aada..8f79b324a 100644 --- a/go/internal/state/history_source_bridge_test.go +++ b/go/internal/state/history_source_bridge_test.go @@ -40,6 +40,17 @@ func TestSourceLoadDayForgivesShortGaps(t *testing.T) { }, loadWh: 2000 + 2000*15.0/3600, importWh: 1000 + 1000*5.0/3600 + 3000*3595.0/3600}, {name: "PV unknown 145 s", receipt: unknownFor(MissingPV, 145000), loadWh: 2000, importWh: 2000}, {name: "EV unknown 30 min", receipt: unknownFor(MissingEV, 30*minute), loadWh: 2000, importWh: 2000}, + // A 2 kW charge ends 25 minutes into a 30-minute gap. The held 2 kW + // would make the house negative, so it is not held from then; Home + // use loses the six minutes this touches instead of 2 kW for five. + {name: "EV finishes while unknown", receipt: func(off int64, v *HistorySources) { + if off < 25*minute { + v.GridW, v.EVW = 3000, 2000 + } + if off > 0 && off <= 30*minute { + v.Missing, v.EVW = MissingEV, 0 + } + }, loadWh: 2000 - 1000*6.0/60, importWh: 1000 + 3000*25.0/60 + 1000*35.0/60}, {name: "meter replaced", receipt: func(off int64, v *HistorySources) { if off > 0 { v.Topology = "new meter+charger" diff --git a/go/internal/state/history_sources.go b/go/internal/state/history_sources.go index d25605474..1599a167d 100644 --- a/go/internal/state/history_sources.go +++ b/go/internal/state/history_sources.go @@ -155,14 +155,25 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History // every other interval carries EstimatedLoad, which readers never clear. func holdSourceIntervalTx(ctx context.Context, tx *sql.Tx, last sourceCursor, cur sourceReceipt) error { v := last.Sources - var missing HistoryMissing - for flag, w := range sourceFlows(&v) { + var missing, held HistoryMissing + flows := sourceFlows(&v) + for flag, w := range flows { switch { case v.Missing&flag == 0: case last.Held == nil || last.Held.Missing&flag != 0 || flag == MissingGrid && cur.AtMS-last.GridAt > maxSourceBridgeMS: missing, *w = missing|flag, 0 default: *w = *sourceFlows(last.Held)[flag] + held |= flag + } + } + // A household never feeds the site. If a held value says it does, the + // source changed while unknown, such as a charger that finished. + if held != 0 && v.GridW-v.PVW-v.BatteryW-v.EVW-v.V2XW < 0 { + for flag, w := range flows { + if held&flag != 0 { + missing, *w = missing|flag, 0 + } } } // A snapshot with an invalid configuration has no household balance. From 85100d65aa897392e39fc0d7a4fe89376633d03b Mon Sep 17 00:00:00 2001 From: Fredrik Ahlgren Date: Mon, 19 Oct 2026 20:50:14 +0200 Subject: [PATCH 4/4] fix(history): a failed source leaves only its own flow unknown historySources set MissingLoad for every reason except invalid_balance. Telemetry puts a source's own failure there too, such as "offline:sma:pv" or "stale:easee:ev", so a real snapshot with a silent inverter or charger voided the whole balance and the source writer held nothing. Only configuration and duplicate-flow reasons now set MissingLoad; a failed source is unknown on its own flag. Found by the #1755 session through historySources. Co-Authored-By: Claude Opus 5.5 --- go/cmd/ftw/history_source_load_test.go | 45 ++++++++++++++++++++++++++ go/cmd/ftw/main.go | 5 +-- go/internal/telemetry/forecast.go | 14 ++++++++ 3 files changed, 62 insertions(+), 2 deletions(-) diff --git a/go/cmd/ftw/history_source_load_test.go b/go/cmd/ftw/history_source_load_test.go index bfa7df3e9..fb290398a 100644 --- a/go/cmd/ftw/history_source_load_test.go +++ b/go/cmd/ftw/history_source_load_test.go @@ -57,6 +57,48 @@ func TestHistoryLoadIndependentReportingCadence(t *testing.T) { } } +// A silent inverter is unknown on its own. Its balance with the meter is +// held, not voided, so the household keeps its estimated energy. +func TestHistoryHoldsASilentSourceThroughTheBalance(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) + } + tel := telemetry.NewStore() + tel.Update("grid", telemetry.DerMeter, 1000, nil, nil) + tel.RecordDriverSuccess("grid") + tel.Update("sma", telemetry.DerPV, 0, nil, nil) + tel.RecordDriverSuccess("sma") + base := time.Now().Add(-20 * time.Minute).Truncate(5 * time.Minute).UnixMilli() + for i := int64(0); i <= 60; i++ { + if i == 6 { + tel.Get("sma", telemetry.DerPV).UpdatedAt = time.Now().Add(-10 * time.Minute) + } + p, _ := buildHistoryPoint(tel, tickPersistControl{SiteMeterDriver: "grid"}, base+i*10000, 90*time.Second) + if i >= 6 && p.Sources.Missing != state.MissingPV { + t.Fatalf("silent inverter marked %b, want only its own flow", p.Sources.Missing) + } + p.Sources.AtMS = base + i*10000 + if err := st.RecordTickWithEnergy(p, nil, nil); err != nil { + t.Fatal(err) + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + d, err := st.DailyEnergy(base, base+600000) + if err != nil { + t.Fatal(err) + } + if d.Missing&state.MissingLoad != 0 || math.Abs(d.LoadWh-1000.0/6) > 0.001 { + t.Fatalf("household lost through a silent inverter: %v Wh, quality %s", d.LoadWh, d.Missing.Quality()) + } +} + func TestHistorySourceQualificationDoesNotInventMissingFlows(t *testing.T) { for _, tc := range []struct { reason string @@ -65,8 +107,11 @@ func TestHistorySourceQualificationDoesNotInventMissingFlows(t *testing.T) { want state.HistoryMissing }{ {"missing_config", 0, true, state.MissingLoad}, + {"duplicate_charger:easee", 0, true, state.MissingLoad}, {"invalid_balance", 1000, true, 0}, {"nonfinite_sum", math.Inf(1), false, state.MissingLoad | state.MissingEV}, + {"stale:easee:ev", 0, false, state.MissingEV}, + {"offline:easee:ev", 0, false, state.MissingEV}, } { known := map[telemetry.DerType]bool{telemetry.DerMeter: true, telemetry.DerPV: true, telemetry.DerBattery: true, telemetry.DerEV: tc.known, telemetry.DerV2X: true} p := historySources(telemetry.ForecastReading{At: time.Now(), Known: known, Reason: tc.reason, EVW: tc.watts}) diff --git a/go/cmd/ftw/main.go b/go/cmd/ftw/main.go index 7fcbfc827..d50b719ad 100644 --- a/go/cmd/ftw/main.go +++ b/go/cmd/ftw/main.go @@ -4161,8 +4161,9 @@ func historySources(reading telemetry.ForecastReading) *state.HistorySources { } } // A negative instantaneous residual does not invalidate its independent - // sources. Missing configuration and duplicate electrical flows do. - if reading.Reason != "" && reading.Reason != "invalid_balance" { + // sources. Missing configuration and duplicate electrical flows do. A + // failed source is unknown on its own flag, so history may hold it. + if reading.Reason != "" && reading.Reason != "invalid_balance" && !telemetry.SourceFailure(reading.Reason) { p.Missing |= state.MissingLoad } for flag, w := range map[state.HistoryMissing]*float64{state.MissingGrid: &p.GridW, state.MissingPV: &p.PVW, state.MissingBattery: &p.BatteryW, state.MissingEV: &p.EVW, state.MissingV2X: &p.V2XW} { diff --git a/go/internal/telemetry/forecast.go b/go/internal/telemetry/forecast.go index 2ac8457af..d5e3743e7 100644 --- a/go/internal/telemetry/forecast.go +++ b/go/internal/telemetry/forecast.go @@ -5,6 +5,7 @@ import ( "fmt" "math" "sort" + "strings" "time" ) @@ -96,6 +97,19 @@ func (s *Store) LiveMeasurementNow(siteMeter string, opts ForecastOptions) Forec return s.measurementLocked(time.Now(), siteMeter, opts, false) } +// SourceFailure reports whether a reading's Reason names one failed source. +// That source is already unknown in Known; the balance of the others still +// holds. Configuration and duplicate-flow reasons are not source failures. +// Keep in step with the reasons measurementLocked gives. +func SourceFailure(reason string) bool { + for _, prefix := range []string{"missing:", "offline:", "unknown_power:", "nonfinite:", "stale:", "future:", "invalid_sign:"} { + if strings.HasPrefix(reason, prefix) { + return true + } + } + return reason == "nonfinite_pv_sum" +} + func (s *Store) forecastMeasurementLocked(now time.Time, siteMeter string, opts ForecastOptions) ForecastReading { return s.measurementLocked(now, siteMeter, opts, true) }