Files
lmxopcua/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/TimeWeightedAggregator.cs
T
Joseph Doherty 0d206a0884 fix(historian): #680 review — recovery/only-sample TA2/Total2 value, outage-last + Boolean parity, docs
TimeAverage2/Total2: the interval the data ends inside with no non-Bad time
before its last sample (comms-loss recovery, a single sample) answers that
sample's value (Total2 0), Uncertain + Partial, instead of Bad/BadNoData —
as the SDK. Parity datasets RecoveryEnds, OutageLast, BoolSwitch with
documented masks; ProcessedConformance's reverse TA/Total mask narrowed to
the later-edge-sample interval. Docs: report-by-exception consequence,
outage-last and Boolean differences; CLAUDE.md historian bullet.
2026-10-11 16:02:22 -04:00

601 lines
34 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common;
/// <summary>
/// The OPC UA Part 13 time-weighted aggregates (<see cref="AggregateFamily.TimeWeighted"/>): <c>TimeAverage</c>,
/// <c>TimeAverage2</c>, <c>Total</c>, <c>Total2</c>, <c>Interpolative</c> and the duration-based <c>PercentGood</c> /
/// <c>PercentBad</c> (#680), evaluated over a time-ordered stream of stored samples. Both SQL backends fetch the same rows and feed them here, so the two produce identical results
/// by construction.
/// </summary>
/// <remarks>
/// <para>
/// <b>Input.</b> Samples in strictly ascending time: the bounding rows before the window (the latest
/// <i>usable</i> sample before <c>start</c>, then the latest sample of any quality before <c>start</c> — they differ
/// when an outage straddles the start), every in-window sample, then the bounding rows at/after the window end
/// (the first sample of any quality at or after <c>end</c>, then the first <i>usable</i> one). <see cref="Add"/>
/// ignores a sample at or before the last one, so a bound fetched twice is harmless. Memory is O(buckets), not O(samples).
/// </para>
/// <para>
/// <b>The signal (server defaults: TreatUncertainAsBad = false, PercentDataBad = PercentDataGood = 100,
/// UseSlopedExtrapolation = false; the node manager refuses any other client <c>AggregateConfiguration</c>).</b> Between a usable
/// sample <c>p</c> and the next sample <c>s</c>:
/// <list type="bullet">
/// <item><c>s</c> usable, both Double/Int64 ⇒ <b>sloped</b> (linear <c>p → s</c>); quality = worse of the two.</item>
/// <item><c>s</c> usable, either Boolean ⇒ <b>stepped</b> (<c>p</c> held); quality = <c>p</c>'s.</item>
/// <item><c>s</c> Bad ⇒ <c>p</c> held up to the outage (never interpolated into it, as AtTime does); Uncertain.</item>
/// <item>no later sample at all (nothing at or after the window end) ⇒ the data has ENDED at <c>p</c>: no
/// extrapolation (Part 13 "After End of Data", #680).</item>
/// </list>
/// From a Bad sample (incl. an <c>Empty</c> outage marker) to the next sample the signal is Bad (outage) — to the
/// window end when nothing follows it; before the first sample there is no data. This is the <i>simple bounding</i>
/// signal.
/// </para>
/// <para>
/// The <i>interpolated bounding</i> signal (Part 13 "interpolated bounding values") skips Bad samples: it joins
/// consecutive <b>usable</b> samples the same way (sloped / stepped) straight across any outage, and holds the
/// last usable value to the window end. It is undefined before the first usable sample.
/// </para>
/// <list type="table">
/// <item><term>TimeAverage</term><description>∫ interpolated-bounding signal ÷ its defined duration in the interval.</description></item>
/// <item><term>Total</term><description>∫ interpolated-bounding signal over its defined part, in value·s — Part 13
/// §5.4.3.8 (= <c>TimeAverage × interval seconds</c> on a covered interval; only the covered part when the data
/// starts inside it, #680).</description></item>
/// <item><term>TimeAverage2</term><description>∫ simple-bounding signal over its non-Bad parts ÷ non-Bad duration (Bad intervals excluded).</description></item>
/// <item><term>Total2</term><description>∫ simple-bounding signal over its non-Bad parts, in value·s (= TimeAverage2 × non-Bad seconds).</description></item>
/// <item><term>PercentGood / PercentBad</term><description><c>100 ×</c> Good / Bad region time ÷ interval duration
/// (Part 13 §5.4.3.31-34): stepped regions, each with the status of the sample that starts it; the region before an
/// interval's first own sample has the simple bounding value's status (Bad after a Bad sample; Uncertain on a
/// segment running into an outage or sloping to/from an Uncertain sample; else the earlier sample's — the SDK's
/// <c>GetSimpleBound</c>); time before the first sample is Bad; at the end of the data the last region runs to the
/// end of its own interval (Part 13; the SDK gives it 1 ms) and later intervals are <c>BadNoData</c>; an interval
/// ending before the first sample ⇒ <c>BadNoData</c>. Status Good; any value kind.</description></item>
/// <item><term>Interpolative</term><description>The interpolated-bounding signal's value at each interval start —
/// for a reverse read, at <c>End − i·interval</c> (Part 13 anchors a reverse read at its later time), emitted
/// latest first.</description></item>
/// </list>
/// <para>
/// <b>Status (TimeAverage / Total, #680)</b> — Part 13 ("Uncertain if Bad values are skipped or Uncertain values are
/// used") and the SDK's <c>AverageAggregateCalculator</c>: no defined value in the interval ⇒ <c>BadNoData</c>; any
/// part of it on a bridged segment with an Uncertain end (or Uncertain start when stepped), bridged across a Bad
/// sample, extrapolated past the last usable sample, or not covered at all ⇒ <c>UncertainDataSubNormal</c> (so the
/// latest interval of a live tag is Uncertain, and an interval wholly inside an outage has the bridged value,
/// Uncertain); else <c>Good</c>.
/// </para>
/// <para>
/// <b>Status (TimeAverage2 / Total2)</b> — the SDK <c>AggregateCalculator.GetTimeBasedStatusCode</c> rule with
/// TreatUncertainAsBad = false, so Uncertain time (Uncertain samples, a value held up to an outage) counts as Good
/// time, over the interval cut at the end of the data: after the end of the data ⇒ <c>BadNoData</c>; the data
/// ending exactly on the interval start ⇒ the last sample's value (Total2 0) with its status; every covered instant
/// non-Bad ⇒ <c>Good</c>; no non-Bad time but some Bad (outage) time ⇒ <c>Bad</c> (Part 13's PercentDataBad rule;
/// the SDK answers <c>BadNoData</c>); no data at all ⇒ <c>BadNoData</c>; otherwise (PercentDataGood = PercentDataBad
/// = 100) <c>UncertainDataSubNormal</c>. Time before the first sample counts as Bad (no data) time.
/// </para>
/// <para>
/// For the four, a usable String/DateTime sample contributing to the interval ⇒ <c>BadAggregateInvalidInputs</c>.
/// Every status of this family but Interpolative's carries the Calculated bit, plus — on a non-Bad result only — the
/// Partial bit on: the truncated interval (Part 13 §5.3.3.2 Case 3; the SDK does not set it — deliberate); the
/// interval the data starts inside in the read direction (forward: the first sample fed lies strictly within it,
/// #677; reverse: the latest sample fed lies strictly within it, nothing after — the SDK's <c>m_startOfData</c>,
/// keyed on the first value it is fed, #680); and, forward, the end-of-data interval of TimeAverage2 / Total2 /
/// PercentGood / PercentBad (the SDK's <c>m_endOfData</c>, which it skips for an extrapolated TimeAverage / Total).
/// The timestamp is the interval's <see cref="BucketPlan.BucketTimestamp"/> (its start forward, its later edge
/// reverse); the value is a <see cref="double"/> (Boolean = 0/1, Int64 as double).
/// </para>
/// <para>
/// <b>Status (Interpolative).</b> A usable sample exactly at the point ⇒ its value with its own status
/// (raw, no historian bit). Otherwise no usable sample at or before ⇒ <c>BadNoData</c>; else the interpolated
/// value with the Interpolated bit and the status of the simple-bounding signal at that instant: <c>Good</c> on
/// a Good segment, <c>UncertainDataSubNormal</c> on an Uncertain one, inside an outage (interpolated across it)
/// or past the last sample (extrapolated). A reverse read returns the raw sample at a point exactly on a sample
/// too — the SDK's reverse answer (Interpolated, from the segment beyond) differs from its own forward one; that
/// asymmetry is the SDK's defect, deliberately not copied.
/// </para>
/// </remarks>
public sealed class TimeWeightedAggregator
{
private const double TicksPerSecond = TimeSpan.TicksPerSecond;
/// <summary>
/// Default cap on the raw in-window rows one time-weighted read may stream (<c>ServerHistorian:MaxTimeWeightedRawRows</c>).
/// </summary>
public const int DefaultMaxWindowRows = 1_000_000;
/// <summary>
/// The <c>TOP</c>/<c>LIMIT</c> a backend puts on its window query for a configured cap: <c>cap + 1</c> (so one
/// row past the cap proves it was exceeded without reading on), or <see cref="int.MaxValue"/> when the cap is
/// &lt;= 0 (unbounded).
/// </summary>
/// <param name="maxWindowRows">The configured cap.</param>
/// <returns>The row limit to put on the window query.</returns>
public static int WindowFetchLimit(int maxWindowRows) =>
maxWindowRows <= 0 || maxWindowRows == int.MaxValue ? int.MaxValue : maxWindowRows + 1;
/// <summary>Throws when <paramref name="rowsRead"/> has passed a positive <paramref name="maxWindowRows"/>.</summary>
/// <param name="rowsRead">Window rows read so far.</param>
/// <param name="maxWindowRows">The configured cap (&lt;= 0 = unbounded).</param>
/// <param name="tag">The tag, for the message.</param>
/// <exception cref="HistoryReadLimitExceededException">The cap is exceeded.</exception>
public static void CheckWindowRows(long rowsRead, int maxWindowRows, string tag)
{
if (maxWindowRows > 0 && rowsRead > maxWindowRows)
throw new HistoryReadLimitExceededException(maxWindowRows,
$"The time-weighted aggregate read of '{tag}' covers more than {maxWindowRows} raw samples " +
"(ServerHistorian:MaxTimeWeightedRawRows); narrow the window or use a pushed-down aggregate such as Average.");
}
private readonly HistoryAggregateType _aggregate;
private readonly BucketPlan _plan;
private readonly long _start;
private readonly long _end;
// Simple-bounding (raw) signal, per bucket: ∫ over non-Bad time, non-Bad ticks, ticks inside an outage.
private readonly double[] _rawArea;
private readonly long[] _usableTicks;
private readonly long[] _badTicks;
// Interpolated-bounding (bridged) signal, per bucket.
private readonly double[] _bridgedArea;
private readonly long[] _bridgedTicks;
// A bridged segment touching the bucket was non-Good: an Uncertain end, a Bad sample skipped, or extrapolated (#680).
private readonly bool[] _bridgedNonGood;
private readonly bool[] _invalid;
// PercentGood / PercentBad (#680): stepped status regions, per bucket.
private readonly long[] _goodRegionTicks;
private readonly long[] _badRegionTicks;
// Interpolative, per point (index = ascending time; see PointTicks).
private readonly DataValueSnapshot?[] _points;
private StoredSample? _prevAny;
private StoredSample? _prevUsable;
private DateTime? _firstSampleUtc; // the earliest sample fed (any quality): where the data starts
private long? _dataEndTicks; // the last sample fed, when usable and nothing at/after the end followed
private long _nextPoint; // next point (ascending time) not yet classified
private readonly List<PendingPoint> _pending = []; // points awaiting the next usable sample
private bool _finished;
/// <summary>
/// What the simple-bounding signal is at a pending bucket start: no sample yet; on a segment from a usable sample
/// whose end is not known yet; or degraded (inside an outage, or held up to one).
/// </summary>
private enum PointQuality { NoPrior, OnSegment, Degraded }
private readonly record struct PendingPoint(long Bucket, long Ticks, PointQuality Quality);
/// <summary>Starts an evaluation of <paramref name="aggregate"/> over <paramref name="plan"/>.</summary>
/// <param name="aggregate">One of the <see cref="AggregateFamily.TimeWeighted"/> aggregates.</param>
/// <param name="plan">The bucket plan (forward window).</param>
/// <exception cref="HistoryAggregateNotSupportedException">The aggregate is not time-weighted.</exception>
public TimeWeightedAggregator(HistoryAggregateType aggregate, BucketPlan plan)
{
ArgumentNullException.ThrowIfNull(plan);
if (AggregateFamilies.FamilyOf(aggregate) != AggregateFamily.TimeWeighted)
throw new HistoryAggregateNotSupportedException(aggregate);
_aggregate = aggregate;
_plan = plan;
_start = plan.StartUtc.Ticks;
_end = plan.EndUtc.Ticks;
var n = (int)plan.BucketCount;
_rawArea = new double[n];
_usableTicks = new long[n];
_badTicks = new long[n];
_bridgedArea = new double[n];
_bridgedTicks = new long[n];
_bridgedNonGood = new bool[n];
_invalid = new bool[n];
_goodRegionTicks = new long[n];
_badRegionTicks = new long[n];
_points = new DataValueSnapshot?[n];
}
/// <summary>Evaluates <paramref name="aggregate"/> over an already time-ordered sample sequence.</summary>
/// <param name="aggregate">One of the <see cref="AggregateFamily.TimeWeighted"/> aggregates.</param>
/// <param name="plan">The bucket plan.</param>
/// <param name="samples">Bounds + window samples, ascending by time.</param>
/// <param name="now">Server timestamp for the results.</param>
/// <returns>One value per bucket, in the plan's order (descending for a reverse window).</returns>
public static IReadOnlyList<DataValueSnapshot> Evaluate(
HistoryAggregateType aggregate, BucketPlan plan, IEnumerable<StoredSample> samples, DateTime now)
{
ArgumentNullException.ThrowIfNull(samples);
var acc = new TimeWeightedAggregator(aggregate, plan);
foreach (var s in samples) acc.Add(s);
return acc.Finish(now);
}
/// <summary>Feeds the next sample (ascending time; a sample at or before the previous one is ignored).</summary>
/// <param name="s">The sample.</param>
public void Add(StoredSample s)
{
if (_finished) throw new InvalidOperationException("The aggregation is already finished.");
var t = s.TimeUtc.Ticks;
if (_prevAny is { } last && t <= last.TimeUtc.Ticks) return;
if (_firstSampleUtc is null)
{
_firstSampleUtc = s.TimeUtc;
AddRegion(_badRegionTicks, _start, t); // PercentBad: the time before the first sample is Bad (no data)
}
// Interpolative: classify every bucket start before this sample by the simple-bounding signal there.
ClassifyPointsBefore(t);
// Simple-bounding segment [prevAny, s).
if (_prevAny is { IsUsable: true } p)
{
// Toward a usable sample: sloped or stepped; toward a Bad one: held up to the outage.
var sloped = s.IsUsable && IsSloped(p, s);
AddRaw(p, t, sloped ? s.Value.Number : p.Value.Number);
}
else if (_prevAny is { } bad)
{
AddBad(bad, t);
}
if (_prevAny is { } r) AddStatusRegion(r, s, t);
if (!s.IsUsable) DegradePending(); // the segment the pending points sit on runs into an outage
if (s.IsUsable)
{
if (_prevUsable is { } pu)
{
var sloped = IsSloped(pu, s);
// Part 13: Uncertain when a Bad value was skipped or an Uncertain value is used (the end sample only
// contributes its value to a sloped segment).
var nonGood = !pu.IsGood || (sloped && !s.IsGood) || _prevAny is { IsUsable: false };
AddBridged(pu, t, sloped ? s.Value.Number : pu.Value.Number, nonGood);
}
ResolvePending(_prevUsable, s);
_prevUsable = s;
}
_prevAny = s;
}
/// <summary>Completes the evaluation (extrapolating past the last sample) and assembles the per-bucket results.</summary>
/// <param name="now">Server timestamp for the results.</param>
/// <returns>One value per bucket, in the plan's order (descending for a reverse window).</returns>
public IReadOnlyList<DataValueSnapshot> Finish(DateTime now)
{
if (!_finished)
{
_finished = true;
ClassifyPointsBefore(long.MaxValue);
if (_prevAny is { IsUsable: true } p && p.TimeUtc.Ticks < _end)
{
// No sample at or after the window end: the data ENDS here (Part 13 "After End of Data"). The simple-
// bounding signal is not extrapolated (TimeAverage2/Total2 cover up to it; later intervals BadNoData), and
// the status region it starts runs to the end of its own interval only (PercentGood/PercentBad).
_dataEndTicks = p.TimeUtc.Ticks;
if (p.TimeUtc.Ticks >= _start && _plan.BucketCount > 0)
AddRegion(RegionTicks(p), p.TimeUtc.Ticks, BucketEnd(FirstBucketAt(p.TimeUtc.Ticks)));
}
else if (_prevAny is { } bad && bad.TimeUtc.Ticks < _end)
{
AddBad(bad, _end); // an outage marker last: the outage runs to the window end
AddRegion(_badRegionTicks, bad.TimeUtc.Ticks, _end);
}
if (_prevUsable is { } pu && pu.TimeUtc.Ticks < _end)
AddBridged(pu, _end, pu.Value.Number, nonGood: true); // stepped extrapolation ⇒ Uncertain
ResolvePending(_prevUsable, next: null);
}
var byBucket = new Dictionary<long, DataValueSnapshot>((int)_plan.BucketCount);
for (long i = 0; i < _plan.BucketCount; i++)
byBucket[i] = Result(i, now);
return _plan.Assemble(byBucket, t => AggregateEvaluator.NoData(t, now));
}
// ------------------------------------------------------------------------------------------ Results
private DataValueSnapshot Result(long i, DateTime now)
{
if (_aggregate == HistoryAggregateType.Interpolative)
return _points[i] is { } point
? point with { ServerTimestampUtc = now }
: new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, PointTime(i), now);
var stamp = _plan.BucketTimestamp(i);
var lo = BucketStartTicks(i);
var hi = BucketEnd(i);
var duration = hi - lo;
// Partial: the truncated interval (the latest forward, the earliest reverse — Part 13 §5.3.3.2 Case 3, which the
// SDK does not implement), or the one the data starts inside IN THE READ DIRECTION (no bounding value on that
// side — the SDK's m_startOfData rule, #677): forward, the earliest sample strictly inside; reverse, the latest
// sample strictly inside (nothing after it — the SDK keys m_startOfData on the first value it is fed, #680).
// Forward, the end-of-data interval (the latest sample inside it, nothing after) is Partial too — never on an
// extrapolated TimeAverage/Total (the SDK's m_endOfData rule, #680). Never on a Bad result.
var latestInside = _plan.Reversed && _prevAny is { } latest
&& latest.TimeUtc.Ticks > lo && latest.TimeUtc.Ticks < hi;
var partial = _plan.IsTruncated(i) || latestInside
|| (!_plan.Reversed && _plan.DataStartsInside(i, _firstSampleUtc));
var endOfData = !_plan.Reversed && _dataEndTicks is { } de && de >= lo && de < hi;
DataValueSnapshot Calc(object? value, uint status, bool endPartial = false)
{
var code = HistorianInfoBits.WithCalculated(status);
if ((partial || endPartial) && !StatusBits.IsBad(status)) code |= HistorianInfoBits.HistorianPartial;
return new DataValueSnapshot(value, code, stamp, now);
}
switch (_aggregate)
{
case HistoryAggregateType.TimeAverage or HistoryAggregateType.Total:
{
// Interpolated bounding: Uncertain when any part of the interval was bridged across a Bad value, used an
// Uncertain value, was extrapolated, or is not covered at all (Part 13; the SDK's AverageAggregateCalculator).
if (_invalid[i]) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs);
if (_bridgedTicks[i] == 0) return AggregateEvaluator.NoData(stamp, now);
var status = _bridgedNonGood[i] || _bridgedTicks[i] < duration
? HistorianStatusCodes.UncertainDataSubNormal
: HistorianStatusCodes.Good;
return _aggregate == HistoryAggregateType.TimeAverage
? Calc(_bridgedArea[i] / _bridgedTicks[i], status)
: Calc(_bridgedArea[i] / TicksPerSecond, status); // ∫ over the covered part only (Part 13 partial width)
}
case HistoryAggregateType.TimeAverage2 or HistoryAggregateType.Total2:
{
if (_invalid[i]) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs);
if (_dataEndTicks is { } dataEnd)
{
// After the end of the data: nothing to aggregate (Part 13 "After End of Data: Bad_NoData").
if (lo > dataEnd) return AggregateEvaluator.NoData(stamp, now);
if (_usableTicks[i] == 0 && dataEnd < hi && _prevAny is { } last)
{
// The data ends inside this interval and no non-Bad time precedes its last sample here (the data
// ends on the interval's start; or the last sample is the first usable one after an outage — the
// comms-loss recovery shape — or the tag's only sample): nothing covered, so the value at that
// instant (Total2 0). Good only when the data ends exactly on the interval start with a Good
// sample (nothing before it in the interval); otherwise the interval also holds Bad / no-data time
// ⇒ Uncertain (the SDK answers the same, modulo its 1 ms artifact).
if (!IsNumeric(last)) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs);
var st = lo == dataEnd && last.IsGood ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
return Calc(_aggregate == HistoryAggregateType.TimeAverage2 ? last.Value.Number : 0d, st, endOfData);
}
duration = Math.Min(hi, dataEnd) - lo;
}
if (_usableTicks[i] == 0)
return _badTicks[i] > 0 ? Calc(null, HistorianStatusCodes.Bad) : AggregateEvaluator.NoData(stamp, now);
var status = _usableTicks[i] == duration ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
return _aggregate == HistoryAggregateType.TimeAverage2
? Calc(_rawArea[i] / _usableTicks[i], status, endOfData)
: Calc(_rawArea[i] / TicksPerSecond, status, endOfData);
}
case HistoryAggregateType.PercentGood or HistoryAggregateType.PercentBad:
{
// No data in or around the interval (it ends before the first sample — the SDK's out-of-range test — or
// starts after the end of the data) ⇒ BadNoData; otherwise the share of the interval's duration.
if (_firstSampleUtc is not { } first || hi < first.Ticks || (_dataEndTicks is { } de2 && lo > de2))
return AggregateEvaluator.NoData(stamp, now);
var ticks = _aggregate == HistoryAggregateType.PercentGood ? _goodRegionTicks[i] : _badRegionTicks[i];
return Calc(100d * ticks / duration, HistorianStatusCodes.Good, endOfData);
}
default:
throw new HistoryAggregateNotSupportedException(_aggregate);
}
}
// ------------------------------------------------------------------------------------------ Segments
private static bool IsNumeric(StoredSample s) =>
s.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64 or HistorianValueKind.Boolean;
/// <summary>Sloped between two usable samples only when both are Double/Int64; anything Boolean is stepped.</summary>
private static bool IsSloped(StoredSample a, StoredSample b) =>
a.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64
&& b.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64;
private long BucketStartTicks(long i) => _plan.LowerTicks(i);
private long BucketEnd(long i) => _plan.UpperTicks(i);
/// <summary>The ascending index of the bucket whose (half-open) span holds tick <paramref name="t"/> (t ≥ window start).</summary>
private long FirstBucketAt(long t) => Math.Max(0, (t - _plan.OriginTicks) / _plan.IntervalTicks);
private void AddRaw(StoredSample from, long toTicks, double toValue) =>
Distribute(from, toTicks, toValue, (i, area, ticks) =>
{
_rawArea[i] += area;
_usableTicks[i] += ticks;
});
/// <summary>Counts <c>[from.t, toTicks)</c> as outage (Bad) time; no value, never marks a bucket invalid.</summary>
private void AddBad(StoredSample from, long toTicks)
{
var lo = Math.Max(from.TimeUtc.Ticks, _start);
var hi = Math.Min(toTicks, _end);
if (lo >= hi || _plan.BucketCount == 0) return;
for (var i = FirstBucketAt(lo); i < _plan.BucketCount; i++)
{
var bs = BucketStartTicks(i);
if (bs >= hi) break;
var a = Math.Max(lo, bs);
var b = Math.Min(hi, BucketEnd(i));
if (a < b) _badTicks[i] += b - a;
}
}
private void AddBridged(StoredSample from, long toTicks, double toValue, bool nonGood) =>
Distribute(from, toTicks, toValue, (i, area, ticks) =>
{
_bridgedArea[i] += area;
_bridgedTicks[i] += ticks;
if (nonGood) _bridgedNonGood[i] = true;
});
// ------------------------------------------------------------------------------------------ PercentGood / PercentBad
/// <summary>The region array a stepped region of <paramref name="s"/>'s status counts toward (null: Uncertain, neither).</summary>
private long[]? RegionTicks(StoredSample s) =>
s.IsGood ? _goodRegionTicks : !s.IsUsable ? _badRegionTicks : null;
/// <summary>
/// The status regions between sample <paramref name="p"/> and the next sample <paramref name="s"/> (at
/// <paramref name="t"/>). Inside the interval <paramref name="p"/> lies in, the region has <paramref name="p"/>'s
/// status (stepped, Part 13 §5.4.3.31-34). In a later interval the region starts at that interval's simple bounding
/// value, whose status is the simple-bounding signal's: Bad after a Bad <paramref name="p"/>; Uncertain when the
/// segment runs into an outage or a sloped one has a non-Good end; else <paramref name="p"/>'s (the SDK's
/// <c>GetSimpleBound</c>).
/// </summary>
private void AddStatusRegion(StoredSample p, StoredSample s, long t)
{
var lo = Math.Max(p.TimeUtc.Ticks, _start);
var hi = Math.Min(t, _end);
if (lo >= hi || _plan.BucketCount == 0) return;
long[]? bound;
if (!p.IsUsable) bound = _badRegionTicks;
else if (!s.IsUsable) bound = null; // held into an outage ⇒ Uncertain
else if (IsSloped(p, s)) bound = p.IsGood && s.IsGood ? _goodRegionTicks : null;
else bound = RegionTicks(p);
var own = RegionTicks(p);
for (var i = FirstBucketAt(lo); i < _plan.BucketCount; i++)
{
var bs = BucketStartTicks(i);
if (bs >= hi) break;
var a = Math.Max(lo, bs);
var b = Math.Min(hi, BucketEnd(i));
if (a >= b) continue;
var target = p.TimeUtc.Ticks >= bs ? own : bound;
if (target is not null) target[i] += b - a;
}
}
/// <summary>Adds <c>[from, to)</c>, clipped to the window, to <paramref name="target"/> per bucket.</summary>
private void AddRegion(long[]? target, long from, long to)
{
if (target is null) return;
var lo = Math.Max(from, _start);
var hi = Math.Min(to, _end);
if (lo >= hi || _plan.BucketCount == 0) return;
for (var i = FirstBucketAt(lo); i < _plan.BucketCount; i++)
{
var bs = BucketStartTicks(i);
if (bs >= hi) break;
var a = Math.Max(lo, bs);
var b = Math.Min(hi, BucketEnd(i));
if (a < b) target[i] += b - a;
}
}
/// <summary>
/// Splits the linear segment <c>[from.t, toTicks)</c> (value <c>from.v → toValue</c>) at the bucket edges, clipped to
/// the window, and hands each bucket its trapezoid area (value·ticks) and duration. A non-numeric start sample
/// marks every bucket it touches invalid instead.
/// </summary>
private void Distribute(StoredSample from, long toTicks, double toValue, Action<long, double, long> add)
{
var t0 = from.TimeUtc.Ticks;
var lo = Math.Max(t0, _start);
var hi = Math.Min(toTicks, _end);
if (lo >= hi || _plan.BucketCount == 0) return;
var numeric = IsNumeric(from);
var v0 = from.Value.Number;
var span = (double)(toTicks - t0);
double ValueAt(long x) => toValue == v0 ? v0 : v0 + (toValue - v0) * ((x - t0) / span);
for (var i = FirstBucketAt(lo); i < _plan.BucketCount; i++)
{
var bs = BucketStartTicks(i);
if (bs >= hi) break;
var a = Math.Max(lo, bs);
var b = Math.Min(hi, BucketEnd(i));
if (a >= b) continue;
if (!numeric) { _invalid[i] = true; continue; }
add(i, (ValueAt(a) + ValueAt(b)) / 2d * (b - a), b - a);
}
}
// ------------------------------------------------------------------------------------------ Interpolative
/// <summary>
/// The time of Interpolative point <paramref name="j"/> (ascending time): the bucket's timestamp. Forward: the
/// bucket starts <c>start + j·interval</c>. Reverse (Part 13 anchors a reverse read at its later time, the
/// request's start): the later edges <c>End, End − interval, …</c> once <see cref="BucketPlan.Assemble"/> has
/// reversed the order — so a reverse read 10:00 → 09:00 at 1 h yields the value at 10:00.
/// </summary>
private long PointTicks(long j) => _plan.BucketTimestamp(j).Ticks;
private DateTime PointTime(long j) => new(PointTicks(j), DateTimeKind.Utc);
/// <summary>Classifies every unclassified point strictly before <paramref name="beforeTicks"/>.</summary>
private void ClassifyPointsBefore(long beforeTicks)
{
if (_aggregate != HistoryAggregateType.Interpolative) return;
for (; _nextPoint < _plan.BucketCount; _nextPoint++)
{
var a = PointTicks(_nextPoint);
if (a >= beforeTicks) break;
if (_prevAny is not { } p)
{
_pending.Add(new PendingPoint(_nextPoint, a, PointQuality.NoPrior));
}
else if (p.IsUsable && p.TimeUtc.Ticks == a)
{
// A usable sample exactly at the interval start: its own value and status.
_points[_nextPoint] = IsNumeric(p)
? new DataValueSnapshot(p.Value.Number, p.Status, PointTime(_nextPoint), default)
: InvalidPoint(_nextPoint);
}
else if (!p.IsUsable)
{
_pending.Add(new PendingPoint(_nextPoint, a, PointQuality.Degraded));
}
else
{
// Inside a segment that starts at a usable sample; its quality is only known once the next sample
// arrives (beforeTicks is that sample's time, or MaxValue at Finish ⇒ extrapolated).
_pending.Add(new PendingPoint(_nextPoint, a, PointQuality.OnSegment));
}
}
}
/// <summary>Resolves the pending bucket starts against the bridging pair <c>(p, next)</c>.</summary>
private void ResolvePending(StoredSample? p, StoredSample? next)
{
if (_pending.Count == 0) return;
foreach (var pt in _pending)
{
var bucketStart = PointTime(pt.Bucket);
if (p is not { } prev || pt.Quality == PointQuality.NoPrior)
{
_points[pt.Bucket] = new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, bucketStart, default);
continue;
}
if (!IsNumeric(prev) || next is { } nn && !IsNumeric(nn))
{
_points[pt.Bucket] = InvalidPoint(pt.Bucket);
continue;
}
// Good only on a segment that runs straight from prev to next (no outage skipped, not extrapolated), and only
// when the samples that define the value are Good: both ends when sloped, prev alone when stepped.
double value;
bool good;
if (next is { } n && IsSloped(prev, n))
{
value = prev.Value.Number + (n.Value.Number - prev.Value.Number)
* ((double)(pt.Ticks - prev.TimeUtc.Ticks) / (n.TimeUtc.Ticks - prev.TimeUtc.Ticks));
good = pt.Quality == PointQuality.OnSegment && prev.IsGood && n.IsGood;
}
else
{
value = prev.Value.Number;
good = pt.Quality == PointQuality.OnSegment && next is not null && prev.IsGood;
}
var status = good ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal;
_points[pt.Bucket] = new DataValueSnapshot(value, HistorianInfoBits.WithInterpolated(status), bucketStart, default);
}
_pending.Clear();
}
/// <summary>Marks every pending on-segment point degraded: its segment ends at a Bad sample.</summary>
private void DegradePending()
{
for (var k = 0; k < _pending.Count; k++)
if (_pending[k].Quality == PointQuality.OnSegment)
_pending[k] = _pending[k] with { Quality = PointQuality.Degraded };
}
private DataValueSnapshot InvalidPoint(long bucket) =>
new(null, HistorianStatusCodes.BadAggregateInvalidInputs, PointTime(bucket), default);
}