Edge AI with .NET, Part 4: Time Series That Outlives the Hardware

-- Locations and meters
CREATE TABLE locations (
    location_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    address     TEXT NOT NULL
);
 
CREATE TABLE meters (
    meter_id     UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    location_id  UUID NOT NULL REFERENCES locations(location_id),
    device_serial TEXT NOT NULL,
    meter_type   TEXT NOT NULL CHECK (meter_type IN ('water','power')),
    installed_at TIMESTAMPTZ NOT NULL,
    removed_at   TIMESTAMPTZ
);
 
-- Raw readings hypertable
CREATE TABLE readings (
    time        TIMESTAMPTZ     NOT NULL,
    meter_id    UUID            NOT NULL,
    value       DOUBLE PRECISION NOT NULL  -- litres or watt-hours
);
 
SELECT create_hypertable('readings', 'time',
    chunk_time_interval => INTERVAL '7 days');
 
-- Hypercore columnstore compression after 7 days
SELECT add_columnstore_policy('readings',
    after => INTERVAL '7 days',
    segmentby => ARRAY['meter_id']);
 
-- Tariff periods for cost calculation
CREATE TABLE tariff_periods (
    tariff_id   UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    valid_from  TIMESTAMPTZ NOT NULL,
    valid_until TIMESTAMPTZ,          -- NULL means current tariff
    eur_per_kwh NUMERIC(10,6) NOT NULL
);

-- Hourly rollup
CREATE MATERIALIZED VIEW readings_hourly
WITH (timescaledb.continuous,
      timescaledb.materialized_only = false) AS
SELECT
    time_bucket('1 hour', time)  AS bucket,
    meter_id,
    SUM(value)                   AS total,
    MIN(value)                   AS min_val,
    MAX(value)                   AS max_val,
    COUNT(*)                     AS sample_count
FROM readings
GROUP BY bucket, meter_id
WITH NO DATA;
 
SELECT add_continuous_aggregate_policy('readings_hourly',
    start_offset  => INTERVAL '2 hours',
    end_offset    => INTERVAL '1 hour',
    schedule_interval => INTERVAL '1 hour');
 
-- Daily rollup sourced from hourly
CREATE MATERIALIZED VIEW readings_daily
WITH (timescaledb.continuous,
      timescaledb.materialized_only = false) AS
SELECT
    time_bucket('1 day', bucket)  AS bucket,
    meter_id,
    SUM(total)                    AS total,
    MIN(min_val)                  AS min_val,
    MAX(max_val)                  AS max_val,
    SUM(sample_count)             AS sample_count
FROM readings_hourly
GROUP BY time_bucket('1 day', bucket), meter_id
WITH NO DATA;
 
SELECT add_continuous_aggregate_policy('readings_daily',
    start_offset  => INTERVAL '2 days',
    end_offset    => INTERVAL '1 day',
    schedule_interval => INTERVAL '1 day');

// Reading.cs
public class Reading
{
    public DateTimeOffset Time     { get; set; }
    public Guid           MeterId  { get; set; }
    public double         Value    { get; set; }
 
    public Meter Meter { get; set; } = null!;
}
 
// Meter.cs
public class Meter
{
    public Guid            MeterId      { get; set; }
    public Guid            LocationId   { get; set; }
    public string          DeviceSerial { get; set; } = string.Empty;
    public string          MeterType    { get; set; } = string.Empty;
    public DateTimeOffset  InstalledAt  { get; set; }
    public DateTimeOffset? RemovedAt    { get; set; }
 
    public Location          Location { get; set; } = null!;
    public ICollection<Reading> Readings { get; set; } = [];
}
 
