diff --git a/Directory.Packages.props b/Directory.Packages.props index 05c3e74..de217f0 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -14,6 +14,7 @@ + diff --git a/src/App/Program.cs b/src/App/Program.cs index fefd949..de49a1e 100644 --- a/src/App/Program.cs +++ b/src/App/Program.cs @@ -26,6 +26,13 @@ try ?? "Host=localhost;Port=5432;Database=metervault;Username=metervault;Password=metervault"; builder.Services.AddMeterVaultInfrastructure(connectionString); + var options = builder.Configuration.GetSection(MeterVaultOptions.SectionName).Get() + ?? new MeterVaultOptions(); + if (options.EnableLiveIngestion) + { + builder.Services.AddMeterVaultIngestion(); + } + builder.Services.AddRazorComponents() .AddInteractiveServerComponents(); diff --git a/src/Infrastructure/DependencyInjection.cs b/src/Infrastructure/DependencyInjection.cs index d196a80..b68b7c4 100644 --- a/src/Infrastructure/DependencyInjection.cs +++ b/src/Infrastructure/DependencyInjection.cs @@ -1,5 +1,6 @@ using MeterVault.Core.Normalization; using MeterVault.Infrastructure.Import; +using MeterVault.Infrastructure.Ingestion; using MeterVault.Infrastructure.Normalization; using MeterVault.Infrastructure.Persistence; using Microsoft.EntityFrameworkCore; @@ -24,7 +25,21 @@ public static class DependencyInjection services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddScoped(); + services.AddScoped(); return services; } + + /// + /// Registers the live-ingestion background workers (MQTT/Tasmota + Home Assistant). Kept + /// separate from so tests can opt out of brokers. + /// + public static IServiceCollection AddMeterVaultIngestion(this IServiceCollection services) + { + services.AddHttpClient(); + services.AddHostedService(); + services.AddHostedService(); + return services; + } } diff --git a/src/Infrastructure/Ingestion/EndpointConfig.cs b/src/Infrastructure/Ingestion/EndpointConfig.cs new file mode 100644 index 0000000..03dd666 --- /dev/null +++ b/src/Infrastructure/Ingestion/EndpointConfig.cs @@ -0,0 +1,55 @@ +using System.Text.Json; +using System.Text.Json.Serialization; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// The parsed JSON for an MQTT broker. Secrets +/// are stored by reference only (SDD §6.4): / +/// name environment variables resolved at runtime, never plaintext credentials in the database. +/// +public sealed record EndpointConfig +{ + private static readonly JsonSerializerOptions Options = new() + { + PropertyNameCaseInsensitive = true, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, + }; + + public string Host { get; init; } = "localhost"; + + public int Port { get; init; } = 1883; + + public bool Tls { get; init; } + + /// Extra topic filters to subscribe (beyond the sources' own topics), e.g. tele/+/SENSOR. + public IReadOnlyList ExtraTopics { get; init; } = []; + + public string? UsernameEnv { get; init; } + + public string? PasswordEnv { get; init; } + + public static EndpointConfig Parse(string? json) + { + if (string.IsNullOrWhiteSpace(json)) + { + return new EndpointConfig(); + } + + try + { + return JsonSerializer.Deserialize(json, Options) ?? new EndpointConfig(); + } + catch (JsonException) + { + return new EndpointConfig(); + } + } + + public string? ResolveUsername() => Resolve(UsernameEnv); + + public string? ResolvePassword() => Resolve(PasswordEnv); + + private static string? Resolve(string? envVarName) => + string.IsNullOrWhiteSpace(envVarName) ? null : Environment.GetEnvironmentVariable(envVarName); +} diff --git a/src/Infrastructure/Ingestion/HaStateClient.cs b/src/Infrastructure/Ingestion/HaStateClient.cs new file mode 100644 index 0000000..21821ad --- /dev/null +++ b/src/Infrastructure/Ingestion/HaStateClient.cs @@ -0,0 +1,78 @@ +using System.Globalization; +using System.Net.Http.Headers; +using System.Text.Json; + +namespace MeterVault.Infrastructure.Ingestion; + +/// A resolved Home Assistant state: numeric value and the entity's last-updated time. +public readonly record struct HaState(double Value, DateTimeOffset Time); + +/// +/// Reads a single Home Assistant entity state over the REST API +/// (GET /api/states/{entity_id} with a long-lived bearer token) and parses the numeric +/// state or a named attribute (SDD §6.2). This is the poll fallback to the WebSocket push path. +/// +public sealed class HaStateClient(HttpClient httpClient) +{ + private readonly HttpClient _httpClient = httpClient; + + public async Task GetStateAsync( + string baseUrl, string token, string entityId, string? attribute, + CancellationToken cancellationToken = default) + { + var url = $"{baseUrl.TrimEnd('/')}/api/states/{entityId}"; + using var request = new HttpRequestMessage(HttpMethod.Get, url); + request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", token); + + using var response = await _httpClient.SendAsync(request, cancellationToken).ConfigureAwait(false); + response.EnsureSuccessStatusCode(); + + await using var stream = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false); + return ParseState(await JsonDocument.ParseAsync(stream, cancellationToken: cancellationToken).ConfigureAwait(false), attribute); + } + + internal static HaState? ParseState(JsonDocument document, string? attribute) + { + var root = document.RootElement; + + double value; + if (string.IsNullOrWhiteSpace(attribute)) + { + if (!root.TryGetProperty("state", out var state) || !TryReadNumber(state, out value)) + { + return null; // 'unavailable', 'unknown', or non-numeric. + } + } + else + { + if (!root.TryGetProperty("attributes", out var attrs) + || !attrs.TryGetProperty(attribute, out var attr) + || !TryReadNumber(attr, out value)) + { + return null; + } + } + + var time = DateTimeOffset.UtcNow; + if (root.TryGetProperty("last_updated", out var lastUpdated) + && lastUpdated.ValueKind == JsonValueKind.String + && DateTimeOffset.TryParse(lastUpdated.GetString(), CultureInfo.InvariantCulture, + DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal, out var parsed)) + { + time = parsed; + } + + return new HaState(value, time); + } + + private static bool TryReadNumber(JsonElement element, out double value) + { + value = 0; + return element.ValueKind switch + { + JsonValueKind.Number => element.TryGetDouble(out value), + JsonValueKind.String => double.TryParse(element.GetString(), NumberStyles.Float, CultureInfo.InvariantCulture, out value), + _ => false, + }; + } +} diff --git a/src/Infrastructure/Ingestion/HomeAssistantWorker.cs b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs new file mode 100644 index 0000000..724a415 --- /dev/null +++ b/src/Infrastructure/Ingestion/HomeAssistantWorker.cs @@ -0,0 +1,145 @@ +using System.Collections.Concurrent; +using System.Text.Json; +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// Polls Home Assistant entity states on each source's configured interval and ingests them +/// (SDD §6.2 REST fallback). Home Assistant can alternatively push to MQTT (handled by +/// ) or push to the app's REST API. Tokens are resolved from +/// environment variables named in the endpoint config — never stored in the database. +/// +public sealed class HomeAssistantWorker( + IServiceScopeFactory scopeFactory, + IHttpClientFactory httpClientFactory, + ILogger logger) : BackgroundService +{ + private static readonly TimeSpan TickInterval = TimeSpan.FromSeconds(10); + + private readonly IServiceScopeFactory _scopeFactory = scopeFactory; + private readonly IHttpClientFactory _httpClientFactory = httpClientFactory; + private readonly ILogger _logger = logger; + private readonly ConcurrentDictionary _nextPoll = new(); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + using var timer = new PeriodicTimer(TickInterval); + do + { + try + { + await PollDueSourcesAsync(stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Home Assistant poll tick failed; will retry"); + } + } + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)); + } + + private async Task PollDueSourcesAsync(CancellationToken cancellationToken) + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var ingestion = scope.ServiceProvider.GetRequiredService(); + var client = new HaStateClient(_httpClientFactory.CreateClient()); + + var endpoints = await db.IngestionEndpoints + .Where(e => e.IsEnabled && e.Type == EndpointType.HomeAssistant) + .ToDictionaryAsync(e => e.Id, cancellationToken).ConfigureAwait(false); + if (endpoints.Count == 0) + { + return; + } + + var sources = await db.MeterSources + .Where(s => s.IsEnabled && s.SourceType == SourceType.HomeAssistant && s.EndpointId != null) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + var now = DateTimeOffset.UtcNow; + foreach (var source in sources) + { + if (_nextPoll.TryGetValue(source.Id, out var due) && due > now) + { + continue; + } + + if (source.EndpointId is not { } endpointId || !endpoints.TryGetValue(endpointId, out var endpoint)) + { + continue; + } + + var config = SourceConfig.Parse(source.Config); + var interval = TimeSpan.FromSeconds(Math.Max(5, config.PollSeconds ?? 60)); + _nextPoll[source.Id] = now + interval; + + await PollSourceAsync(client, ingestion, endpoint, source, config, cancellationToken).ConfigureAwait(false); + } + } + + private async Task PollSourceAsync( + HaStateClient client, IngestionService ingestion, IngestionEndpoint endpoint, + MeterSource source, SourceConfig config, CancellationToken cancellationToken) + { + var endpointConfig = ParseHaEndpoint(endpoint.Config); + var token = string.IsNullOrWhiteSpace(endpointConfig.TokenEnv) + ? null + : Environment.GetEnvironmentVariable(endpointConfig.TokenEnv); + + if (string.IsNullOrWhiteSpace(endpointConfig.BaseUrl) || string.IsNullOrWhiteSpace(token) + || string.IsNullOrWhiteSpace(config.EntityId)) + { + return; + } + + try + { + var state = await client.GetStateAsync( + endpointConfig.BaseUrl, token, config.EntityId, config.Attribute, cancellationToken).ConfigureAwait(false); + if (state is { } value) + { + await ingestion.IngestAsync(source.Id, value.Time, value.Value, cancellationToken).ConfigureAwait(false); + } + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Home Assistant poll failed for entity {EntityId}", config.EntityId); + } + } + + private static HaEndpoint ParseHaEndpoint(string? json) + { + if (string.IsNullOrWhiteSpace(json)) + { + return new HaEndpoint(); + } + + try + { + return JsonSerializer.Deserialize(json, + new JsonSerializerOptions { PropertyNameCaseInsensitive = true }) ?? new HaEndpoint(); + } + catch (JsonException) + { + return new HaEndpoint(); + } + } + + private sealed record HaEndpoint + { + public string? BaseUrl { get; init; } + + public string? TokenEnv { get; init; } + } +} diff --git a/src/Infrastructure/Ingestion/IngestionService.cs b/src/Infrastructure/Ingestion/IngestionService.cs new file mode 100644 index 0000000..4f7f550 --- /dev/null +++ b/src/Infrastructure/Ingestion/IngestionService.cs @@ -0,0 +1,115 @@ +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; + +namespace MeterVault.Infrastructure.Ingestion; + +/// The outcome of ingesting one reading. +public enum IngestionOutcome +{ + Written, + Updated, + RejectedDecrease, + UnknownSource, +} + +/// +/// Persists a single incoming reading (SDD §6.1, FR-4): applies the source's scale/offset, is +/// idempotent on (meter_id, time), and guards monotonic registers against spurious decreases +/// 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) +{ + private static readonly HashSet MonotonicModes = + [MeterMode.CumulativeCounter, MeterMode.GenerationCounter, MeterMode.RuntimeCounter]; + + private readonly MeterVaultDbContext _db = db; + + public async Task IngestAsync( + int sourceId, DateTimeOffset time, double rawValue, CancellationToken cancellationToken = default) + { + var source = await _db.MeterSources + .FirstOrDefaultAsync(s => s.Id == sourceId, cancellationToken).ConfigureAwait(false); + if (source is null) + { + return IngestionOutcome.UnknownSource; + } + + var meter = await _db.Meters + .FirstOrDefaultAsync(m => m.Id == source.MeterId, cancellationToken).ConfigureAwait(false); + if (meter is null) + { + return IngestionOutcome.UnknownSource; + } + + var value = (rawValue * source.Scale) + source.Offset; + var utc = time.ToUniversalTime(); + + if (MonotonicModes.Contains(meter.Mode) + && await IsSpuriousDecreaseAsync(meter.Id, utc, value, cancellationToken).ConfigureAwait(false)) + { + await UpdateSourceStatusAsync(source, utc, value, "rejected: decrease", cancellationToken).ConfigureAwait(false); + return IngestionOutcome.RejectedDecrease; + } + + var existing = await _db.Readings + .FirstOrDefaultAsync(r => r.MeterId == meter.Id && r.Time == utc, cancellationToken).ConfigureAwait(false); + + IngestionOutcome outcome; + if (existing is null) + { + _db.Readings.Add(new Reading + { + MeterId = meter.Id, + Time = utc, + Value = value, + SourceId = source.Id, + Quality = ReadingQuality.Measured, + }); + outcome = IngestionOutcome.Written; + } + else + { + existing.Value = value; + existing.SourceId = source.Id; + outcome = IngestionOutcome.Updated; + } + + await UpdateSourceStatusAsync(source, utc, value, "ok", cancellationToken).ConfigureAwait(false); + return outcome; + } + + private async Task IsSpuriousDecreaseAsync( + int meterId, DateTimeOffset time, double value, CancellationToken cancellationToken) + { + var previous = await _db.Readings + .Where(r => r.MeterId == meterId && r.Time < time) + .OrderByDescending(r => r.Time) + .Select(r => (double?)r.Value) + .FirstOrDefaultAsync(cancellationToken).ConfigureAwait(false); + + if (previous is null || value >= previous.Value) + { + return false; + } + + // A reset/swap event between the previous reading and this one explains the decrease. + var explained = await _db.MeterEvents.AnyAsync( + e => e.MeterId == meterId + && (e.EventType == MeterEventType.CounterReset || e.EventType == MeterEventType.MeterSwap) + && e.Time <= time, + cancellationToken).ConfigureAwait(false); + + return !explained; + } + + private async Task UpdateSourceStatusAsync( + MeterSource source, DateTimeOffset time, double value, string status, CancellationToken cancellationToken) + { + source.LastSeenAt = time; + source.LastValue = value; + source.LastStatus = status; + await _db.SaveChangesAsync(cancellationToken).ConfigureAwait(false); + } +} diff --git a/src/Infrastructure/Ingestion/MqttIngestionWorker.cs b/src/Infrastructure/Ingestion/MqttIngestionWorker.cs new file mode 100644 index 0000000..8c674ed --- /dev/null +++ b/src/Infrastructure/Ingestion/MqttIngestionWorker.cs @@ -0,0 +1,176 @@ +using System.Collections.Concurrent; +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using MQTTnet; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// Maintains a persistent MQTTnet connection to each enabled MQTT broker endpoint, subscribes to +/// the union of enabled sources' topics (plus any extra Tasmota patterns), and routes each message +/// through (SDD §6.1, FR-4). Degrades gracefully: broker outages +/// are logged and retried on the next tick rather than crashing the app. +/// +public sealed class MqttIngestionWorker( + IServiceScopeFactory scopeFactory, ILogger logger) : BackgroundService +{ + private static readonly TimeSpan ReconnectInterval = TimeSpan.FromSeconds(15); + + private readonly IServiceScopeFactory _scopeFactory = scopeFactory; + private readonly ILogger _logger = logger; + private readonly MqttClientFactory _factory = new(); + private readonly ConcurrentDictionary _clients = new(); + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + using var timer = new PeriodicTimer(ReconnectInterval); + do + { + try + { + await EnsureConnectionsAsync(stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + break; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "MQTT ingestion tick failed; will retry"); + } + } + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)); + + foreach (var client in _clients.Values) + { + try + { + if (client.IsConnected) + { + await client.DisconnectAsync(cancellationToken: CancellationToken.None).ConfigureAwait(false); + } + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Error disconnecting MQTT client during shutdown"); + } + + client.Dispose(); + } + } + + private async Task EnsureConnectionsAsync(CancellationToken cancellationToken) + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var db = scope.ServiceProvider.GetRequiredService(); + + var endpoints = await db.IngestionEndpoints + .Where(e => e.IsEnabled && e.Type == EndpointType.MqttBroker) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + foreach (var endpoint in endpoints) + { + var client = _clients.GetOrAdd(endpoint.Id, _ => CreateClient()); + if (client.IsConnected) + { + continue; + } + + var topics = await ResolveTopicsAsync(db, endpoint, cancellationToken).ConfigureAwait(false); + await ConnectAndSubscribeAsync(endpoint, client, topics, cancellationToken).ConfigureAwait(false); + } + } + + private IMqttClient CreateClient() + { + var client = _factory.CreateMqttClient(); + client.ApplicationMessageReceivedAsync += OnMessageAsync; + return client; + } + + private async Task ConnectAndSubscribeAsync( + IngestionEndpoint endpoint, IMqttClient client, IReadOnlyList topics, CancellationToken cancellationToken) + { + var config = EndpointConfig.Parse(endpoint.Config); + var builder = new MqttClientOptionsBuilder() + .WithTcpServer(config.Host, config.Port) + .WithCleanSession(); + + var username = config.ResolveUsername(); + if (!string.IsNullOrWhiteSpace(username)) + { + builder = builder.WithCredentials(username, config.ResolvePassword() ?? string.Empty); + } + + if (config.Tls) + { + builder = builder.WithTlsOptions(o => { }); + } + + try + { + await client.ConnectAsync(builder.Build(), cancellationToken).ConfigureAwait(false); + foreach (var topic in topics) + { + await client.SubscribeAsync(topic, cancellationToken: cancellationToken).ConfigureAwait(false); + } + + _logger.LogInformation("MQTT connected to {Host}:{Port} ({TopicCount} topics)", + config.Host, config.Port, topics.Count); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "MQTT broker {Host}:{Port} unreachable; ingestion degraded", + config.Host, config.Port); + } + } + + private static async Task> ResolveTopicsAsync( + MeterVaultDbContext db, IngestionEndpoint endpoint, CancellationToken cancellationToken) + { + var sourceConfigs = await db.MeterSources + .Where(s => s.IsEnabled + && (s.SourceType == SourceType.Mqtt || s.SourceType == SourceType.Tasmota) + && (s.EndpointId == endpoint.Id || s.EndpointId == null)) + .Select(s => s.Config) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + var topics = new HashSet(StringComparer.Ordinal); + foreach (var raw in sourceConfigs) + { + var topic = SourceConfig.Parse(raw).Topic; + if (!string.IsNullOrWhiteSpace(topic)) + { + topics.Add(topic); + } + } + + foreach (var extra in EndpointConfig.Parse(endpoint.Config).ExtraTopics) + { + topics.Add(extra); + } + + return [.. topics]; + } + + private async Task OnMessageAsync(MqttApplicationMessageReceivedEventArgs args) + { + var topic = args.ApplicationMessage.Topic; + var payload = args.ApplicationMessage.ConvertPayloadToString() ?? string.Empty; + + try + { + await using var scope = _scopeFactory.CreateAsyncScope(); + var router = scope.ServiceProvider.GetRequiredService(); + await router.RouteAsync(topic, payload).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, "Failed to route MQTT message on topic {Topic}", topic); + } + } +} diff --git a/src/Infrastructure/Ingestion/MqttMessageRouter.cs b/src/Infrastructure/Ingestion/MqttMessageRouter.cs new file mode 100644 index 0000000..5eb6d62 --- /dev/null +++ b/src/Infrastructure/Ingestion/MqttMessageRouter.cs @@ -0,0 +1,54 @@ +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// Routes an incoming MQTT message to every enabled MQTT/Tasmota source whose topic filter covers +/// it, extracts the value (and payload timestamp), and ingests it (SDD §6.1). Decoupled from the +/// broker client so it can be exercised directly against the database in tests. +/// +public sealed class MqttMessageRouter( + MeterVaultDbContext db, IngestionService ingestion, ILogger logger) +{ + private readonly MeterVaultDbContext _db = db; + private readonly IngestionService _ingestion = ingestion; + private readonly ILogger _logger = logger; + + public async Task RouteAsync(string topic, string payload, CancellationToken cancellationToken = default) + { + var sources = await _db.MeterSources + .Where(s => s.IsEnabled && (s.SourceType == SourceType.Mqtt || s.SourceType == SourceType.Tasmota)) + .ToListAsync(cancellationToken).ConfigureAwait(false); + + var routed = 0; + foreach (var source in sources) + { + var config = SourceConfig.Parse(source.Config); + if (!MqttTopicMatcher.Matches(config.Topic, topic)) + { + continue; + } + + if (!PayloadExtractor.TryExtractValue(payload, config.Path, out var value)) + { + _logger.LogDebug("Source {SourceId}: no value at path '{Path}' in topic {Topic}", + source.Id, config.Path, topic); + continue; + } + + // Prefer the payload's own time; Tasmota SENSOR messages carry "Time"; else receive time. + var timePath = config.TimePath ?? (source.SourceType == SourceType.Tasmota ? "Time" : null); + var time = PayloadExtractor.TryExtractTime(payload, timePath, out var payloadTime) + ? payloadTime + : DateTimeOffset.UtcNow; + + await _ingestion.IngestAsync(source.Id, time, value, cancellationToken).ConfigureAwait(false); + routed++; + } + + return routed; + } +} diff --git a/src/Infrastructure/Ingestion/MqttTopicMatcher.cs b/src/Infrastructure/Ingestion/MqttTopicMatcher.cs new file mode 100644 index 0000000..b6d1c24 --- /dev/null +++ b/src/Infrastructure/Ingestion/MqttTopicMatcher.cs @@ -0,0 +1,47 @@ +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// Matches an MQTT topic against a subscription filter using the standard + (single level) +/// and # (multi level, trailing only) wildcards. Used to route an incoming message to the +/// meter sources whose configured topic filter covers it (e.g. tele/+/SENSOR). +/// +public static class MqttTopicMatcher +{ + public static bool Matches(string? filter, string? topic) + { + if (string.IsNullOrEmpty(filter) || string.IsNullOrEmpty(topic)) + { + return false; + } + + if (string.Equals(filter, topic, StringComparison.Ordinal)) + { + return true; + } + + var filterLevels = filter.Split('/'); + var topicLevels = topic.Split('/'); + + for (var i = 0; i < filterLevels.Length; i++) + { + var f = filterLevels[i]; + if (f == "#") + { + // '#' must be the last level and matches the remainder (including zero levels). + return i == filterLevels.Length - 1; + } + + if (i >= topicLevels.Length) + { + return false; + } + + if (f != "+" && !string.Equals(f, topicLevels[i], StringComparison.Ordinal)) + { + return false; + } + } + + return filterLevels.Length == topicLevels.Length; + } +} diff --git a/src/Infrastructure/Ingestion/PayloadExtractor.cs b/src/Infrastructure/Ingestion/PayloadExtractor.cs new file mode 100644 index 0000000..991b5ea --- /dev/null +++ b/src/Infrastructure/Ingestion/PayloadExtractor.cs @@ -0,0 +1,100 @@ +using System.Globalization; +using System.Text.Json; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// Extracts a numeric value and optional timestamp from a JSON message payload via a dot-path +/// (SDD §6.1). Tasmota energy lives under ENERGY.Total / ENERGY.Today / +/// ENERGY.Power; generic sensors are addressed by path. A bare numeric payload (no JSON) +/// is accepted when the path is empty — Home Assistant states are often just "123.4". +/// +public static class PayloadExtractor +{ + public static bool TryExtractValue(string payload, string? path, out double value) + { + value = 0; + if (string.IsNullOrWhiteSpace(payload)) + { + return false; + } + + // Bare scalar (HA state, plain sensor) when no path is given. + if (string.IsNullOrWhiteSpace(path)) + { + return TryParseNumber(payload.Trim(), out value); + } + + try + { + using var doc = JsonDocument.Parse(payload); + if (!TryNavigate(doc.RootElement, path, out var element)) + { + return false; + } + + return TryReadNumber(element, out value); + } + catch (JsonException) + { + return false; + } + } + + public static bool TryExtractTime(string payload, string? timePath, out DateTimeOffset time) + { + time = default; + if (string.IsNullOrWhiteSpace(payload) || string.IsNullOrWhiteSpace(timePath)) + { + return false; + } + + try + { + using var doc = JsonDocument.Parse(payload); + if (!TryNavigate(doc.RootElement, timePath, out var element) || element.ValueKind != JsonValueKind.String) + { + return false; + } + + var raw = element.GetString(); + return DateTimeOffset.TryParse(raw, CultureInfo.InvariantCulture, + DateTimeStyles.AssumeUniversal | DateTimeStyles.AdjustToUniversal, out time); + } + catch (JsonException) + { + return false; + } + } + + private static bool TryNavigate(JsonElement root, string path, out JsonElement result) + { + result = root; + foreach (var segment in path.Split('.', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)) + { + if (result.ValueKind != JsonValueKind.Object || !result.TryGetProperty(segment, out result)) + { + return false; + } + } + + return true; + } + + private static bool TryReadNumber(JsonElement element, out double value) + { + switch (element.ValueKind) + { + case JsonValueKind.Number: + return element.TryGetDouble(out value); + case JsonValueKind.String: + return TryParseNumber(element.GetString(), out value); + default: + value = 0; + return false; + } + } + + private static bool TryParseNumber(string? raw, out double value) => + double.TryParse(raw, NumberStyles.Float, CultureInfo.InvariantCulture, out value); +} diff --git a/src/Infrastructure/Ingestion/SourceConfig.cs b/src/Infrastructure/Ingestion/SourceConfig.cs new file mode 100644 index 0000000..2c4b6b3 --- /dev/null +++ b/src/Infrastructure/Ingestion/SourceConfig.cs @@ -0,0 +1,53 @@ +using System.Text.Json; +using System.Text.Json.Serialization; + +namespace MeterVault.Infrastructure.Ingestion; + +/// +/// The parsed JSON. Which fields matter depends on the +/// source type: MQTT/Tasmota use //; +/// Home Assistant uses //. +/// +public sealed record SourceConfig +{ + private static readonly JsonSerializerOptions Options = new() + { + PropertyNameCaseInsensitive = true, + DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull, + }; + + /// MQTT topic filter this source listens on (e.g. tele/plug1/SENSOR). + public string? Topic { get; init; } + + /// Dot-path to the value in the payload (e.g. ENERGY.Total). Null = bare scalar. + public string? Path { get; init; } + + /// Dot-path to a timestamp in the payload (Tasmota Time). Null = use receive time. + public string? TimePath { get; init; } + + /// Home Assistant entity id (e.g. sensor.house_power). + public string? EntityId { get; init; } + + /// 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; } + + public static SourceConfig Parse(string? json) + { + if (string.IsNullOrWhiteSpace(json)) + { + return new SourceConfig(); + } + + try + { + return JsonSerializer.Deserialize(json, Options) ?? new SourceConfig(); + } + catch (JsonException) + { + return new SourceConfig(); + } + } +} diff --git a/src/Infrastructure/MeterVault.Infrastructure.csproj b/src/Infrastructure/MeterVault.Infrastructure.csproj index ec58a95..0702083 100644 --- a/src/Infrastructure/MeterVault.Infrastructure.csproj +++ b/src/Infrastructure/MeterVault.Infrastructure.csproj @@ -5,6 +5,9 @@ + + + diff --git a/src/Infrastructure/Options/MeterVaultOptions.cs b/src/Infrastructure/Options/MeterVaultOptions.cs index 80990e7..f6be39f 100644 --- a/src/Infrastructure/Options/MeterVaultOptions.cs +++ b/src/Infrastructure/Options/MeterVaultOptions.cs @@ -19,6 +19,9 @@ public sealed class MeterVaultOptions /// Run EF migrations on startup. Disable for tests that migrate out-of-band. public bool RunMigrationsAtStartup { get; set; } = true; + /// Start the MQTT/Home Assistant ingestion workers. Disable for tests. + public bool EnableLiveIngestion { get; set; } = true; + /// How long full-resolution raw readings are retained (SDD §5.5, default 3 years). public int RawRetentionDays { get; set; } = 1095; } diff --git a/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs b/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs new file mode 100644 index 0000000..17c0dfb --- /dev/null +++ b/tests/Integration.Tests/Ingestion/IngestionServiceTests.cs @@ -0,0 +1,134 @@ +using System.Text.Json; +using MeterVault.Core.Domain; +using MeterVault.Infrastructure.Ingestion; +using MeterVault.Infrastructure.Persistence; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging.Abstractions; + +namespace MeterVault.Integration.Tests.Ingestion; + +[Collection("Timescale")] +public sealed class IngestionServiceTests(TimescaleFixture fx) +{ + private static readonly DateTimeOffset T0 = new(2024, 1, 1, 0, 0, 0, TimeSpan.Zero); + + [Fact] + public async Task Writes_and_updates_idempotently_with_scale_and_offset() + { + await using var db = fx.CreateContext(); + var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter, scale: 0.001, offset: 0); + var service = new IngestionService(db); + + // 1000 raw × 0.001 = 1.0. + Assert.Equal(IngestionOutcome.Written, await service.IngestAsync(sourceId, T0, 1000)); + Assert.Equal(IngestionOutcome.Updated, await service.IngestAsync(sourceId, T0, 2000)); // same time → update + + var reading = await db.Readings.SingleAsync(r => r.MeterId == meterId && r.Time == T0); + Assert.Equal(2.0, reading.Value, 6); + + var source = await db.MeterSources.SingleAsync(s => s.Id == sourceId); + Assert.Equal("ok", source.LastStatus); + Assert.Equal(2.0, source.LastValue!.Value, 6); + + await CleanupAsync(db, meterId); + } + + [Fact] + public async Task Rejects_spurious_decrease_on_a_cumulative_register() + { + await using var db = fx.CreateContext(); + var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter); + var service = new IngestionService(db); + + await service.IngestAsync(sourceId, T0, 500); + var outcome = await service.IngestAsync(sourceId, T0.AddHours(1), 400); // decrease, no event + + Assert.Equal(IngestionOutcome.RejectedDecrease, outcome); + Assert.False(await db.Readings.AnyAsync(r => r.MeterId == meterId && r.Time == T0.AddHours(1))); + + await CleanupAsync(db, meterId); + } + + [Fact] + public async Task Allows_decrease_when_a_swap_event_explains_it() + { + await using var db = fx.CreateContext(); + var (meterId, sourceId) = await SetupAsync(db, MeterMode.CumulativeCounter); + var service = new IngestionService(db); + + await service.IngestAsync(sourceId, T0, 500); + db.MeterEvents.Add(new MeterEvent + { + MeterId = meterId, + Time = T0.AddMinutes(30), + EventType = MeterEventType.MeterSwap, + PrevValue = 500, + NewValue = 0, + }); + await db.SaveChangesAsync(); + + var outcome = await service.IngestAsync(sourceId, T0.AddHours(1), 20); // new meter reads low + + Assert.Equal(IngestionOutcome.Written, outcome); + + await CleanupAsync(db, meterId); + } + + [Fact] + public async Task Mqtt_router_ingests_a_tasmota_payload() + { + await using var db = fx.CreateContext(); + var (meterId, _) = await SetupAsync(db, MeterMode.CumulativeCounter, topic: "tele/plug7/SENSOR"); + var router = new MqttMessageRouter(db, new IngestionService(db), NullLogger.Instance); + + var routed = await router.RouteAsync( + "tele/plug7/SENSOR", + """{"Time":"2024-03-01T10:00:00","ENERGY":{"Total":8421.0}}"""); + + Assert.Equal(1, routed); + var reading = await db.Readings.SingleAsync(r => r.MeterId == meterId); + Assert.Equal(8421.0, reading.Value, 3); + Assert.Equal(new DateTimeOffset(2024, 3, 1, 10, 0, 0, TimeSpan.Zero), reading.Time); + + await CleanupAsync(db, meterId); + } + + 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") + { + await DatabaseSeeder.SeedAsync(db); + var type = await db.EnergyTypes.FirstAsync(t => t.Key == "electricity"); + + var meter = new Meter + { + Name = $"ingest-{Guid.NewGuid():N}", + EnergyTypeId = type.Id, + Mode = mode, + Unit = "kWh", + }; + db.Meters.Add(meter); + await db.SaveChangesAsync(); + + var source = new MeterSource + { + MeterId = meter.Id, + SourceType = SourceType.Tasmota, + ValueKind = SourceValueKind.Register, + Scale = scale, + Offset = offset, + Config = JsonSerializer.Serialize(new { topic, path }), + }; + db.MeterSources.Add(source); + await db.SaveChangesAsync(); + + return (meter.Id, source.Id); + } + + private static async Task CleanupAsync(MeterVaultDbContext db, int meterId) + { + await db.Readings.Where(r => r.MeterId == meterId).ExecuteDeleteAsync(); + await db.MeterEvents.Where(e => e.MeterId == meterId).ExecuteDeleteAsync(); + await db.Meters.Where(m => m.Id == meterId).ExecuteDeleteAsync(); + } +} diff --git a/tests/Integration.Tests/Ingestion/PayloadExtractionTests.cs b/tests/Integration.Tests/Ingestion/PayloadExtractionTests.cs new file mode 100644 index 0000000..9c85cb9 --- /dev/null +++ b/tests/Integration.Tests/Ingestion/PayloadExtractionTests.cs @@ -0,0 +1,73 @@ +using System.Text.Json; +using MeterVault.Infrastructure.Ingestion; + +namespace MeterVault.Integration.Tests.Ingestion; + +/// Pure ingestion helpers — no broker or database needed. +public sealed class PayloadExtractionTests +{ + private const string TasmotaSensor = + """{"Time":"2024-01-15T12:30:00","ENERGY":{"Total":1234.56,"Today":1.2,"Power":50}}"""; + + [Fact] + public void Extracts_tasmota_energy_total() + { + Assert.True(PayloadExtractor.TryExtractValue(TasmotaSensor, "ENERGY.Total", out var value)); + Assert.Equal(1234.56, value, 3); + } + + [Fact] + public void Extracts_tasmota_time() + { + Assert.True(PayloadExtractor.TryExtractTime(TasmotaSensor, "Time", out var time)); + Assert.Equal(new DateTimeOffset(2024, 1, 15, 12, 30, 0, TimeSpan.Zero), time); + } + + [Fact] + public void Extracts_bare_scalar_when_no_path() + { + Assert.True(PayloadExtractor.TryExtractValue("42.7", null, out var value)); + Assert.Equal(42.7, value, 3); + } + + [Fact] + public void Missing_path_yields_false() + { + Assert.False(PayloadExtractor.TryExtractValue(TasmotaSensor, "ENERGY.Nope", out _)); + } + + [Theory] + [InlineData("tele/+/SENSOR", "tele/plug1/SENSOR", true)] + [InlineData("tele/plug1/SENSOR", "tele/plug1/SENSOR", true)] + [InlineData("tele/#", "tele/plug1/SENSOR", true)] + [InlineData("tele/+/SENSOR", "tele/plug1/STATE", false)] + [InlineData("tele/+/SENSOR", "tele/a/b/SENSOR", false)] + [InlineData("home/#", "office/x", false)] + public void Topic_matcher_follows_mqtt_wildcards(string filter, string topic, bool expected) + { + Assert.Equal(expected, MqttTopicMatcher.Matches(filter, topic)); + } + + [Fact] + public void Ha_state_client_parses_numeric_state() + { + using var doc = JsonDocument.Parse( + """{"entity_id":"sensor.p","state":"987.6","attributes":{"unit":"kWh"},"last_updated":"2024-02-01T08:00:00+00:00"}"""); + + var state = HaStateClient.ParseState(doc, attribute: null); + + Assert.NotNull(state); + Assert.Equal(987.6, state!.Value.Value, 3); + Assert.Equal(new DateTimeOffset(2024, 2, 1, 8, 0, 0, TimeSpan.Zero), state.Value.Time); + } + + [Fact] + public void Ha_state_client_parses_attribute_and_rejects_unavailable() + { + using var doc = JsonDocument.Parse( + """{"entity_id":"sensor.p","state":"unavailable","attributes":{"power":123.4}}"""); + + Assert.Null(HaStateClient.ParseState(doc, attribute: null)); // non-numeric state + Assert.Equal(123.4, HaStateClient.ParseState(doc, "power")!.Value.Value, 3); + } +} diff --git a/tests/Integration.Tests/MeterVaultAppFactory.cs b/tests/Integration.Tests/MeterVaultAppFactory.cs index 5f5e71c..44b7902 100644 --- a/tests/Integration.Tests/MeterVaultAppFactory.cs +++ b/tests/Integration.Tests/MeterVaultAppFactory.cs @@ -14,5 +14,6 @@ public sealed class MeterVaultAppFactory(string connectionString) : WebApplicati builder.UseEnvironment("Testing"); builder.UseSetting("ConnectionStrings:Default", connectionString); builder.UseSetting("MeterVault:RunMigrationsAtStartup", "false"); + builder.UseSetting("MeterVault:EnableLiveIngestion", "false"); } }