From 62d102c33515e63719cfc44d141a0436578d7afd Mon Sep 17 00:00:00 2001 From: Florian Schmidt Date: Sat, 18 Jul 2026 18:45:57 +0200 Subject: [PATCH] Ingestion: derive consumption on ingest, and poll HA in minutes not seconds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Live ingestion wrote the raw reading and stopped there. Import, the REST push endpoint and the meter editor all recompute afterwards; the MQTT/Tasmota/HA path was the one that did not, so a polled reading landed in `reading` and every derived figure stayed frozen at the last import. Observed on a GenerationCounter: 45 readings, 44 consumption rows, generation pinned to the register value of the last imported reading. Recompute inline rather than behind a debounce. Normalizing a whole meter is cheap at metering cadence and a background dirty-set worker is machinery this does not yet need; the remark on RenormalizeAsync records when it would. Fixes a latent bug this surfaced in NormalizationService: ExecuteDelete drops the consumption rows in the database but leaves them in the change tracker, so a second recompute on the same context threw an identity conflict on (meter, time, kind). One worker scope ingesting two readings was enough to hit it. Detach the stale entries after the delete. Poll interval is now minutes, default 60, replacing seconds/60. A meter answers "how much this month, what will it cost" — an hourly sample answers that exactly as well as a per-second one, with far less raw volume (SDD §5.5). The `pollSeconds` key no longer binds, so existing sources fall back to the 60 default and move from every-60-seconds to hourly, which is the intent. A source that had deliberately set e.g. 300 seconds also lands on 60 minutes. Two test cleanups now delete consumption before the meter: live ingestion never produced any before, so the FK had nothing to trip on. Claude-Session: https://claude.ai/code/session_01V6joyergfvVLFEizH1hJLd --- docs/wiring.md | 8 ++-- src/App/Components/Pages/MeterDetail.razor | 11 +++-- .../Ingestion/HomeAssistantWorker.cs | 2 +- .../Ingestion/IngestionService.cs | 34 +++++++++++++- src/Infrastructure/Ingestion/SourceConfig.cs | 16 +++++-- .../Normalization/NormalizationService.cs | 11 +++++ tests/Integration.Tests/ApiTests.cs | 1 + .../HomeAssistantWebSocketWorkerTests.cs | 5 +++ .../Ingestion/IngestionServiceTests.cs | 45 ++++++++++++++++--- 9 files changed, 114 insertions(+), 19 deletions(-) diff --git a/docs/wiring.md b/docs/wiring.md index 04f1325..b710fd2 100644 --- a/docs/wiring.md +++ b/docs/wiring.md @@ -55,11 +55,13 @@ it as an MQTT source (above). No HA endpoint needed. and a `HomeAssistant` source on the meter: ```json -{ "entityId": "sensor.house_power", "attribute": null, "pollSeconds": 60 } +{ "entityId": "sensor.house_power", "attribute": null, "pollMinutes": 60 } ``` -Set `HA_TOKEN` (a long-lived access token) in the environment. Numeric state (or a named -`attribute`) is read every `pollSeconds`; `unavailable`/`unknown` states are skipped. +Set `HA_TOKEN` (a long-lived access token) in the environment, or type the token into the connector +dialog to have it encrypted at rest (SDD §6.4). Numeric state (or a named `attribute`) is read every +`pollMinutes` — default 60, because monthly totals and cost are identical whether a meter is sampled +hourly or per-second. `unavailable`/`unknown` states are skipped. **C — HA pushes to the REST API.** POST to `/api/v1/readings` with an `X-Api-Key` header (see the README). Good when HA should drive the cadence. diff --git a/src/App/Components/Pages/MeterDetail.razor b/src/App/Components/Pages/MeterDetail.razor index ec33cfa..ad009d6 100644 --- a/src/App/Components/Pages/MeterDetail.razor +++ b/src/App/Components/Pages/MeterDetail.razor @@ -243,7 +243,10 @@ else { - + + + Hourly is plenty for a meter — monthly totals and cost come out identical, with far less raw data. + } else if (_sourceEdit.SourceType is SourceType.Mqtt or SourceType.Tasmota) { @@ -332,7 +335,7 @@ else IsEnabled = source.IsEnabled, EntityId = config.EntityId, Attribute = config.Attribute, - PollSeconds = config.PollSeconds, + PollMinutes = config.PollMinutes, Topic = config.Topic, Path = config.Path, TimePath = config.TimePath, @@ -369,7 +372,7 @@ else { EntityId = Trim(_sourceEdit.EntityId), Attribute = Trim(_sourceEdit.Attribute), - PollSeconds = _sourceEdit.PollSeconds, + PollMinutes = _sourceEdit.PollMinutes, Topic = Trim(_sourceEdit.Topic), Path = Trim(_sourceEdit.Path), TimePath = Trim(_sourceEdit.TimePath), @@ -477,7 +480,7 @@ else public bool IsEnabled { get; set; } = true; public string? EntityId { get; set; } public string? Attribute { get; set; } - public int? PollSeconds { get; set; } = 60; + public int? PollMinutes { get; set; } = 60; public string? Topic { get; set; } public string? Path { get; set; } public string? TimePath { get; set; } diff --git a/src/Infrastructure/Ingestion/HomeAssistantWorker.cs b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs index cc32280..db41f11 100644 --- a/src/Infrastructure/Ingestion/HomeAssistantWorker.cs +++ b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs @@ -94,7 +94,7 @@ public sealed class HomeAssistantWorker( } var config = SourceConfig.Parse(source.Config); - var interval = TimeSpan.FromSeconds(Math.Max(5, config.PollSeconds ?? 60)); + var interval = TimeSpan.FromMinutes(Math.Max(1, config.PollMinutes ?? 60)); _nextPoll[source.Id] = now + interval; await PollSourceAsync(client, ingestion, endpoint, source, config, cancellationToken).ConfigureAwait(false); diff --git a/src/Infrastructure/Ingestion/IngestionService.cs b/src/Infrastructure/Ingestion/IngestionService.cs index f86b10d..b56d475 100644 --- a/src/Infrastructure/Ingestion/IngestionService.cs +++ b/src/Infrastructure/Ingestion/IngestionService.cs @@ -19,12 +19,14 @@ public enum IngestionOutcome /// unless an active reset/swap event explains them. Updates the source's last-seen status. /// Consumption normalization is recomputed separately (batch/scheduled), not per message. /// -public sealed class IngestionService(MeterVaultDbContext db) +public sealed class IngestionService( + MeterVaultDbContext db, Normalization.NormalizationService normalization) { private static readonly HashSet MonotonicModes = [MeterMode.CumulativeCounter, MeterMode.GenerationCounter, MeterMode.RuntimeCounter]; private readonly MeterVaultDbContext _db = db; + private readonly Normalization.NormalizationService _normalization = normalization; /// Ingests through a configured source (MQTT/HA workers): applies scale/offset and updates source status. public async Task IngestAsync( @@ -56,6 +58,7 @@ public sealed class IngestionService(MeterVaultDbContext db) var outcome = await UpsertAsync(meter, utc, value, source.Id, cancellationToken).ConfigureAwait(false); await UpdateSourceStatusAsync(source, utc, value, "ok", cancellationToken).ConfigureAwait(false); + await RenormalizeAsync(meter.Id, outcome, cancellationToken).ConfigureAwait(false); return outcome; } @@ -79,9 +82,38 @@ public sealed class IngestionService(MeterVaultDbContext db) var outcome = await UpsertAsync(meter, utc, value, sourceId: null, cancellationToken).ConfigureAwait(false); await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + await RenormalizeAsync(meter.Id, outcome, cancellationToken).ConfigureAwait(false); return outcome; } + /// + /// Derives consumption from the reading just written. Without this a live-ingested reading sits + /// in reading forever and every derived figure — consumption, generation, cost — stays + /// frozen at the last import, because nothing else recomputes that meter. + /// + /// + /// Recomputes inline rather than on a debounce. + /// rewrites a meter's whole consumption series, which is cheap at metering cadence — HA polls + /// hourly — but would be wasteful under a chatty MQTT source publishing every few seconds. If + /// such a source is ever added, batch this behind a dirty-set worker rather than making the + /// normalizer incremental: consumption being a pure function of readings + events is what makes + /// it reproducible. + /// + private async Task RenormalizeAsync( + int meterId, IngestionOutcome outcome, CancellationToken cancellationToken) + { + // A rejected decrease changed nothing, so the existing series is still correct. + if (outcome is not (IngestionOutcome.Written or IngestionOutcome.Updated)) + { + return; + } + + // The reading must already be persisted: RecomputeMeterAsync re-reads the meter's readings + // from the database, so anything still pending in the change tracker would be missed. + await _normalization.RecomputeMeterAsync(meterId, batchId: null, cancellationToken).ConfigureAwait(false); + await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } + private async Task UpsertAsync( Meter meter, DateTimeOffset utc, double value, int? sourceId, CancellationToken cancellationToken) { diff --git a/src/Infrastructure/Ingestion/SourceConfig.cs b/src/Infrastructure/Ingestion/SourceConfig.cs index 2c4b6b3..946ee5c 100644 --- a/src/Infrastructure/Ingestion/SourceConfig.cs +++ b/src/Infrastructure/Ingestion/SourceConfig.cs @@ -6,7 +6,7 @@ namespace MeterVault.Infrastructure.Ingestion; /// /// The parsed JSON. Which fields matter depends on the /// source type: MQTT/Tasmota use //; -/// Home Assistant uses //. +/// Home Assistant uses //. /// public sealed record SourceConfig { @@ -31,8 +31,18 @@ public sealed record SourceConfig /// Home Assistant attribute name; null reads the entity state. public string? Attribute { get; init; } - /// Home Assistant REST poll interval in seconds (fallback when not using WebSocket push). - public int? PollSeconds { get; init; } + /// + /// Home Assistant REST poll interval in minutes (fallback when not using WebSocket + /// push). Default 60. + /// + /// + /// Deliberately not seconds. A meter is read to answer "how much did I use this month, and what + /// will it cost" — questions an hourly sample answers exactly as well as a per-second one, at a + /// fraction of the raw volume (SDD §5.5). This field replaced pollSeconds; the old key no + /// longer binds, so sources written before the change fall back to the 60 default and are read + /// hourly instead of every 60 seconds. + /// + public int? PollMinutes { get; init; } public static SourceConfig Parse(string? json) { diff --git a/src/Infrastructure/Normalization/NormalizationService.cs b/src/Infrastructure/Normalization/NormalizationService.cs index d71f037..2bc90e3 100644 --- a/src/Infrastructure/Normalization/NormalizationService.cs +++ b/src/Infrastructure/Normalization/NormalizationService.cs @@ -46,6 +46,17 @@ public sealed class NormalizationService(MeterVaultDbContext db, INormalizationE await _db.Consumption.Where(c => c.MeterId == meterId) .ExecuteDeleteAsync(cancellationToken).ConfigureAwait(false); + // ExecuteDelete goes straight to the database and leaves the change tracker untouched, so + // rows added by an earlier recompute on this context are still tracked but no longer exist. + // Detach them, or re-adding the same (meter, time, kind) key throws an identity conflict — + // which is what happens when one context recomputes a meter twice, e.g. a worker ingesting + // two readings in a single scope. + foreach (var stale in _db.ChangeTracker.Entries() + .Where(e => e.Entity.MeterId == meterId).ToList()) + { + stale.State = EntityState.Detached; + } + foreach (var row in consumption) { row.ImportBatchId = batchId; diff --git a/tests/Integration.Tests/ApiTests.cs b/tests/Integration.Tests/ApiTests.cs index c713598..48cb13f 100644 --- a/tests/Integration.Tests/ApiTests.cs +++ b/tests/Integration.Tests/ApiTests.cs @@ -47,6 +47,7 @@ public sealed class ApiTests(TimescaleFixture fx) { var reading = await db.Readings.SingleAsync(r => r.MeterId == meterId); Assert.Equal(1500, reading.Value, 3); + await db.Consumption.Where(c => c.MeterId == meterId).ExecuteDeleteAsync(); await db.Readings.Where(r => r.MeterId == meterId).ExecuteDeleteAsync(); await db.Meters.Where(m => m.Id == meterId).ExecuteDeleteAsync(); } diff --git a/tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs b/tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs index f7e0132..413a4b3 100644 --- a/tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs +++ b/tests/Integration.Tests/Ingestion/HomeAssistantWebSocketWorkerTests.cs @@ -92,6 +92,7 @@ public sealed class HomeAssistantWebSocketWorkerTests(TimescaleFixture fx) finally { Environment.SetEnvironmentVariable(TokenEnvVar, null); + await db.Consumption.ExecuteDeleteAsync(); await db.Readings.ExecuteDeleteAsync(); await db.MeterSources.ExecuteDeleteAsync(); await db.IngestionEndpoints.ExecuteDeleteAsync(); @@ -109,6 +110,10 @@ public sealed class HomeAssistantWebSocketWorkerTests(TimescaleFixture fx) .UseSnakeCaseNamingConvention()); services.AddScoped(sp => sp.GetRequiredService>().CreateDbContext()); services.AddScoped(); + // Ingestion derives consumption inline, so the normalizer has to be resolvable here too. + services.AddSingleton( + _ => MeterVault.Core.Normalization.NormalizationEngine.CreateDefault()); + services.AddScoped(); // Ephemeral keys: this test's token comes from an env var, so nothing needs to outlive the run. services.AddSingleton( new Microsoft.AspNetCore.DataProtection.EphemeralDataProtectionProvider()); diff --git a/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs b/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs index f6a8df5..873db15 100644 --- a/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs +++ b/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs @@ -1,4 +1,4 @@ -using System.Text.Json; +using System.Text.Json; using MeterVault.Core.Domain; using MeterVault.Infrastructure.Ingestion; using MeterVault.Infrastructure.Persistence; @@ -17,7 +17,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) { await using var db = fx.CreateContext(); var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter, scale: 0.001, offset: 0); - var service = new IngestionService(db); + var service = NewIngestion(db); // 1000 raw × 0.001 = 1.0. Assert.Equal(IngestionOutcome.Written, await service.IngestAsync(sourceId, T0, 1000)); @@ -38,7 +38,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) { await using var db = fx.CreateContext(); var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter); - var service = new IngestionService(db); + var service = NewIngestion(db); await service.IngestAsync(sourceId, T0, 500); var outcome = await service.IngestAsync(sourceId, T0.AddHours(1), 400); // decrease, no event @@ -54,7 +54,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) { await using var db = fx.CreateContext(); var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter); - var service = new IngestionService(db); + var service = NewIngestion(db); // A reset early on explains an early decrease... await service.IngestAsync(sourceId, T0, 100); @@ -76,7 +76,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) { await using var db = fx.CreateContext(); var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter); - var service = new IngestionService(db); + var service = NewIngestion(db); await service.IngestAsync(sourceId, T0, 500); db.MeterEvents.Add(new MeterEvent @@ -103,7 +103,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) var brokerId = await CreateBrokerAsync(db); var (meterId, _) = await SetupAsync( db, MeterMode.CumulativeCounter, topic: "tele/plug7/SENSOR", endpointId: brokerId); - var router = new MqttMessageRouter(db, new IngestionService(db), NullLogger.Instance); + var router = new MqttMessageRouter(db, NewIngestion(db), NullLogger.Instance); var routed = await router.RouteAsync( brokerId, @@ -129,7 +129,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) // separating them. var (meterId, _) = await SetupAsync( db, MeterMode.CumulativeCounter, topic: "tele/+/SENSOR", endpointId: brokerB); - var router = new MqttMessageRouter(db, new IngestionService(db), NullLogger.Instance); + var router = new MqttMessageRouter(db, NewIngestion(db), NullLogger.Instance); var routed = await router.RouteAsync( brokerA, @@ -148,6 +148,36 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) await CleanupAsync(db, meterId); } + [Fact] + public async Task Ingesting_a_reading_derives_consumption_without_a_separate_recompute() + { + // Regression: live ingestion used to write only the raw reading, so consumption/generation + // stayed frozen at the last import until something else recomputed the meter. + await using var db = fx.CreateContext(); + var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter); + var service = NewIngestion(db); + + await service.IngestAsync(sourceId, T0, 1000); + await service.IngestAsync(sourceId, T0.AddHours(1), 1250); + + var consumption = await db.Consumption.AsNoTracking() + .Where(c => c.MeterId == meterId) + .OrderBy(c => c.Time) + .ToListAsync(); + + // The first reading is anchored against the meter's baseline (0), so it contributes 1000; + // what proves the fix is the second reading's 250 delta being there at all. + Assert.Equal(2, consumption.Count); + Assert.Equal(250d, consumption.Single(c => c.Time == T0.AddHours(1)).Amount, 3); + Assert.Equal(1250d, consumption.Sum(c => c.Amount), 3); + + await CleanupAsync(db, meterId); + } + + private static IngestionService NewIngestion(MeterVaultDbContext db) => + new(db, new MeterVault.Infrastructure.Normalization.NormalizationService( + db, MeterVault.Core.Normalization.NormalizationEngine.CreateDefault())); + private static async Task<(int MeterId, int SourceId)> SetupAsync( MeterVaultDbContext db, MeterMode mode, double scale = 1, double offset = 0, string topic = "tele/x/SENSOR", string? path = "ENERGY.Total", int? endpointId = null) @@ -198,6 +228,7 @@ public sealed class IngestionServiceTests(TimescaleFixture fx) { await db.Readings.Where(r => r.MeterId == meterId).ExecuteDeleteAsync(); await db.MeterEvents.Where(e => e.MeterId == meterId).ExecuteDeleteAsync(); + await db.Consumption.Where(c => c.MeterId == meterId).ExecuteDeleteAsync(); await db.Meters.Where(m => m.Id == meterId).ExecuteDeleteAsync(); await db.IngestionEndpoints.Where(e => e.Name.StartsWith("broker-")).ExecuteDeleteAsync(); }