diff --git a/.changeset/home-use-subtotal.md b/.changeset/home-use-subtotal.md new file mode 100644 index 000000000..7a31c378b --- /dev/null +++ b/.changeset/home-use-subtotal.md @@ -0,0 +1,11 @@ +--- +"ftw": patch +--- + +Keep an estimated Home use subtotal visible after missing source readings or a recording gap. Publish its accepted interval coverage separately from the unknown full total and measured energy. Preserve original history and exclude gaps rather than filling them. + +Add an offline historical repair tool that snapshots the original databases, prepares a hashed plan from selected retained AC power series, and applies audited estimates only to a working copy. Core uses these corrections in household daily totals and history without changing source measurements or their quality. Unsupported gaps remain unknown. + +Keep the native launcher and update API available when a local patch identifies the installed release with SemVer build metadata. Match the current slot and clean-exit marker using the base release while keeping published trial identity checks exact. + +Support audited endpoint-mean repairs for both historical Home use and Solar. Prepare while Core collects data, review the exact hash, and publish additive correction metadata atomically without replacing a live database. Show per-flow estimate labels and completed-through timestamps, preserving source observations and update support. diff --git a/docs/history-endpoint-repair.md b/docs/history-endpoint-repair.md new file mode 100644 index 000000000..649a361fd --- /dev/null +++ b/docs/history-endpoint-repair.md @@ -0,0 +1,74 @@ +# Repair Home use and Solar from bounding power + +`ftw-history-endpoints` estimates only incomplete retained minutes. It uses +`Wh = (before_W + after_W) / 2 × duration_hours`, keeping the same mean through +a contiguous gap. Home use endpoints are the closest valid retained minute +source-energy means before and after the gap. Solar endpoints are valid +retained ten-second PV means, expressed as positive generation. Two observed +night-time zeros give 0 Wh. Missing endpoints never imply zero. + +The endpoint distance is bounded explicitly (default 15 minutes, maximum 30). +This is historical estimation, not recovered measurement. Changes between +the two readings are unknown. A corrected minute replaces that flow's whole +incomplete minute in the read view; its originally recorded part is not added +a second time. Complete minutes and other flows keep their original evidence. + +Build Core and the helper from the same source. Native local Core builds may +retain valid SemVer build metadata while using the installed base-release +slot; keep the normal launcher so updates remain available. Prepare the build +and transfer it before a short stopped-Core binary swap. Do not stop Core +while copying or reviewing the database. Never start a second controlling Core. + +Take an independent, integrity-checked SQLite online backup of `history.db` +while the live Core continues collecting. Use SQLite's backup API with a read +snapshot, not a filesystem copy of an active WAL database. Keep matching state +configuration and the prior binary for rollback. Leave disk headroom for the +backup and WAL growth. Then prepare a completed minute range, up to 48 hours: + +```sh +ftw-history-endpoints prepare --state /var/lib/ftw/state.db \ + --plan ./endpoint-plan.json --since 2026-10-19T00:00:00+02:00 \ + --until 2026-10-19T23:15:00+02:00 --max-gap 20m \ + --reason 'Owner requested endpoint mean, issue 1750' +``` + +Inspect `minutes` (flow, whole-minute Wh, endpoint times and powers), `skipped`, +source generation and evidence digest. The boundary timestamps are not raw +polls: their retained resolution is part of the method. Choose the smallest +bound that covers the reviewed gaps. An unsupported edge remains unknown. +Apply the actual reviewed hash and verified independent backup: + +```sh +ftw-history-endpoints apply --state /var/lib/ftw/state.db \ + --plan ./endpoint-plan.json --sha256 ACTUAL_REVIEWED_SHA256 \ + --backup /private/backup/history-before.db +``` + +Apply is safe alongside the live collector. It briefly reserves the SQLite +writer, rechecks historical evidence, inserts one audited metadata record and +invalidates derived daily caches in one transaction. It neither replaces a +live database nor changes raw energy/power/quality rows, device configuration, +control or forecasts. New unrelated measurements survive. An identical retry +is idempotent; tampered, stale or conflicting plans fail without partial writes. + +Plans are stored in the existing `energy_ledger_meta` namespace +`endpoint-repair-v1:`, with backup hash and application time. This +metadata already participates in portable backup/restore. Original minute +energy in affected quarters is protected from rollup so replacements retain +their boundaries. No new database schema or original quality flags are needed. +Home corrections take read precedence over earlier source-power repairs; +the earlier audit remains present. + +Daily totals, system ledger buckets and aggregate charts use the corrections. +Solar has `pv_quality: estimated`; inferred Solar is excluded from the +`measured` fallback. Home use remains `load_quality: estimated`. Current totals +end at the latest committed complete minute, with independent `load_until_ms` +and `pv_until_ms`; the unfinished writer tail is not extrapolated. Aggregate +charts carry estimated/not-for-training markers. Raw samples remain unchanged. + +Verify both totals, flow quality, ledger provenance, writer progress and native +update availability after application. If a client has cached a closed-day +aggregate, reload it. To undo a plan, stop Core briefly or take a SQLite writer +transaction, remove only its exact `energy_ledger_meta` key and invalidate the +three derived day-cache tables. Keep later measurements; do not overwrite them +with an older whole-database backup. diff --git a/docs/history-repair.md b/docs/history-repair.md new file mode 100644 index 000000000..f5286a7e3 --- /dev/null +++ b/docs/history-repair.md @@ -0,0 +1,110 @@ +# Repair historical household energy + +`ftw-history-repair` reconstructs missing household energy from retained, +explicitly selected power series. It writes durable, audited corrections to +`history_load_estimates` on a **separate working copy**, not the source +databases. Core then reads the corrections in daily household totals, chart +aggregates and the household energy ledger. Original samples, other energy +flows, quality masks, forecasts and control remain unchanged. + +These are estimates, not exact recovery of lost measurements. A correction +does not make a missing PV measurement known. The API marks household energy +with `load_quality: estimated`, independently of other missing flows. It keeps +derived energy out of `measured.load_wh` and self-powered percentages. A live +day corrected this way reports `load_until_ms` at its last completed minute. + +## Prepare a copy and inspect its plan + +Build the tool from the same patched source as Core: + +```sh +cd go +go build -o ../bin/ftw-history-repair ./cmd/ftw-history-repair +``` + +For a deployment, stop Core using the installation's actual service manager +before taking the snapshot, and keep it stopped until the corrected history +is installed or the operation is abandoned. A copy prepared while Core keeps +collecting data can be used for analysis, but must not later replace the live +history: that would discard newer measurements. Keep a verified full FTW +backup and the original release for deployment rollback. This tool additionally +preserves checked copies of the affected state/history databases. + +Select a completed range, at most 48 hours, with explicit time zones and +minute boundaries. Select every electrical flow or explicitly declare it +absent. Use site-side **AC** battery power; do not combine cabinet DC energy +with AC grid/PV readings. Check the running box's metric catalogue and physical +topology before using these example driver/metric names: + +```sh +bin/ftw-history-repair prepare \ + --state /var/lib/ftw/state.db \ + --workdir ./repair-2026-10-19 \ + --since 2026-10-19T00:00:00+02:00 \ + --until 2026-10-19T14:00:00+02:00 \ + --grid pixii:meter_w --pv sma:pv_w \ + --battery pixii:battery_ac_w --ev easee-cloud:ev_w --v2x none \ + --max-gap 15s --reason 'Reconstruct missing Home use intervals, issue 1750' +``` + +The work directory must be new. `backup/` contains state/history snapshots; +`corrected/` starts as a separate copy. `plan.json` records their hashes, +selected sources, period, reason, per-minute Wh, interpolation evidence, +unsupported minutes and the exact plan SHA-256. Preparing does not apply +corrections. The originals are opened read-only; both copies are checked +through Core's database integrity/generation checks. + +The reconstruction uses retained power means at minute resolution or finer. +It integrates between their observed time centres over the same minute for +every flow, using `grid - PV - battery - EV - V2X`. It does not extrapolate +edges, split coarse hourly evidence, clip signed residuals, or replace an +existing estimate. It refuses duplicate source mappings and non-watt inputs. + +The default maximum point gap is 15 seconds. Unsupported minutes remain +unknown. If the owner explicitly accepts bounded interpolation, prepare a +**new** plan with a larger bound, at most 180 seconds, and inspect the changed +minutes and `max_interpolated_ms_per_source` values. This is an estimate of +allocation, not proof of constant power or an instrument's delay. Signed +minute residuals are retained; each displayed aggregate must still have +complete supported coverage and a nonnegative resulting balance. + +## Apply the reviewed plan to the copy + +After reviewing the actual plan and its source selections, supply its exact +hash: + +```sh +bin/ftw-history-repair apply --workdir ./repair-2026-10-19 \ + --sha256 REPLACE_WITH_THE_REVIEWED_PLAN_SHA256 +``` + +The operation verifies the backup hashes and unchanged working inputs, +inserts all correction minutes and clears derived daily caches in one SQLite +transaction. A conflict rolls the transaction back. Retrying the same applied +plan does not add energy again. `apply-result.json` reports before/after +household Wh and quality for the **selected range**, not an assumed full day. +If another gap remains unsupported, the complete total stays unknown and the +accepted subtotal remains available. + +The correction rows carry the plan hash, backup hash, timestamp, reason and +source mapping in their method metadata. They survive portable backup/restore +and Core restart. Minute energy needed to interpret a corrected quarter is +retained rather than merged away by energy rollup. Original measurement +coverage remains separate from reconstructed household coverage. + +## Validate and deploy separately + +Use the patched Core to read `corrected/state.db` and `corrected/history.db` +in an isolated test setup without equipment control. Check the selected +period's daily total, ledger provenance, estimate labels and remaining gaps. +The tool never starts a second Core and has no install or restart action. + +For the owner's stopped installation, an operator must verify the generation, +release layout, file ownership, backup and absence of newer source data before +installing the corrected history and patched Core together. The same source +state configuration must still bind that history generation. Do not replace +the owner's configuration with a stale test copy. To abandon or undo the +repair before deployment, discard the working copy; the source is unchanged. +For deployment rollback, restore the verified original history snapshot and +matching original release while Core is stopped. Never copy SQLite files +over a running Core or leave mismatched WAL/SHM files beside a restored DB. diff --git a/go/cmd/ftw-history-endpoints/main.go b/go/cmd/ftw-history-endpoints/main.go new file mode 100644 index 000000000..cba47530c --- /dev/null +++ b/go/cmd/ftw-history-endpoints/main.go @@ -0,0 +1,95 @@ +// ftw-history-endpoints repairs completed historical minutes from bounding +// power means. Preparing and applying leave equipment control to the live Core. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "github.com/srcfl/ftw/go/internal/state" + "os" + "time" +) + +func run(ctx context.Context, args []string) error { + if len(args) == 0 { + return errors.New("use prepare or apply") + } + f := flag.NewFlagSet(args[0], flag.ContinueOnError) + source := f.String("state", "", "existing state.db") + planFile := f.String("plan", "", "reviewable plan JSON") + switch args[0] { + case "prepare": + since := f.String("since", "", "RFC3339 start") + until := f.String("until", "", "RFC3339 completed end") + gap := f.Duration("max-gap", 15*time.Minute, "maximum distance between valid endpoints (1–30m)") + reason := f.String("reason", "", "audit reason") + if err := f.Parse(args[1:]); err != nil { + return err + } + if *source == "" || *planFile == "" || f.NArg() != 0 { + return errors.New("state and new plan path required") + } + from, err := time.Parse(time.RFC3339, *since) + if err != nil { + return err + } + to, err := time.Parse(time.RFC3339, *until) + if err != nil { + return err + } + p, err := state.PrepareEndpointRepair(ctx, *source, state.EndpointRepairRequest{SinceMS: from.UnixMilli(), UntilMS: to.UnixMilli(), MaxGapMS: gap.Milliseconds(), Reason: *reason}) + if err != nil { + return err + } + b, err := json.MarshalIndent(p, "", " ") + if err != nil { + return err + } + file, err := os.OpenFile(*planFile, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600) + if err != nil { + return err + } + _, err = file.Write(append(b, '\n')) + closeErr := file.Close() + if err != nil { + return err + } + if closeErr != nil { + return closeErr + } + return json.NewEncoder(os.Stdout).Encode(map[string]any{"plan_sha256": p.SHA256, "correction_minutes": len(p.Minutes), "skipped": p.Skipped}) + case "apply": + sha := f.String("sha256", "", "exact reviewed plan digest") + backup := f.String("backup", "", "independent verified history.db backup") + if err := f.Parse(args[1:]); err != nil { + return err + } + if *source == "" || *planFile == "" || *backup == "" || f.NArg() != 0 { + return errors.New("state, plan and backup required") + } + b, err := os.ReadFile(*planFile) + if err != nil { + return err + } + var p state.EndpointRepairPlan + if err = json.Unmarshal(b, &p); err != nil { + return err + } + n, err := state.ApplyEndpointRepair(ctx, *source, *backup, p, *sha) + if err != nil { + return err + } + return json.NewEncoder(os.Stdout).Encode(map[string]any{"plan_sha256": p.SHA256, "applied_minutes": n, "method": "endpoint_mean_v1"}) + default: + return errors.New("unknown operation") + } +} +func main() { + if err := run(context.Background(), os.Args[1:]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} diff --git a/go/cmd/ftw-history-repair/main.go b/go/cmd/ftw-history-repair/main.go new file mode 100644 index 000000000..a07cd20e5 --- /dev/null +++ b/go/cmd/ftw-history-repair/main.go @@ -0,0 +1,92 @@ +// ftw-history-repair prepares and applies historical household corrections on +// a separate database copy. It never writes the owner's source databases. +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "os" + "strings" + "time" + + "github.com/srcfl/ftw/go/internal/state" +) + +func run(ctx context.Context, args []string, out io.Writer) error { + if len(args) == 0 { + return errors.New("use prepare --state PATH --workdir NEW_DIR --since RFC3339 --until RFC3339 --grid DRIVER:METRIC --pv DRIVER:METRIC|none --battery DRIVER:AC_METRIC|none --ev DRIVER:METRIC|none --v2x DRIVER:METRIC|none --reason TEXT [--max-gap 15s], then apply --workdir DIR --sha256 REVIEWED_PLAN_HASH") + } + f := flag.NewFlagSet(args[0], flag.ContinueOnError) + workdir := f.String("workdir", "", "new private working directory for prepare; same directory for apply") + if args[0] == "apply" { + sha := f.String("sha256", "", "SHA-256 of the exact reviewed plan") + if err := f.Parse(args[1:]); err != nil { + return err + } + if *workdir == "" || f.NArg() != 0 { + return errors.New("workdir is required; no positional arguments") + } + result, err := state.ApplyHistoryRepair(ctx, *workdir, *sha) + if err != nil { + return err + } + return json.NewEncoder(out).Encode(result) + } + if args[0] != "prepare" { + return errors.New("unknown repair operation") + } + source := f.String("state", "", "source state.db; opened read-only") + since := f.String("since", "", "start RFC3339 with timezone, on a minute boundary") + until := f.String("until", "", "end RFC3339 with timezone, on a completed minute boundary") + reason := f.String("reason", "", "reason for this historical estimate") + gap := f.Duration("max-gap", 15*time.Second, "largest source-point gap to interpolate; up to 180s, explicitly reviewed") + flows := map[string]*string{} + for _, name := range []string{"grid", "pv", "battery", "ev", "v2x"} { + flows[name] = f.String(name, "", "explicit driver:power_metric or none; battery must be site-side AC") + } + if err := f.Parse(args[1:]); err != nil { + return err + } + if *source == "" || *workdir == "" || f.NArg() != 0 { + return errors.New("state and workdir required; no positional arguments") + } + start, err := time.Parse(time.RFC3339, *since) + if err != nil { + return err + } + end, err := time.Parse(time.RFC3339, *until) + if err != nil { + return err + } + r := state.HistoryRepairRequest{SinceMS: start.UnixMilli(), UntilMS: end.UnixMilli(), MaxGapMS: gap.Milliseconds(), Reason: *reason} + for name, value := range flows { + v := state.RepairPowerSource{Flow: name} + if *value == "none" { + v.Absent = true + } else { + parts := strings.SplitN(*value, ":", 2) + if len(parts) != 2 { + return fmt.Errorf("%s needs driver:power_metric or explicit none", name) + } + v.Driver, v.Metric = parts[0], parts[1] + } + r.Sources = append(r.Sources, v) + } + state.SortRepairSources(r.Sources) + plan, err := state.PrepareHistoryRepair(ctx, *source, *workdir, r) + if err != nil { + return err + } + return json.NewEncoder(out).Encode(plan) +} + +func main() { + if err := run(context.Background(), os.Args[1:], os.Stdout); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} diff --git a/go/cmd/ftw-history-repair/main_test.go b/go/cmd/ftw-history-repair/main_test.go new file mode 100644 index 000000000..0b0b297af --- /dev/null +++ b/go/cmd/ftw-history-repair/main_test.go @@ -0,0 +1,94 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "github.com/srcfl/ftw/go/internal/state" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestRepairCommandRequiresExplicitFlowsAndReviewedPlan(t *testing.T) { + for _, args := range [][]string{nil, {"write"}, {"apply", "--workdir", "/not-a-repair"}, {"prepare", "--state", "state.db", "--workdir", "new", "--since", "2026-10-19T00:00:00+02:00", "--until", "2026-10-19T14:00:00+02:00", "--grid", "meter:meter_w"}} { + var out bytes.Buffer + if err := run(context.Background(), args, &out); err == nil { + t.Fatal("incomplete repair command succeeded", args) + } + } + var out bytes.Buffer + err := run(context.Background(), []string{"prepare", "--state", "state.db", "--workdir", "new", "--since", "2026-10-19T00:00:00", "--until", "2026-10-19T14:00:00+02:00"}, &out) + if err == nil || !strings.Contains(err.Error(), "cannot parse") { + t.Fatal("timezone-less date accepted", err) + } +} + +func TestRepairCommandPrepareThenApplyChangesHistoricalCopy(t *testing.T) { + ctx := context.Background() + dir := t.TempDir() + path := filepath.Join(dir, "state.db") + s, err := state.Open(path) + if err != nil { + t.Fatal(err) + } + if err := s.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + base := time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour) + for i := int64(0); i <= 90; i++ { + at := base.Add(time.Duration(i) * 10 * time.Second).UnixMilli() + v := &state.HistorySources{AtMS: at, Topology: "meter", GridW: 1000} + if i == 42 { + v.Missing = state.MissingPV + } + p := &state.HistoryPoint{TsMs: at, GridW: 1000, LoadW: 1000, JSON: `{}`, Sources: v} + if err := s.EnqueueTelemetryTick(p, nil, nil); err != nil { + t.Fatal(err) + } + if i%10 == 0 { + if err := s.FlushHistory(ctx); err != nil { + t.Fatal(err) + } + } + } + if err := s.FlushHistory(ctx); err != nil { + t.Fatal(err) + } + var samples []state.Sample + for i := int64(-6); i <= 186; i++ { + samples = append(samples, state.Sample{Driver: "meter", Metric: "meter_w", TsMs: base.Add(time.Duration(i) * 5 * time.Second).UnixMilli(), Value: 1000, Unit: "W"}) + } + if err := s.RecordSamples(samples); err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + work := filepath.Join(dir, "repair") + var out bytes.Buffer + args := []string{"prepare", "--state", path, "--workdir", work, "--since", base.Format(time.RFC3339), "--until", base.Add(15 * time.Minute).Format(time.RFC3339), "--grid", "meter:meter_w", "--pv", "none", "--battery", "none", "--ev", "none", "--v2x", "none", "--reason", "CLI repair regression"} + if err := run(ctx, args, &out); err != nil { + t.Fatal(err) + } + var plan state.HistoryRepairPlan + if err := json.Unmarshal(out.Bytes(), &plan); err != nil { + t.Fatal(err) + } + out.Reset() + if err := run(ctx, []string{"apply", "--workdir", work, "--sha256", plan.SHA256}, &out); err != nil { + t.Fatal(err) + } + var result state.HistoryRepairResult + if err := json.Unmarshal(out.Bytes(), &result); err != nil { + t.Fatal(err) + } + if result.BeforeWh != nil || result.AfterWh == nil || !result.Estimated { + t.Fatal("CLI did not repair history", result) + } + if wh := result.AfterWh.(float64); wh < 249.99 || wh > 250.01 { + t.Fatal("wrong corrected total", wh) + } + t.Logf("REPAIR_RESULT_JSON=%s", strings.TrimSpace(out.String())) +} diff --git a/go/cmd/ftw/native_update.go b/go/cmd/ftw/native_update.go index 211e74bd5..333e310f0 100644 --- a/go/cmd/ftw/native_update.go +++ b/go/cmd/ftw/native_update.go @@ -6,6 +6,7 @@ import ( "log/slog" "os" "path/filepath" + "strings" "time" "github.com/srcfl/ftw/go/internal/nativeupdate" @@ -14,6 +15,26 @@ import ( const nativeTrialTimeout = 6 * time.Hour +// Build metadata identifies local patches without changing the installed +// release's SemVer identity. Published trial binaries still require an exact tag. +func nativeSlotVersion(version string) string { + base, metadata, found := strings.Cut(version, "+") + if !found || !nativeupdate.ValidTag(base) { + return version + } + for _, id := range strings.Split(metadata, ".") { + if id == "" { + return version + } + for _, c := range id { + if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '-') { + return version + } + } + } + return base +} + type nativeTrial struct { manager nativeupdate.Manager tag string @@ -36,7 +57,7 @@ func beginNativeTrial(root, tag, version string) (*nativeTrial, error) { return nil, err } if tag == "" { - if state.Current != version { + if state.Current != nativeSlotVersion(version) { return nil, fmt.Errorf("native current slot %s does not match binary %s", state.Current, version) } return nil, nil @@ -101,7 +122,7 @@ func reconcileNativeFallback(root, version string, checker *selfupdate.Checker) return } state, err := (nativeupdate.Manager{Root: root}).Read() - if err != nil || state.Current != version || state.LastFailed == "" || status.Target != state.LastFailed { + if err != nil || state.Current != nativeSlotVersion(version) || state.LastFailed == "" || status.Target != state.LastFailed { return } switch status.State { @@ -128,7 +149,7 @@ func markNativeCleanExit(root string) { if root == "" { return } - if err := (nativeupdate.Manager{Root: root}).MarkCleanExit(Version); err != nil { + if err := (nativeupdate.Manager{Root: root}).MarkCleanExit(nativeSlotVersion(Version)); err != nil { slog.Warn("native clean-exit mark could not be saved", "err", err) } } diff --git a/go/cmd/ftw/native_update_test.go b/go/cmd/ftw/native_update_test.go index 05e921b21..f121a5af6 100644 --- a/go/cmd/ftw/native_update_test.go +++ b/go/cmd/ftw/native_update_test.go @@ -80,6 +80,37 @@ func TestNativeTrialCommitsOnlyAfterReadiness(t *testing.T) { } } +func TestNativeCurrentBuildMetadataKeepsReleaseUpdates(t *testing.T) { + root := t.TempDir() + current := "v0.154.0-beta.1" + nativeFixtureRelease(t, root, current) + if err := (nativeupdate.Manager{Root: root}).Init(current); err != nil { + t.Fatal(err) + } + for _, version := range []string{current, current + "+history-repair.6f71db1"} { + trial, err := beginNativeTrial(root, "", version) + if err != nil || trial != nil { + t.Fatalf("installed release build %q rejected: %v", version, err) + } + } + previousVersion := Version + Version = current + "+history-repair.6f71db1" + t.Cleanup(func() { Version = previousVersion }) + markNativeCleanExit(root) + mark, err := os.ReadFile(filepath.Join(root, ".clean-exit")) + if err != nil || strings.TrimSpace(string(mark)) != current { + t.Fatalf("clean exit did not identify installed release: %q %v", mark, err) + } + for _, version := range []string{"v0.153.0-beta.1+history-repair", current + "+", current + "+bad..metadata", current + "+bad/metadata"} { + if _, err := beginNativeTrial(root, "", version); err == nil { + t.Fatalf("wrong release or malformed metadata accepted: %q", version) + } + } + if _, err := beginNativeTrial(root, current, current+"+history-repair"); err == nil { + t.Fatal("published trial identity must remain exact") + } +} + func TestNativeFallbackReportsFailedTrial(t *testing.T) { root := t.TempDir() current, next := "v0.131.0-beta.1", "v0.131.0-beta.2" diff --git a/go/internal/api/api.go b/go/internal/api/api.go index d9e2a9b4b..5537628a0 100644 --- a/go/internal/api/api.go +++ b/go/internal/api/api.go @@ -1205,6 +1205,9 @@ func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { // instead of "no data yet". Mirror the old `len(pts) > 1` guard. if d.Intervals > 0 { energyToday = energyTodayFields(d) + if err := s.addHouseholdSubtotal(r.Context(), midnight.UnixMilli(), now.UnixMilli(), energyToday); err != nil { + slog.Warn("household energy subtotal unavailable", "err", err) + } } } else { slog.Warn("failed to integrate today's energy", "err", err) @@ -2603,7 +2606,7 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { http.Error(w, "history coverage unavailable", http.StatusInternalServerError) return } - out = append(out, map[string]any{ + dayEnergy := map[string]any{ "coverage_ms": coverage, "period_ms": periodEnd - dayStart.UnixMilli(), "history_quality": de.Missing.Quality(), @@ -2614,7 +2617,27 @@ func (s *Server) handleEnergyDaily(w http.ResponseWriter, r *http.Request) { "bat_charged_wh": de.Missing.Value(state.MissingBattery, de.BatChargedWh), "bat_discharged_wh": de.Missing.Value(state.MissingBattery, de.BatDischargedWh), "load_wh": de.Missing.Value(state.MissingLoad, de.LoadWh), - }) + } + if de.Missing&state.EstimatedLoad != 0 { + dayEnergy["load_quality"] = "estimated" + } + if de.Missing&state.EndpointEstimatedPV != 0 { + dayEnergy["pv_quality"] = "estimated" + if dayEnergy["history_quality"] == "complete" { + dayEnergy["history_quality"] = "estimated" + } + } + if de.PVUntilMS > 0 { + dayEnergy["pv_until_ms"] = de.PVUntilMS + } + if de.LoadUntilMS > 0 { + dayEnergy["load_until_ms"] = de.LoadUntilMS + } + if err := s.addHouseholdSubtotal(r.Context(), dayStart.UnixMilli(), periodEnd, dayEnergy); err != nil { + http.Error(w, "household energy subtotal unavailable", http.StatusInternalServerError) + return + } + out = append(out, dayEnergy) } writeJSON(w, 200, map[string]any{"days": out, "tz": localzone.Name()}) } diff --git a/go/internal/api/api_energy_today.go b/go/internal/api/api_energy_today.go index de8e5cc58..0149b5a66 100644 --- a/go/internal/api/api_energy_today.go +++ b/go/internal/api/api_energy_today.go @@ -7,6 +7,21 @@ func energyTodayFields(d state.DayEnergy) map[string]any { return nil } out := map[string]any{"history_quality": d.Missing.Quality()} + if d.Missing&state.EstimatedLoad != 0 { + out["load_quality"] = "estimated" + } + if d.LoadUntilMS > 0 { + out["load_until_ms"] = d.LoadUntilMS + } + if d.Missing&state.EndpointEstimatedPV != 0 { + out["pv_quality"] = "estimated" + if out["history_quality"] == "complete" { + out["history_quality"] = "estimated" + } + } + if d.PVUntilMS > 0 { + out["pv_until_ms"] = d.PVUntilMS + } measured := make(map[string]any) for key, field := range map[string]struct { flag state.HistoryMissing @@ -28,6 +43,9 @@ func energyTodayFields(d state.DayEnergy) map[string]any { // in the measured-only fallback used by clients after a data gap. measured["load_wh"] = nil } + if d.Missing&state.EndpointEstimatedPV != 0 { + measured["pv_wh"] = nil + } out["measured"] = measured return out } diff --git a/go/internal/api/api_history_repair_test.go b/go/internal/api/api_history_repair_test.go new file mode 100644 index 000000000..f76cd5467 --- /dev/null +++ b/go/internal/api/api_history_repair_test.go @@ -0,0 +1,29 @@ +package api + +import ( + "github.com/srcfl/ftw/go/internal/state" + "testing" +) + +func TestHistoricalRepairEnergyFieldsPreserveOtherUnknownFlows(t *testing.T) { + got := energyTodayFields(state.DayEnergy{Missing: state.MissingPV | state.EstimatedLoad, Intervals: 10, LoadWh: 500, LoadUntilMS: 1800000}) + if got["load_wh"] != float64(500) || got["load_quality"] != "estimated" || got["load_until_ms"] != int64(1800000) { + t.Fatal(got) + } + if got["pv_wh"] != nil || got["history_quality"] != "incomplete" { + t.Fatal("repair relabelled missing PV as known", got) + } + if got["measured"].(map[string]any)["load_wh"] != nil { + t.Fatal("repair became a measurement", got) + } +} + +func TestEndpointSolarEnergyFieldsAreEstimatedNotMeasured(t *testing.T) { + got := energyTodayFields(state.DayEnergy{Missing: state.EndpointEstimatedPV, Intervals: 10, PVWh: 1234, PVUntilMS: 1800000}) + if got["pv_wh"] != float64(1234) || got["pv_quality"] != "estimated" || got["pv_until_ms"] != int64(1800000) || got["history_quality"] != "estimated" { + t.Fatal(got) + } + if got["measured"].(map[string]any)["pv_wh"] != nil { + t.Fatal("estimated solar labelled measured", got) + } +} diff --git a/go/internal/api/api_household_subtotal.go b/go/internal/api/api_household_subtotal.go new file mode 100644 index 000000000..2681a66a5 --- /dev/null +++ b/go/internal/api/api_household_subtotal.go @@ -0,0 +1,16 @@ +package api + +import "context" + +// This separate field must never populate measured.load_wh, the complete +// load_wh total, forecast training or a self-powered percentage. +func (s *Server) addHouseholdSubtotal(ctx context.Context, from, to int64, fields map[string]any) error { + if fields == nil || fields["load_wh"] != nil { + return nil + } + subtotal, err := s.deps.State.HouseholdEnergySubtotal(ctx, from, to) + if err == nil && subtotal != nil { + fields["load_subtotal"] = subtotal + } + return err +} diff --git a/go/internal/api/api_household_subtotal_test.go b/go/internal/api/api_household_subtotal_test.go new file mode 100644 index 000000000..3ad90c5e2 --- /dev/null +++ b/go/internal/api/api_household_subtotal_test.go @@ -0,0 +1,103 @@ +package api + +import ( + "context" + "encoding/json" + "net/http/httptest" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/srcfl/ftw/go/internal/config" + "github.com/srcfl/ftw/go/internal/control" + "github.com/srcfl/ftw/go/internal/state" + "github.com/srcfl/ftw/go/internal/telemetry" +) + +func TestHouseholdSubtotalStatusAndDailyKeepIncompleteTotalSeparate(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) + } + now := time.Now() + midnight := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) + if now.Sub(midnight) < 16*time.Minute { + t.Skip("need fifteen completed minutes today") + } + for _, day := range []time.Time{midnight.AddDate(0, 0, -1), midnight} { + for i := int64(0); i <= 90; i++ { + at := day.Add(time.Duration(i) * 10 * time.Second).UnixMilli() + v := &state.HistorySources{AtMS: at, Topology: "site", GridW: 1000} + if i == 42 { + v.Missing = state.MissingPV + } + p := &state.HistoryPoint{TsMs: at, GridW: 1000, LoadW: 1000, JSON: `{}`, Sources: v} + if err := st.EnqueueTelemetryTick(p, nil, nil); err != nil { + t.Fatal(err) + } + if i%10 == 0 { + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + } + } + } + if err := st.FlushHistory(context.Background()); err != nil { + t.Fatal(err) + } + srv := New(&Deps{State: st, Tel: telemetry.NewStore(), Ctrl: &control.State{}, CtrlMu: &sync.Mutex{}, + CapMu: &sync.RWMutex{}, CfgMu: &sync.RWMutex{}, Cfg: &config.Config{}}) + check := func(fields map[string]any) { + t.Helper() + if fields["load_wh"] != nil || fields["history_quality"] != "incomplete" { + t.Fatal("partial data claimed as a complete day", fields) + } + sub, ok := fields["load_subtotal"].(map[string]any) + if !ok || sub["quality"] != "estimated" || sub["coverage_ms"] != float64(600000) { + t.Fatal("accepted subtotal or its coverage missing", fields) + } + if wh, ok := sub["wh"].(float64); !ok || wh < 166.66 || wh > 166.67 { + t.Fatal(sub) + } + } + rr := httptest.NewRecorder() + srv.handleStatus(rr, httptest.NewRequest("GET", "/api/status", nil)) + var status struct { + Energy struct { + Today map[string]any `json:"today"` + } `json:"energy"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &status); err != nil { + t.Fatal(err) + } + check(status.Energy.Today) + if status.Energy.Today["measured"].(map[string]any)["load_wh"] != nil { + t.Fatal("estimated subtotal leaked into measured energy") + } + // Repeat the read: yesterday takes the persistent/in-memory cache path, + // while its independently queried subtotal must retain the same coverage. + for i := 0; i < 2; i++ { + rr = httptest.NewRecorder() + srv.handleEnergyDaily(rr, httptest.NewRequest("GET", "/api/energy/daily?days=2", nil)) + if rr.Code != 200 { + t.Fatal(rr.Code, rr.Body.String()) + } + var daily struct { + Days []map[string]any `json:"days"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &daily); err != nil { + t.Fatal(err) + } + if len(daily.Days) != 2 { + t.Fatal(daily) + } + for _, fields := range daily.Days { + check(fields) + } + } +} diff --git a/go/internal/state/energy_ledger.go b/go/internal/state/energy_ledger.go index 4b4f9c5cf..490f1203b 100644 --- a/go/internal/state/energy_ledger.go +++ b/go/internal/state/energy_ledger.go @@ -501,6 +501,19 @@ func (s *Store) LoadEnergyHistoryContext(ctx context.Context, q EnergyHistoryQue if err != nil { return nil, false, err } + out, err = s.applyHistoricalLoadLedger(ctx, q, out) + if err != nil { + return nil, false, err + } + rank := 0 + for i, p := range out { + if i == 0 || p.BucketStartMS != out[i-1].BucketStartMS { + rank++ + } + if rank > q.Limit { + return out[:i], true, nil + } + } return out, truncated || sourceTruncated, nil } diff --git a/go/internal/state/history_dashboard_retention.go b/go/internal/state/history_dashboard_retention.go index 2cf1f7f59..07de225e5 100644 --- a/go/internal/state/history_dashboard_retention.go +++ b/go/internal/state/history_dashboard_retention.go @@ -194,7 +194,12 @@ func (s *Store) rollSiteEnergyStorage(ctx context.Context, cutoff int64, table, cutoff = bucketStart(cutoff, width) for { var first sql.NullInt64 - if err := s.history.QueryRowContext(ctx, `SELECT MIN(start_ms) FROM `+table+` WHERE resolution_ms=? AND start_ms=` + table + `.start_ms-(` + table + `.start_ms%900000) AND r.start_ms<` + table + `.start_ms-(` + table + `.start_ms%900000)+900000)` + protected += ` AND NOT EXISTS (SELECT 1 FROM energy_ledger_meta a, json_each(CASE WHEN a.key GLOB 'endpoint-repair-v1:*' THEN a.value ELSE '{}' END,'$.plan.minutes') m WHERE a.key GLOB 'endpoint-repair-v1:*' AND json_extract(m.value,'$.start_ms')>=` + table + `.start_ms-(` + table + `.start_ms%900000) AND json_extract(m.value,'$.start_ms')<` + table + `.start_ms-(` + table + `.start_ms%900000)+900000)` + if err := s.history.QueryRowContext(ctx, `SELECT MIN(start_ms) FROM `+table+` WHERE resolution_ms=? AND start_ms=? AND start_ms 0 { + end = min(end, bucketStart(last, 60000)) + } + return end, nil +} +func (s *Store) endpointPVBuckets(ctx context.Context, from, to, width int64) ([]sourceLoadBucket, map[int64]EndpointRepairMinute, error) { + all, err := s.endpointRepairs(ctx, from, to) + if err != nil { + return nil, nil, err + } + repairs := all["pv"] + if len(repairs) == 0 { + return nil, repairs, nil + } + by := map[int64]*sourceLoadBucket{} + add := func(r siteEnergyRow) error { + w := max(width, r.width) + start := from + (r.start-from)/w*w + b := by[start] + if b == nil { + b = &sourceLoadBucket{start: start, width: w, first: r.first, last: r.last} + by[start] = b + } + if b.width != w || r.first < start || r.last > start+w { + b.missing |= MissingPV + } + b.first = min(b.first, r.first) + b.last = max(b.last, r.last) + b.covered += r.covered + b.n += r.n + b.wh += r.pv + b.missing |= r.missing & MissingPV + return nil + } + err = s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { + if _, ok := repairs[r.start]; ok && r.width == 60000 { + return nil + } + return add(r) + }) + if err != nil { + return nil, nil, err + } + for at, p := range repairs { + _ = add(siteEnergyRow{start: at, width: 60000, first: at, last: at + 60000, covered: 60000, pv: p.Wh}) + } + var out []sourceLoadBucket + for _, b := range by { + if b.covered != b.width || b.first != b.start || b.last != b.start+b.width || b.wh < 0 || !finiteHistoryValue(b.wh) { + b.missing |= MissingPV + } + out = append(out, *b) + } + sort.Slice(out, func(i, j int) bool { return out[i].start < out[j].start }) + return out, repairs, nil +} +func (s *Store) applyEndpointDailyPV(ctx context.Context, from, to int64, d *DayEnergy) error { + end, err := s.endpointReadUntil(ctx, from, to) + if err != nil || end <= from { + return err + } + b, repairs, err := s.endpointPVBuckets(ctx, from, end, end-from) + if err != nil || len(repairs) == 0 { + return err + } + if len(b) == 1 && b[0].missing&MissingPV == 0 && b[0].covered == end-from { + d.PVWh = b[0].wh + d.Missing &^= MissingPV + d.Missing |= EndpointEstimatedPV + d.PVUntilMS = end + } + return nil +} +func (s *Store) applyHistoricalLoadChart(ctx context.Context, points []HistoryPoint, from, to, width int64) ([]HistoryPoint, error) { + out, err := s.applyHistoricalLoadChartOnly(ctx, points, from, to, width) + if err != nil || width < 60000 { + return out, err + } + all, err := s.endpointRepairs(ctx, from, to) + if err != nil { + return nil, err + } + changedLoad := map[int64]bool{} + for at := range all["load"] { + changedLoad[from+(at-from)/width*width] = true + } + for i := range out { + p := &out[i] + at := from + (p.TsMs-from)/width*width + if !changedLoad[at] || p.Missing&MissingLoad != 0 { + continue + } + var detail map[string]any + _ = json.Unmarshal([]byte(p.JSON), &detail) + if detail == nil { + detail = map[string]any{} + } + detail["history_load_method"] = "endpoint_mean_v1" + detail["load_quality"] = "estimated" + detail["forecast_measurement_quality"] = "aggregate_not_for_training" + raw, _ := json.Marshal(detail) + p.JSON = string(raw) + } + buckets, repairs, err := s.endpointPVBuckets(ctx, from, to, width) + if err != nil || len(repairs) == 0 { + return out, err + } + changed := map[int64]bool{} + for at := range repairs { + changed[from+(at-from)/width*width] = true + } + by := map[int64]sourceLoadBucket{} + for _, b := range buckets { + by[b.start] = b + } + seen := map[int64]bool{} + for i := range out { + p := &out[i] + at := from + (p.TsMs-from)/width*width + b, ok := by[at] + if !ok || !changed[at] || b.missing&MissingPV != 0 { + continue + } + p.PVW = -b.wh * 3600000 / float64(b.width) + p.Missing &^= MissingPV + p.Missing |= EstimatedLoad + seen[at] = true + var detail map[string]any + _ = json.Unmarshal([]byte(p.JSON), &detail) + if detail == nil { + detail = map[string]any{} + } + detail["pv_quality"] = "estimated" + detail["history_pv_method"] = "endpoint_mean_v1" + detail["forecast_measurement_quality"] = "aggregate_not_for_training" + raw, _ := json.Marshal(detail) + p.JSON = string(raw) + } + for _, b := range buckets { + if !changed[b.start] || seen[b.start] || b.missing&MissingPV != 0 { + continue + } + out = append(out, HistoryPoint{TsMs: b.start + b.width - 1, FirstMS: b.start, ResolutionMS: b.width, PVW: -b.wh * 3600000 / float64(b.width), Missing: MissingGrid | MissingBattery | MissingEV | MissingV2X | MissingSoC | MissingLoad | EstimatedLoad, JSON: `{"pv_quality":"estimated","history_pv_method":"endpoint_mean_v1","forecast_measurement_quality":"aggregate_not_for_training"}`}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].TsMs < out[j].TsMs }) + return out, nil +} +func (s *Store) applyHistoricalLoadLedger(ctx context.Context, q EnergyHistoryQuery, points []EnergyLedgerPoint) ([]EnergyLedgerPoint, error) { + out, err := s.applyHistoricalLoadLedgerOnly(ctx, q, points) + if err != nil { + return nil, err + } + all, err := s.endpointRepairs(ctx, q.SinceMS, q.UntilMS) + if err != nil { + return nil, err + } + changedLoad := map[int64]bool{} + for at := range all["load"] { + changedLoad[q.SinceMS+(at-q.SinceMS)/q.BucketMS*q.BucketMS] = true + } + for i := range out { + if out[i].Flow == FlowConsumerUse && changedLoad[out[i].BucketStartMS] { + out[i].Provenance = "endpoint_mean_v1" + } + } + if q.AssetID != "" { + return out, nil + } + buckets, repairs, err := s.endpointPVBuckets(ctx, q.SinceMS, q.UntilMS, q.BucketMS) + if err != nil || len(repairs) == 0 { + return out, err + } + changed := map[int64]bool{} + for at := range repairs { + changed[q.SinceMS+(at-q.SinceMS)/q.BucketMS*q.BucketMS] = true + } + filtered := out[:0] + for _, p := range out { + if p.Flow != FlowPVGeneration || !changed[p.BucketStartMS] { + filtered = append(filtered, p) + } + } + out = filtered + for _, b := range buckets { + if changed[b.start] && b.missing&MissingPV == 0 { + out = append(out, EnergyLedgerPoint{SchemaVersion: EnergyLedgerSchemaVersion, AssetID: "system", Flow: FlowPVGeneration, BucketStartMS: b.start, BucketLenMS: b.width, EnergyWh: b.wh, Source: "historical_repair", Quality: "estimated", Provenance: "endpoint_mean_v1", SampleCount: b.n}) + } + } + sort.SliceStable(out, func(i, j int) bool { return out[i].BucketStartMS < out[j].BucketStartMS }) + return out, nil +} diff --git a/go/internal/state/history_endpoint_repair.go b/go/internal/state/history_endpoint_repair.go new file mode 100644 index 000000000..cba154a52 --- /dev/null +++ b/go/internal/state/history_endpoint_repair.go @@ -0,0 +1,376 @@ +package state + +import ( + "context" + "crypto/sha256" + "database/sql" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "time" +) + +const endpointRepairPrefix = "endpoint-repair-v1:" + +type EndpointRepairRequest struct { + SinceMS int64 `json:"since_ms"` + UntilMS int64 `json:"until_ms"` + MaxGapMS int64 `json:"max_gap_ms"` + Reason string `json:"reason"` +} +type EndpointRepairMinute struct { + Flow string `json:"flow"` + StartMS int64 `json:"start_ms"` + Wh float64 `json:"wh"` + BeforeMS int64 `json:"before_ms"` + AfterMS int64 `json:"after_ms"` + BeforeW float64 `json:"before_w"` + AfterW float64 `json:"after_w"` +} +type EndpointRepairPlan struct { + SHA256 string `json:"sha256"` + EvidenceSHA256 string `json:"evidence_sha256"` + Generation string `json:"generation"` + Request EndpointRepairRequest `json:"request"` + Minutes []EndpointRepairMinute `json:"minutes"` + Skipped map[string]string `json:"skipped"` +} +type endpointApplied struct { + Plan EndpointRepairPlan `json:"plan"` + BackupSHA256 string `json:"backup_sha256"` + AppliedAt string `json:"applied_at"` +} + +func endpointDigest(v any) string { + b, _ := json.Marshal(v) + h := sha256.Sum256(b) + return hex.EncodeToString(h[:]) +} +func (p EndpointRepairPlan) digest() string { p.SHA256 = ""; return endpointDigest(p) } +func (r EndpointRepairRequest) validate() error { + if r.SinceMS%60000 != 0 || r.UntilMS%60000 != 0 || r.UntilMS <= r.SinceMS || r.UntilMS-r.SinceMS > 48*3600000 || r.UntilMS > time.Now().Add(-time.Minute).UnixMilli() || r.MaxGapMS < 60000 || r.MaxGapMS > 30*60000 || strings.TrimSpace(r.Reason) == "" { + return errors.New("choose a completed minute range up to 48h, max-gap 1–30m and a reason") + } + return nil +} + +type endpointOriginal struct { + Start int64 + Covered int64 + First int64 + Last int64 + Load float64 + PV float64 + Missing HistoryMissing +} + +func (s *Store) endpointOriginals(ctx context.Context, from, to int64) ([]endpointOriginal, error) { + by := map[int64]*endpointOriginal{} + err := s.walkSiteEnergy(ctx, from, to, func(r siteEnergyRow) error { + if r.width != 60000 { + return errors.New("endpoint repair requires retained minute energy") + } + b := by[r.start] + if b == nil { + b = &endpointOriginal{Start: r.start, First: r.first} + by[r.start] = b + } + b.Covered += r.covered + b.First = min(b.First, r.first) + b.Last = max(b.Last, r.last) + b.Load += r.load + b.PV += r.pv + b.Missing |= r.missing + return nil + }) + if err != nil { + return nil, err + } + out := make([]endpointOriginal, 0, (to-from)/60000) + for at := from; at < to; at += 60000 { + if b := by[at]; b != nil { + out = append(out, *b) + } else { + out = append(out, endpointOriginal{Start: at, Missing: MissingPower}) + } + } + return out, nil +} +func endpointOriginalValid(b endpointOriginal, flow string) bool { + if b.Covered != 60000 || b.First != b.Start || b.Last != b.Start+60000 { + return false + } + if flow == "pv" { + return b.Missing&MissingPV == 0 && finiteHistoryValue(b.PV) && b.PV >= 0 + } + return b.Missing&(MissingPower|BalanceMismatch) == 0 && finiteHistoryValue(b.Load) && b.Load >= 0 +} + +type endpointPower struct { + At int64 + W float64 +} + +func (s *Store) endpointPowers(ctx context.Context, from, to int64, flow string) ([]endpointPower, error) { + if flow == "load" { + originals, err := s.endpointOriginals(ctx, bucketStart(from, 60000), bucketStart(to, 60000)+60000) + if err != nil { + return nil, err + } + var out []endpointPower + for _, b := range originals { + if endpointOriginalValid(b, "load") { + out = append(out, endpointPower{At: b.Start + 30000, W: b.Load * 60}) + } + } + return out, nil + } + column, mask := "load_w", MissingPower|BalanceMismatch + if flow == "pv" { + column = "-pv_w" + mask = MissingPV + } + // Retained ten-second means are explicit endpoints, not invented raw polls. + rows, err := s.history.QueryContext(ctx, `SELECT e.last_ms,e.`+strings.TrimPrefix(column, "-")+`,COALESCE(q.missing,0) FROM ( + SELECT *, 'dashboard' kind FROM history_dashboard UNION ALL SELECT *, 'partial_dashboard' kind FROM history_dashboard_partial + ) e LEFT JOIN history_quality q ON q.kind=e.kind AND q.start_ms=e.start_ms AND q.resolution_ms=e.resolution_ms + WHERE e.resolution_ms=10000 AND e.last_ms>=? AND e.first_ms<=? ORDER BY e.last_ms`, from, to) + if err != nil { + return nil, err + } + defer rows.Close() + var out []endpointPower + for rows.Next() { + var p endpointPower + var missing HistoryMissing + if err := rows.Scan(&p.At, &p.W, &missing); err != nil { + return nil, err + } + if flow == "pv" { + p.W = -p.W + } + if missing&mask == 0 && finiteHistoryValue(p.W) && p.W >= 0 { + out = append(out, p) + } + } + return out, rows.Err() +} +func (s *Store) planEndpointRepair(ctx context.Context, r EndpointRepairRequest) (EndpointRepairPlan, error) { + p := EndpointRepairPlan{Request: r, Skipped: map[string]string{}} + if err := r.validate(); err != nil { + return p, err + } + if err := s.history.QueryRowContext(ctx, `SELECT name FROM history_migrations WHERE name LIKE 'generation:%'`).Scan(&p.Generation); err != nil { + return p, err + } + original, err := s.endpointOriginals(ctx, r.SinceMS, r.UntilMS) + if err != nil { + return p, err + } + p.EvidenceSHA256 = endpointDigest(original) + for _, flow := range []string{"load", "pv"} { + powers, err := s.endpointPowers(ctx, r.SinceMS-r.MaxGapMS, r.UntilMS+r.MaxGapMS, flow) + if err != nil { + return p, err + } + for i := 0; i < len(original); { + if endpointOriginalValid(original[i], flow) { + i++ + continue + } + start := i + for i < len(original) && !endpointOriginalValid(original[i], flow) { + i++ + } + from, to := original[start].Start, original[i-1].Start+60000 + before := sort.Search(len(powers), func(j int) bool { return powers[j].At > from }) - 1 + after := sort.Search(len(powers), func(j int) bool { return powers[j].At >= to }) + key := fmt.Sprintf("%s:%d:%d", flow, from, to) + if before < 0 || after == len(powers) { + p.Skipped[key] = "missing valid bounding power" + continue + } + a, b := powers[before], powers[after] + if b.At <= a.At || b.At-a.At > r.MaxGapMS { + p.Skipped[key] = "bounding observations exceed max-gap" + continue + } + wh := (a.W + b.W) / 2 / 60 + if !finiteHistoryValue(wh) || wh < 0 || !plausibleEnergy(wh, 60000) { + p.Skipped[key] = "implausible endpoint estimate" + continue + } + for at := from; at < to; at += 60000 { + p.Minutes = append(p.Minutes, EndpointRepairMinute{flow, at, wh, a.At, b.At, a.W, b.W}) + } + } + } + p.SHA256 = p.digest() + return p, nil +} + +// PrepareEndpointRepair reads the live box without migrations or a long copy. +// Only completed evidence is considered. It never starts equipment control. +func PrepareEndpointRepair(ctx context.Context, statePath string, r EndpointRepairRequest) (EndpointRepairPlan, error) { + s, err := OpenBackupSource(statePath) + if err != nil { + return EndpointRepairPlan{}, err + } + defer s.Close() + if s.history == nil { + return EndpointRepairPlan{}, errors.New("primary SQLite history required") + } + return s.planEndpointRepair(ctx, r) +} +func (s *Store) endpointRepairs(ctx context.Context, from, to int64) (map[string]map[int64]EndpointRepairMinute, error) { + out := map[string]map[int64]EndpointRepairMinute{"load": {}, "pv": {}} + if s.history == nil { + return out, nil + } + rows, err := s.history.QueryContext(ctx, `SELECT value FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*' ORDER BY key`) + if err != nil { + return nil, err + } + defer rows.Close() + for rows.Next() { + var raw string + var a endpointApplied + if err := rows.Scan(&raw); err != nil { + return nil, err + } + if err := json.Unmarshal([]byte(raw), &a); err != nil { + return nil, err + } + if a.Plan.SHA256 != a.Plan.digest() || len(a.BackupSHA256) != 64 { + return nil, errors.New("invalid endpoint repair audit") + } + for _, m := range a.Plan.Minutes { + if m.StartMS < from || m.StartMS+60000 > to { + continue + } + if out[m.Flow] == nil || m.StartMS%60000 != 0 || !finiteHistoryValue(m.Wh) || m.Wh < 0 { + return nil, errors.New("invalid endpoint correction") + } + if old, ok := out[m.Flow][m.StartMS]; ok && old != m { + return nil, errors.New("conflicting endpoint corrections") + } + out[m.Flow][m.StartMS] = m + } + } + return out, rows.Err() +} + +// ApplyEndpointRepair writes only audited metadata and derived caches, in one +// short SQLite transaction. It never replaces a live database with a snapshot. +// The independently verified backup and exact plan hash are mandatory. +func ApplyEndpointRepair(ctx context.Context, statePath, backupPath string, p EndpointRepairPlan, sha string) (int, error) { + if len(sha) != 64 || p.SHA256 != sha || p.digest() != sha { + return 0, errors.New("exact reviewed plan hash required") + } + if err := p.Request.validate(); err != nil { + return 0, err + } + if len(p.Minutes) == 0 { + return 0, errors.New("no supported corrections") + } + source := historyDatabasePath(statePath) + src, err := filepath.EvalSymlinks(source) + if err != nil { + return 0, err + } + bak, err := filepath.EvalSymlinks(backupPath) + if err != nil { + return 0, err + } + srcInfo, err := os.Stat(src) + if err != nil { + return 0, err + } + bakInfo, err := os.Stat(bak) + if err != nil { + return 0, err + } + if os.SameFile(srcInfo, bakInfo) { + return 0, errors.New("backup must be independent") + } + backupSHA, err := repairFileHash(bak) + if err != nil { + return 0, err + } + backup, err := sql.Open("sqlite", ReadOnlyDatabaseURI(bak)) + if err != nil { + return 0, err + } + var generation string + err = backup.QueryRowContext(ctx, `SELECT name FROM history_migrations WHERE name=?`, p.Generation).Scan(&generation) + backup.Close() + if err != nil { + return 0, errors.New("backup generation differs") + } + s, err := OpenBackupSource(statePath) + if err != nil { + return 0, err + } + defer s.Close() + db, err := sql.Open("sqlite", source+"?_pragma=busy_timeout(5000)&_pragma=synchronous(FULL)") + if err != nil { + return 0, err + } + defer db.Close() + conn, err := db.Conn(ctx) + if err != nil { + return 0, err + } + defer conn.Close() + if _, err = conn.ExecContext(ctx, "BEGIN IMMEDIATE"); err != nil { + return 0, err + } + defer conn.ExecContext(context.Background(), "ROLLBACK") + // Holding the write reservation keeps Core from changing the evidence between + // review verification and publication, while WAL readers remain available. + current, err := s.planEndpointRepair(ctx, p.Request) + if err != nil { + return 0, err + } + if current.SHA256 != sha { + return 0, errors.New("historical evidence changed; prepare a new plan") + } + existing, err := s.endpointRepairs(ctx, p.Request.SinceMS, p.Request.UntilMS) + if err != nil { + return 0, err + } + for _, m := range p.Minutes { + if old, ok := existing[m.Flow][m.StartMS]; ok && old != m { + return 0, errors.New("overlapping plan conflicts") + } + } + var saved string + err = conn.QueryRowContext(ctx, `SELECT value FROM energy_ledger_meta WHERE key=?`, endpointRepairPrefix+sha).Scan(&saved) + if err == nil { + return len(p.Minutes), nil + } + if !errors.Is(err, sql.ErrNoRows) { + return 0, err + } + raw, err := json.Marshal(endpointApplied{p, backupSHA, time.Now().UTC().Format(time.RFC3339Nano)}) + if err != nil { + return 0, err + } + if _, err = conn.ExecContext(ctx, `INSERT INTO energy_ledger_meta(key,value) VALUES(?,?)`, endpointRepairPrefix+sha, string(raw)); err != nil { + return 0, err + } + for _, table := range []string{"energy_daily", "energy_daily_partial", "history_daily_quality"} { + if _, err = conn.ExecContext(ctx, `DELETE FROM `+table); err != nil { + return 0, err + } + } + if _, err = conn.ExecContext(ctx, "COMMIT"); err != nil { + return 0, err + } + return len(p.Minutes), nil +} diff --git a/go/internal/state/history_endpoint_repair_test.go b/go/internal/state/history_endpoint_repair_test.go new file mode 100644 index 000000000..122e0054c --- /dev/null +++ b/go/internal/state/history_endpoint_repair_test.go @@ -0,0 +1,286 @@ +package state + +import ( + "context" + "encoding/json" + "fmt" + "path/filepath" + "testing" + "time" +) + +func endpointFixture(t *testing.T, night bool, at ...int64) (*Store, EndpointRepairRequest, string) { + t.Helper() + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour).UnixMilli() + if len(at) > 0 { + base = at[0] + } + for at := base - 60000; at <= base+31*60000; at += 5000 { + if at >= base+10*60000 && at < base+22*60000 { + continue + } + load, pv := 2000.0, 1000.0 + if at >= base+22*60000 { + load, pv = 4000, 3000 + } + if night { + pv = 0 + } + v := HistorySources{AtMS: at, Topology: "grid+pv", GridW: load - pv, PVW: -pv} + p := HistoryPoint{TsMs: at, GridW: v.GridW, PVW: v.PVW, LoadW: load, JSON: `{"ev_w":0,"v2x_w":0}`, Sources: &v} + dashboardTick(t, s, fmt.Sprint(at), &p) + } + r := EndpointRepairRequest{base, base + 30*60000, 15 * 60000, "owner-authorized endpoint mean"} + backup := filepath.Join(t.TempDir(), "history-before.db") + if _, err := s.history.Exec(`VACUUM INTO ?`, backup); err != nil { + t.Fatal(err) + } + return s, r, backup +} +func TestEndpointRepairMeansBothFlowsWhileLiveAndPreservesOriginals(t *testing.T) { + ctx := context.Background() + s, r, backup := endpointFixture(t, false) + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if len(p.Minutes) == 0 || len(p.Skipped) != 0 { + t.Fatal(p) + } + original, err := s.endpointOriginals(ctx, r.SinceMS, r.UntilMS) + if err != nil { + t.Fatal(err) + } + beforeDigest := endpointDigest(original) + // Seed the independent measured PV ledger; the repair must preserve buckets + // outside its affected windows and replace whole affected system buckets. + pvLedger := map[int64]float64{} + for _, b := range original { + pvLedger[bucketStart(b.Start, 300000)] += b.PV + } + for at, wh := range pvLedger { + insertLedgerEntryTest(t, s, "pv-device", FlowPVGeneration, at, 300000, wh, "hardware_counter", "measured", "counter", 1) + } + for _, m := range p.Minutes { + want := 50.0 + if m.Flow == "pv" { + want = 2000.0 / 60 + } + near(t, "endpoint formula", m.Wh, want) + if m.AfterMS-m.BeforeMS > r.MaxGapMS { + t.Fatal("bound ignored", m) + } + } + // Simulate new receipts arriving after review: they must survive publication. + next := r.UntilMS + 120000 + sourceTick(t, s, next, HistorySources{GridW: 3210}) + var through string + if err := s.history.QueryRow(`SELECT value FROM history_source_cursor WHERE key='through'`).Scan(&through); err != nil { + t.Fatal(err) + } + n, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256) + if err != nil || n != len(p.Minutes) { + t.Fatal(n, err) + } + again, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256) + if err != nil || again != n { + t.Fatal("retry", again, err) + } + afterOriginal, _ := s.endpointOriginals(ctx, r.SinceMS, r.UntilMS) + if endpointDigest(afterOriginal) != beforeDigest { + t.Fatal("original energy/quality changed") + } + var afterThrough string + _ = s.history.QueryRow(`SELECT value FROM history_source_cursor WHERE key='through'`).Scan(&afterThrough) + if through != afterThrough { + t.Fatal("new data lost") + } + var plans int + _ = s.history.QueryRow(`SELECT count(*) FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*'`).Scan(&plans) + if plans != 1 { + t.Fatal("double counted", plans) + } + d, err := s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&(MissingLoad|MissingPV) != 0 || d.Missing&EndpointEstimatedPV == 0 || d.Missing&EstimatedLoad == 0 { + t.Fatal("daily", d, err) + } + expected := map[string]float64{"load": 0, "pv": 0} + corrections := map[string]map[int64]float64{"load": {}, "pv": {}} + for _, m := range p.Minutes { + corrections[m.Flow][m.StartMS] = m.Wh + } + for _, b := range original { + for _, flow := range []string{"load", "pv"} { + if wh, ok := corrections[flow][b.Start]; ok { + expected[flow] += wh + } else if flow == "load" { + expected[flow] += b.Load + } else { + expected[flow] += b.PV + } + } + } + near(t, "home total", d.LoadWh, expected["load"]) + near(t, "solar total", d.PVWh, expected["pv"]) + ledger, truncated, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: r.SinceMS, UntilMS: r.UntilMS, BucketMS: 300000, Limit: 20}) + if err != nil || truncated { + t.Fatal(err) + } + sums := map[EnergyFlow]float64{} + for _, v := range ledger { + if v.Flow == FlowConsumerUse || v.Flow == FlowPVGeneration { + sums[v.Flow] += v.EnergyWh + if v.Source == "historical_repair" && v.Provenance != "endpoint_mean_v1" { + t.Fatal(v) + } + } + } + near(t, "ledger home", sums[FlowConsumerUse], d.LoadWh) + near(t, "ledger solar", sums[FlowPVGeneration], d.PVWh) + points, err := s.LoadHistory(r.SinceMS, r.UntilMS, 30) + if err != nil { + t.Fatal(err) + } + var marked bool + for _, v := range points { + var detail map[string]any + _ = json.Unmarshal([]byte(v.JSON), &detail) + if detail["pv_quality"] == "estimated" { + marked = true + if v.Missing&EstimatedLoad == 0 { + t.Fatal("training mark missing") + } + } + } + if !marked { + t.Fatal("chart correction missing") + } + if err := s.SaveDailyEnergy("endpoint-cache", d); err != nil { + t.Fatal(err) + } + cached, hit, err := s.LoadDailyEnergy("endpoint-cache") + if err != nil || !hit || cached.Missing&EndpointEstimatedPV == 0 { + t.Fatal("solar estimate marker lost in cache", cached, hit, err) + } + // Retention must preserve the minute boundaries used to replace each flow. + if err := s.rollSiteEnergy(ctx, r.UntilMS+900000); err != nil { + t.Fatal(err) + } + retained, err := s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil { + t.Fatal(err) + } + near(t, "retained solar", retained.PVWh, d.PVWh) + near(t, "retained home", retained.LoadWh, d.LoadWh) + exported := filepath.Join(t.TempDir(), "restored.db") + if err := s.copyStateForBackup(ctx, exported); err != nil { + t.Fatal(err) + } + if err := s.exportHistoryToSQLite(ctx, exported); err != nil { + t.Fatal(err) + } + restored, err := Open(exported) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + got, err := restored.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || got.Missing&(MissingLoad|MissingPV) != 0 { + t.Fatal(got, err) + } + near(t, "restored home", got.LoadWh, d.LoadWh) + near(t, "restored solar", got.PVWh, d.PVWh) +} +func TestEndpointRepairNightZeroAndBounds(t *testing.T) { + s, r, backup := endpointFixture(t, true) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + for _, m := range p.Minutes { + if m.Flow == "pv" && m.Wh != 0 { + t.Fatal("night must use observed zero endpoints", m) + } + } + bad := p + bad.Minutes = append([]EndpointRepairMinute(nil), p.Minutes...) + bad.Minutes[0].Wh++ + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, bad, p.SHA256); err == nil { + t.Fatal("tampering accepted") + } + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, s.historyPath, p, p.SHA256); err == nil { + t.Fatal("source accepted as backup") + } + r.MaxGapMS = 5 * 60000 + p, err = PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil || len(p.Minutes) != 0 || len(p.Skipped) != 2 { + t.Fatal("long gap was filled", p, err) + } +} +func TestEndpointRepairRejectsChangedEvidence(t *testing.T) { + s, r, backup := endpointFixture(t, false) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if _, err = s.history.Exec(`UPDATE history_source_energy SET import_wh=import_wh+1 WHERE start_ms=?`, r.SinceMS); err != nil { + t.Fatal(err) + } + if _, err = ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256); err == nil { + t.Fatal("stale evidence accepted") + } + var n int + _ = s.history.QueryRow(`SELECT count(*) FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*'`).Scan(&n) + if n != 0 { + t.Fatal("partial application") + } +} + +func TestEndpointRepairRollsBackWhenCacheInvalidationFails(t *testing.T) { + s, r, backup := endpointFixture(t, false) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if err := s.SaveDailyEnergy("guard", DayEnergy{Intervals: 1}); err != nil { + t.Fatal(err) + } + if _, err := s.history.Exec(`CREATE TRIGGER endpoint_cache_guard BEFORE DELETE ON energy_daily_partial BEGIN SELECT RAISE(ABORT,'cache locked for test'); END`); err != nil { + t.Fatal(err) + } + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256); err == nil { + t.Fatal("expected transaction failure") + } + var n int + _ = s.history.QueryRow(`SELECT count(*) FROM energy_ledger_meta WHERE key GLOB 'endpoint-repair-v1:*'`).Scan(&n) + if n != 0 { + t.Fatal("audit partially committed") + } + if _, hit, err := s.LoadDailyEnergy("guard"); err != nil || !hit { + t.Fatal("cache partially deleted", hit, err) + } +} + +func TestEndpointRepairLiveTotalsStopAtCommittedMinute(t *testing.T) { + base := time.Now().Truncate(time.Minute).Add(-40 * time.Minute).UnixMilli() + s, r, backup := endpointFixture(t, false, base) + ctx := context.Background() + p, err := PrepareEndpointRepair(ctx, s.mainDBPath, r) + if err != nil { + t.Fatal(err) + } + if _, err := ApplyEndpointRepair(ctx, s.mainDBPath, backup, p, p.SHA256); err != nil { + t.Fatal(err) + } + d, err := s.DailyEnergy(base, time.Now().UnixMilli()) + if err != nil || d.Missing&(MissingLoad|MissingPV) != 0 { + t.Fatal("writer tail voided repaired totals", d, err) + } + if d.LoadUntilMS != base+31*60000 || d.PVUntilMS != d.LoadUntilMS { + t.Fatal("extrapolated unfinished tail", d) + } +} diff --git a/go/internal/state/history_repair.go b/go/internal/state/history_repair.go new file mode 100644 index 000000000..21dc56128 --- /dev/null +++ b/go/internal/state/history_repair.go @@ -0,0 +1,470 @@ +package state + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + "time" +) + +// Every electrical flow must be selected or explicitly declared absent. In +// particular, select site-side AC battery power, never a cabinet DC counter. +type RepairPowerSource struct { + Flow string `json:"flow"` + Driver string `json:"driver,omitempty"` + Metric string `json:"metric,omitempty"` + Absent bool `json:"absent,omitempty"` +} + +type HistoryRepairRequest struct { + SinceMS int64 `json:"since_ms"` + UntilMS int64 `json:"until_ms"` + MaxGapMS int64 `json:"max_gap_ms"` + Reason string `json:"reason"` + Sources []RepairPowerSource `json:"sources"` +} + +type HistoryRepairPlan struct { + SHA256 string `json:"sha256"` + Request HistoryRepairRequest `json:"request"` + CreatedAt string `json:"created_at"` + BackupStateSHA256 string `json:"backup_state_sha256"` + BackupHistorySHA256 string `json:"backup_history_sha256"` + WorkingHistorySHA256 string `json:"working_history_sha256"` + WorkingStateSHA256 string `json:"working_state_sha256"` + Minutes []repairMinute `json:"minutes"` + Skipped map[int64]string `json:"skipped"` +} + +func (p HistoryRepairPlan) digest() string { + p.SHA256 = "" + b, _ := json.Marshal(p) + h := sha256.Sum256(b) + return hex.EncodeToString(h[:]) +} + +func (r HistoryRepairRequest) validate() error { + if r.SinceMS%60000 != 0 || r.UntilMS%60000 != 0 || r.UntilMS <= r.SinceMS || r.UntilMS-r.SinceMS > 48*3600000 || r.UntilMS > time.Now().Add(-time.Minute).UnixMilli() { + return errors.New("choose completed minute boundaries, at most 48 hours, ending at least one minute ago") + } + if r.MaxGapMS < 1000 || r.MaxGapMS > 180000 || strings.TrimSpace(r.Reason) == "" { + return errors.New("reason and max-gap between 1 and 180 seconds are required") + } + seen := map[string]bool{} + electrical := map[string]bool{} + for _, v := range r.Sources { + if v.Flow != "grid" && v.Flow != "pv" && v.Flow != "battery" && v.Flow != "ev" && v.Flow != "v2x" { + return errors.New("unknown electrical flow") + } + if seen[v.Flow] { + return errors.New("duplicate electrical flow") + } + seen[v.Flow] = true + if v.Absent { + if v.Flow == "grid" || v.Driver != "" || v.Metric != "" { + return errors.New("grid cannot be absent; absent flows cannot name a source") + } + continue + } + if v.Driver == "" || !strings.HasSuffix(v.Metric, "_w") { + return errors.New("each present flow needs an explicit driver and power metric ending in _w") + } + key := v.Driver + ":" + v.Metric + if electrical[key] { + return errors.New("one electrical source cannot be counted twice") + } + electrical[key] = true + } + if len(seen) != 5 { + return errors.New("select grid, pv, battery, ev and v2x; explicitly use none for absent flows") + } + return nil +} + +type repairPowerPoint struct { + at int64 + watts float64 +} + +// Aggregates are retained means, not reconstructed raw polls. Linear +// allocation and any bridge are estimates recorded in the correction plan. +func repairIntegral(points []repairPowerPoint, from, to, maxGap int64) (float64, int64, bool) { + var wh float64 + var covered, interpolated int64 + first := max(1, sort.Search(len(points), func(i int) bool { return points[i].at >= from })) + for i := first; i < len(points); i++ { + a, b := points[i-1], points[i] + if a.at >= to { + break + } + lo, hi := max(from, a.at), min(to, b.at) + if hi <= lo { + continue + } + dt := b.at - a.at + if dt <= 0 || dt > maxGap { + return 0, 0, false + } + left := a.watts + (b.watts-a.watts)*float64(lo-a.at)/float64(dt) + right := a.watts + (b.watts-a.watts)*float64(hi-a.at)/float64(dt) + wh += (left + right) / 2 * float64(hi-lo) / 3600000 + covered += hi - lo + if dt > maxLoadEstimateTickMS { + interpolated += hi - lo + } + } + return wh, interpolated, covered == to-from && finiteHistoryValue(wh) +} + +func (s *Store) planHistoryRepair(ctx context.Context, r HistoryRepairRequest) ([]repairMinute, map[int64]string, error) { + if err := r.validate(); err != nil { + return nil, nil, err + } + original := map[int64]*sourceLoadBucket{} + err := s.walkSiteEnergy(ctx, r.SinceMS, r.UntilMS, func(v siteEnergyRow) error { + if v.width != ArchiveResolutionMS { + return errors.New("repair needs retained minute energy; coarse historical intervals are not split") + } + b := original[v.start] + if b == nil { + b = &sourceLoadBucket{} + original[v.start] = b + } + b.covered += v.covered + b.wh += v.load + b.missing |= v.missing + return nil + }) + if err != nil { + return nil, nil, err + } + series := map[string][]repairPowerPoint{} + for _, v := range r.Sources { + if v.Absent { + continue + } + var unit string + if err := s.history.QueryRowContext(ctx, `SELECT COALESCE(unit,'') FROM ts_metrics WHERE name=?`, v.Metric).Scan(&unit); err != nil { + return nil, nil, fmt.Errorf("source %s:%s: %w", v.Driver, v.Metric, err) + } + if unit != "" && unit != "W" { + return nil, nil, errors.New("repair source is not in watts") + } + p, err := s.LoadSeriesBucketsOrRawContext(ctx, v.Driver, v.Metric, r.SinceMS-r.MaxGapMS-60000, r.UntilMS+r.MaxGapMS+60000, 0) + if err != nil { + return nil, nil, err + } + for _, q := range p { + if q.ResolutionMS > 60000 || !finiteHistoryValue(q.V) { + return nil, nil, errors.New("repair requires finite power evidence at minute resolution or finer") + } + at := q.TsMs + if q.FirstMS > 0 { + at = q.FirstMS + (q.TsMs-q.FirstMS)/2 + } + series[v.Flow] = append(series[v.Flow], repairPowerPoint{at, q.V}) + } + for i := 1; i < len(series[v.Flow]); i++ { + if series[v.Flow][i].at <= series[v.Flow][i-1].at { + return nil, nil, errors.New("conflicting source observation times") + } + } + } + var minutes []repairMinute + skipped := map[int64]string{} + for at := r.SinceMS; at < r.UntilMS; at += 60000 { + b := original[at] + if b != nil && b.covered == 60000 && b.missing&MissingPower == 0 && b.wh >= 0 { + continue + } + var existing int + if err := s.history.QueryRowContext(ctx, `SELECT COUNT(*) FROM history_load_estimates WHERE start_ms=?`, at).Scan(&existing); err != nil { + return nil, nil, err + } + if existing != 0 { + skipped[at] = "existing estimate preserved" + continue + } + p := repairMinute{StartMS: at} + valid := true + for _, v := range r.Sources { + if v.Absent { + continue + } + wh, bridge, ok := repairIntegral(series[v.Flow], at, at+60000, r.MaxGapMS) + if !ok { + skipped[at] = "insufficient bounded power evidence: " + v.Flow + valid = false + break + } + if v.Flow == "grid" { + p.Wh += wh + } else { + p.Wh -= wh + } + p.InterpolatedMS = max(p.InterpolatedMS, bridge) + } + if valid { + if !plausibleEnergy(absRepairWh(p.Wh), 60000) { + skipped[at] = "implausible reconstructed energy" + continue + } + minutes = append(minutes, p) + } + } + return minutes, skipped, nil +} + +func absRepairWh(v float64) float64 { + if v < 0 { + return -v + } + return v +} + +func repairFileHash(path string) (string, error) { + info, err := os.Lstat(path) + if err != nil { + return "", err + } + if !info.Mode().IsRegular() { + return "", errors.New("repair files must be regular files, not links") + } + f, err := os.Open(path) + if err != nil { + return "", err + } + defer f.Close() + h := sha256.New() + if _, err = io.Copy(h, f); err != nil { + return "", err + } + return hex.EncodeToString(h.Sum(nil)), nil +} + +func repairCopyFile(from, to string) error { + a, err := os.Open(from) + if err != nil { + return err + } + defer a.Close() + b, err := os.OpenFile(to, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0600) + if err != nil { + return err + } + _, copyErr := io.Copy(b, a) + syncErr := b.Sync() + return errors.Join(copyErr, syncErr, b.Close()) +} + +func writeRepairJSON(path string, v any) error { + b, err := json.MarshalIndent(v, "", " ") + if err != nil { + return err + } + return os.WriteFile(path, append(b, '\n'), 0600) +} + +// PrepareHistoryRepair never opens the owner's databases for writing. It +// snapshots state and history, then operates exclusively on the new copies. +func PrepareHistoryRepair(ctx context.Context, statePath, workdir string, r HistoryRepairRequest) (HistoryRepairPlan, error) { + var plan HistoryRepairPlan + if err := r.validate(); err != nil { + return plan, err + } + if err := os.Mkdir(workdir, 0700); err != nil { + return plan, err + } + for _, dir := range []string{"backup", "corrected"} { + if err := os.Mkdir(filepath.Join(workdir, dir), 0700); err != nil { + return plan, err + } + } + s, err := OpenBackupSource(statePath) + if err != nil { + return plan, err + } + if s.history == nil { + s.Close() + return plan, errors.New("repair requires primary SQLite history") + } + if err := EnsureDiskSpace(workdir, 3*s.BackupSourceBytes()+backupScratchHeadroom); err != nil { + s.Close() + return plan, err + } + _, err = s.db.ExecContext(ctx, `VACUUM INTO ?`, filepath.Join(workdir, "backup", "state.db")) + if err == nil { + _, err = s.history.ExecContext(ctx, `VACUUM INTO ?`, filepath.Join(workdir, "backup", HistoryFilename)) + } + closeErr := s.Close() + if err != nil { + return plan, err + } + if closeErr != nil { + return plan, closeErr + } + for _, name := range []string{"state.db", HistoryFilename} { + if err := repairCopyFile(filepath.Join(workdir, "backup", name), filepath.Join(workdir, "corrected", name)); err != nil { + return plan, err + } + } + plan.BackupStateSHA256, err = repairFileHash(filepath.Join(workdir, "backup", "state.db")) + if err != nil { + return plan, err + } + plan.BackupHistorySHA256, err = repairFileHash(filepath.Join(workdir, "backup", HistoryFilename)) + if err != nil { + return plan, err + } + c, err := Open(filepath.Join(workdir, "corrected", "state.db")) + if err != nil { + return plan, err + } + var integrity string + err = c.history.QueryRowContext(ctx, `PRAGMA quick_check`).Scan(&integrity) + if err == nil && integrity != "ok" { + err = errors.New("history integrity check failed") + } + if err == nil { + plan.Minutes, plan.Skipped, err = c.planHistoryRepair(ctx, r) + } + closeErr = c.Close() + if err != nil { + return plan, err + } + if closeErr != nil { + return plan, closeErr + } + plan.WorkingHistorySHA256, err = repairFileHash(filepath.Join(workdir, "corrected", HistoryFilename)) + if err != nil { + return plan, err + } + plan.WorkingStateSHA256, err = repairFileHash(filepath.Join(workdir, "corrected", "state.db")) + if err != nil { + return plan, err + } + plan.Request = r + plan.CreatedAt = time.Now().UTC().Format(time.RFC3339Nano) + plan.SHA256 = plan.digest() + return plan, writeRepairJSON(filepath.Join(workdir, "plan.json"), plan) +} + +type HistoryRepairResult struct { + PlanSHA256 string `json:"plan_sha256"` + AppliedMinutes int `json:"applied_minutes"` + BeforeWh any `json:"before_load_wh"` + AfterWh any `json:"after_load_wh"` + AfterQuality string `json:"after_quality"` + Estimated bool `json:"load_estimated"` +} + +// ApplyHistoryRepair writes the approved additive estimates to the working +// copy only. A matching retry is idempotent; stale/tampered plans fail closed. +func ApplyHistoryRepair(ctx context.Context, workdir, approvedSHA string) (HistoryRepairResult, error) { + var result HistoryRepairResult + var p HistoryRepairPlan + b, err := os.ReadFile(filepath.Join(workdir, "plan.json")) + if err != nil { + return result, err + } + if err = json.Unmarshal(b, &p); err != nil { + return result, err + } + if len(approvedSHA) != 64 || p.SHA256 != approvedSHA || p.digest() != approvedSHA { + return result, errors.New("exact reviewed plan SHA-256 is required") + } + if err = p.Request.validate(); err != nil { + return result, err + } + for _, v := range []struct{ name, want string }{{"state.db", p.BackupStateSHA256}, {HistoryFilename, p.BackupHistorySHA256}} { + got, err := repairFileHash(filepath.Join(workdir, "backup", v.name)) + if err != nil || got != v.want { + return result, errors.New("verified backup changed or is unavailable") + } + } + working := filepath.Join(workdir, "corrected", HistoryFilename) + stateHash, err := repairFileHash(filepath.Join(workdir, "corrected", "state.db")) + if err != nil || stateHash != p.WorkingStateSHA256 { + return result, errors.New("working state changed after the plan") + } + hash, err := repairFileHash(working) + if err != nil { + return result, err + } + s, err := Open(filepath.Join(workdir, "corrected", "state.db")) + if err != nil { + return result, err + } + defer s.Close() + methodJSON, _ := json.Marshal(map[string]any{"plan_sha256": p.SHA256, "backup_history_sha256": p.BackupHistorySHA256, "request": p.Request, "created_at": p.CreatedAt}) + method := historicalRepairMethod + string(methodJSON) + if len(p.Minutes) == 0 { + return result, errors.New("plan has no supported corrections") + } + allApplied := true + for _, v := range p.Minutes { + var wh float64 + var saved string + err := s.history.QueryRowContext(ctx, `SELECT load_wh,method FROM history_load_estimates WHERE start_ms=? AND resolution_ms=60000`, v.StartMS).Scan(&wh, &saved) + if err != nil || saved != method || wh != v.Wh { + allApplied = false + break + } + } + if hash != p.WorkingHistorySHA256 && !allApplied { + return result, errors.New("working history changed after the plan; prepare a new copy") + } + beforeSource, err := OpenBackupSource(filepath.Join(workdir, "backup", "state.db")) + if err != nil { + return result, err + } + before, err := beforeSource.DailyEnergy(p.Request.SinceMS, p.Request.UntilMS) + beforeSource.Close() + if err != nil { + return result, err + } + if !allApplied { + tx, err := s.history.BeginTx(ctx, nil) + if err != nil { + return result, err + } + defer tx.Rollback() + for _, v := range p.Minutes { + if v.StartMS < p.Request.SinceMS || v.StartMS+60000 > p.Request.UntilMS || v.StartMS%60000 != 0 || !finiteHistoryValue(v.Wh) { + return result, errors.New("invalid correction minute") + } + if _, err = tx.ExecContext(ctx, `INSERT INTO history_load_estimates(start_ms,resolution_ms,covered_ms,load_wh,method) VALUES(?,60000,60000,?,?)`, v.StartMS, v.Wh, method); err != nil { + return result, err + } + } + // All these tables are derived caches. Original observations and their + // quality are untouched; a new Core process recomputes corrected days. + for _, table := range []string{"energy_daily", "energy_daily_partial", "history_daily_quality"} { + if _, err = tx.ExecContext(ctx, `DELETE FROM `+table); err != nil { + return result, err + } + } + if err = tx.Commit(); err != nil { + return result, err + } + } + after, err := s.DailyEnergy(p.Request.SinceMS, p.Request.UntilMS) + if err != nil { + return result, err + } + result = HistoryRepairResult{p.SHA256, len(p.Minutes), before.Missing.Value(MissingLoad, before.LoadWh), after.Missing.Value(MissingLoad, after.LoadWh), after.Missing.Quality(), after.Missing&EstimatedLoad != 0} + return result, writeRepairJSON(filepath.Join(workdir, "apply-result.json"), result) +} + +// Keep plan order deterministic for callers that construct requests in maps. +func SortRepairSources(sources []RepairPowerSource) { + sort.Slice(sources, func(i, j int) bool { return sources[i].Flow < sources[j].Flow }) +} diff --git a/go/internal/state/history_repair_read.go b/go/internal/state/history_repair_read.go new file mode 100644 index 000000000..85917c111 --- /dev/null +++ b/go/internal/state/history_repair_read.go @@ -0,0 +1,247 @@ +package state + +import ( + "context" + "encoding/json" + "errors" + "sort" + "strings" +) + +const historicalRepairMethod = "historical_power_repair_v1:" + +type repairMinute struct { + StartMS int64 `json:"start_ms"` + Wh float64 `json:"wh"` + InterpolatedMS int64 `json:"max_interpolated_ms_per_source"` +} + +func (s *Store) historicalLoadRepairs(ctx context.Context, from, to int64) (map[int64]repairMinute, error) { + out := map[int64]repairMinute{} + if s.history == nil { + return out, nil + } + rows, err := s.history.QueryContext(ctx, `SELECT start_ms,resolution_ms,covered_ms,load_wh,method FROM history_load_estimates WHERE start_ms>=? AND start_ms to { + continue + } + if err := visit(siteEnergyRow{start: start, width: ArchiveResolutionMS, first: start, last: start + ArchiveResolutionMS, + covered: ArchiveResolutionMS, load: p.Wh, missing: EstimatedLoad}); err != nil { + return err + } + } + return nil +} + +func (s *Store) repairedLoadBuckets(ctx context.Context, from, to, width int64) ([]sourceLoadBucket, bool, error) { + repairs, err := s.historicalLoadRepairs(ctx, from, to) + if err != nil || len(repairs) == 0 { + return nil, false, err + } + byStart := map[int64]*sourceLoadBucket{} + err = s.walkHouseholdLoad(ctx, from, to, repairs, func(r siteEnergyRow) error { + w := max(width, r.width) + start := from + (r.start-from)/w*w + b := byStart[start] + if b == nil { + b = &sourceLoadBucket{start: start, width: w, first: r.first, last: r.last} + byStart[start] = b + } + if b.width != w || r.first < start || r.last > start+w { + b.missing |= MissingLoad + } + b.first = min(b.first, r.first) + b.last = max(b.last, r.last) + b.covered += r.covered + b.n += r.n + b.wh += r.load + b.missing |= r.missing + return nil + }) + if err != nil { + return nil, true, err + } + out := make([]sourceLoadBucket, 0, len(byStart)) + for _, b := range byStart { + if b.covered != b.width || b.first != b.start || b.last != b.start+b.width || b.wh < 0 || !finiteHistoryValue(b.wh) { + b.missing |= MissingLoad + } + out = append(out, *b) + } + sort.Slice(out, func(i, j int) bool { return out[i].start < out[j].start }) + return out, true, nil +} + +func (s *Store) applyHistoricalDailyLoad(ctx context.Context, from, to int64, d *DayEnergy) error { + // A historical correction cannot invent the currently unfinished minute. + end, err := s.endpointReadUntil(ctx, from, to) + if err != nil { + return err + } + if end <= from { + return nil + } + b, have, err := s.repairedLoadBuckets(ctx, from, end, end-from) + if err != nil || !have { + return err + } + if len(b) == 1 && b[0].missing&(MissingLoad|BalanceMismatch) == 0 && b[0].covered == end-from { + d.LoadWh = b[0].wh + d.Missing &^= MissingLoad | BalanceMismatch + d.Missing |= EstimatedLoad + d.LoadUntilMS = end + } + return nil +} + +func (s *Store) applyHistoricalLoadChartOnly(ctx context.Context, points []HistoryPoint, from, to, width int64) ([]HistoryPoint, error) { + if width < ArchiveResolutionMS { + return points, nil + } // Do not invent raw poll values. + b, have, err := s.repairedLoadBuckets(ctx, from, to, width) + if err != nil || !have { + return points, err + } + byStart := map[int64]sourceLoadBucket{} + seen := map[int64]bool{} + repairs, err := s.historicalLoadRepairs(ctx, from, to) + if err != nil { + return points, err + } + changed := map[int64]bool{} + for start := range repairs { + changed[from+(start-from)/width*width] = true + } + for _, p := range b { + byStart[p.start] = p + } + for i := range points { + p := &points[i] + start := from + (p.TsMs-from)/width*width + if !changed[start] { + continue + } + v, ok := byStart[start] + if !ok || v.missing&(MissingLoad|BalanceMismatch) != 0 { + continue + } + p.LoadW = v.wh * 3600000 / float64(v.width) + seen[start] = true + p.Missing &^= MissingLoad | BalanceMismatch + p.Missing |= EstimatedLoad + var detail map[string]any + if json.Unmarshal([]byte(p.JSON), &detail) == nil && detail != nil { + detail["history_load_method"] = "historical_power_repair_v1" + detail["load_quality"] = "estimated" + if data, err := json.Marshal(detail); err == nil { + p.JSON = string(data) + } + } + } + for _, v := range b { + if !changed[v.start] || seen[v.start] || v.missing&(MissingLoad|BalanceMismatch) != 0 { + continue + } + points = append(points, HistoryPoint{TsMs: v.start + v.width - 1, FirstMS: v.start, ResolutionMS: v.width, LoadW: v.wh * 3600000 / float64(v.width), + Missing: MissingGrid | MissingPV | MissingBattery | MissingEV | MissingV2X | MissingSoC | EstimatedLoad, + JSON: `{"history_load_method":"historical_power_repair_v1","load_quality":"estimated","forecast_measurement_quality":"aggregate_not_for_training"}`}) + } + sort.SliceStable(points, func(i, j int) bool { return points[i].TsMs < points[j].TsMs }) + return points, nil +} + +func (s *Store) applyHistoricalLoadLedgerOnly(ctx context.Context, q EnergyHistoryQuery, points []EnergyLedgerPoint) ([]EnergyLedgerPoint, error) { + if q.AssetID != "" && q.AssetID != ObservedConsumerAssetID { + return points, nil + } + b, have, err := s.repairedLoadBuckets(ctx, q.SinceMS, q.UntilMS, q.BucketMS) + if err != nil || !have { + return points, err + } + repairs, err := s.historicalLoadRepairs(ctx, q.SinceMS, q.UntilMS) + if err != nil { + return nil, err + } + changed := map[int64]bool{} + for start := range repairs { + changed[q.SinceMS+(start-q.SinceMS)/q.BucketMS*q.BucketMS] = true + } + out := points[:0] + for _, p := range points { + if p.Flow != FlowConsumerUse || !changed[p.BucketStartMS] { + out = append(out, p) + } + } + for _, p := range b { + if !changed[p.start] { + continue + } + if p.missing&(MissingLoad|BalanceMismatch) != 0 { + continue + } + asset := "system" + if q.AssetID != "" { + asset = q.AssetID + } + out = append(out, EnergyLedgerPoint{SchemaVersion: EnergyLedgerSchemaVersion, AssetID: asset, Flow: FlowConsumerUse, + BucketStartMS: p.start, BucketLenMS: p.width, EnergyWh: p.wh, Source: "historical_repair", Quality: "estimated", Provenance: "historical_power_repair_v1", SampleCount: p.n}) + } + sort.SliceStable(out, func(i, j int) bool { return out[i].BucketStartMS < out[j].BucketStartMS }) + return out, nil +} diff --git a/go/internal/state/history_repair_test.go b/go/internal/state/history_repair_test.go new file mode 100644 index 000000000..64b4d9c2c --- /dev/null +++ b/go/internal/state/history_repair_test.go @@ -0,0 +1,350 @@ +package state + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "reflect" + "testing" + "time" +) + +func historicalRepairFixture(t *testing.T, powerGap bool) (string, HistoryRepairRequest) { + t.Helper() + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).Add(-24 * time.Hour).UnixMilli() + for i := int64(0); i <= 90; i++ { + v := HistorySources{GridW: 2000, PVW: -500, BatteryW: 100, EVW: 400} + if i == 42 { + v.Missing = MissingPV + } + if i == 66 { + continue + } // retained scalar series can repair a writer gap too. + sourceTick(t, s, base+i*10000, v) + } + var samples []Sample + for at := base - 30000; at <= base+930000; at += 5000 { + for metric, w := range map[string]float64{"meter_w": 2000, "pv_w": -500, "battery_ac_w": 100, "ev_w": 400} { + if powerGap && metric == "pv_w" && at >= base+360000 && at <= base+480000 { + continue + } + samples = append(samples, Sample{Driver: "equipment", Metric: metric, TsMs: at, Value: w, Unit: "W"}) + } + } + if err := s.RecordSamples(samples); err != nil { + t.Fatal(err) + } + path := s.mainDBPath + if err := s.SaveDailyEnergy("stale-cache", DayEnergy{Missing: MissingLoad, Intervals: 1}); err != nil { + t.Fatal(err) + } + if err := s.Close(); err != nil { + t.Fatal(err) + } + r := HistoryRepairRequest{SinceMS: base, UntilMS: base + 900000, MaxGapMS: 15000, Reason: "fixture: reconstruct retained AC source energy", + Sources: []RepairPowerSource{{Flow: "grid", Driver: "equipment", Metric: "meter_w"}, {Flow: "pv", Driver: "equipment", Metric: "pv_w"}, {Flow: "battery", Driver: "equipment", Metric: "battery_ac_w"}, {Flow: "ev", Driver: "equipment", Metric: "ev_w"}, {Flow: "v2x", Absent: true}}} + return path, r +} + +func TestHistoryRepairCopyRecalculatesHistoricalTotalsAndPreservesOriginal(t *testing.T) { + ctx := context.Background() + source, r := historicalRepairFixture(t, false) + originalState, _ := repairFileHash(source) + originalHistory, _ := repairFileHash(filepath.Join(filepath.Dir(source), HistoryFilename)) + work := filepath.Join(t.TempDir(), "repair") + plan, err := PrepareHistoryRepair(ctx, source, work, r) + if err != nil { + t.Fatal(err) + } + if len(plan.Minutes) == 0 || len(plan.Skipped) != 0 { + t.Fatal("fixture has reconstructable gaps", plan) + } + if _, err := ApplyHistoryRepair(ctx, work, "unreviewed"); err == nil { + t.Fatal("unreviewed plan applied") + } + result, err := ApplyHistoryRepair(ctx, work, plan.SHA256) + if err != nil { + t.Fatal(err) + } + if result.BeforeWh != nil || result.AfterWh == nil || !result.Estimated { + t.Fatal("historical total not corrected", result) + } + near(t, "reconstructed period Wh", result.AfterWh.(float64), 500) + again, err := ApplyHistoryRepair(ctx, work, plan.SHA256) + if err != nil || !reflect.DeepEqual(result, again) { + t.Fatal("retry double counted", again, err) + } + if h, _ := repairFileHash(source); h != originalState { + t.Fatal("source state changed") + } + if h, _ := repairFileHash(filepath.Join(filepath.Dir(source), HistoryFilename)); h != originalHistory { + t.Fatal("source history changed") + } + s, err := Open(filepath.Join(work, "corrected", "state.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + var quality int + if err := s.history.QueryRow(`SELECT COUNT(*) FROM history_quality WHERE missing&?!=0`, MissingPV).Scan(&quality); err != nil || quality == 0 { + t.Fatal("original rejected evidence erased", quality, err) + } + var cached int + if err := s.history.QueryRow(`SELECT COUNT(*) FROM energy_daily_partial WHERE day='stale-cache'`).Scan(&cached); err != nil || cached != 0 { + t.Fatal("stale cache retained", cached, err) + } + d, err := s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&MissingLoad != 0 || d.Missing&EstimatedLoad == 0 { + t.Fatal("restart lost repair", d, err) + } + if d.Missing&MissingPV == 0 { + t.Fatal("load correction falsely repaired the PV measurement") + } + if err := s.SaveDailyEnergy("corrected-cache", d); err != nil { + t.Fatal(err) + } + c, hit, err := s.LoadDailyEnergy("corrected-cache") + if err != nil || !hit || c.Missing&EstimatedLoad == 0 { + t.Fatal(c, hit, err) + } + near(t, "persisted corrected Wh", c.LoadWh, 500) + points, truncated, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: r.SinceMS, UntilMS: r.UntilMS, BucketMS: 300000, Limit: 10}) + if err != nil || truncated { + t.Fatal(err) + } + var wh float64 + var n int + var estimated int + for _, p := range points { + if p.Flow == FlowConsumerUse { + wh += p.EnergyWh + n++ + if p.Quality == "estimated" { + estimated++ + } + } + } + if n != 3 { + t.Fatal("historical ledger not corrected", points) + } + if estimated == 0 { + t.Fatal("repair provenance missing from ledger") + } + near(t, "ledger Wh", wh, 500) + limited, truncated, err := s.LoadEnergyHistory(EnergyHistoryQuery{SinceMS: r.SinceMS, UntilMS: r.UntilMS, BucketMS: 300000, Limit: 1}) + if err != nil || !truncated { + t.Fatal("repair bypassed limit", err) + } + for _, p := range limited { + if p.BucketStartMS != r.SinceMS { + t.Fatal("extra buckets exceeded limit", p) + } + } + chart, err := s.LoadHistory(r.SinceMS, r.UntilMS, 3) + if err != nil { + t.Fatal(err) + } + for _, p := range chart { + if p.Missing&MissingLoad != 0 { + t.Fatal("chart kept a repaired load gap", p) + } + } + if len(chart) == 0 { + t.Fatal("corrected chart is empty") + } + // Retention must keep the originals needed to interpret the additive + // minute corrections, while ordinary uncorrected quarters may roll up. + if err := s.rollSiteEnergy(ctx, r.UntilMS); err != nil { + t.Fatal(err) + } + d, err = s.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&MissingLoad != 0 { + t.Fatal("retention lost repair", d, err) + } + near(t, "retained corrected Wh", d.LoadWh, 500) + backup := filepath.Join(t.TempDir(), "restored.db") + if err := s.copyStateForBackup(ctx, backup); err != nil { + t.Fatal(err) + } + if err := s.exportHistoryToSQLite(ctx, backup); err != nil { + t.Fatal(err) + } + restored, err := Open(backup) + if err != nil { + t.Fatal(err) + } + defer restored.Close() + d, err = restored.DailyEnergy(r.SinceMS, r.UntilMS) + if err != nil || d.Missing&MissingLoad != 0 || d.Missing&EstimatedLoad == 0 { + t.Fatal("portable backup lost repair", d, err) + } + near(t, "restored corrected Wh", d.LoadWh, 500) +} + +func TestHistoryRepairAddsDerivedChartIntervalWithoutInventingOtherMeasurements(t *testing.T) { + source, r := historicalRepairFixture(t, false) + work := filepath.Join(t.TempDir(), "repair") + p, err := PrepareHistoryRepair(context.Background(), source, work, r) + if err != nil { + t.Fatal(err) + } + if _, err := ApplyHistoryRepair(context.Background(), work, p.SHA256); err != nil { + t.Fatal(err) + } + s, err := Open(filepath.Join(work, "corrected", "state.db")) + if err != nil { + t.Fatal(err) + } + defer s.Close() + start := p.Minutes[0].StartMS + for _, table := range []string{"history_dashboard", "history_dashboard_partial"} { + if _, err := s.history.Exec(`DELETE FROM `+table+` WHERE last_ms>=? AND last_ms= 0 { + t.Fatal("negative residual clipped", negative, ok) + } + if _, _, ok := repairIntegral(p, -1, 180000, 60000); ok { + t.Fatal("missing edge was extrapolated") + } + if _, _, ok := repairIntegral(p, 0, 180000, 15000); ok { + t.Fatal("long unapproved gaps filled") + } +} + +func TestHistoryRepairRefusesInsufficientCopySpace(t *testing.T) { + source, r := historicalRepairFixture(t, false) + probe := backupDiskAvail + backupDiskAvail = func(string) (int64, error) { return 0, nil } + defer func() { backupDiskAvail = probe }() + if _, err := PrepareHistoryRepair(context.Background(), source, filepath.Join(t.TempDir(), "repair"), r); err == nil { + t.Fatal("repair started without room for backup and working copy") + } +} diff --git a/go/internal/state/household_subtotal.go b/go/internal/state/household_subtotal.go new file mode 100644 index 000000000..8d587959c --- /dev/null +++ b/go/internal/state/household_subtotal.go @@ -0,0 +1,66 @@ +package state + +import "context" + +// HouseholdSubtotal describes accepted, complete energy intervals only. It is +// not a complete range total, an independent meter, or a gap-filled estimate. +type HouseholdSubtotal struct { + Wh float64 `json:"wh"` + CoverageMS int64 `json:"coverage_ms"` + PeriodMS int64 `json:"period_ms"` + // Minimum interval width; older retained evidence can be coarser. + BucketMS int64 `json:"minimum_bucket_ms"` + Quality string `json:"quality"` + Method string `json:"method"` +} + +// HouseholdEnergySubtotal reads the same independent source energy and legacy +// site intervals as DailyEnergy, without changing their quality or caches. +// Balance signed energy before deciding whether an interval is usable; never +// clip negative samples, interpolate missing sources or bridge recording gaps. +func (s *Store) HouseholdEnergySubtotal(ctx context.Context, from, to int64) (*HouseholdSubtotal, error) { + if to <= from || s.history == nil { + return nil, nil + } + const width int64 = 5 * 60000 + repairs, err := s.historicalLoadRepairs(ctx, from, to) + if err != nil { + return nil, err + } + buckets := map[int64]*sourceLoadBucket{} + err = s.walkHouseholdLoad(ctx, from, to, repairs, func(r siteEnergyRow) error { + // Retained coarse intervals remain indivisible; never split an older + // observed interval into invented five-minute measurements. + w := max(width, r.width) + start := bucketStart(r.start, w) + b := buckets[start] + if b == nil { + b = &sourceLoadBucket{start: start, width: w, first: r.first, last: r.last} + buckets[start] = b + } + if b.width != w || r.first < start || r.last > start+w || start < from || start+w > to { + b.missing |= MissingLoad + } + b.first = min(b.first, r.first) + b.last = max(b.last, r.last) + b.covered += r.covered + b.wh += r.load + b.missing |= r.missing + return nil + }) + if err != nil { + return nil, err + } + out := HouseholdSubtotal{PeriodMS: to - from, BucketMS: width, Quality: "estimated", Method: "accepted_interval_balance_v1"} + for _, b := range buckets { + if b.missing&MissingPower != 0 || b.covered != b.width || b.first != b.start || b.last != b.start+b.width || b.wh < 0 || !finiteHistoryValue(b.wh) { + continue + } + out.Wh += b.wh + out.CoverageMS += b.covered + } + if out.CoverageMS == 0 { + return nil, nil + } + return &out, nil +} diff --git a/go/internal/state/household_subtotal_test.go b/go/internal/state/household_subtotal_test.go new file mode 100644 index 000000000..da7057363 --- /dev/null +++ b/go/internal/state/household_subtotal_test.go @@ -0,0 +1,170 @@ +package state + +import ( + "context" + "fmt" + "reflect" + "testing" + "time" +) + +func TestHouseholdSubtotalKeepsAcceptedIntervalsAcrossSourceAndWriterGaps(t *testing.T) { + for _, kind := range []string{"PV missing", "EV missing", "writer gap", "topology"} { + t.Run(kind, func(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i <= 90; i++ { + if kind == "writer gap" && i == 42 { + continue + } + v := HistorySources{GridW: 1200, EVW: 200} + if kind == "topology" && i >= 42 { + v.Topology = "replacement meter" + } + if i == 42 && kind == "PV missing" { + v.Missing = MissingPV + } + if i == 42 && kind == "EV missing" { + v.Missing = MissingEV + } + sourceTick(t, s, base+i*10000, v) + } + before, err := s.DailyEnergy(base, base+900000) + if err != nil || before.Missing&MissingLoad == 0 { + t.Fatal("fixture must have an incomplete full total", before, err) + } + check := func(s *Store) { + t.Helper() + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+900000) + if err != nil || got == nil { + t.Fatal("accepted intervals hidden", got, err) + } + near(t, "accepted Wh", got.Wh, 1000.0/6) + if got.CoverageMS != 600000 || got.PeriodMS != 900000 || got.Quality != "estimated" { + t.Fatal("gap became full coverage or measured energy", got) + } + after, err := s.DailyEnergy(base, base+900000) + if err != nil || !reflect.DeepEqual(before, after) { + t.Fatal("subtotal rewrote full total or original quality", before, after, err) + } + } + check(s) + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + check(reopened) + }) + } +} + +func TestHouseholdSubtotalCombinesLegacyAndSourceWithoutDoubleCounting(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + for i := int64(0); i <= 30; i++ { + at := base + i*10000 + p := &HistoryPoint{TsMs: at, GridW: 1000, LoadW: 1000, JSON: `{}`} + if i == 30 { + p.Sources = &HistorySources{AtMS: at, Topology: "grid+charger", GridW: 3000, EVW: 2000} + } + dashboardTick(t, s, fmt.Sprint(at), p) + } + sourceSeries(t, s, base+300000, []float64{1000, 1000, 1000, 1000, 1000}) + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+600000) + if err != nil || got == nil || got.CoverageMS != 600000 { + t.Fatal(got, err) + } + near(t, "mixed subtotal", got.Wh, 1000.0/6) +} + +func TestHouseholdSubtotalRetainsCoarseEvidenceWithoutSplittingItsBounds(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + minutes := make([]float64, 15) + for i := range minutes { + minutes[i] = 1000 + } + sourceSeries(t, s, base, minutes) + if err := s.rollSiteEnergy(context.Background(), base+900000); err != nil { + t.Fatal(err) + } + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+900000) + if err != nil || got == nil || got.CoverageMS != 900000 { + t.Fatal(got, err) + } + near(t, "retained subtotal", got.Wh, 250) + if got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+600000); err != nil || got != nil { + t.Fatal("coarse interval was split into invented observations", got, err) + } +} + +func TestHouseholdSubtotalPreservesSignedBalanceAndDistinguishesZeroFromUnknown(t *testing.T) { + for _, tc := range []struct { + name string + minutes []float64 + wantWh float64 + wantCoverage int64 + }{ + {"signed balance", []float64{-1000, 3000, 1000, 1000, 1000}, 1000.0 / 12, 300000}, + {"zero", []float64{0, 0, 0, 0, 0}, 0, 300000}, + {"negative interval", []float64{-1000, -1000, -1000, -1000, -1000}, 0, 0}, + {"partial tail", []float64{1000, 1000, 1000, 1000}, 0, 0}, + } { + t.Run(tc.name, func(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + sourceSeries(t, s, base, tc.minutes) + got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+int64(len(tc.minutes))*60000) + if err != nil { + t.Fatal(err) + } + if tc.wantCoverage == 0 { + if got != nil { + t.Fatal("unknown became a known zero", got) + } + return + } + if got == nil || got.CoverageMS != tc.wantCoverage { + t.Fatal(got) + } + near(t, "signed subtotal", got.Wh, tc.wantWh) + }) + } +} + +func TestHouseholdSubtotalNeverBridgesRestartOrClaimsAbsentEvidence(t *testing.T) { + s := sourceStore(t) + base := time.Now().UTC().Truncate(24 * time.Hour).UnixMilli() + if got, err := s.HouseholdEnergySubtotal(context.Background(), base, base+900000); err != nil || got != nil { + t.Fatal("empty history became zero", got, err) + } + sourceSeries(t, s, base, []float64{1000, 1000, 1000, 1000, 1000}) + path := s.mainDBPath + if err := s.Close(); err != nil { + t.Fatal(err) + } + reopened, err := Open(path) + if err != nil { + t.Fatal(err) + } + defer reopened.Close() + if err := reopened.EnableHistoryAggregation(); err != nil { + t.Fatal(err) + } + sourceSeries(t, reopened, base+600000, []float64{1000, 1000, 1000, 1000, 1000}) + got, err := reopened.HouseholdEnergySubtotal(context.Background(), base, base+900000) + if err != nil || got == nil || got.CoverageMS != 600000 { + t.Fatal(got, err) + } + near(t, "restart subtotal", got.Wh, 1000.0/6) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := reopened.HouseholdEnergySubtotal(ctx, base, base+900000); err == nil { + t.Fatal("cancelled diagnostic read succeeded") + } +} diff --git a/go/internal/state/store.go b/go/internal/state/store.go index 0ab26cafb..0dd3f9a8d 100644 --- a/go/internal/state/store.go +++ b/go/internal/state/store.go @@ -1505,6 +1505,11 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, return nil, err } sourceMode = sourceMode && untilMs >= cut + if !sourceMode { + if err := s.history.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM history_load_estimates WHERE start_ms>=? AND start_ms 0 { @@ -1615,6 +1620,10 @@ func (s *Store) loadArchiveHistory(ctx context.Context, sinceMs, untilMs int64, return nil, err } } + all, err = s.applyHistoricalLoadChart(ctx, all, queryFrom, untilMs, bucketMs) + if err != nil { + return nil, err + } return all, nil } @@ -1633,6 +1642,10 @@ type DayEnergy struct { BatChargedWh float64 BatDischargedWh float64 LoadWh float64 + // Historical repairs use completed minute evidence, not an extrapolated + // current tail. Status exposes this bound independently of live power. + LoadUntilMS int64 + PVUntilMS int64 // Intervals is the number of integration intervals that contributed to // the totals — i.e. the count of rows that had a predecessor (prev_ts IS // NOT NULL). Zero means there was at most one history row in the range, so @@ -1678,7 +1691,14 @@ func (s *Store) DailyEnergy(sinceMs, untilMs int64) (DayEnergy, error) { return DayEnergy{}, err } if !ok || untilMs <= from { - return s.legacyDailyEnergy(sinceMs, untilMs) + d, err := s.legacyDailyEnergy(sinceMs, untilMs) + if err == nil { + err = s.applyHistoricalDailyLoad(ctx, sinceMs, untilMs, &d) + if err == nil { + err = s.applyEndpointDailyPV(ctx, sinceMs, untilMs, &d) + } + } + return d, err } out, err := s.aggregateDayEnergy(ctx, max(from, sinceMs), untilMs) if err != nil { @@ -1691,6 +1711,12 @@ func (s *Store) DailyEnergy(sinceMs, untilMs int64) (DayEnergy, error) { } out = addDayEnergy(old, out) } + if err := s.applyEndpointDailyPV(ctx, sinceMs, untilMs, &out); err != nil { + return out, err + } + if err := s.applyHistoricalDailyLoad(ctx, sinceMs, untilMs, &out); err != nil { + return out, err + } return out, nil } diff --git a/web/components/energy-flow-readings.js b/web/components/energy-flow-readings.js index d32ad6092..dfb82f8df 100644 --- a/web/components/energy-flow-readings.js +++ b/web/components/energy-flow-readings.js @@ -252,7 +252,7 @@ export function flowReadingsFromStatus(status, opts) { // quiet meter blanks % SELF-POWERED NOW (unknown instantaneous load) // but must not hide a day that already happened. let selfPoweredPctToday = null; - if (loadWh > 1 && importWh !== null && today.history_quality !== "estimated") { + if (loadWh > 1 && importWh !== null && today.history_quality !== "estimated" && today.load_quality !== "estimated") { selfPoweredPctToday = Math.max(0, Math.min(100, (1 - importWh / loadWh) * 100)); } diff --git a/web/energy-today-subtotal.test.mjs b/web/energy-today-subtotal.test.mjs new file mode 100644 index 000000000..f26891f29 --- /dev/null +++ b/web/energy-today-subtotal.test.mjs @@ -0,0 +1,47 @@ +import assert from "node:assert/strict"; +import {test} from "node:test"; +import {energyTodayReading, formatEnergyToday, updateEnergyToday} from "./energy-today.js"; + +const subtotal = {wh: 26603, coverage_ms: 48900000, period_ms: 50400000, + minimum_bucket_ms: 300000, quality: "estimated", method: "accepted_interval_balance_v1"}; +const incomplete = {load_wh: null, history_quality: "incomplete", + measured: {load_wh: null}, load_subtotal: subtotal}; + +test("a source-integrated partial day displays its own estimated subtotal, never a complete total", () => { + assert.equal(formatEnergyToday(incomplete, "load_wh"), "≈ 26.6 kWh (partial)"); + assert.equal(incomplete.load_wh, null); + assert.equal(incomplete.measured.load_wh, null); + assert.equal(formatEnergyToday({...incomplete, load_wh: 30000, history_quality: "estimated"}, "load_wh"), "≈ 30.0 kWh"); + assert.equal(formatEnergyToday({...incomplete, load_wh: 0}, "load_wh"), "0.00 kWh"); + assert.equal(formatEnergyToday({...incomplete, load_subtotal: {...subtotal, wh: 0}}, "load_wh"), "≈ 0.00 kWh (partial)"); + assert.equal(formatEnergyToday(incomplete, "pv_wh"), "Unknown"); +}); + +test("missing or malformed interval evidence cannot manufacture a partial zero", () => { + for (const invalid of [undefined, null, {...subtotal, wh: NaN}, {...subtotal, wh: -1}, + {...subtotal, coverage_ms: 0}, {...subtotal, coverage_ms: subtotal.period_ms + 1}, + {...subtotal, period_ms: "50400000"}, {...subtotal, quality: "measured"}, + {...subtotal, method: "interpolated"}]) { + assert.equal(energyTodayReading({...incomplete, load_subtotal: invalid}, "load_wh").wh, null); + } +}); + +test("cards explain accepted coverage and clear the partial state after recovery or missing data", () => { + const elements = new Map(); + const group = {append(el) { elements.set(el.id, el); }}; + for (const id of ["e-load", "e-pv"]) elements.set(id, {closest() { return group; }}); + const root = {getElementById(id) { return elements.get(id); }, createElement() { return {}; }}; + updateEnergyToday(incomplete, root); + assert.equal(elements.get("e-load").textContent, "≈ 26.6 kWh (partial)"); + assert.match(elements.get("e-load").title, /gaps are excluded/); + assert.match(elements.get("energy-today-quality").textContent, /815 of 840 min/); + const complete = {load_wh: 30000, pv_wh: 10000, import_wh: 10000, export_wh: 0, + bat_charged_wh: 0, bat_discharged_wh: 0}; + updateEnergyToday(complete, root); + assert.equal(elements.get("e-load").textContent, "30.0 kWh"); + assert.equal(elements.get("e-load").title, ""); + assert.equal(elements.get("energy-today-quality").hidden, true); + updateEnergyToday(null, root); + assert.equal(elements.get("e-load").textContent, "Unknown"); + assert.equal(elements.get("energy-today-quality").textContent, "Today's energy is not available yet."); +}); diff --git a/web/energy-today.js b/web/energy-today.js index 4b37e2ed0..1de2322b5 100644 --- a/web/energy-today.js +++ b/web/energy-today.js @@ -16,10 +16,18 @@ function knownWh(value) { export function energyTodayReading(today, key) { const total = today?.[key]; - if (knownWh(total)) return key === "load_wh" && today?.history_quality === "estimated" + if (knownWh(total)) return ((key === "load_wh" && (today?.history_quality === "estimated" || today?.load_quality === "estimated")) || (key === "pv_wh" && today?.pv_quality === "estimated")) ? {wh: total, partial: false, estimated: true} : {wh: total, partial: false}; const measured = today?.measured?.[key]; if (knownWh(measured)) return {wh: measured, partial: true}; + const subtotal = today?.load_subtotal; + if (key === "load_wh" && knownWh(subtotal?.wh) && subtotal.quality === "estimated" && + subtotal.method === "accepted_interval_balance_v1" && + Number.isFinite(subtotal.coverage_ms) && Number.isFinite(subtotal.period_ms) && + subtotal.coverage_ms > 0 && subtotal.coverage_ms <= subtotal.period_ms) { + return {wh: subtotal.wh, partial: true, estimated: true, + coverageMs: subtotal.coverage_ms, periodMs: subtotal.period_ms}; + } return {wh: null, partial: false}; } @@ -28,7 +36,7 @@ export function formatEnergyToday(today, key, withUnit = true) { if (wh === null) return "Unknown"; const kwh = wh / 1000; const digits = kwh >= 100 ? 0 : kwh >= 10 ? 1 : 2; - return `${estimated ? "≈ " : partial ? "≥ " : ""}${kwh.toFixed(digits)}${withUnit ? " kWh" : ""}`; + return `${estimated ? "≈ " : partial ? "≥ " : ""}${kwh.toFixed(digits)}${withUnit ? " kWh" : ""}${estimated && partial ? " (partial)" : ""}`; } export function updateEnergyToday(today, root = document) { @@ -37,7 +45,8 @@ export function updateEnergyToday(today, root = document) { if (!el) continue; const {partial, estimated} = energyTodayReading(today, key); el.textContent = formatEnergyToday(today, key); - el.title = estimated ? "Estimated from source energy during brief balance failures." + el.title = estimated && partial ? "Estimated subtotal from accepted energy intervals only; gaps are excluded." + : estimated ? "Estimated from source energy; may include audited historical corrections." : partial ? "Recorded energy only; missing readings are excluded." : ""; } const group = root.getElementById("e-pv")?.closest(".energy-group"); @@ -51,10 +60,16 @@ export function updateEnergyToday(today, root = document) { } const keys = new Set(Object.values(ENERGY_FIELDS)); const incomplete = [...keys].some(key => !knownWh(today?.[key])); - const estimated = today?.history_quality === "estimated"; + const estimated = today?.history_quality === "estimated" || today?.load_quality === "estimated"; note.hidden = !incomplete && !estimated; note.textContent = today ? "Gaps in today's readings. ≥ shows recorded energy only." : "Today's energy is not available yet."; - if (estimated) note.textContent = (incomplete ? note.textContent + " " : "") + "Home use includes estimates for brief balance failures."; + if (estimated) note.textContent = (incomplete ? note.textContent + " " : "") + "Home use includes estimates from source energy."; + if (today?.pv_quality === "estimated") {note.hidden = false; note.textContent += " Solar includes estimated energy across historical gaps.";} + const home = energyTodayReading(today, "load_wh"); + if (home.estimated && home.partial) { + note.hidden = false; + note.textContent += ` Home use: estimated subtotal for ${Math.floor(home.coverageMs / 60000)} of ${Math.ceil(home.periodMs / 60000)} min; gaps excluded.`; + } } diff --git a/web/history-repair-quality.test.mjs b/web/history-repair-quality.test.mjs new file mode 100644 index 000000000..be41ec5c6 --- /dev/null +++ b/web/history-repair-quality.test.mjs @@ -0,0 +1,18 @@ +import assert from "node:assert/strict"; +import {test} from "node:test"; +import {formatEnergyToday} from "./energy-today.js"; +import {flowReadingsFromStatus} from "./components/energy-flow-readings.js"; + +test("corrected household energy stays estimated even when another original flow is missing",()=>{ + const today={load_wh:28000,load_quality:"estimated",history_quality:"incomplete",pv_wh:null,import_wh:12000,measured:{load_wh:null}}; + assert.equal(formatEnergyToday(today,"load_wh"),"≈ 28.0 kWh"); + assert.equal(formatEnergyToday(today,"pv_wh"),"Unknown"); + assert.equal(flowReadingsFromStatus({load_w:1200,energy:{today}}).selfPoweredPctToday,null); + assert.equal(today.measured.load_wh,null); +}); + +test("endpoint solar and home totals both show estimate markers",()=>{ + const today={load_wh:41000,load_quality:"estimated",pv_wh:37000,pv_quality:"estimated",history_quality:"incomplete",measured:{load_wh:null,pv_wh:null}}; + assert.equal(formatEnergyToday(today,"load_wh"),"≈ 41.0 kWh"); + assert.equal(formatEnergyToday(today,"pv_wh"),"≈ 37.0 kWh"); +});