refactor(#655): delete the dead driver-side history capability
HistoryRead is served server-side by IHistorianDataSource; nothing ever dispatched to a driver's IHistoryProvider. Removes IHistoryProvider, the OpcUaClient history passthrough, DriverCapability.HistoryRead (+ resilience tier rows and AdminUI row), HistoryReadStatusException and its node-manager catches, and the tests that only covered them. Shared result types and HistoryAggregateNotSupportedException stay (moved to Historian/HistoryReadContracts.cs). A stored ResilienceConfig still carrying a HistoryRead key parses (diagnostic only; new test). Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
@@ -29,7 +29,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.OpcUaClient;
|
||||
/// monitored-item handles. That mechanic lands in PR 69.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, IHistoryProvider, ISkippedTagReporter, IDisposable, IAsyncDisposable
|
||||
public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWritable, ISubscribable, IHostConnectivityProbe, IAlarmSource, ISkippedTagReporter, IDisposable, IAsyncDisposable
|
||||
{
|
||||
/// <summary>Authored tags whose TagConfig carried no usable <c>nodeId</c>, so there is no upstream node
|
||||
/// to dial (Gitea #541). This driver's skip path was the quietest of the twelve: <c>NamespaceMap</c>
|
||||
@@ -2258,335 +2258,6 @@ public sealed class OpcUaClientDriver : IDriver, ITagDiscovery, IReadable, IWrit
|
||||
public string DiagnosticId => $"opcua-alarm-sub-{Id}";
|
||||
}
|
||||
|
||||
// ---- IHistoryProvider (passthrough to upstream server) ----
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<Core.Abstractions.HistoryReadResult> ReadRawAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var details = new ReadRawModifiedDetails
|
||||
{
|
||||
IsReadModified = false,
|
||||
StartTime = startUtc,
|
||||
EndTime = endUtc,
|
||||
NumValuesPerNode = maxValuesPerNode,
|
||||
ReturnBounds = false,
|
||||
};
|
||||
return await ExecuteHistoryReadAsync(fullReference, new ExtensionObject(details), cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<Core.Abstractions.HistoryReadResult> ReadProcessedAsync(
|
||||
string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
|
||||
HistoryAggregateType aggregate, CancellationToken cancellationToken)
|
||||
{
|
||||
var aggregateId = MapAggregateToNodeId(aggregate);
|
||||
var details = new ReadProcessedDetails
|
||||
{
|
||||
StartTime = startUtc,
|
||||
EndTime = endUtc,
|
||||
ProcessingInterval = interval.TotalMilliseconds,
|
||||
AggregateType = [aggregateId],
|
||||
};
|
||||
return await ExecuteHistoryReadAsync(fullReference, new ExtensionObject(details), cancellationToken, aggregate)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<Core.Abstractions.HistoryReadResult> ReadAtTimeAsync(
|
||||
string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken)
|
||||
{
|
||||
var reqTimes = new DateTimeCollection(timestampsUtc);
|
||||
var details = new ReadAtTimeDetails
|
||||
{
|
||||
ReqTimes = reqTimes,
|
||||
UseSimpleBounds = true,
|
||||
};
|
||||
return await ExecuteHistoryReadAsync(fullReference, new ExtensionObject(details), cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Shared HistoryRead wire path — used by Raw/Processed/AtTime. Handles NodeId parse,
|
||||
/// Session.HistoryReadAsync call, per-sample StatusCode passthrough (no translation per §8
|
||||
/// cascading-quality rule), and HistoryData unwrap into <see cref="DataValueSnapshot"/>. A Bad
|
||||
/// per-NODE <c>HistoryReadResult.StatusCode</c> is never read as empty data: <c>BadAggregateNotSupported</c>
|
||||
/// throws <see cref="HistoryAggregateNotSupportedException"/>, any other Bad throws
|
||||
/// <see cref="HistoryReadStatusException"/> (Gitea #649). Good / GoodNoData / GoodMoreData / Uncertain are
|
||||
/// unchanged.
|
||||
/// </summary>
|
||||
private async Task<Core.Abstractions.HistoryReadResult> ExecuteHistoryReadAsync(
|
||||
string fullReference, ExtensionObject historyReadDetails, CancellationToken ct,
|
||||
HistoryAggregateType? aggregate = null)
|
||||
{
|
||||
// Make sure a session exists before queuing on the gate, but do NOT bind the wire call
|
||||
// (or the namespace-relative NodeId parse) to this reference — a reconnect can swap
|
||||
// Session while we wait on _gate. The session actually used is re-read inside the gate.
|
||||
_ = RequireSession();
|
||||
|
||||
await _gate.WaitAsync(ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
// Re-read Session inside the critical section: if a reconnect completed while we
|
||||
// were blocked on _gate, OnReconnectComplete has already swapped in the new session.
|
||||
// NodeId parsing is namespace-relative, so it must also use the current session's
|
||||
// namespace table.
|
||||
var session = Session;
|
||||
if (session is null || !TryResolveTagNode(session, fullReference, out var nodeId))
|
||||
{
|
||||
return new Core.Abstractions.HistoryReadResult([], null);
|
||||
}
|
||||
|
||||
var nodesToRead = new HistoryReadValueIdCollection
|
||||
{
|
||||
new HistoryReadValueId { NodeId = nodeId },
|
||||
};
|
||||
|
||||
var resp = await session.HistoryReadAsync(
|
||||
requestHeader: null,
|
||||
historyReadDetails: historyReadDetails,
|
||||
timestampsToReturn: TimestampsToReturn.Both,
|
||||
releaseContinuationPoints: false,
|
||||
nodesToRead: nodesToRead,
|
||||
ct: ct).ConfigureAwait(false);
|
||||
|
||||
if (resp.Results.Count == 0) return new Core.Abstractions.HistoryReadResult([], null);
|
||||
var r = resp.Results[0];
|
||||
|
||||
// The per-node status decides whether the payload means anything. BadNoData is the one Bad code
|
||||
// that legitimately means "nothing in the window", so it stays an empty result.
|
||||
// CodeBits (not Code) so an info/overflow sub-status bit on the upstream code does not dodge the match.
|
||||
if (StatusCode.IsBad(r.StatusCode) && r.StatusCode.CodeBits != StatusCodes.BadNoData)
|
||||
{
|
||||
if (r.StatusCode.CodeBits == StatusCodes.BadAggregateNotSupported && aggregate is { } agg)
|
||||
throw new HistoryAggregateNotSupportedException(agg);
|
||||
throw new HistoryReadStatusException(r.StatusCode.Code);
|
||||
}
|
||||
|
||||
// Unwrap HistoryData from the ExtensionObject-encoded payload the SDK returns.
|
||||
// Samples stay in chronological order per OPC UA Part 11; cascading-quality
|
||||
// rule: preserve each DataValue's upstream StatusCode + timestamps verbatim.
|
||||
var samples = new List<DataValueSnapshot>();
|
||||
if (r.HistoryData?.Body is HistoryData hd)
|
||||
{
|
||||
var now = DateTime.UtcNow;
|
||||
foreach (var dv in hd.DataValues)
|
||||
{
|
||||
samples.Add(new DataValueSnapshot(
|
||||
Value: dv.Value,
|
||||
StatusCode: dv.StatusCode.Code,
|
||||
SourceTimestampUtc: dv.SourceTimestamp == DateTime.MinValue ? null : dv.SourceTimestamp,
|
||||
ServerTimestampUtc: dv.ServerTimestamp == DateTime.MinValue ? now : dv.ServerTimestamp));
|
||||
}
|
||||
}
|
||||
|
||||
var contPt = r.ContinuationPoint is { Length: > 0 } ? r.ContinuationPoint : null;
|
||||
return new Core.Abstractions.HistoryReadResult(samples, contPt);
|
||||
}
|
||||
finally { _gate.Release(); }
|
||||
}
|
||||
|
||||
/// <summary>Map <see cref="HistoryAggregateType"/> to the OPC UA Part 13 standard aggregate NodeId.</summary>
|
||||
/// <param name="aggregate">The aggregation function type to map.</param>
|
||||
/// <returns>The standard aggregate function NodeId.</returns>
|
||||
internal static NodeId MapAggregateToNodeId(HistoryAggregateType aggregate) => aggregate switch
|
||||
{
|
||||
HistoryAggregateType.Average => ObjectIds.AggregateFunction_Average,
|
||||
HistoryAggregateType.Minimum => ObjectIds.AggregateFunction_Minimum,
|
||||
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),
|
||||
};
|
||||
|
||||
// Canonical BaseEventType select-clause order — MapHistoryEvents maps by these indices.
|
||||
private static readonly string[] EventFieldBrowseNames =
|
||||
[
|
||||
BrowseNames.EventId, // 0
|
||||
BrowseNames.SourceName, // 1
|
||||
BrowseNames.Time, // 2
|
||||
BrowseNames.ReceiveTime, // 3
|
||||
BrowseNames.Message, // 4
|
||||
BrowseNames.Severity, // 5
|
||||
];
|
||||
|
||||
/// <summary>
|
||||
/// Builds the fixed canonical EventFilter the driver sends upstream for HistoryReadEvents —
|
||||
/// the six BaseEventType fields the OtOpcUa server projects (<see cref="HistoricalEvent"/>).
|
||||
/// The clause order is load-bearing: <see cref="MapHistoryEvents"/> reads results by index.
|
||||
/// </summary>
|
||||
/// <returns>An EventFilter with six SimpleAttributeOperand value clauses.</returns>
|
||||
internal static EventFilter BuildBaseEventFilter()
|
||||
{
|
||||
var filter = new EventFilter();
|
||||
foreach (var browseName in EventFieldBrowseNames)
|
||||
{
|
||||
filter.SelectClauses.Add(new SimpleAttributeOperand
|
||||
{
|
||||
TypeDefinitionId = ObjectTypeIds.BaseEventType,
|
||||
BrowsePath = new QualifiedNameCollection { new QualifiedName(browseName) },
|
||||
AttributeId = Attributes.Value,
|
||||
});
|
||||
}
|
||||
return filter;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Maps an upstream <see cref="HistoryEvent"/> (field arrays ordered to match
|
||||
/// <see cref="BuildBaseEventFilter"/>) onto <see cref="HistoricalEvent"/> records. Defensive:
|
||||
/// short / null / wrong-typed fields degrade to null/default rather than throwing.
|
||||
/// </summary>
|
||||
/// <param name="historyEvent">The upstream history-event payload.</param>
|
||||
/// <returns>The mapped historical events in upstream order.</returns>
|
||||
internal static IReadOnlyList<Core.Abstractions.HistoricalEvent> MapHistoryEvents(HistoryEvent historyEvent)
|
||||
{
|
||||
if (historyEvent?.Events is not { Count: > 0 } rows) return [];
|
||||
var result = new List<Core.Abstractions.HistoricalEvent>(rows.Count);
|
||||
foreach (var row in rows)
|
||||
{
|
||||
var fields = row?.EventFields;
|
||||
result.Add(new Core.Abstractions.HistoricalEvent(
|
||||
EventId: CoerceEventId(FieldAt(fields, 0)),
|
||||
SourceName: CoerceString(FieldAt(fields, 1)),
|
||||
EventTimeUtc: CoerceDateTime(FieldAt(fields, 2)),
|
||||
ReceivedTimeUtc: CoerceDateTime(FieldAt(fields, 3)),
|
||||
Message: CoerceString(FieldAt(fields, 4)),
|
||||
Severity: CoerceSeverity(FieldAt(fields, 5))));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
private static object? FieldAt(VariantCollection? fields, int index)
|
||||
=> fields is not null && index < fields.Count ? fields[index].Value : null;
|
||||
|
||||
private static string CoerceEventId(object? value) => value switch
|
||||
{
|
||||
byte[] bytes => Convert.ToBase64String(bytes),
|
||||
string s => s,
|
||||
null => string.Empty,
|
||||
_ => value.ToString() ?? string.Empty,
|
||||
};
|
||||
|
||||
private static string? CoerceString(object? value) => value switch
|
||||
{
|
||||
LocalizedText lt => lt.Text,
|
||||
string s => s,
|
||||
null => null,
|
||||
_ => value.ToString(),
|
||||
};
|
||||
|
||||
// Missing-field sentinel is UTC-kinded so a downstream .ToUniversalTime() can't shift it.
|
||||
private static readonly DateTime MissingTimeSentinel =
|
||||
DateTime.SpecifyKind(DateTime.MinValue, DateTimeKind.Utc);
|
||||
|
||||
private static DateTime CoerceDateTime(object? value)
|
||||
=> value is DateTime dt ? dt : MissingTimeSentinel;
|
||||
|
||||
private static ushort CoerceSeverity(object? value)
|
||||
{
|
||||
try { return value is null ? (ushort)0 : Convert.ToUInt16(value); }
|
||||
catch (Exception ex) when (ex is FormatException or InvalidCastException or OverflowException) { return 0; }
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
// Forwards OPC UA HistoryReadEvents to the upstream server. Sends the fixed canonical
|
||||
// BuildBaseEventFilter and maps the result onto HistoricalEvent (the OtOpcUa server projects
|
||||
// only those six BaseEventType fields, so a richer client filter would be discarded
|
||||
// server-side — the driver supplies the canonical set itself). sourceName is the upstream
|
||||
// event-notifier NodeId to read from (mirrors how fullReference is the upstream NodeId for
|
||||
// raw reads); null/empty resolves to the upstream Server object (i=2253), the standard
|
||||
// server-wide event notifier, and an unparseable id short-circuits to an empty result,
|
||||
// matching the raw path's malformed-NodeId behavior.
|
||||
public async Task<Core.Abstractions.HistoricalEventsResult> ReadEventsAsync(
|
||||
string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
// Confirm a session exists before queuing; the session actually used — and the
|
||||
// namespace-relative source-NodeId parse — are re-read inside the gate so a reconnect
|
||||
// mid-wait can't leave us reading from a closed session.
|
||||
_ = RequireSession();
|
||||
|
||||
await _gate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
// Re-read Session inside the critical section: OnReconnectComplete may have swapped
|
||||
// in a new session while we were blocked on _gate, and the source-NodeId parse is
|
||||
// namespace-relative so it must bind against the current session's namespace table.
|
||||
var session = Session;
|
||||
if (session is null) return new Core.Abstractions.HistoricalEventsResult([], null);
|
||||
|
||||
NodeId notifierNodeId;
|
||||
if (string.IsNullOrEmpty(sourceName))
|
||||
{
|
||||
notifierNodeId = ObjectIds.Server;
|
||||
}
|
||||
// sourceName is an upstream event-notifier node id (not a tag RawPath), so it
|
||||
// re-binds directly against the session via the static resolve.
|
||||
else if (!NamespaceMap.TryResolve(session, sourceName, out var parsed))
|
||||
{
|
||||
return new Core.Abstractions.HistoricalEventsResult([], null);
|
||||
}
|
||||
else
|
||||
{
|
||||
notifierNodeId = parsed;
|
||||
}
|
||||
|
||||
var details = new ReadEventDetails
|
||||
{
|
||||
StartTime = startUtc,
|
||||
EndTime = endUtc,
|
||||
NumValuesPerNode = maxEvents <= 0 ? 0u : (uint)maxEvents,
|
||||
Filter = BuildBaseEventFilter(),
|
||||
};
|
||||
|
||||
var nodesToRead = new HistoryReadValueIdCollection
|
||||
{
|
||||
new HistoryReadValueId { NodeId = notifierNodeId },
|
||||
};
|
||||
|
||||
var resp = await session.HistoryReadAsync(
|
||||
requestHeader: null,
|
||||
historyReadDetails: new ExtensionObject(details),
|
||||
timestampsToReturn: TimestampsToReturn.Both,
|
||||
releaseContinuationPoints: false,
|
||||
nodesToRead: nodesToRead,
|
||||
ct: cancellationToken).ConfigureAwait(false);
|
||||
|
||||
if (resp.Results.Count == 0) return new Core.Abstractions.HistoricalEventsResult([], null);
|
||||
var r = resp.Results[0];
|
||||
|
||||
var events = r.HistoryData?.Body is HistoryEvent he
|
||||
? MapHistoryEvents(he)
|
||||
: [];
|
||||
|
||||
var contPt = r.ContinuationPoint is { Length: > 0 } ? r.ContinuationPoint : null;
|
||||
return new Core.Abstractions.HistoricalEventsResult(events, contPt);
|
||||
}
|
||||
finally { _gate.Release(); }
|
||||
}
|
||||
|
||||
// ---- IHostConnectivityProbe ----
|
||||
|
||||
/// <summary>
|
||||
|
||||
Reference in New Issue
Block a user