diff --git a/.changeset/bridge-short-history-gaps.md b/.changeset/bridge-short-history-gaps.md new file mode 100644 index 000000000..276ba0f06 --- /dev/null +++ b/.changeset/bridge-short-history-gaps.md @@ -0,0 +1,5 @@ +--- +"ftw": patch +--- + +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 77b9d80d7..72872e5f8 100644 --- a/docs/site-convention.md +++ b/docs/site-convention.md @@ -56,8 +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. 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/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/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 new file mode 100644 index 000000000..8f79b324a --- /dev/null +++ b/go/internal/state/history_source_bridge_test.go @@ -0,0 +1,210 @@ +package state + +import ( + "context" + "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 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) + // Zero Wh means the total must be unknown. + loadWh, importWh float64 + }{ + {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" + } + }, 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 + // 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 + } + if off > 0 && off <= 15000 { + v.Missing, v.EVW = MissingEV, 0 + } + }, 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" + } + }, 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) + 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) + } + 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) + } + } + 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") + } + }) + } +} + +// 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) + 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 > base+10*60000 && at <= base+10*60000+15000 { + v.Missing = MissingEV + } + sourceTick(t, s, at, v) + } + 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) + } + 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 + } + save() + s = reopenSourceStore(t, s) + if cached() { + t.Fatal("day cached under the old rule survived") + } + save() + s = reopenSourceStore(t, s) + if !cached() { + t.Fatal("recomputed day dropped again") + } +} + +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..1599a167d 100644 --- a/go/internal/state/history_sources.go +++ b/go/internal/state/history_sources.go @@ -29,20 +29,37 @@ const sourceCursorSchema = `CREATE TABLE IF NOT EXISTS history_source_cursor(key const sourceEnergyTable = "history_source_energy" const maxSourceEnergyWidth int64 = 900000 +// 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. Stopped marks the first one after a +// process stop. +type sourceReceipt struct { + HistorySources + Stopped bool `json:",omitempty"` +} + +type sourceCursor struct { + At int64 + Start int64 + Sources HistorySources + // 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 { - 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,14 +70,23 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History return err } } + if p == nil || p.Sources == nil { + _, err := tx.ExecContext(ctx, `DELETE FROM history_source_cursor WHERE key IN ('last','stopped')`) + return err + } + 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 at == through { + return nil } + _, 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 { @@ -72,49 +98,44 @@ 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 - } + switch { + case last.At == 0: + last.Start = at + case at-last.At > maxSourceBridgeMS: + // The gap stays uncovered. A long stop starts a new period, as an + // older writer may have run. + if cur.Stopped { + last.Start = at } - v := last.Sources - missing := (v.Missing | p.Sources.Missing) & MissingPower - if missing != 0 { - missing |= MissingLoad + default: + if err := holdSourceIntervalTx(ctx, tx, last, cur); err != nil { + return err } - 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 { - return err - } - if err := putHistoryQuality(ctx, tx, "source_energy", start, ArchiveResolutionMS, missing); err != nil { - return err - } - from = to + } + 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 } } - last.At, last.Sources = at, *p.Sources + if cur.Missing&MissingGrid == 0 { + last.GridAt = at + } + 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 b, err := json.Marshal(last) if err != nil { return err @@ -126,6 +147,76 @@ func (s *Store) recordSourceEnergyTx(ctx context.Context, tx *sql.Tx, p *History return err } +// 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, 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. + if missing != 0 || v.Missing&MissingLoad != 0 { + missing |= MissingLoad + } + if (v.Missing|cur.Missing)&(MissingPower|EstimatedLoad) != 0 || cur.Stopped || cur.AtMS-v.AtMS > maxLoadEstimateTickMS || v.Topology != cur.Topology { + missing |= EstimatedLoad + } + 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 { + // 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) @@ -146,11 +237,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 c5eb57448..99120ff38 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" || kind == "topology" 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..23375702c 100644 --- a/go/internal/state/history_sqlite.go +++ b/go/internal/state/history_sqlite.go @@ -154,13 +154,17 @@ 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 } + if err := s.refreshSourceDayCache(context.Background()); err != nil { + return err + } ok = true return nil } 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) }