feat(history): full Part 13 aggregate set; unsupported-by-backend aggregates answer BadAggregateNotSupported (#285)
Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
@@ -132,6 +132,14 @@ public sealed record HistoryReadResult(
|
||||
byte[]? ContinuationPoint);
|
||||
|
||||
/// <summary>Aggregate function for processed history reads. Mirrors OPC UA Part 13 standard aggregates.</summary>
|
||||
/// <remarks>
|
||||
/// The first five members are the original set every historian backend serves. The remainder (#285) are
|
||||
/// the further standard Part 13 aggregates a server-side HistoryRead can forward to a backend that
|
||||
/// supports them (the OPC UA Client driver forwards all of them upstream). A backend that cannot serve a
|
||||
/// member must throw <see cref="HistoryAggregateNotSupportedException"/> — the server maps that to
|
||||
/// <c>BadAggregateNotSupported</c> rather than a generic failure. <b>Append only</b>: the numeric values
|
||||
/// are not persisted, but every <c>switch</c> over this enum must keep a total default.
|
||||
/// </remarks>
|
||||
public enum HistoryAggregateType
|
||||
{
|
||||
Average,
|
||||
@@ -139,6 +147,42 @@ public enum HistoryAggregateType
|
||||
Maximum,
|
||||
Total,
|
||||
Count,
|
||||
Interpolative,
|
||||
TimeAverage,
|
||||
Start,
|
||||
End,
|
||||
Delta,
|
||||
Range,
|
||||
MinimumActualTime,
|
||||
MaximumActualTime,
|
||||
StartBound,
|
||||
EndBound,
|
||||
DeltaBounds,
|
||||
DurationGood,
|
||||
DurationBad,
|
||||
PercentGood,
|
||||
PercentBad,
|
||||
WorstQuality,
|
||||
NumberOfTransitions,
|
||||
AnnotationCount,
|
||||
StandardDeviationPopulation,
|
||||
VariancePopulation,
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Thrown by a history backend when it cannot serve the requested <see cref="HistoryAggregateType"/>.
|
||||
/// The Server HistoryRead path translates it to <c>BadAggregateNotSupported</c> for that node only.
|
||||
/// </summary>
|
||||
public sealed class HistoryAggregateNotSupportedException : NotSupportedException
|
||||
{
|
||||
/// <summary>The aggregate the backend could not serve.</summary>
|
||||
public HistoryAggregateType Aggregate { get; }
|
||||
|
||||
/// <summary>Creates the exception for <paramref name="aggregate"/>.</summary>
|
||||
/// <param name="aggregate">The unsupported aggregate.</param>
|
||||
public HistoryAggregateNotSupportedException(HistoryAggregateType aggregate)
|
||||
: base($"The historian backend does not support the {aggregate} aggregate.")
|
||||
=> Aggregate = aggregate;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
|
||||
+3
-1
@@ -79,11 +79,13 @@ public sealed class GatewayHistorianDataSource : IHistorianDataSource, IAsyncDis
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||||
{
|
||||
// Mapped OUTSIDE the failure-recording try: an aggregate the gateway has no mode for is a client
|
||||
// request the server answers BadAggregateNotSupported, not a backend fault to count (#285).
|
||||
var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
|
||||
try
|
||||
{
|
||||
// Total/Count are now native gateway retrieval modes — no client-side scaling
|
||||
// (unlike the Wonderware path that derived Total as Average × interval-seconds).
|
||||
var mode = AggregateModeMapper.ToRetrievalMode(aggregate);
|
||||
var buckets = new List<HistorianAggregateSample>();
|
||||
await foreach (var bucket in _client
|
||||
.ReadAggregateAsync(fullReference, startUtc, endUtc, mode, interval, cancellationToken)
|
||||
|
||||
+6
-1
@@ -25,7 +25,8 @@ internal static class AggregateModeMapper
|
||||
/// <summary>Maps an aggregate function to the gateway retrieval mode.</summary>
|
||||
/// <param name="aggregate">The driver-agnostic aggregate function.</param>
|
||||
/// <returns>The matching gateway <see cref="RetrievalMode"/>.</returns>
|
||||
/// <exception cref="ArgumentOutOfRangeException">A future, unmapped enum member (fails the matrix guard).</exception>
|
||||
/// <exception cref="HistoryAggregateNotSupportedException">A defined member the gateway has no retrieval mode for.</exception>
|
||||
/// <exception cref="ArgumentOutOfRangeException">An undefined enum value.</exception>
|
||||
public static RetrievalMode ToRetrievalMode(HistoryAggregateType aggregate) => aggregate switch
|
||||
{
|
||||
HistoryAggregateType.Average => RetrievalMode.TimeWeightedAverage,
|
||||
@@ -33,6 +34,10 @@ internal static class AggregateModeMapper
|
||||
HistoryAggregateType.Maximum => RetrievalMode.MaximumWithTime,
|
||||
HistoryAggregateType.Total => RetrievalMode.Integral,
|
||||
HistoryAggregateType.Count => RetrievalMode.Counter,
|
||||
// Part 13 aggregates the gateway has no retrieval mode for (#285) — an UNDEFINED enum value is a
|
||||
// programming error, but a defined-yet-unserved member is a normal client request the server
|
||||
// answers with BadAggregateNotSupported.
|
||||
_ when Enum.IsDefined(aggregate) => throw new HistoryAggregateNotSupportedException(aggregate),
|
||||
_ => throw new ArgumentOutOfRangeException(
|
||||
nameof(aggregate), aggregate, "Unmapped HistoryAggregateType — add a RetrievalMode mapping."),
|
||||
};
|
||||
|
||||
@@ -1779,6 +1779,26 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
HistoryAggregateType.Maximum => ObjectIds.AggregateFunction_Maximum,
|
||||
HistoryAggregateType.Total => ObjectIds.AggregateFunction_Total,
|
||||
HistoryAggregateType.Count => ObjectIds.AggregateFunction_Count,
|
||||
HistoryAggregateType.Interpolative => ObjectIds.AggregateFunction_Interpolative,
|
||||
HistoryAggregateType.TimeAverage => ObjectIds.AggregateFunction_TimeAverage,
|
||||
HistoryAggregateType.Start => ObjectIds.AggregateFunction_Start,
|
||||
HistoryAggregateType.End => ObjectIds.AggregateFunction_End,
|
||||
HistoryAggregateType.Delta => ObjectIds.AggregateFunction_Delta,
|
||||
HistoryAggregateType.Range => ObjectIds.AggregateFunction_Range,
|
||||
HistoryAggregateType.MinimumActualTime => ObjectIds.AggregateFunction_MinimumActualTime,
|
||||
HistoryAggregateType.MaximumActualTime => ObjectIds.AggregateFunction_MaximumActualTime,
|
||||
HistoryAggregateType.StartBound => ObjectIds.AggregateFunction_StartBound,
|
||||
HistoryAggregateType.EndBound => ObjectIds.AggregateFunction_EndBound,
|
||||
HistoryAggregateType.DeltaBounds => ObjectIds.AggregateFunction_DeltaBounds,
|
||||
HistoryAggregateType.DurationGood => ObjectIds.AggregateFunction_DurationGood,
|
||||
HistoryAggregateType.DurationBad => ObjectIds.AggregateFunction_DurationBad,
|
||||
HistoryAggregateType.PercentGood => ObjectIds.AggregateFunction_PercentGood,
|
||||
HistoryAggregateType.PercentBad => ObjectIds.AggregateFunction_PercentBad,
|
||||
HistoryAggregateType.WorstQuality => ObjectIds.AggregateFunction_WorstQuality,
|
||||
HistoryAggregateType.NumberOfTransitions => ObjectIds.AggregateFunction_NumberOfTransitions,
|
||||
HistoryAggregateType.AnnotationCount => ObjectIds.AggregateFunction_AnnotationCount,
|
||||
HistoryAggregateType.StandardDeviationPopulation => ObjectIds.AggregateFunction_StandardDeviationPopulation,
|
||||
HistoryAggregateType.VariancePopulation => ObjectIds.AggregateFunction_VariancePopulation,
|
||||
_ => throw new ArgumentOutOfRangeException(nameof(aggregate), aggregate, null),
|
||||
};
|
||||
|
||||
|
||||
@@ -5169,6 +5169,14 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
errors[handle.Index] = StatusCodes.BadTimeout;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout };
|
||||
}
|
||||
catch (HistoryAggregateNotSupportedException ex)
|
||||
{
|
||||
// The backend cannot serve this Part 13 aggregate (#285) — the spec status for THIS node,
|
||||
// not a logged server fault.
|
||||
_logger.LogDebug(ex, "OtOpcUaNodeManager: backend does not support aggregate for node {NodeId}", handle.NodeId);
|
||||
errors[handle.Index] = StatusCodes.BadAggregateNotSupported;
|
||||
results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadAggregateNotSupported };
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
// One node's backend failure (throw / timeout) must not throw out of the batch — surface a
|
||||
@@ -5424,6 +5432,26 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_Maximum) return HistoryAggregateType.Maximum;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_Total) return HistoryAggregateType.Total;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_Count) return HistoryAggregateType.Count;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_Interpolative) return HistoryAggregateType.Interpolative;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_TimeAverage) return HistoryAggregateType.TimeAverage;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_Start) return HistoryAggregateType.Start;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_End) return HistoryAggregateType.End;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_Delta) return HistoryAggregateType.Delta;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_Range) return HistoryAggregateType.Range;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_MinimumActualTime) return HistoryAggregateType.MinimumActualTime;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_MaximumActualTime) return HistoryAggregateType.MaximumActualTime;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_StartBound) return HistoryAggregateType.StartBound;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_EndBound) return HistoryAggregateType.EndBound;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_DeltaBounds) return HistoryAggregateType.DeltaBounds;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_DurationGood) return HistoryAggregateType.DurationGood;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_DurationBad) return HistoryAggregateType.DurationBad;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_PercentGood) return HistoryAggregateType.PercentGood;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_PercentBad) return HistoryAggregateType.PercentBad;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_WorstQuality) return HistoryAggregateType.WorstQuality;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_NumberOfTransitions) return HistoryAggregateType.NumberOfTransitions;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_AnnotationCount) return HistoryAggregateType.AnnotationCount;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_StandardDeviationPopulation) return HistoryAggregateType.StandardDeviationPopulation;
|
||||
if (aggregateNodeId == ObjectIds.AggregateFunction_VariancePopulation) return HistoryAggregateType.VariancePopulation;
|
||||
return null;
|
||||
}
|
||||
|
||||
|
||||
+17
-3
@@ -16,10 +16,24 @@ public sealed class AggregateModeMapperTests
|
||||
public void Maps_each_aggregate(HistoryAggregateType a, RetrievalMode expected)
|
||||
=> Assert.Equal(expected, AggregateModeMapper.ToRetrievalMode(a));
|
||||
|
||||
[Fact] // matrix guard: a new HistoryAggregateType member must fail here
|
||||
public void Every_aggregate_member_is_mapped()
|
||||
[Fact] // matrix guard: every member is either mapped or explicitly reported as not supported
|
||||
public void Every_aggregate_member_is_mapped_or_reports_not_supported()
|
||||
{
|
||||
var original = new[]
|
||||
{
|
||||
HistoryAggregateType.Average, HistoryAggregateType.Minimum, HistoryAggregateType.Maximum,
|
||||
HistoryAggregateType.Total, HistoryAggregateType.Count,
|
||||
};
|
||||
foreach (var a in Enum.GetValues<HistoryAggregateType>())
|
||||
_ = AggregateModeMapper.ToRetrievalMode(a); // must not throw for any defined member
|
||||
{
|
||||
if (original.Contains(a))
|
||||
_ = AggregateModeMapper.ToRetrievalMode(a);
|
||||
else
|
||||
Assert.Throws<HistoryAggregateNotSupportedException>(() => AggregateModeMapper.ToRetrievalMode(a));
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Undefined_value_is_a_programming_error_not_a_not_supported()
|
||||
=> Assert.Throws<ArgumentOutOfRangeException>(() => AggregateModeMapper.ToRetrievalMode((HistoryAggregateType)9999));
|
||||
}
|
||||
|
||||
+14
-5
@@ -8,14 +8,23 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient.Tests;
|
||||
[Trait("Category", "Unit")]
|
||||
public sealed class OpcUaClientHistoryTests
|
||||
{
|
||||
/// <summary>Every <see cref="HistoryAggregateType"/> member (#285: the full Part 13 set, not just five).</summary>
|
||||
public static IEnumerable<object[]> AllAggregates() =>
|
||||
Enum.GetValues<HistoryAggregateType>().Select(a => new object[] { a });
|
||||
|
||||
/// <summary>Every member maps to a DISTINCT aggregate NodeId (a copy/paste mapping would collide).</summary>
|
||||
[Fact]
|
||||
public void MapAggregateToNodeId_is_injective_over_the_whole_enum()
|
||||
{
|
||||
var ids = Enum.GetValues<HistoryAggregateType>().Select(OpcUaClientDriver.MapAggregateToNodeId).ToList();
|
||||
ids.Distinct().Count().ShouldBe(ids.Count);
|
||||
Enum.GetValues<HistoryAggregateType>().Length.ShouldBeGreaterThan(5);
|
||||
}
|
||||
|
||||
/// <summary>Verifies MapAggregateToNodeId returns standard Part 13 aggregate for every enum.</summary>
|
||||
/// <param name="agg">The history aggregate type to test.</param>
|
||||
[Theory]
|
||||
[InlineData(HistoryAggregateType.Average)]
|
||||
[InlineData(HistoryAggregateType.Minimum)]
|
||||
[InlineData(HistoryAggregateType.Maximum)]
|
||||
[InlineData(HistoryAggregateType.Total)]
|
||||
[InlineData(HistoryAggregateType.Count)]
|
||||
[MemberData(nameof(AllAggregates))]
|
||||
public void MapAggregateToNodeId_returns_standard_Part13_aggregate_for_every_enum(HistoryAggregateType agg)
|
||||
{
|
||||
var nodeId = OpcUaClientDriver.MapAggregateToNodeId(agg);
|
||||
|
||||
@@ -268,6 +268,52 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>#285: a Part 13 aggregate beyond the original five (Interpolative) is forwarded to the
|
||||
/// backend as its own <see cref="HistoryAggregateType"/> member.</summary>
|
||||
[Fact]
|
||||
public async Task Processed_extended_aggregate_dispatches_to_the_source()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
var nm = server.NodeManager!;
|
||||
var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() };
|
||||
nm.HistorianDataSource = fake;
|
||||
nm.EnsureVariable("eq-1/interp", parentFolderNodeId: null, displayName: "I", dataType: "Float",
|
||||
writable: false, historianTagname: "WW.Interp", realm: AddressSpaceRealm.Uns);
|
||||
var nodeId = nm.TryGetVariable("eq-1/interp")!.NodeId;
|
||||
|
||||
InvokeHistoryRead(server, nm, new ReadProcessedDetails
|
||||
{
|
||||
StartTime = DateTime.UtcNow.AddHours(-1), EndTime = DateTime.UtcNow, ProcessingInterval = 10_000.0,
|
||||
AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_Interpolative },
|
||||
}, nodeId);
|
||||
|
||||
fake.LastCall.ShouldBe("Processed");
|
||||
fake.LastAggregate.ShouldBe(HistoryAggregateType.Interpolative);
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>#285: a backend that cannot serve a mapped aggregate answers BadAggregateNotSupported
|
||||
/// (not the generic BadHistoryOperationUnsupported a plain throw produces).</summary>
|
||||
[Fact]
|
||||
public async Task Processed_backend_unsupported_aggregate_yields_BadAggregateNotSupported()
|
||||
{
|
||||
var (host, server) = await BootAsync();
|
||||
var nm = server.NodeManager!;
|
||||
nm.HistorianDataSource = new RecordingHistorianDataSource { AggregateNotSupported = true };
|
||||
nm.EnsureVariable("eq-1/ns", parentFolderNodeId: null, displayName: "N", dataType: "Float",
|
||||
writable: false, historianTagname: "WW.Ns", realm: AddressSpaceRealm.Uns);
|
||||
var nodeId = nm.TryGetVariable("eq-1/ns")!.NodeId;
|
||||
|
||||
var (_, errors) = InvokeHistoryRead(server, nm, new ReadProcessedDetails
|
||||
{
|
||||
StartTime = DateTime.UtcNow.AddHours(-1), EndTime = DateTime.UtcNow, ProcessingInterval = 10_000.0,
|
||||
AggregateType = new NodeIdCollection { ObjectIds.AggregateFunction_PercentGood },
|
||||
}, nodeId);
|
||||
|
||||
errors[0].StatusCode.Code.ShouldBe(StatusCodes.BadAggregateNotSupported);
|
||||
await host.DisposeAsync();
|
||||
}
|
||||
|
||||
/// <summary>AtTime read: the requested timestamps reach ReadAtTimeAsync in order.</summary>
|
||||
[Fact]
|
||||
public async Task AtTime_dispatches_requested_timestamps()
|
||||
@@ -430,6 +476,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
||||
private sealed class RecordingHistorianDataSource : IHistorianDataSource
|
||||
{
|
||||
public bool ThrowOnRead { get; init; }
|
||||
public bool AggregateNotSupported { get; init; }
|
||||
public HistorianRead RawResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
|
||||
public HistorianRead ProcessedResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
|
||||
public HistorianRead AtTimeResult { get; set; } = new(Array.Empty<DataValueSnapshot>(), null);
|
||||
@@ -461,6 +508,7 @@ public sealed class NodeManagerHistoryReadTests : IDisposable
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||||
{
|
||||
if (ThrowOnRead) throw new InvalidOperationException("backend boom");
|
||||
if (AggregateNotSupported) throw new HistoryAggregateNotSupportedException(aggregate);
|
||||
LastCall = "Processed";
|
||||
LastTagname = fullReference;
|
||||
LastStart = startUtc;
|
||||
|
||||
Reference in New Issue
Block a user