// MeterDbContext.cs (relevant excerpt)
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<Reading>(e =>
    {
        e.HasKey(r => new { r.Time, r.MeterId });
        e.Property(r => r.Time).HasColumnType("timestamptz");
        e.HasOne(r => r.Meter)
         .WithMany(m => m.Readings)
         .HasForeignKey(r => r.MeterId);
    });
 
    modelBuilder.Entity<Meter>(e =>
    {
        e.HasKey(m => m.MeterId);
        e.Property(m => m.RemovedAt).HasColumnType("timestamptz");
    });
 
    // If not using the community package, call the hypertable
    // DDL from the migration's Up() method via migrationBuilder.Sql()
}

public class AnomalyWorker(MeterDbContext db,
                           ILogger<AnomalyWorker> logger,
                           IAlertService alerts)
    : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromMinutes(1));
        while (await timer.WaitForNextTickAsync(ct))
        {
            await CheckLeaksAsync(ct);
            await CheckPowerSpikesAsync(ct);
            await CheckSilentMetersAsync(ct);
        }
    }
 
    private async Task CheckLeaksAsync(CancellationToken ct)
    {
        var cutoff = DateTimeOffset.UtcNow.AddHours(-6);
 
        // readings_hourly is a cagg, queried like a normal table
        var leaking = await db.Database
            .SqlQuery<Guid>($"""
                SELECT h.meter_id AS "Value"
                FROM   readings_hourly h
                JOIN   meters m ON m.meter_id = h.meter_id
                WHERE  m.meter_type = 'water'
                  AND  m.removed_at IS NULL
                  AND  h.bucket >= {cutoff}
                  AND  h.min_val > 0
                GROUP  BY h.meter_id
                HAVING COUNT(*) >= 6
                """)
            .ToListAsync(ct);
 
        foreach (var meterId in leaking)
            await alerts.RaiseAsync(meterId, "LeakDetected", ct);
    }
 
    private async Task CheckPowerSpikesAsync(CancellationToken ct)
    {
        // 7-day baseline: flag today if today's total > avg + 2*stddev
        var spikes = await db.Database
            .SqlQuery<Guid>($"""
                WITH baseline AS (
                    SELECT meter_id,
                           AVG(total)    AS avg_total,
                           STDDEV(total) AS std_total
                    FROM   readings_daily
                    WHERE  bucket >= NOW() - INTERVAL '8 days'
                      AND  bucket <  NOW() - INTERVAL '1 day'
                    GROUP  BY meter_id
                ),
                today AS (
                    SELECT meter_id, SUM(total) AS day_total
                    FROM   readings_hourly
                    WHERE  bucket >= date_trunc('day', NOW())
                    GROUP  BY meter_id
                )
                SELECT t.meter_id AS "Value"
                FROM   today t
                JOIN   baseline b ON b.meter_id = t.meter_id
                WHERE  t.day_total > b.avg_total + 2 * COALESCE(b.std_total, 0)
                """)
            .ToListAsync(ct);
 
        foreach (var meterId in spikes)
            await alerts.RaiseAsync(meterId, "PowerSpike", ct);
    }
 
    private async Task CheckSilentMetersAsync(CancellationToken ct)
    {
        var threshold = DateTimeOffset.UtcNow.AddMinutes(-15);
 
        var silent = await db.Database
            .SqlQuery<Guid>($"""
                SELECT m.meter_id AS "Value"
                FROM   meters m
                WHERE  m.removed_at IS NULL
                  AND  NOT EXISTS (
                    SELECT 1 FROM readings r
                    WHERE  r.meter_id = m.meter_id
                      AND  r.time    >= {threshold}
                  )
                """)
            .ToListAsync(ct);
 
        foreach (var meterId in silent)
            await alerts.RaiseAsync(meterId, "SilentMeter", ct);
    }
}

SELECT
    d.meter_id,
    d.bucket,
    d.total / 1000.0                   AS kwh,
    tp.eur_per_kwh,
    (d.total / 1000.0) * tp.eur_per_kwh AS eur_cost
FROM readings_daily d
JOIN tariff_periods tp
  ON  d.bucket >= tp.valid_from
  AND (tp.valid_until IS NULL OR d.bucket < tp.valid_until)
WHERE d.meter_id = @meterId
ORDER BY d.bucket;