using MeterVault.Core.Domain;
using MeterVault.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
namespace MeterVault.Infrastructure.Dashboard;
///
/// Builds the per-energy-type flow graph (Sankey) from the meter topology ()
/// and consumption over a period. Each meter is a node sized by its consumption; each configured
/// edge carries the downstream meter's consumption (split proportionally when a meter has several
/// upstreams); the unaccounted remainder under a meter becomes a synthetic "Other" node. Nothing is
/// hardcoded per energy type — it works for electricity, water, gas, … alike. DbContext factory
/// keeps it Blazor-circuit safe.
///
public sealed class FlowService(IDbContextFactory contextFactory)
{
private const double Epsilon = 0.01;
private readonly IDbContextFactory _contextFactory = contextFactory;
public async Task GetFlowAsync(short energyTypeId, DateOnly from, DateOnly to, CancellationToken cancellationToken = default)
{
await using var db = await _contextFactory.CreateDbContextAsync(cancellationToken).ConfigureAwait(false);
var energyType = await db.EnergyTypes.AsNoTracking().FirstOrDefaultAsync(t => t.Id == energyTypeId, cancellationToken).ConfigureAwait(false);
var meters = await db.Meters.AsNoTracking().Where(m => m.EnergyTypeId == energyTypeId).ToListAsync(cancellationToken).ConfigureAwait(false);
if (energyType is null || meters.Count == 0)
{
return new FlowGraph(energyTypeId, energyType?.DisplayName ?? "", energyType?.BaseUnit ?? "", 0, [], []);
}
var meterIds = meters.Select(m => m.Id).ToHashSet();
var fromUtc = ToUtc(from);
var toUtc = ToUtc(to);
var sums = await db.Consumption.AsNoTracking()
.Where(c => c.Time >= fromUtc && c.Time < toUtc && c.Kind == ConsumptionKind.Consumption)
.GroupBy(c => c.MeterId)
.Select(g => new { MeterId = g.Key, Total = g.Sum(x => x.Amount) })
.ToListAsync(cancellationToken).ConfigureAwait(false);
var value = sums.Where(s => meterIds.Contains(s.MeterId)).ToDictionary(s => s.MeterId, s => s.Total);
double V(int id) => value.GetValueOrDefault(id);
var links = await db.MeterLinks.AsNoTracking()
.Where(l => meterIds.Contains(l.FromMeterId) && meterIds.Contains(l.ToMeterId))
.ToListAsync(cancellationToken).ConfigureAwait(false);
var parents = meters.ToDictionary(m => m.Id, _ => new List());
var children = meters.ToDictionary(m => m.Id, _ => new List());
foreach (var link in links)
{
children[link.FromMeterId].Add(link.ToMeterId);
parents[link.ToMeterId].Add(link.FromMeterId);
}
var depth = ComputeDepths(meters.Select(m => m.Id).ToList(), parents, children);
// Link value: a child's consumption flows in from its parent(s); with several parents it is
// split proportionally to the parents' own consumption (equal split if those are all zero).
var flowLinks = new List();
var outgoingByParent = meters.ToDictionary(m => m.Id, _ => 0d);
foreach (var (childId, parentIds) in parents)
{
if (parentIds.Count == 0)
{
continue;
}
var parentTotal = parentIds.Sum(V);
foreach (var parentId in parentIds)
{
var share = parentIds.Count == 1 ? 1d
: parentTotal > Epsilon ? V(parentId) / parentTotal
: 1d / parentIds.Count;
var linkValue = V(childId) * share;
if (linkValue > Epsilon)
{
flowLinks.Add(new FlowLink(NodeId(parentId), NodeId(childId), linkValue));
outgoingByParent[parentId] += linkValue;
}
}
}
var nodes = new List();
foreach (var meter in meters)
{
// Keep a meter node if it carries flow or participates in the topology.
if (V(meter.Id) <= Epsilon && children[meter.Id].Count == 0 && parents[meter.Id].Count == 0)
{
continue;
}
nodes.Add(new FlowNode(NodeId(meter.Id), meter.Name, V(meter.Id), depth.GetValueOrDefault(meter.Id), energyType.ColorHex, false, meter.Id));
// Unaccounted remainder under a meter with sub-meters → "Other".
if (children[meter.Id].Count > 0)
{
var remainder = V(meter.Id) - outgoingByParent[meter.Id];
if (remainder > Epsilon)
{
var otherId = $"other{meter.Id}";
nodes.Add(new FlowNode(otherId, $"Other ({meter.Name})", remainder, depth.GetValueOrDefault(meter.Id) + 1, "#78909C", true, null));
flowLinks.Add(new FlowLink(NodeId(meter.Id), otherId, remainder));
}
}
}
var total = meters.Where(m => parents[m.Id].Count == 0).Sum(m => V(m.Id));
return new FlowGraph(energyTypeId, energyType.DisplayName, energyType.BaseUnit, total, nodes, flowLinks);
}
/// Longest-path depth from the roots (Kahn topological relaxation); robust to stray cycles.
private static Dictionary ComputeDepths(
List ids, Dictionary> parents, Dictionary> children)
{
var depth = ids.ToDictionary(id => id, _ => 0);
var indegree = ids.ToDictionary(id => id, id => parents[id].Count);
var queue = new Queue(ids.Where(id => indegree[id] == 0));
var processed = 0;
while (queue.Count > 0)
{
var node = queue.Dequeue();
processed++;
foreach (var child in children[node])
{
depth[child] = Math.Max(depth[child], depth[node] + 1);
if (--indegree[child] == 0)
{
queue.Enqueue(child);
}
}
}
// Any nodes left (a cycle) keep depth 0 — the admin prevents cycles, this is just a guard.
return depth;
}
private static string NodeId(int meterId) => $"m{meterId}";
private static DateTimeOffset ToUtc(DateOnly date) => new(date.Year, date.Month, date.Day, 0, 0, 0, TimeSpan.Zero);
}