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(); }