M2: German CSV importer + reconciliation of all 4 fixtures

- German-dialect scalar parsers in Core (GermanNumber/Money/Date, ValueCell):
  decimal comma, thousands dot, unit suffixes, € currency, both date shapes.
- Declarative MappingProfile + RowClassifier (skip summary/blank/all-zero rows) +
  CsvImporter (CsvHelper) staging readings/events/manual-costs, with auto swap
  detection on register decreases and month-end anchoring for interleaved oil dates.
- Four built-in ReferenceProfiles (Strom/Wasser/Heizöl/Kosten).
- ImportService: commit as revertible import_batch + wholesale consumption recompute
  per affected meter (NormalizationService/MeterConfigFactory), revert by batch.
- Reconciliation tests: all 4 CSVs match the sheet's own columns within tolerance
  (electricity 5 meters + Netz Einsparung, water swap→12, oil tank incl. deliveries +
  burner hours, cost category totals). Commit/revert round-trip verified on Timescale.

69 tests green (53 Core + 16 integration).

Known follow-up (polish): historical imports can contend with the 30-day compression
policy's background job; tests pause it. Consider retry-on-deadlock or deferred
compression for large historical imports in production.

Claude-Session: https://claude.ai/code/session_01WujdMtMJPbxDpDnMeK22rr
This commit is contained in:
2026-07-13 11:35:54 +02:00
parent d972f67bad
commit 5977c81002
24 changed files with 1571 additions and 0 deletions
@@ -0,0 +1,88 @@
using MeterVault.Core.Domain;
using MeterVault.Core.Normalization;
using MeterVault.Infrastructure.Import;
using MeterVault.Infrastructure.Normalization;
using MeterVault.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
namespace MeterVault.Integration.Tests;
/// <summary>
/// End-to-end persistence: stage the water CSV → commit as a batch → normalized consumption lands
/// in the hypertable → revert removes everything and rebases consumption (SDD §6.3, FR-6).
/// </summary>
[Collection("Timescale")]
public sealed class ImportRoundTripTests(TimescaleFixture fx)
{
[Fact]
public async Task Commit_persists_consumption_and_revert_removes_it()
{
await using var db = fx.CreateContext();
await DatabaseSeeder.SeedAsync(db);
var waterType = await db.EnergyTypes.FirstAsync(t => t.Key == "water");
var meter = new Meter
{
Name = "Zähler Wasser (round-trip test)",
EnergyTypeId = waterType.Id,
Mode = MeterMode.CumulativeCounter,
Unit = "m3",
InitialBaseline = 820,
};
db.Meters.Add(meter);
await db.SaveChangesAsync();
var profile = new MappingProfile
{
Name = "test-water",
DateColumn = 0,
DateKind = DateKind.MonthName,
FirstDataRowIndex = 1,
DetectCumulativeSwaps = true,
Columns =
[
new ColumnMapping
{
Index = 1,
Role = MappingRole.Reading,
MeterId = meter.Id,
Unit = "m3",
SwapConsumptionColumn = 2,
},
],
};
StagedImport staged;
using (var reader = new StreamReader(FixturePath(Water)))
{
staged = new CsvImporter().Stage(profile, reader);
}
var normalization = new NormalizationService(db, NormalizationEngine.CreateDefault());
var service = new ImportService(db, normalization);
// Commit.
var batchId = await service.CommitAsync(staged, sourceName: "Wasser.csv", mappingJson: null);
Assert.True(await db.Readings.AnyAsync(r => r.MeterId == meter.Id));
Assert.True(await db.MeterEvents.AnyAsync(e => e.MeterId == meter.Id && e.EventType == MeterEventType.MeterSwap));
var maerz = await db.Consumption.SingleAsync(c =>
c.MeterId == meter.Id && c.Time == new DateTimeOffset(2023, 3, 1, 0, 0, 0, TimeSpan.Zero));
Assert.Equal(12d, maerz.Amount, 3);
// Revert.
await service.RevertAsync(batchId);
Assert.False(await db.Readings.AnyAsync(r => r.MeterId == meter.Id));
Assert.False(await db.MeterEvents.AnyAsync(e => e.MeterId == meter.Id));
Assert.False(await db.Consumption.AnyAsync(c => c.MeterId == meter.Id));
var batch = await db.ImportBatches.SingleAsync(b => b.Id == batchId);
Assert.NotNull(batch.RevertedAt);
// Cleanup so the shared container stays tidy for other tests. ExecuteDelete bypasses the
// change tracker (which holds stale entries after the revert's set-based deletes).
await db.Meters.Where(m => m.Id == meter.Id).ExecuteDeleteAsync();
}
}
@@ -14,6 +14,11 @@
<Using Include="Xunit" />
</ItemGroup>
<!-- The four reference CSVs are golden fixtures; link (don't copy) from sampledata/. -->
<ItemGroup>
<Content Include="..\..\sampledata\*.csv" Link="fixtures\%(Filename)%(Extension)" CopyToOutputDirectory="PreserveNewest" />
</ItemGroup>
<ItemGroup>
<ProjectReference Include="..\..\src\App\MeterVault.App.csproj" />
</ItemGroup>
@@ -0,0 +1,41 @@
using MeterVault.Infrastructure.Import;
using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
namespace MeterVault.Integration.Tests.Reconciliation;
/// <summary>
/// The Kosten sheet is pre-computed category costs (SDD §2.1). The importer stages one manual
/// cost per category column; their monthly sum must equal the sheet's total Kosten column, and
/// the meterless "Pool Betrieb" column becomes manual costs with no meter.
/// </summary>
public sealed class CostsReconciliationTests
{
[Fact]
public void Category_costs_sum_to_the_monthly_total()
{
var staged = Stage(ReferenceProfiles.Costs(), Costs);
var computed = staged.ManualCosts
.GroupBy(c => new DateOnly(c.PeriodStart.Year, c.PeriodStart.Month, 1))
.ToDictionary(g => g.Key, g => g.Sum(c => c.Amount));
var oracle = OracleByMonth(ReadRows(Costs), dateColumn: 0, valueColumn: 2, firstDataRow: 1);
AssertReconciles(computed, oracle, tolerance: 0.02, "category totals", minMatches: 20);
}
[Fact]
public void Manual_costs_are_staged_meterless_by_category()
{
var staged = Stage(ReferenceProfiles.Costs(), Costs);
Assert.NotEmpty(staged.ManualCosts);
Assert.All(staged.ManualCosts, c => Assert.Null(c.MeterId));
// Wasser Dez 2022 = 70,00 € (SDD §2.1: categories decoupled from meters).
var wasserDec = staged.ManualCosts.Single(c =>
c.CategoryId == ReferenceProfiles.CategoryWasser
&& c.PeriodStart == new DateOnly(2022, 12, 1));
Assert.Equal(70d, wasserDec.Amount, 2);
}
}
@@ -0,0 +1,72 @@
using MeterVault.Core.Domain;
using MeterVault.Core.Normalization;
using MeterVault.Infrastructure.Import;
using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
namespace MeterVault.Integration.Tests.Reconciliation;
/// <summary>
/// Reconciles the five electricity meters and a derived virtual meter against the Strom sheet's
/// own Verbrauch / Netz Einsparung columns (SDD §2.2). Runs without a database.
/// </summary>
public sealed class ElectricityReconciliationTests
{
private static MeterConfig Cumulative(int id) =>
new() { MeterId = id, Mode = MeterMode.CumulativeCounter, Unit = "kWh" };
private static MeterConfig Generation(int id) =>
new() { MeterId = id, Mode = MeterMode.GenerationCounter, Unit = "kWh" };
[Theory]
[InlineData(ReferenceProfiles.Haus, 7)] // Haus Verbrauch
[InlineData(ReferenceProfiles.Netz, 8)] // Netz Verbrauch
[InlineData(ReferenceProfiles.Auto, 9)] // Auto Verbrauch
[InlineData(ReferenceProfiles.Solar1, 10)] // Solar Erzeugung 1
[InlineData(ReferenceProfiles.Solar2, 11)] // Solar Erzeugung 2
public void Meter_consumption_matches_the_sheet(int meterId, int oracleColumn)
{
var staged = Stage(ReferenceProfiles.Electricity(), Electricity);
var config = meterId is ReferenceProfiles.Solar1 or ReferenceProfiles.Solar2
? Generation(meterId)
: Cumulative(meterId);
var computed = ByMonth(Normalize(staged, config));
var oracle = OracleByMonth(ReadRows(Electricity), dateColumn: 0, valueColumn: oracleColumn, firstDataRow: 1);
AssertReconciles(computed, oracle, tolerance: 1.0, $"meter {meterId}", minMatches: 20);
}
[Fact]
public void Netz_einsparung_virtual_matches_the_sheet()
{
var staged = Stage(ReferenceProfiles.Electricity(), Electricity);
var haus = Normalize(staged, Cumulative(ReferenceProfiles.Haus));
var netz = Normalize(staged, Cumulative(ReferenceProfiles.Netz));
var engine = NormalizationEngine.CreateDefault();
var virtualContext = new NormalizationContext
{
Meter = new MeterConfig
{
MeterId = 100,
Mode = MeterMode.Virtual,
Unit = "kWh",
Virtual = new VirtualSpec
{
Expression = $"m{ReferenceProfiles.Haus} - m{ReferenceProfiles.Netz}",
ReferencedMeterIds = [ReferenceProfiles.Haus, ReferenceProfiles.Netz],
},
},
ReferencedSeries = new Dictionary<int, IReadOnlyList<Consumption>>
{
[ReferenceProfiles.Haus] = haus,
[ReferenceProfiles.Netz] = netz,
},
};
var computed = ByMonth(engine.Normalize(virtualContext));
var oracle = OracleByMonth(ReadRows(Electricity), dateColumn: 0, valueColumn: 13, firstDataRow: 1); // Netz Einsparung
AssertReconciles(computed, oracle, tolerance: 1.0, "Netz Einsparung", minMatches: 20);
}
}
@@ -0,0 +1,56 @@
using MeterVault.Core.Domain;
using MeterVault.Core.Normalization;
using MeterVault.Infrastructure.Import;
using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
namespace MeterVault.Integration.Tests.Reconciliation;
/// <summary>
/// Reconciles the heating-oil tank against the Heizöl sheet (SDD §2.4). The sheet's
/// <c>Differenz Tank</c> (col 9) is the negated actual consumption, already netting deliveries —
/// exactly what the consumable-balance normalizer computes (prevLevel + deliveries currLevel).
/// The burner runtime meter's Δhours is reconciled against <c>Differenz Betrieb</c> (col 2).
/// </summary>
public sealed class OilReconciliationTests
{
[Fact]
public void Tank_consumption_matches_differenz_tank_including_deliveries()
{
var staged = Stage(ReferenceProfiles.HeatingOil(), Oil);
var config = new MeterConfig
{
MeterId = ReferenceProfiles.OilTank,
Mode = MeterMode.ConsumableBalance,
Unit = "L",
Tank = new TankConfig
{
Capacity = 7000,
Calibration = new CalibrationCurve(ReferenceProfiles.OilLitresPerCm),
},
};
var computed = ByDate(Normalize(staged, config));
// Differenz Tank is negative consumption → negate to compare with positive draw.
var oracle = OracleByDate(ReadRows(Oil), dateColumn: 0, valueColumn: 9, firstDataRow: 4, v => -v);
// ±2 L covers cm→litre rounding on both ends of each interval.
AssertReconciles(computed, oracle, tolerance: 2.0, "oil tank", minMatches: 30);
}
[Fact]
public void Burner_delta_hours_match_differenz_betrieb()
{
var staged = Stage(ReferenceProfiles.HeatingOil(), Oil);
var config = new MeterConfig
{
MeterId = ReferenceProfiles.Burner,
Mode = MeterMode.RuntimeCounter,
Unit = "h",
};
var computed = ByDate(Normalize(staged, config));
var oracle = OracleByDate(ReadRows(Oil), dateColumn: 0, valueColumn: 2, firstDataRow: 4);
AssertReconciles(computed, oracle, tolerance: 0.5, "burner hours", minMatches: 30);
}
}
@@ -0,0 +1,146 @@
using MeterVault.Core.Domain;
using MeterVault.Core.Normalization;
using MeterVault.Core.Parsing;
using MeterVault.Infrastructure.Import;
namespace MeterVault.Integration.Tests.Reconciliation;
/// <summary>
/// Shared helpers for the golden-fixture reconciliation tests. The CSVs are self-oracling: the
/// same file carries both the input (registers/levels/hours) and the expected output
/// (Verbrauch / Differenz Tank / Kosten columns). We parse input → normalize → compare against
/// the sheet's own columns (SDD §0.3, §13). No database is involved.
/// </summary>
internal static class ReconciliationSupport
{
// Fixture file names (linked into fixtures/ in the test output).
public const string Electricity = "Energiebilanz - Strom Verbrauch.csv";
public const string Water = "Energiebilanz - Wasser.csv";
public const string Oil = "Energiebilanz - Heizöl Verbrauch.csv";
public const string Costs = "Energiebilanz - Kosten.csv";
public static string FixturePath(string fileName) =>
Path.Combine(AppContext.BaseDirectory, "fixtures", fileName);
public static List<string[]> ReadRows(string fileName)
{
using var reader = new StreamReader(FixturePath(fileName));
return CsvImporter.ReadRows(reader);
}
public static StagedImport Stage(MappingProfile profile, string fileName)
{
using var reader = new StreamReader(FixturePath(fileName));
return new CsvImporter().Stage(profile, reader);
}
/// <summary>Normalizes one meter from a staged import using the given config.</summary>
public static IReadOnlyList<Consumption> Normalize(StagedImport staged, MeterConfig config)
{
var engine = NormalizationEngine.CreateDefault();
var context = new NormalizationContext
{
Meter = config,
Readings = staged.Readings.Where(r => r.MeterId == config.MeterId).ToList(),
Events = staged.Events.Where(e => e.MeterId == config.MeterId).ToList(),
};
return engine.Normalize(context);
}
/// <summary>Extracts a sheet oracle column keyed by month, using German number parsing.</summary>
public static Dictionary<DateOnly, double> OracleByMonth(
IReadOnlyList<string[]> rows, int dateColumn, int valueColumn, int firstDataRow)
{
var result = new Dictionary<DateOnly, double>();
for (var r = firstDataRow; r < rows.Count; r++)
{
var row = rows[r];
if (dateColumn >= row.Length || valueColumn >= row.Length)
{
continue;
}
if (!GermanDate.TryParse(row[dateColumn], out var date))
{
continue;
}
if (GermanNumber.TryParse(row[valueColumn], out var value))
{
result[new DateOnly(date.Year, date.Month, 1)] = value;
}
}
return result;
}
public static DateOnly MonthKey(DateTimeOffset time) => new(time.Year, time.Month, 1);
/// <summary>
/// Asserts every month present in both computed and oracle agrees within tolerance, and that
/// a meaningful number of months were actually compared (so an empty result can't pass).
/// </summary>
public static void AssertReconciles(
IReadOnlyDictionary<DateOnly, double> computed,
IReadOnlyDictionary<DateOnly, double> oracle,
double tolerance,
string label,
int minMatches)
{
var matched = 0;
foreach (var (month, expected) in oracle)
{
if (!computed.TryGetValue(month, out var actual))
{
continue;
}
matched++;
Assert.True(
Math.Abs(actual - expected) <= tolerance,
$"{label} {month:yyyy-MM}: computed {actual:0.##} vs sheet {expected:0.##} (tol {tolerance}).");
}
Assert.True(matched >= minMatches, $"{label}: only {matched} months reconciled (expected ≥ {minMatches}).");
}
public static Dictionary<DateOnly, double> ByMonth(IReadOnlyList<Consumption> series) =>
series.GroupBy(c => MonthKey(c.Time)).ToDictionary(g => g.Key, g => g.Sum(c => c.Amount));
/// <summary>Consumption keyed by exact reading date (oil rows are event-dated, sometimes two per month).</summary>
public static Dictionary<DateOnly, double> ByDate(IReadOnlyList<Consumption> series) =>
series.GroupBy(c => DateOnly.FromDateTime(c.Time.UtcDateTime))
.ToDictionary(g => g.Key, g => g.Sum(c => c.Amount));
/// <summary>
/// Extracts a sheet oracle column keyed by exact date, optionally transformed. Uses the same
/// month-end anchoring as the importer so day-dated and month-dated rows align.
/// </summary>
public static Dictionary<DateOnly, double> OracleByDate(
IReadOnlyList<string[]> rows, int dateColumn, int valueColumn, int firstDataRow,
Func<double, double>? transform = null, bool anchorMonthsToEnd = true)
{
var result = new Dictionary<DateOnly, double>();
for (var r = firstDataRow; r < rows.Count; r++)
{
var row = rows[r];
if (dateColumn >= row.Length || valueColumn >= row.Length)
{
continue;
}
if (!ImportDate.TryResolve(row[dateColumn], anchorMonthsToEnd, out var date))
{
continue;
}
if (GermanNumber.TryParse(row[valueColumn], out var value))
{
result[date] = transform is null ? value : transform(value);
}
}
return result;
}
}
@@ -0,0 +1,39 @@
using MeterVault.Core.Domain;
using MeterVault.Core.Normalization;
using MeterVault.Infrastructure.Import;
using static MeterVault.Integration.Tests.Reconciliation.ReconciliationSupport;
namespace MeterVault.Integration.Tests.Reconciliation;
/// <summary>
/// Reconciles the water meter against the Wasser sheet's Wasserverbrauch column, exercising the
/// mid-series register swap …861 → 2 (SDD §2.3). The importer detects the register drop and seeds
/// a swap event whose consumption override comes from the sheet's own column.
/// </summary>
public sealed class WaterReconciliationTests
{
[Fact]
public void Water_consumption_matches_the_sheet_across_the_swap()
{
var staged = Stage(ReferenceProfiles.Water(), Water);
// A swap event must have been auto-detected at the 861→2 drop.
Assert.Contains(staged.Events, e => e.EventType == MeterEventType.MeterSwap);
var config = new MeterConfig
{
MeterId = ReferenceProfiles.Wasser,
Mode = MeterMode.CumulativeCounter,
Unit = "m3",
InitialBaseline = 820, // pre-existing meter's register when tracking began (Nov 2022).
};
var computed = ByMonth(Normalize(staged, config));
var oracle = OracleByMonth(ReadRows(Water), dateColumn: 0, valueColumn: 2, firstDataRow: 1);
AssertReconciles(computed, oracle, tolerance: 1.0, "water", minMatches: 12);
// Explicitly assert the swap month reconciles to 12.
Assert.Equal(12d, computed[new DateOnly(2023, 3, 1)], 3);
}
}
@@ -35,6 +35,14 @@ public sealed class TimescaleFixture : IAsyncLifetime
await _db.StartAsync();
await using var ctx = CreateContext();
await ctx.Database.MigrateAsync();
// The fixtures are historical (20222023), so freshly-inserted chunks immediately fall
// past the 30-day compression horizon. Pause the compression job's scheduler so its
// background worker can't deadlock a test's import transaction. The policy still exists
// (SchemaTests verifies that); only its scheduling is disabled.
await ctx.Database.ExecuteSqlRawAsync(
"SELECT alter_job(job_id, scheduled => false) FROM timescaledb_information.jobs " +
"WHERE proc_name IN ('policy_compression', 'policy_columnstore');");
}
public async Task DisposeAsync() => await _db.DisposeAsync();