diff --git a/CLAUDE.md b/CLAUDE.md index baa899c2..105699bb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1341,7 +1341,7 @@ historian tagname, which is the tag's driver `FullName`. ### Pluggable backends, SQL Server and HistoryUpdate (2026-10) - **Selection:** `ServerHistorian:Backend` (`Gateway` default | `SqlServer` | `QuestDb`; unknown fails host start). `HistorianBackendRegistry` → `IHistorianBackendProvider` builds all five seams; the gateway keys are only required for `Gateway`. `SqlServer` uses `ServerHistorian:SqlServer:{ConnectionString (required), Schema=hist, CommandTimeout=30s, AutoCreateSchema=true}`. -- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; aggregates (13, simple per-sample definitions, not Part 13) are pushed down to T-SQL with bucket math in **100 ns ticks**. Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. +- **SQL Server** (`Driver.Historian.SqlServer`): `hist.Sample` PK `(TagId, SourceTimeUtc)` — writes are insert-if-absent, so outbox redelivery is idempotent. Stores every value kind with its **real status**; per-sample aggregates (12, simple definitions) are pushed down to T-SQL with bucket math in **100 ns ticks**; the Part 13 time-weighted five (TimeAverage/TimeAverage2/Total/Total2/Interpolative, #671 — ⚠️ `Total` is now the time-weighted integral, no longer `SUM`) stream window + 4 bound rows into the shared `TimeWeightedAggregator` (both SQL backends, identical results; client `AggregateConfiguration` ignored). Docker-dev recipe, schema and aggregate semantics: `docs/Historian.md` §"SQL Server backend". Integration tests need `HISTORIAN_SQL_TEST_CONNECTION`. - **QuestDB** (`Driver.Historian.QuestDb`, `ServerHistorian:QuestDb:*`, Npgsql over PG wire; docker-dev overlay `docker-compose.questdb.yml`): same kinds/status/aggregates/HistoryUpdate as SqlServer, soft delete by tombstone (INSERT only — never `UPDATE`/`DELETE`; every read filters `NOT deleted`). ⚠️ **Timestamps are µs** (`QuestDbConnection.Round`/`BindTimestamp`; `IHistorianDataSource.TimestampResolutionTicks` = 10 steps HistoryRead paging), **pre-1970 cannot be stored** (writer drops, HistoryUpdate `BadOutOfRange`), every write waits for WAL visibility, TTL is enforced on write commits only, set Npgsql `Maximum Pool Size` < QuestDB's 64. `docs/Historian.md` §"QuestDB backend". - **Real status fix:** the recorder used to stamp every sample Good (`192`); it now carries the driver's status. ⚠️ On the **Gateway** backend this changes nothing stored: the gateway discards quality (C-002), so AVEVA still stores Good. - **Comms loss is historized as an outage marker** (`Empty` + `BadNoCommunication`, one per tag per outage) on SqlServer; dropped (`DroppedUnsupportedKind`) on Gateway. Outbox records are **format v2** — drain or delete `ContinuousHistorization:OutboxPath` before downgrading a node. diff --git a/docs/Historian.md b/docs/Historian.md index 38d9bd37..0a718d05 100644 --- a/docs/Historian.md +++ b/docs/Historian.md @@ -77,7 +77,7 @@ consumes the historian. The existing enable gates (`ServerHistorian:Enabled`, `A | Value kinds recorded | `Double`, `Int64`, `Boolean` (numeric analog only) | all: `Double`, `Int64`, `Boolean`, `String`, `DateTime`, plus `Empty` (a Bad sample with no value) | same as `SqlServer` | | Status code stored | Sent as DA quality but **discarded** by the gateway (C-002); AVEVA stores every sample Good | Stored bit-for-bit (`StatusCode` int column) | Stored bit-for-bit (unsigned, in a `LONG`) | | Comms-loss / driver-Bad publishes | Dropped, counted in `DroppedUnsupportedKind` | Stored as an `Empty` outage-marker sample carrying the Bad code (see [below](#typed-values-real-status-and-outage-markers)) | Stored as an `Empty` outage-marker sample, as on `SqlServer` | -| Processed aggregates | Average, Minimum, Maximum, Total, Count | The thirteen in the [aggregate table](#aggregates-processed-reads) | The thirteen, pushed down to QuestDB SQL | +| Processed aggregates | Average, Minimum, Maximum, Total, Count | The seventeen in the [aggregate table](#aggregates-processed-reads) (incl. Part 13 time-weighted, #671) | The same seventeen, identical results | | HistoryUpdate | No (`BadHistoryOperationUnsupported`) | Yes (Insert / Replace / Update / DeleteRaw / DeleteAtTime) | Yes — soft delete (tombstones), see [below](#historyupdate-soft-delete) | | Alarm history | gateway `SendEvent` | `Event` table (insert-if-absent) | `{prefix}event` table (dedup upsert on `(ts, event_id)`) | | Provisioning (`EnsureTags`) | numeric types only; others skipped | `MERGE` upsert of every request, all types | Upsert into `{prefix}tag`, all types | @@ -480,7 +480,7 @@ The server supports all four OPC UA HistoryRead variants: | Variant | Node type | CLI `--aggregate` | |---|---|---| | **Raw** | Historized variable | (omit `--aggregate`) | -| **Processed** | Historized variable | `Average`, `Minimum`, `Maximum`, `Count`, `Start`, `End`, `StandardDeviation` (the CLI's own list; `SqlServer` serves thirteen aggregates — see the [aggregate table](#aggregates-processed-reads)) | +| **Processed** | Historized variable | `Average`, `Minimum`, `Maximum`, `Count`, `Start`, `End`, `StandardDeviation` (the CLI's own list; `SqlServer` / `QuestDb` serve seventeen aggregates — see the [aggregate table](#aggregates-processed-reads)) | | **AtTime** | Historized variable | n/a (client supplies exact timestamps) | | **Events** | Equipment folder (event notifier) | n/a | @@ -597,9 +597,9 @@ buckets surface as `BadNoData`. `HistoryAggregateType` carries the wider Part 13 set (Interpolative, TimeAverage, Start, End, Delta, Range, Min/MaxActualTime, Start/EndBound, DeltaBounds, DurationGood/Bad, PercentGood/Bad, WorstQuality, NumberOfTransitions, AnnotationCount, StandardDeviationPopulation, VariancePopulation); `OtOpcUaNodeManager.MapAggregate` -maps each standard aggregate NodeId. The **`SqlServer`** backend serves the thirteen listed in the -[aggregate table](#aggregates-processed-reads); everything else (e.g. `Interpolative`, `TimeAverage`, -`StandardDeviationPopulation`) is `BadAggregateNotSupported` for that node. The **HistorianGateway** backend only has retrieval modes for the original +maps each standard aggregate NodeId (`TimeAverage2` and `Total2` were appended for #671). The **`SqlServer`** and +**`QuestDb`** backends serve the seventeen listed in the [aggregate table](#aggregates-processed-reads); everything else +(e.g. `StartBound`, `DurationGood`, `StandardDeviationPopulation`) is `BadAggregateNotSupported` for that node. The **HistorianGateway** backend only has retrieval modes for the original five (Average, Minimum, Maximum, Total, Count): for any other member `AggregateModeMapper` throws `HistoryAggregateNotSupportedException`, which the HistoryRead path turns into `BadAggregateNotSupported` for that node only (not a generic failure, and not counted as a backend fault). The OPC UA Client driver forwards every @@ -730,10 +730,18 @@ continuation points, so a client-capped read returns only the first page. ### Aggregates (Processed reads) -> **These are simple per-sample definitions, not full OPC UA Part 13.** There is no time-weighting, no interpolated interval-boundary -> value and no partial-interval status. They are pushed down to T-SQL and suit trend-style consumption. Any `HistoryAggregateType` not -> in the table throws `HistoryAggregateNotSupportedException` ⇒ `BadAggregateNotSupported` for that node (e.g. `Interpolative`, -> `TimeAverage`, `StartBound`, `DurationGood`, `StandardDeviationPopulation`). +> **Two kinds.** The *per-sample* aggregates (Average, Minimum, Maximum, Count, Range, Start/End/Delta, Min/MaxActualTime, +> PercentGood/Bad) are simple per-sample definitions — no time-weighting, no interval-boundary values — pushed down to SQL. The +> *time-weighted* aggregates (TimeAverage, TimeAverage2, Total, Total2, Interpolative; #671) follow OPC UA Part 13 with bounding values +> — see [Time-weighted aggregates](#time-weighted-aggregates-part-13-671). Any `HistoryAggregateType` not in the tables throws +> `HistoryAggregateNotSupportedException` ⇒ `BadAggregateNotSupported` for that node (e.g. `StartBound`, `DurationGood`, +> `StandardDeviationPopulation`). +> +> ⚠️ **Release note / migration (#671): `Total` changed meaning on the `SqlServer` and `QuestDb` backends.** It was the plain `SUM` of +> the usable samples in the interval; it is now the Part 13 time-weighted integral, `TimeAverage × interval seconds` (value·seconds), +> matching what the `Gateway` backend's native `Integral` mode always returned. A client that relied on the old sum gets a different +> number for the same data (e.g. two samples 10 and 20 ten seconds apart no longer total 30). There is no switch back; `Count` × +> `Average` reproduces the old sum. **Bucketing.** One result per interval of `[start, end)`; bucket *i* = `[start + i·interval, start + (i+1)·interval)`, start-inclusive / end-exclusive to the tick, the last bucket truncated at `end`. `interval <= 0` ⇒ a single interval `[start, end)`; `start == end` ⇒ no @@ -749,7 +757,7 @@ status carries the historian *Calculated* info bit, `BadNoData` included. | Aggregate | Computed over | Value timestamp | |---|---|---| -| `Average`, `Minimum`, `Maximum`, `Total` | `AVG` / `MIN` / `MAX` / `SUM` of `NumValue` over usable samples | bucket start | +| `Average`, `Minimum`, `Maximum` | `AVG` / `MIN` / `MAX` of `NumValue` over usable samples | bucket start | | `Count` | number of usable samples (`Int32`) | bucket start | | `Range` | `Maximum − Minimum` | bucket start | | `Start`, `End` | first / last usable sample by time (keeps the sample's type) | bucket start | @@ -766,6 +774,52 @@ Status and edge rules: - **`PercentGood` / `PercentBad` and Uncertain samples:** Uncertain samples are counted in the denominator but are **neither Good nor Bad**, so `PercentGood + PercentBad < 100` whenever a bucket holds Uncertain samples. Both report status `Good` for any non-empty bucket. - Everything except `Count` and `Start`/`End` is a `double`; `Int64` samples aggregate as doubles and `Boolean` as 0/1. +### Time-weighted aggregates (Part 13, #671) + +`TimeAverage`, `TimeAverage2`, `Total`, `Total2` and `Interpolative`, evaluated by the shared +`Driver.Historian.Sql.Common/TimeWeightedAggregator.cs` for **both** SQL backends (so `SqlServer` and `QuestDb` agree exactly). +Bucketing (edges, truncated last bucket, reverse windows, the 100 000-interval cap) is the same as above. + +**Server defaults, fixed:** `TreatUncertainAsBad = false`, `PercentDataBad = PercentDataGood = 100`, `UseSlopedExtrapolation = false`. +**A client's `AggregateConfiguration` in the HistoryRead request is ignored** (the node manager does not forward it); every read uses +these defaults. + +**Bounding values.** Besides the window `[start, end)`, each read fetches four single-row seeks: the latest *usable* sample before +`start`, the latest sample of any quality before `start` (they differ when an outage straddles the start), the first sample of any +quality at or after `end`, and the first *usable* one at or after `end`. So a bucket with no samples of its own, between two samples, +still gets a value. + +**The signal.** Between a usable sample `p` and the next sample `s`: both `Double`/`Int64` ⇒ **sloped** (linear), segment quality the +worse of the two; either `Boolean` ⇒ **stepped** (`p` held), quality `p`'s; `s` Bad ⇒ `p` held up to the outage (never interpolated into +it — the AtTime rule), Uncertain; no later sample at all ⇒ `p` held to the window end (stepped extrapolation), Uncertain. From a Bad +sample (incl. an `Empty` outage marker) to the next sample, and before the first sample, the signal is **Bad**. That is the +*simple-bounding* signal. The *interpolated-bounding* signal skips Bad samples: it joins consecutive usable samples (sloped / stepped) +straight across an outage. + +| Aggregate | Value | Bad intervals | +|---|---|---| +| `TimeAverage` | ∫ interpolated-bounding signal ÷ its defined duration | interpolated across (Part 13) | +| `Total` | `TimeAverage × interval seconds` (value·s; the truncated last bucket uses its own length) | interpolated across | +| `TimeAverage2` | ∫ simple-bounding signal over non-Bad time ÷ non-Bad duration | excluded | +| `Total2` | ∫ simple-bounding signal over non-Bad time (value·s) | excluded | +| `Interpolative` | interpolated-bounding value **at the interval start** | interpolated across | + +Status, the four time-weighted values: no non-Bad time in the interval (including an interval entirely inside an outage) ⇒ +`BadNoData`; the whole interval Good ⇒ `Good`; otherwise `UncertainDataSubNormal`. A usable `String`/`DateTime` sample feeding the +interval ⇒ `BadAggregateInvalidInputs`. All carry the Calculated bit; timestamp = interval start; value `double` (Boolean 0/1). + +Status, `Interpolative`: a usable sample exactly at the interval start ⇒ that value with **its own stored status** (no historian bit); +no usable sample at or before ⇒ plain `BadNoData`; otherwise the value with the **Interpolated** bit, `Good` on a segment whose +defining samples are Good, `UncertainDataSubNormal` inside or up to an outage (bridged) and past the last sample (extrapolated). + +**Pushdown trade-off.** Unlike the per-sample aggregates, the time-weighted family is **not** pushed down to SQL: the backend streams +the window's rows plus the four bounds (one batch of five result sets on SQL Server; a bounds `UNION ALL` + a window select on QuestDB) +and the evaluator splits segments at bucket edges in C#. Network traffic is therefore O(samples in the window) rather than O(buckets); +evaluation memory stays O(buckets) (rows are streamed, never materialised). The alternative — the same Part 13 segment +interpolation, edge splitting and two bounding definitions written twice, in T-SQL and QuestDB SQL, and kept bit-identical — was +judged the larger risk. For very long, densely sampled windows prefer the pushed-down `Average` (or a coarser read) over +`TimeAverage`. + ### Health The read client reports `GetHealthSnapshot()` (query/failure counters, consecutive failures, `LastError` — e.g. a missing schema under @@ -822,7 +876,7 @@ the spike that pinned every QuestDB-specific form: [plans/2026-10-10-questdb-spi `questdb/questdb:10.0.1`. QuestDB serves every value kind (`Double`, `Int64`, `Boolean`, `String`, `DateTime`, `Empty`), stores the real status, records comms-loss -outage markers, serves the thirteen aggregates and supports **HistoryUpdate** (soft delete), exactly like `SqlServer`. The same conformance +outage markers, serves the seventeen aggregates (time-weighted ones through the same shared evaluator, so results are identical) and supports **HistoryUpdate** (soft delete), exactly like `SqlServer`. The same conformance suite (`tests/Drivers/Historian.Conformance`) runs against both backends; the differences below are the only ones. ### Configuration (`ServerHistorian:QuestDb`) diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs index 81096ec3..aa602e09 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/HistoryReadContracts.cs @@ -47,6 +47,12 @@ public enum HistoryAggregateType AnnotationCount, StandardDeviationPopulation, VariancePopulation, + + /// Part 13 TimeAverage2: time-weighted average over the interval's non-Bad data only (#671). + TimeAverage2, + + /// Part 13 Total2: time-weighted integral (value·seconds) over the interval's non-Bad data only (#671). + Total2, } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs index afe02efa..1b094cf0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbAggregateQueries.cs @@ -105,6 +105,41 @@ internal static class QuestDbAggregateQueries "SELECT st.*, r.num_value, r.ts " + "FROM st LEFT JOIN (SELECT * FROM r WHERE rn = 1) r ON r.b = st.b ORDER BY st.b"; + /// + /// Time-weighted family (#671), bounds statement — not pushed down (see ). + /// Four LIMIT 1 seeks UNION ALL-ed, each tagged with k: 0 = latest usable sample before the + /// start, 1 = latest sample of any quality before the start, 2 = first sample of any quality at or after the end, + /// 3 = first usable one at or after the end. Parameters: $1 tag, $2/$3 the forward window + /// start / end. Columns: k, then ts, kind, num_value, int_value, str_value, status, server_ts. + /// + /// Prefixed sample table name. + /// The SQL text. + public static string TimeWeightedBounds(string sample) + { + var usable = $" AND status < {NotBadLimit} AND kind <> {EmptyKind}"; + string Seek(int k, string cmp, string filter, string order) => + $"SELECT * FROM (SELECT {k.ToString(CultureInfo.InvariantCulture)} AS k, {SampleColumns} FROM {sample} " + + $"WHERE tag = $1 AND ts {cmp}{filter} AND NOT deleted ORDER BY ts {order} LIMIT 1)"; + return Seek(0, "< $2", usable, "DESC") + " UNION ALL " + Seek(1, "< $2", "", "DESC") + " UNION ALL " + + Seek(2, ">= $3", "", "ASC") + " UNION ALL " + Seek(3, ">= $3", usable, "ASC"); + } + + /// + /// Time-weighted family (#671), window statement: every live sample in [$2, $3) for tag $1, ascending. + /// Columns: ts, kind, num_value, int_value, str_value, status, server_ts. The caller feeds the before-bounds, + /// these rows and the after-bounds, in that order, into — the same evaluator + /// the SQL Server backend uses, so the two agree exactly. Trade-off: the window's rows travel to the server rather + /// than one row per bucket; Part 13's segment interpolation and bucket-edge splitting are then written once + /// instead of twice in two SQL dialects that must stay bit-identical. + /// + /// Prefixed sample table name. + /// The SQL text. + public static string TimeWeightedWindow(string sample) => + $"SELECT {SampleColumns} FROM {sample} WHERE tag = $1 AND ts >= $2 AND ts < $3 AND NOT deleted ORDER BY ts ASC"; + + /// The row columns the time-weighted statements return (after k for the bounds). + private const string SampleColumns = "ts, kind, num_value, int_value, str_value, status, server_ts"; + /// The bucket-index expression (exposed for tests). internal const string BucketExpression = "(cast(ts AS LONG) * 10 - $1) / $2"; diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs index b1d92d01..b987fe55 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb/QuestDbHistorianDataSource.cs @@ -332,9 +332,10 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource /// the statistics of each non-empty bucket (bucket index /// (cast(ts AS LONG)*10 − StartUnixTicks) / IntervalTicks, never SAMPLE BY); /// produces each value and fills the - /// empty buckets with BadNoData. + /// empty buckets with BadNoData. The time-weighted family (#671) instead streams the window plus four bounding + /// rows into , exactly as the SQL Server backend does. /// - /// The aggregate is not one of the thirteen served. + /// The aggregate is not one of the seventeen served. /// The window holds more than intervals. public async Task ReadProcessedAsync( string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, @@ -348,6 +349,15 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource { var now = _time.GetUtcNow().UtcDateTime; var evaluated = new Dictionary(); + if (family == AggregateFamily.TimeWeighted) + { + var tw = new TimeWeightedAggregator(aggregate, plan); + if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false)) + await StreamTimeWeightedAsync(fullReference, plan, tw, cancellationToken).ConfigureAwait(false); + var twResult = tw.Finish(now); + RecordSuccess(); + return new HistoryReadResult(twResult, ContinuationPoint: null); + } if (plan.BucketCount > 0 && await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false)) { var sample = _conn.SampleTable; @@ -384,6 +394,40 @@ public sealed class QuestDbHistorianDataSource : IHistorianDataSource } } + /// + /// Streams the time-weighted bounds () and window + /// () into in time order: the two + /// before-bounds, the window, the two after-bounds (duplicates are dropped by the aggregator). + /// + private async Task StreamTimeWeightedAsync(string tag, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct) + { + var bounds = new StoredSample?[4]; + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.TimeWeightedBounds(_conn.SampleTable))) + { + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindTimestamp(cmd, plan.StartUtc); + QuestDbConnection.BindTimestamp(cmd, plan.EndUtc); + await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await reader.ReadAsync(ct).ConfigureAwait(false)) + bounds[Convert.ToInt32(reader.GetValue(0), CultureInfo.InvariantCulture)] = ReadStored(reader, 1); + } + + if (bounds[0] is { } usableBefore) tw.Add(usableBefore); + if (bounds[1] is { } anyBefore) tw.Add(anyBefore); + await using (var cmd = _conn.CreateCommand(conn, QuestDbAggregateQueries.TimeWeightedWindow(_conn.SampleTable))) + { + QuestDbConnection.BindText(cmd, tag); + QuestDbConnection.BindTimestamp(cmd, plan.StartUtc); + QuestDbConnection.BindTimestamp(cmd, plan.EndUtc); + await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + while (await reader.ReadAsync(ct).ConfigureAwait(false)) + tw.Add(ReadStored(reader, 0)); + } + if (bounds[2] is { } anyAfter) tw.Add(anyAfter); + if (bounds[3] is { } usableAfter) tw.Add(usableAfter); + } + private static BucketRow ReadBucket(NpgsqlDataReader reader, AggregateFamily family) { static double? D(NpgsqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs index f79c3747..ff097de9 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateEvaluator.cs @@ -17,7 +17,8 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// /// Values: Count is ; Start/End keep the sample's type; every other aggregate is /// computed over the numeric value (Boolean = 0/1). These are the simple per-sample -/// definitions, not Part 13's interval-boundary (interpolated bound / time-weighted) semantics. +/// definitions; the Part 13 time-weighted aggregates (TimeAverage, TimeAverage2, Total, Total2, Interpolative) +/// are evaluated by , not here. /// /// public static class AggregateEvaluator @@ -28,7 +29,8 @@ public static class AggregateEvaluator /// The bucket's start time (the result timestamp). /// Server timestamp. /// The calculated value. - /// The aggregate is not one of the thirteen served. + /// The aggregate is not one of the per-sample (Simple / + /// StartEnd / ActualTime family) aggregates. public static DataValueSnapshot Evaluate(HistoryAggregateType aggregate, BucketRow r, DateTime bucketStart, DateTime now) { DataValueSnapshot Calc(object? value, uint status, DateTime? t = null) => @@ -53,7 +55,6 @@ public static class AggregateEvaluator case HistoryAggregateType.Average: return Calc(r.Avg, status); case HistoryAggregateType.Minimum: return Calc(r.Min, status); case HistoryAggregateType.Maximum: return Calc(r.Max, status); - case HistoryAggregateType.Total: return Calc(r.Sum, status); case HistoryAggregateType.Range: return Calc(r.Max - r.Min, status); case HistoryAggregateType.Start: return Calc(r.First.ToObject(), status); case HistoryAggregateType.End: return Calc(r.Last.ToObject(), status); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs index 761a0cab..4b94d3d8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/AggregateFamily.cs @@ -5,7 +5,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; /// The query shape that serves an aggregate. One query per family. public enum AggregateFamily { - /// Average / Minimum / Maximum / Total / Count / Range / PercentGood / PercentBad: one GROUP BY. + /// Average / Minimum / Maximum / Count / Range / PercentGood / PercentBad: one GROUP BY. Simple, /// Start / End / Delta: first + last usable sample per bucket. @@ -13,6 +13,13 @@ public enum AggregateFamily /// MinimumActualTime / MaximumActualTime: extreme usable sample per bucket, ties ⇒ earliest. ActualTime, + + /// + /// TimeAverage / TimeAverage2 / Total / Total2 / Interpolative: Part 13 time-weighted, with bounding values. + /// The backend streams the window's samples plus the bounding rows and + /// evaluates them (identically for every backend). + /// + TimeWeighted, } /// Which aggregates the SQL historian backends serve, and by which family. @@ -24,10 +31,12 @@ public static class AggregateFamilies public static AggregateFamily? FamilyOf(HistoryAggregateType aggregate) => aggregate switch { HistoryAggregateType.Average or HistoryAggregateType.Minimum or HistoryAggregateType.Maximum - or HistoryAggregateType.Total or HistoryAggregateType.Count or HistoryAggregateType.Range + or HistoryAggregateType.Count or HistoryAggregateType.Range or HistoryAggregateType.PercentGood or HistoryAggregateType.PercentBad => AggregateFamily.Simple, HistoryAggregateType.Start or HistoryAggregateType.End or HistoryAggregateType.Delta => AggregateFamily.StartEnd, HistoryAggregateType.MinimumActualTime or HistoryAggregateType.MaximumActualTime => AggregateFamily.ActualTime, + HistoryAggregateType.TimeAverage or HistoryAggregateType.TimeAverage2 or HistoryAggregateType.Total + or HistoryAggregateType.Total2 or HistoryAggregateType.Interpolative => AggregateFamily.TimeWeighted, _ => null, }; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/TimeWeightedAggregator.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/TimeWeightedAggregator.cs new file mode 100644 index 00000000..1c66ca69 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common/TimeWeightedAggregator.cs @@ -0,0 +1,364 @@ +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common; + +/// +/// The OPC UA Part 13 time-weighted aggregates (): TimeAverage, +/// TimeAverage2, Total, Total2 and Interpolative, 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. +/// +/// +/// +/// Input. Samples in strictly ascending time: the bounding rows before the window (the latest +/// usable sample before start, then the latest sample of any quality before start — 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 end, then the first usable one). +/// ignores a sample at or before the last one, so a bound fetched twice is harmless. Memory is O(buckets), not O(samples). +/// +/// +/// The signal (server defaults: TreatUncertainAsBad = false, PercentDataBad = PercentDataGood = 100, +/// UseSlopedExtrapolation = false; a client AggregateConfiguration is not honoured). Between a usable +/// sample p and the next sample s: +/// +/// s usable, both Double/Int64 ⇒ sloped (linear p → s); quality = worse of the two. +/// s usable, either Boolean ⇒ stepped (p held); quality = p's. +/// s Bad ⇒ p held up to the outage (never interpolated into it, as AtTime does); Uncertain. +/// no later sample at all ⇒ p held to the window end (stepped extrapolation); Uncertain. +/// +/// From a Bad sample (incl. an Empty outage marker) to the next sample, and before the first sample, the +/// signal is Bad (no data). This is the simple bounding signal. +/// +/// +/// The interpolated bounding signal (Part 13 "interpolated bounding values") skips Bad samples: it joins +/// consecutive usable 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. +/// +/// +/// TimeAverage∫ interpolated-bounding signal ÷ its defined duration in the interval. +/// TotalTimeAverage × interval seconds (value·s) — Part 13 §5.4.3.8. +/// TimeAverage2∫ simple-bounding signal over its non-Bad parts ÷ non-Bad duration (Bad intervals excluded). +/// Total2∫ simple-bounding signal over its non-Bad parts, in value·s (= TimeAverage2 × non-Bad seconds). +/// InterpolativeThe interpolated-bounding signal's value at each interval start. +/// +/// +/// Status (time-weighted four). No non-Bad time in the interval ⇒ BadNoData; the whole interval +/// Good ⇒ Good; otherwise UncertainDataSubNormal (PercentDataGood = PercentDataBad = 100). A +/// usable String/DateTime sample contributing to the interval ⇒ BadAggregateInvalidInputs. Every status +/// carries the Calculated bit; the timestamp is the interval start; the value is a +/// (Boolean = 0/1, Int64 as double). +/// +/// +/// Status (Interpolative). A usable sample exactly at the interval start ⇒ its value with its own status +/// (raw, no historian bit). Otherwise no usable sample at or before ⇒ BadNoData; else the interpolated +/// value with the Interpolated bit and the status of the simple-bounding signal at that instant: Good on +/// a Good segment, UncertainDataSubNormal on an Uncertain one, inside an outage (interpolated across it) +/// or past the last sample (extrapolated). +/// +/// +public sealed class TimeWeightedAggregator +{ + private const double TicksPerSecond = TimeSpan.TicksPerSecond; + + 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, Good ticks. + private readonly double[] _rawArea; + private readonly long[] _usableTicks; + private readonly long[] _goodTicks; + // Interpolated-bounding (bridged) signal, per bucket. + private readonly double[] _bridgedArea; + private readonly long[] _bridgedTicks; + private readonly bool[] _invalid; + // Interpolative, per bucket start. + private readonly DataValueSnapshot?[] _points; + + private StoredSample? _prevAny; + private StoredSample? _prevUsable; + private long _nextPoint; // next bucket start not yet classified + private readonly List _pending = []; // bucket starts awaiting the next usable sample + private bool _finished; + + /// + /// 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). + /// + private enum PointQuality { NoPrior, OnSegment, Degraded } + + private readonly record struct PendingPoint(long Bucket, long Ticks, PointQuality Quality); + + /// Starts an evaluation of over . + /// One of the aggregates. + /// The bucket plan (forward window). + /// The aggregate is not time-weighted. + 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]; + _goodTicks = new long[n]; + _bridgedArea = new double[n]; + _bridgedTicks = new long[n]; + _invalid = new bool[n]; + _points = new DataValueSnapshot?[n]; + } + + /// Evaluates over an already time-ordered sample sequence. + /// One of the aggregates. + /// The bucket plan. + /// Bounds + window samples, ascending by time. + /// Server timestamp for the results. + /// One value per bucket, in the plan's order (descending for a reverse window). + public static IReadOnlyList Evaluate( + HistoryAggregateType aggregate, BucketPlan plan, IEnumerable samples, DateTime now) + { + ArgumentNullException.ThrowIfNull(samples); + var acc = new TimeWeightedAggregator(aggregate, plan); + foreach (var s in samples) acc.Add(s); + return acc.Finish(now); + } + + /// Feeds the next sample (ascending time; a sample at or before the previous one is ignored). + /// The sample. + 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; + + // 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) + { + if (s.IsUsable) + { + var sloped = IsSloped(p, s); + AddRaw(p, t, sloped ? s.Value.Number : p.Value.Number, sloped ? p.IsGood && s.IsGood : p.IsGood); + } + else + { + AddRaw(p, t, p.Value.Number, good: false); // held up to the outage + } + } + + 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); + AddBridged(pu, t, sloped ? s.Value.Number : pu.Value.Number); + } + ResolvePending(_prevUsable, s); + _prevUsable = s; + } + + _prevAny = s; + } + + /// Completes the evaluation (extrapolating past the last sample) and assembles the per-bucket results. + /// Server timestamp for the results. + /// One value per bucket, in the plan's order (descending for a reverse window). + public IReadOnlyList Finish(DateTime now) + { + if (!_finished) + { + _finished = true; + ClassifyPointsBefore(long.MaxValue); + if (_prevAny is { IsUsable: true } p && p.TimeUtc.Ticks < _end) + AddRaw(p, _end, p.Value.Number, good: false); // stepped extrapolation + if (_prevUsable is { } pu && pu.TimeUtc.Ticks < _end) + AddBridged(pu, _end, pu.Value.Number); + ResolvePending(_prevUsable, next: null); + } + + var byBucket = new Dictionary((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) + { + var bucketStart = _plan.BucketStart(i); + if (_aggregate == HistoryAggregateType.Interpolative) + return _points[i] is { } point + ? point with { ServerTimestampUtc = now } + : new DataValueSnapshot(null, HistorianStatusCodes.BadNoData, bucketStart, now); + + DataValueSnapshot Calc(object? value, uint status) => + new(value, HistorianInfoBits.WithCalculated(status), bucketStart, now); + + if (_invalid[i]) return Calc(null, HistorianStatusCodes.BadAggregateInvalidInputs); + if (_usableTicks[i] == 0) return AggregateEvaluator.NoData(bucketStart, now); + + var duration = BucketEnd(i) - BucketStartTicks(i); + var status = _goodTicks[i] == duration ? HistorianStatusCodes.Good : HistorianStatusCodes.UncertainDataSubNormal; + return _aggregate switch + { + HistoryAggregateType.TimeAverage => Calc(_bridgedArea[i] / _bridgedTicks[i], status), + HistoryAggregateType.Total => Calc(_bridgedArea[i] / _bridgedTicks[i] * (duration / TicksPerSecond), status), + HistoryAggregateType.TimeAverage2 => Calc(_rawArea[i] / _usableTicks[i], status), + HistoryAggregateType.Total2 => Calc(_rawArea[i] / TicksPerSecond, status), + _ => throw new HistoryAggregateNotSupportedException(_aggregate), + }; + } + + // ------------------------------------------------------------------------------------------ Segments + + private static bool IsNumeric(StoredSample s) => + s.Value.Kind is HistorianValueKind.Double or HistorianValueKind.Int64 or HistorianValueKind.Boolean; + + /// Sloped between two usable samples only when both are Double/Int64; anything Boolean is stepped. + 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) => _start + i * _plan.IntervalTicks; + + private long BucketEnd(long i) => Math.Min(BucketStartTicks(i) + _plan.IntervalTicks, _end); + + private void AddRaw(StoredSample from, long toTicks, double toValue, bool good) => + Distribute(from, toTicks, toValue, (i, area, ticks) => + { + _rawArea[i] += area; + _usableTicks[i] += ticks; + if (good) _goodTicks[i] += ticks; + }); + + private void AddBridged(StoredSample from, long toTicks, double toValue) => + Distribute(from, toTicks, toValue, (i, area, ticks) => + { + _bridgedArea[i] += area; + _bridgedTicks[i] += ticks; + }); + + /// + /// Splits the linear segment [from.t, toTicks) (value from.v → toValue) 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. + /// + private void Distribute(StoredSample from, long toTicks, double toValue, Action 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 = (lo - _start) / _plan.IntervalTicks; 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 + + /// Classifies every unclassified bucket start strictly before . + private void ClassifyPointsBefore(long beforeTicks) + { + if (_aggregate != HistoryAggregateType.Interpolative) return; + for (; _nextPoint < _plan.BucketCount; _nextPoint++) + { + var a = BucketStartTicks(_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, _plan.BucketStart(_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)); + } + } + } + + /// Resolves the pending bucket starts against the bridging pair (p, next). + private void ResolvePending(StoredSample? p, StoredSample? next) + { + if (_pending.Count == 0) return; + foreach (var pt in _pending) + { + var bucketStart = _plan.BucketStart(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(); + } + + /// Marks every pending on-segment point degraded: its segment ends at a Bad sample. + 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, _plan.BucketStart(bucket), default); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs index 050cb231..874c8479 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlAggregateQueries.cs @@ -91,6 +91,36 @@ internal static class SqlAggregateQueries "SELECT st.*, r.NumValue, r.SourceTimeUtc " + "FROM st LEFT JOIN r ON r.b = st.b AND r.rn = 1 ORDER BY st.b;"; + /// + /// Time-weighted family (#671: TimeAverage / TimeAverage2 / Total / Total2 / Interpolative) — not pushed down. + /// One batch of five result sets, each SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode: + /// (1) the latest usable sample before @start, (2) the latest sample of any quality before @start, + /// (3) every sample in [@start, @end) ascending, (4) the first sample of any quality at or after + /// @end, (5) the first usable one at or after @end. Each bound is a TOP 1 seek on the + /// (TagId, SourceTimeUtc) key. The caller streams them, in that order, into + /// (shared with QuestDB, so both backends agree exactly). + /// + /// + /// Trade-off: the window's rows travel to the server instead of one row per bucket. Part 13's per-segment + /// interpolation, bucket-edge splitting and the two bounding-value definitions would otherwise have to be written + /// twice (T-SQL and QuestDB SQL) and kept bit-identical; evaluating once in C# keeps the backends equal by + /// construction. Memory stays O(buckets): the rows are streamed, never materialised. + /// + /// Schema-qualified Sample table name. + /// The SQL batch. + public static string TimeWeighted(string sample) + { + const string cols = "SourceTimeUtc, ValueKind, NumValue, IntValue, StrValue, StatusCode"; + const string usable = " AND StatusCode >= 0 AND ValueKind <> 0"; + var key = $"FROM {sample} WHERE TagId = @id AND "; + return + $"SELECT TOP 1 {cols} {key}SourceTimeUtc < @start{usable} ORDER BY SourceTimeUtc DESC; " + + $"SELECT TOP 1 {cols} {key}SourceTimeUtc < @start ORDER BY SourceTimeUtc DESC; " + + $"SELECT {cols} {key}SourceTimeUtc >= @start AND SourceTimeUtc < @end ORDER BY SourceTimeUtc ASC; " + + $"SELECT TOP 1 {cols} {key}SourceTimeUtc >= @end ORDER BY SourceTimeUtc ASC; " + + $"SELECT TOP 1 {cols} {key}SourceTimeUtc >= @end{usable} ORDER BY SourceTimeUtc ASC;"; + } + /// The s CTE: the window's samples with bucket index and the usable/good flags. private static string Classified(string sample) => "WITH s AS (SELECT (DATEDIFF_BIG(microsecond, @start, SourceTimeUtc) * 10 " + diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs index c580cdea..71ade926 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.SqlServer/SqlServerHistorianDataSource.cs @@ -298,10 +298,12 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource /// /// Values: Count is ; Start/End keep the sample's type; every other aggregate is /// computed over NumValue (Boolean = 0/1). These are the simple per-sample - /// definitions, not Part 13's interval-boundary (interpolated bound / time-weighted) semantics. + /// definitions. The Part 13 time-weighted family (TimeAverage, TimeAverage2, Total, Total2, Interpolative, #671) is + /// not pushed down: the window's rows plus four bounding rows are streamed into the shared + /// , which fills every interval (an empty one between bounds gets a value). /// /// - /// The aggregate is not one of the thirteen served. + /// The aggregate is not one of the seventeen served. /// The window holds more than intervals. public async Task ReadProcessedAsync( string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, @@ -316,6 +318,14 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource var now = _time.GetUtcNow().UtcDateTime; var evaluated = new Dictionary(); var tagId = plan.BucketCount == 0 ? null : await ResolveAsync(fullReference, cancellationToken).ConfigureAwait(false); + if (family == AggregateFamily.TimeWeighted) + { + var tw = new TimeWeightedAggregator(aggregate, plan); + if (tagId is { } twId) await StreamTimeWeightedAsync(twId, plan, tw, cancellationToken).ConfigureAwait(false); + var twResult = tw.Finish(now); + RecordSuccess(); + return new HistoryReadResult(twResult, ContinuationPoint: null); + } if (tagId is { } id) { var sample = _conn.Q("Sample"); @@ -351,6 +361,25 @@ public sealed class SqlServerHistorianDataSource : IHistorianDataSource } } + /// + /// Streams the time-weighted batch () into : the + /// two before-bounds, the window, the two after-bounds — ascending, with duplicates dropped by the aggregator. + /// + private async Task StreamTimeWeightedAsync(int id, BucketPlan plan, TimeWeightedAggregator tw, CancellationToken ct) + { + await using var conn = await _conn.OpenAsync(ct).ConfigureAwait(false); + await using var cmd = _conn.CreateCommand(conn, SqlAggregateQueries.TimeWeighted(_conn.Q("Sample"))); + cmd.Parameters.Add(new SqlParameter("@id", SqlDbType.Int) { Value = id }); + SqlHistorianConnection.AddDateTime2(cmd, "@start", DateTime.SpecifyKind(plan.StartUtc, DateTimeKind.Utc)); + SqlHistorianConnection.AddDateTime2(cmd, "@end", DateTime.SpecifyKind(plan.EndUtc, DateTimeKind.Utc)); + await using var reader = await cmd.ExecuteReaderAsync(ct).ConfigureAwait(false); + do + { + while (await reader.ReadAsync(ct).ConfigureAwait(false)) + tw.Add(ReadStored(reader, 0)); + } while (await reader.NextResultAsync(ct).ConfigureAwait(false)); + } + private static BucketRow ReadBucket(SqlDataReader reader, AggregateFamily family) { static double? D(SqlDataReader r, int i) => r.IsDBNull(i) ? null : r.GetDouble(i); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 859b69b2..b2c2d16d 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -6063,6 +6063,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 if (aggregateNodeId == ObjectIds.AggregateFunction_AnnotationCount) return HistoryAggregateType.AnnotationCount; if (aggregateNodeId == ObjectIds.AggregateFunction_StandardDeviationPopulation) return HistoryAggregateType.StandardDeviationPopulation; if (aggregateNodeId == ObjectIds.AggregateFunction_VariancePopulation) return HistoryAggregateType.VariancePopulation; + if (aggregateNodeId == ObjectIds.AggregateFunction_TimeAverage2) return HistoryAggregateType.TimeAverage2; + if (aggregateNodeId == ObjectIds.AggregateFunction_Total2) return HistoryAggregateType.Total2; return null; } diff --git a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs index cd579922..8f30596e 100644 --- a/tests/Drivers/Historian.Conformance/ProcessedConformance.cs +++ b/tests/Drivers/Historian.Conformance/ProcessedConformance.cs @@ -39,7 +39,6 @@ public abstract class ProcessedConformance : ConformanceBase [InlineData(HistoryAggregateType.Average, 15d, 40d, 50d)] [InlineData(HistoryAggregateType.Minimum, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.Maximum, 20d, 40d, 50d)] - [InlineData(HistoryAggregateType.Total, 30d, 40d, 50d)] [InlineData(HistoryAggregateType.Start, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.End, 20d, 40d, 50d)] [InlineData(HistoryAggregateType.Delta, 10d, 0d, 0d)] @@ -121,8 +120,8 @@ public abstract class ProcessedConformance : ConformanceBase (edge, HistorianValue.FromDouble(4), Good), // first tick of B1 (At(40).AddTicks(-B.ResolutionTicks), HistorianValue.FromDouble(8), Good), // last tick of B1 (At(40), HistorianValue.FromDouble(2000), Good)); // == end: excluded - var s = (await B.CreateDataSource().ReadProcessedAsync("E", At(0), At(40), Twenty, HistoryAggregateType.Total, Ct)).Samples; - s.Select(x => x.Value).ShouldBe([3d, 12d]); + var s = (await B.CreateDataSource().ReadProcessedAsync("E", At(0), At(40), Twenty, HistoryAggregateType.Average, Ct)).Samples; + s.Select(x => x.Value).ShouldBe([1.5d, 6d]); s.ShouldAllBe(x => x.StatusCode == GoodCalc); } @@ -164,11 +163,15 @@ public abstract class ProcessedConformance : ConformanceBase (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.Count, Ct)) .Samples.Single().Value.ShouldBe(4); // Also a whole-DateTime-range window (0001 → 9999-12-31T23:59:59.9999999). - (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Total, Ct)) - .Samples.Single().Value.ShouldBe(120d); + (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Count, Ct)) + .Samples.Single().Value.ShouldBe(4); // An interval wider than the window (here TimeSpan.MaxValue) is the same single interval, no overflow. - (await src.ReadProcessedAsync("A", start, end, TimeSpan.MaxValue, HistoryAggregateType.Total, Ct)) - .Samples.Single().Value.ShouldBe(120d); + (await src.ReadProcessedAsync("A", start, end, TimeSpan.MaxValue, HistoryAggregateType.Count, Ct)) + .Samples.Single().Value.ShouldBe(4); + // The time-weighted path over the whole range: ≈ the extrapolated 50 that dominates eight millennia. + var tw = (await src.ReadProcessedAsync("A", start, end, TimeSpan.Zero, HistoryAggregateType.TimeAverage2, Ct)).Samples.Single(); + ((double)tw.Value!).ShouldBe(50d, 1e-6); + tw.StatusCode.ShouldBe(UncertainCalc); src.GetHealthSnapshot().TotalFailures.ShouldBe(0); } @@ -303,7 +306,8 @@ public abstract class ProcessedConformance : ConformanceBase async Task One(HistoryAggregateType a) => (await src.ReadProcessedAsync("B", At(0), At(20), Twenty, a, Ct)).Samples.Single(); (await One(HistoryAggregateType.Average)).Value.ShouldBe(0.75d); - (await One(HistoryAggregateType.Total)).Value.ShouldBe(3d); + // Total is time-weighted (#671): stepped 1 over [0,5), 0 over [5,10), 1 over [10,20) ⇒ 15 value·s. + (await One(HistoryAggregateType.Total)).Value.ShouldBe(15d); (await One(HistoryAggregateType.Start)).Value.ShouldBe(true); (await One(HistoryAggregateType.Count)).Value.ShouldBe(4); } @@ -317,8 +321,6 @@ public abstract class ProcessedConformance : ConformanceBase } [Theory] - [InlineData(HistoryAggregateType.Interpolative)] - [InlineData(HistoryAggregateType.TimeAverage)] [InlineData(HistoryAggregateType.StartBound)] [InlineData(HistoryAggregateType.EndBound)] [InlineData(HistoryAggregateType.DeltaBounds)] @@ -338,6 +340,135 @@ public abstract class ProcessedConformance : ConformanceBase src.GetHealthSnapshot().TotalQueries.ShouldBe(0); } + // ------------------------------------------------------------------------------------------ Time-weighted (#671) + + private const uint GoodRaw = 0x00000000u; + private const uint GoodInterp = 0x00000402u; // Good | InfoType 0x400 | Interpolated 0x2 + private const uint UncertainInterp = 0x40A40402u; // UncertainDataSubNormal | 0x402 + private const uint NoData = 0x809B0000u; + + /// + /// Sparse tag W: −30 = 0 G (bound before the window), +20 = 40 G, +30 = outage marker, +50 = 20 G, +100 = 20 G + /// (bound after the window). Window [t0, t0+80s). Simple-bounding signal: [−30,20) 0→40 sloped Good (24 at +0, + /// 32 at +10); [20,30) 40 held into the outage, Uncertain; [30,50) Bad; [50,100) 20 Good. Interpolated bounding: + /// the same, but [20,50) is sloped 40→20 straight across the outage (100/3 at +30, 80/3 at +40). + /// + private Task SeedWAsync() => B.SeedAsync("W", + (At(-30), HistorianValue.FromDouble(0), Good), + (At(20), HistorianValue.FromDouble(40), Good), + (At(30), HistorianValue.Empty, BadNoCommunication), + (At(50), HistorianValue.FromDouble(20), Good), + (At(100), HistorianValue.FromDouble(20), Good)); + + private async Task> ReadW(HistoryAggregateType agg, double intervalS) => + (await B.CreateDataSource().ReadProcessedAsync("W", At(0), At(80), TimeSpan.FromSeconds(intervalS), agg, Ct)).Samples; + + private static void Values(IReadOnlyList s, params double?[] expected) + { + s.Count.ShouldBe(expected.Length); + for (var i = 0; i < expected.Length; i++) + { + if (expected[i] is { } e) ((double)s[i].Value!).ShouldBe(e, 1e-9, $"bucket {i}"); + else s[i].Value.ShouldBeNull($"bucket {i}"); + } + } + + [Theory] + // 40 s intervals (wider than the sample gaps). B0 [0,40): raw [0,20) 24→40 = 640 Good + [20,30) 40 = 400 Uncertain, + // [30,40) Bad; bridged 640 + [20,40) 40→80/3 = 2000/3. B1 [40,80): raw [50,80) 20 = 600 Good, [40,50) Bad; bridged + // [40,50) 80/3→20 = 700/3 + 600. + [InlineData(HistoryAggregateType.TimeAverage, 98d / 3d, 125d / 6d)] // (3920/3)/40, (2500/3)/40 + [InlineData(HistoryAggregateType.Total, 3920d / 3d, 2500d / 3d)] // TimeAverage × 40 s + [InlineData(HistoryAggregateType.TimeAverage2, 104d / 3d, 20d)] // 1040/30 s, 600/30 s + [InlineData(HistoryAggregateType.Total2, 1040d, 600d)] + public async Task Time_weighted_aggregates_over_a_sparse_tag_with_wide_intervals(HistoryAggregateType agg, double b0, double b1) + { + await SeedWAsync(); + var s = await ReadW(agg, 40); + Values(s, b0, b1); + // Neither interval is Good throughout (an outage in each) ⇒ Uncertain. + s.Select(x => x.StatusCode).ShouldBe([UncertainCalc, UncertainCalc]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([B.Round(At(0)), B.Round(At(40))]); + } + + [Fact] + public async Task Interpolative_over_a_sparse_tag_with_wide_intervals() + { + await SeedWAsync(); + var s = await ReadW(HistoryAggregateType.Interpolative, 40); + // +0: from the bound before the window, 0 + 40·(30/50) = 24, Good. +40: inside the outage, bridged 80/3, Uncertain. + Values(s, 24, 80d / 3d); + s.Select(x => x.StatusCode).ShouldBe([GoodInterp, UncertainInterp]); + } + + [Fact] + public async Task TimeAverage2_over_a_sparse_tag_with_narrow_intervals() + { + await SeedWAsync(); + // 10 s intervals (narrower than the gaps): [0,10) 24→32 = 28 G; [10,20) 32→40 = 36 G; [20,30) 40 held, Uncertain; + // [30,40) and [40,50) inside the outage ⇒ BadNoData; [50,60) … [70,80) the flat 20 up to the bound after, Good. + var s = await ReadW(HistoryAggregateType.TimeAverage2, 10); + Values(s, 28, 36, 40, null, null, 20, 20, 20); + s.Select(x => x.StatusCode).ShouldBe( + [GoodCalc, GoodCalc, UncertainCalc, NoDataCalc, NoDataCalc, GoodCalc, GoodCalc, GoodCalc]); + // Total2 of the first interval: 28 × 10 s. + ((double)(await ReadW(HistoryAggregateType.Total2, 10))[0].Value!).ShouldBe(280d, 1e-9); + } + + [Fact] + public async Task Interpolative_over_a_sparse_tag_with_narrow_intervals() + { + await SeedWAsync(); + var s = await ReadW(HistoryAggregateType.Interpolative, 10); + // +0 24, +10 32 (sloped from the bound); +20 exact 40 (raw Good); +30 ON the outage marker ⇒ bridged 100/3; + // +40 bridged 80/3; +50 exact 20; +60, +70 on the Good 20→20 segment to the bound after. + Values(s, 24, 32, 40, 100d / 3d, 80d / 3d, 20, 20, 20); + s.Select(x => x.StatusCode).ShouldBe( + [GoodInterp, GoodInterp, GoodRaw, UncertainInterp, UncertainInterp, GoodRaw, GoodInterp, GoodInterp]); + } + + [Fact] + public async Task Outages_straddling_both_window_edges_use_the_four_bound_seeks() + { + // −10 = 0 G, −5 outage, +10 = 20 G, +20 outage, +30 = 40 G; window [0,20). The four seeks return −10 (latest + // usable before), −5 (latest before), +20 (first at/after end) and +30 (first usable at/after end). + await B.SeedAsync("O", + (At(-10), HistorianValue.FromDouble(0), Good), + (At(-5), HistorianValue.Empty, BadNoCommunication), + (At(10), HistorianValue.FromDouble(20), Good), + (At(20), HistorianValue.Empty, BadNoCommunication), + (At(30), HistorianValue.FromDouble(40), Good)); + var src = B.CreateDataSource(); + async Task> Read(HistoryAggregateType a, double intervalS) => + (await src.ReadProcessedAsync("O", At(0), At(20), TimeSpan.FromSeconds(intervalS), a, Ct)).Samples; + + // Bridged: −10..10 0→20 ⇒ [0,10) mean 15 (150); 10..30 20→40 ⇒ [10,20) mean 25 (250). 400 / 20 = 20. + // (Without the usable-before seek: 250/10 = 25; without the usable-after seek: [10,20) held 20 ⇒ 17.5.) + var ta = (await Read(HistoryAggregateType.TimeAverage, 0)).Single(); + ((double)ta.Value!).ShouldBe(20d, 1e-9); + ta.StatusCode.ShouldBe(UncertainCalc); + // Simple bounding: [0,10) inside the outage from −5; [10,20) 20 held into the outage at +20 ⇒ 200 / 10 s = 20. + // (Without the any-before seek: (150+200)/20 = 17.5; without the any-after seek: [10,20) 20→30 ⇒ 25.) + var ta2 = (await Read(HistoryAggregateType.TimeAverage2, 0)).Single(); + ((double)ta2.Value!).ShouldBe(20d, 1e-9); + ta2.StatusCode.ShouldBe(UncertainCalc); + // +0 inside the outage ⇒ bridged 10, Uncertain; +10 exact 20, raw Good. + var i = await Read(HistoryAggregateType.Interpolative, 10); + Values(i, 10, 20); + i.Select(x => x.StatusCode).ShouldBe([UncertainInterp, GoodRaw]); + } + + [Fact] + public async Task Time_weighted_aggregates_on_an_unknown_tag_are_BadNoData() + { + var src = B.CreateDataSource(); + (await src.ReadProcessedAsync("nope", At(0), At(40), Twenty, HistoryAggregateType.TimeAverage, Ct)) + .Samples.ShouldAllBe(x => x.StatusCode == NoDataCalc); + (await src.ReadProcessedAsync("nope", At(0), At(40), Twenty, HistoryAggregateType.Interpolative, Ct)) + .Samples.ShouldAllBe(x => x.StatusCode == NoData); + src.GetHealthSnapshot().TotalFailures.ShouldBe(0); + } + [Fact] public async Task The_bucket_count_is_capped_at_100000() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs index 6345613b..0f5c52c3 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.QuestDb.IntegrationTests/ReadTests.cs @@ -90,8 +90,8 @@ public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime max.Value.ShouldBe(50d); max.SourceTimestampUtc.ShouldBe(_b.Round(At(40))); // The extreme DateTime range: MinValue/MaxValue are nudged (never bound as ±infinity, which would match nothing). - (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Total, Ct)) - .Samples.Single().Value.ShouldBe(120d); + (await src.ReadProcessedAsync("A", DateTime.MinValue, DateTime.MaxValue, TimeSpan.Zero, HistoryAggregateType.Count, Ct)) + .Samples.Single().Value.ShouldBe(4); // A pre-1970 window holds no data and is not an error. (await src.ReadProcessedAsync("A", start, new DateTime(1969, 1, 1, 0, 0, 0, DateTimeKind.Utc), TimeSpan.FromDays(3650), HistoryAggregateType.Count, Ct)).Samples.ShouldAllBe(x => x.StatusCode == NoDataCalc); @@ -247,7 +247,7 @@ public sealed class ReadTests(QuestDbFixture fx) : IAsyncLifetime await Should.ThrowAsync(() => src.ReadEventsAsync(null, At(0), At(5), 0, Ct)); // Client-request errors are thrown before any I/O and are not backend failures. await Should.ThrowAsync(() => - src.ReadProcessedAsync("A", At(0), At(5), TimeSpan.Zero, HistoryAggregateType.Interpolative, Ct)); + src.ReadProcessedAsync("A", At(0), At(5), TimeSpan.Zero, HistoryAggregateType.StandardDeviationPopulation, Ct)); await Should.ThrowAsync(() => src.ReadProcessedAsync("A", At(0), At(100), TimeSpan.FromTicks(10), HistoryAggregateType.Count, Ct)); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs index 72a4a54b..e7953e8f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/AggregateEvaluatorTests.cs @@ -46,7 +46,6 @@ public sealed class AggregateEvaluatorTests [InlineData(HistoryAggregateType.Average, 15d, 40d, 50d)] [InlineData(HistoryAggregateType.Minimum, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.Maximum, 20d, 40d, 50d)] - [InlineData(HistoryAggregateType.Total, 30d, 40d, 50d)] [InlineData(HistoryAggregateType.Start, 10d, 40d, 50d)] [InlineData(HistoryAggregateType.End, 20d, 40d, 50d)] [InlineData(HistoryAggregateType.Delta, 10d, 0d, 0d)] diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs index cce8846f..a12edf4f 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/ProcessedReadTests.cs @@ -10,7 +10,11 @@ public sealed class ProcessedReadTests [Fact] public void Every_enum_member_is_either_served_or_refused() { - // Thirteen served; anything appended to the enum later falls to the refusal (total default). - Enum.GetValues().Count(a => AggregateFamilies.FamilyOf(a) is not null).ShouldBe(13); + // Seventeen served (#671 added TimeAverage, TimeAverage2, Interpolative and Total2, and moved Total to the + // time-weighted family); anything appended to the enum later falls to the refusal (total default). + Enum.GetValues().Count(a => AggregateFamilies.FamilyOf(a) is not null).ShouldBe(17); + Enum.GetValues().Where(a => AggregateFamilies.FamilyOf(a) == AggregateFamily.TimeWeighted) + .ShouldBe([HistoryAggregateType.Total, HistoryAggregateType.Interpolative, HistoryAggregateType.TimeAverage, + HistoryAggregateType.TimeAverage2, HistoryAggregateType.Total2]); } } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/TimeWeightedAggregatorTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/TimeWeightedAggregatorTests.cs new file mode 100644 index 00000000..4ee56877 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests/TimeWeightedAggregatorTests.cs @@ -0,0 +1,345 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using static ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests.Fixture; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Sql.Common.Tests; + +/// +/// The Part 13 time-weighted aggregates (#671) over hand-built sample streams. Every literal is hand-computed in the +/// comment above it (areas in value·seconds; "raw" = the simple-bounding signal, "bridged" = interpolated bounding). +/// +public sealed class TimeWeightedAggregatorTests +{ + private static readonly DateTime Now = At(3600); + private const uint GoodRaw = 0x00000000u; + private const uint NoData = 0x809B0000u; + private const uint InvalidInputs = 0x80D60000u; + + private static IReadOnlyList Run(HistoryAggregateType agg, double from, double to, double intervalS, + params StoredSample[] samples) => + TimeWeightedAggregator.Evaluate(agg, BucketPlan.Create(At(from), At(to), TimeSpan.FromSeconds(intervalS)), samples, Now); + + private static void Values(IReadOnlyList s, params double?[] expected) + { + s.Count.ShouldBe(expected.Length); + for (var i = 0; i < expected.Length; i++) + { + if (expected[i] is { } e) ((double)s[i].Value!).ShouldBe(e, 1e-9, $"bucket {i}"); + else s[i].Value.ShouldBeNull($"bucket {i}"); + } + } + + // ---------------------------------------------------------------- tag A: +0=10 G, +10=20 G, +20=Bad, +30=40 G, +40=50 G + // Window [0,80), 20 s buckets. Raw: [0,10) 10→20 Good (150); [10,20) 20 held into the outage, Uncertain (200); + // [20,30) Bad; [30,40) 40→50 Good (450); [40,80) 50 extrapolated, Uncertain (2000). + // Bridged: [0,10) 150; [10,30) 20→40 = 250 in B0 + 350 in B1; [30,40) 450; [40,80) 50 held. + + [Fact] + public void TimeAverage_bridges_the_outage_on_tag_A() + { + // B0 (150+250)/20 = 20; B1 (350+450)/20 = 40; B2, B3 = 50. Only B1's [30,40) and B0's [0,10) are Good ⇒ all Uncertain. + var s = Run(HistoryAggregateType.TimeAverage, 0, 80, 20, A); + Values(s, 20, 40, 50, 50); + s.Select(x => x.StatusCode).ShouldBe([UncertainCalc, UncertainCalc, UncertainCalc, UncertainCalc]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(20), At(40), At(60)]); + } + + [Fact] + public void Total_is_TimeAverage_times_interval_seconds_on_tag_A() + { + // 20×20, 40×20, 50×20, 50×20. + Values(Run(HistoryAggregateType.Total, 0, 80, 20, A), 400, 800, 1000, 1000); + } + + [Fact] + public void TimeAverage2_excludes_the_bad_interval_on_tag_A() + { + // B0 350/20 = 17.5; B1 450/10 = 45 (the Bad [20,30) excluded); B2, B3 = 50. + var s = Run(HistoryAggregateType.TimeAverage2, 0, 80, 20, A); + Values(s, 17.5, 45, 50, 50); + s.Select(x => x.StatusCode).ShouldBe([UncertainCalc, UncertainCalc, UncertainCalc, UncertainCalc]); + } + + [Fact] + public void Total2_integrates_only_the_non_bad_data_on_tag_A() + { + // B0 150+200 = 350; B1 450; B2, B3 1000. + Values(Run(HistoryAggregateType.Total2, 0, 80, 20, A), 350, 450, 1000, 1000); + } + + [Fact] + public void Interpolative_on_tag_A() + { + // +0 exact Good 10 (raw status); +20 sits on the Bad sample ⇒ bridged 20→40 at the midpoint = 30, Uncertain; + // +40 exact Good 50; +60 past the last sample ⇒ 50 extrapolated, Uncertain. + var s = Run(HistoryAggregateType.Interpolative, 0, 80, 20, A); + Values(s, 10, 30, 50, 50); + s.Select(x => x.StatusCode).ShouldBe([GoodRaw, UncertainInterpolated, GoodRaw, UncertainInterpolated]); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(0), At(20), At(40), At(60)]); + s.ShouldAllBe(x => x.ServerTimestampUtc == Now); + } + + // ---------------------------------------------------------------- bounding values on both sides, crossing segments + + private static readonly StoredSample[] Bounded = + [ + S(-10, HistorianValue.FromDouble(0), Good), // bound before the window + S(10, HistorianValue.FromDouble(20), Good), + S(30, HistorianValue.FromDouble(20), Good), // bound after the window + ]; + + [Theory] + // Window [0,20), 10 s. Segment -10..10 (0→20) crosses the window start: B0 [0,10) = 10→20, mean 15. + // B1 [10,20) = 20 flat (crossing the window end into the bound). All Good. + [InlineData(HistoryAggregateType.TimeAverage, 15d, 20d)] + [InlineData(HistoryAggregateType.TimeAverage2, 15d, 20d)] + [InlineData(HistoryAggregateType.Total, 150d, 200d)] + [InlineData(HistoryAggregateType.Total2, 150d, 200d)] + public void Bounding_values_before_and_after_the_window_are_used(HistoryAggregateType agg, double b0, double b1) + { + var s = Run(agg, 0, 20, 10, Bounded); + Values(s, b0, b1); + s.ShouldAllBe(x => x.StatusCode == GoodCalc); + } + + [Fact] + public void Interpolative_interpolates_from_the_bound_before_the_window() + { + // +0: 0 + (20−0)·(10/20) = 10, Good both ends; +10 exact Good 20. + var s = Run(HistoryAggregateType.Interpolative, 0, 20, 10, Bounded); + Values(s, 10, 20); + s.Select(x => x.StatusCode).ShouldBe([GoodInterpolated, GoodRaw]); + } + + [Fact] + public void A_segment_spanning_several_buckets_is_split_at_every_edge() + { + // One segment 0..40 (0→40), window [0,40), 10 s: bucket means 5, 15, 25, 35; Total2 50, 150, 250, 350. + StoredSample[] line = [S(0, HistorianValue.FromDouble(0), Good), S(40, HistorianValue.FromDouble(40), Good)]; + Values(Run(HistoryAggregateType.TimeAverage2, 0, 40, 10, line), 5, 15, 25, 35); + Values(Run(HistoryAggregateType.Total2, 0, 40, 10, line), 50, 150, 250, 350); + Values(Run(HistoryAggregateType.Interpolative, 0, 40, 10, line), 0, 10, 20, 30); + } + + // ---------------------------------------------------------------- empty buckets between bounds + + [Fact] + public void An_empty_bucket_with_bounds_on_both_sides_is_interpolated_not_no_data() + { + // -5 = 0, +25 = 30 (slope 1/s); window [0,20), 10 s holds no sample at all. + StoredSample[] sparse = [S(-5, HistorianValue.FromDouble(0), Good), S(25, HistorianValue.FromDouble(30), Good)]; + var interp = Run(HistoryAggregateType.Interpolative, 0, 20, 10, sparse); + Values(interp, 5, 15); + interp.ShouldAllBe(x => x.StatusCode == GoodInterpolated); + // B0 5→15 mean 10; B1 15→25 mean 20. + var avg = Run(HistoryAggregateType.TimeAverage, 0, 20, 10, sparse); + Values(avg, 10, 20); + avg.ShouldAllBe(x => x.StatusCode == GoodCalc); + Values(Run(HistoryAggregateType.Total, 0, 20, 10, sparse), 100, 200); + } + + [Fact] + public void Past_the_last_sample_the_value_is_extrapolated_stepped_and_uncertain() + { + // -5 = 7 and nothing later: held at 7 over [0,10). + StoredSample[] one = [S(-5, HistorianValue.FromDouble(7), Good)]; + var s = Run(HistoryAggregateType.TimeAverage2, 0, 10, 0, one).Single(); + ((double)s.Value!).ShouldBe(7d); + s.StatusCode.ShouldBe(UncertainCalc); + var i = Run(HistoryAggregateType.Interpolative, 0, 10, 0, one).Single(); + ((double)i.Value!).ShouldBe(7d); + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Theory] + [InlineData(HistoryAggregateType.TimeAverage)] + [InlineData(HistoryAggregateType.TimeAverage2)] + [InlineData(HistoryAggregateType.Total)] + [InlineData(HistoryAggregateType.Total2)] + public void No_data_at_all_is_BadNoData_with_the_calculated_bit(HistoryAggregateType agg) + { + var s = Run(agg, 0, 20, 10); + s.ShouldAllBe(x => x.StatusCode == NoDataCalc && x.Value == null); + } + + [Fact] + public void Interpolative_with_no_data_is_plain_BadNoData() + { + // The only sample is at +15: nothing at or before either bucket start (+0, +10). + var s = Run(HistoryAggregateType.Interpolative, 0, 20, 10, S(15, HistorianValue.FromDouble(1), Good)); + s.ShouldAllBe(x => x.StatusCode == NoData && x.Value == null); + } + + // ---------------------------------------------------------------- Bad intervals: TimeAverage vs TimeAverage2/Total2 + + private static readonly StoredSample[] Outage = + [ + S(0, HistorianValue.FromDouble(10), Good), + S(5, HistorianValue.Empty, BadNoCommunication), + S(15, HistorianValue.FromDouble(30), Good), + S(25, HistorianValue.FromDouble(30), Good), // bound after the window + ]; + + [Fact] + public void TimeAverage2_and_Total2_exclude_the_bad_interval() + { + // Window [0,20) one bucket. Raw: [0,5) 10 held into the outage (50, Uncertain); [5,15) Bad; [15,20) 30 Good (150). + // TimeAverage2 = 200 / 10 s = 20; Total2 = 200. + var ta2 = Run(HistoryAggregateType.TimeAverage2, 0, 20, 0, Outage).Single(); + ((double)ta2.Value!).ShouldBe(20d); + ta2.StatusCode.ShouldBe(UncertainCalc); + ((double)Run(HistoryAggregateType.Total2, 0, 20, 0, Outage).Single().Value!).ShouldBe(200d); + } + + [Fact] + public void TimeAverage_and_Total_interpolate_across_the_bad_interval() + { + // Bridged: [0,15) 10→30, mean 20 ⇒ 300; [15,20) 30 ⇒ 150. TimeAverage = 450/20 = 22.5; Total = 22.5 × 20 = 450. + var ta = Run(HistoryAggregateType.TimeAverage, 0, 20, 0, Outage).Single(); + ((double)ta.Value!).ShouldBe(22.5d); + ta.StatusCode.ShouldBe(UncertainCalc); + ((double)Run(HistoryAggregateType.Total, 0, 20, 0, Outage).Single().Value!).ShouldBe(450d); + } + + [Fact] + public void An_interval_entirely_inside_an_outage_is_BadNoData_but_Interpolative_bridges_it() + { + // Window [10,15): raw is Bad throughout ⇒ no non-Bad time ⇒ BadNoData. Interpolative at +10 = 10 + 20·(10/15). + Run(HistoryAggregateType.TimeAverage, 10, 15, 0, Outage).Single().StatusCode.ShouldBe(NoDataCalc); + Run(HistoryAggregateType.TimeAverage2, 10, 15, 0, Outage).Single().StatusCode.ShouldBe(NoDataCalc); + var i = Run(HistoryAggregateType.Interpolative, 10, 15, 0, Outage).Single(); + ((double)i.Value!).ShouldBe(10d + 20d * 10d / 15d, 1e-9); + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public void Interpolative_on_a_segment_that_runs_into_an_outage_is_uncertain() + { + // +2 lies between the Good +0 and the Bad +5: bridged 10→30 over [0,15) ⇒ 10 + 20·(2/15); the simple-bounding + // signal there is held into an outage, so the point is Uncertain even though both bridging ends are Good. + var i = Run(HistoryAggregateType.Interpolative, 2, 20, 0, Outage).Single(); + ((double)i.Value!).ShouldBe(10d + 20d * 2d / 15d, 1e-9); + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + [Fact] + public void The_latest_usable_bound_is_found_behind_a_bad_bound() + { + // -10 = 0 G, -5 Bad, +10 = 20 G, +30 = 20 G. Window [0,20) one bucket. + // Bridged: -10..10 0→20 ⇒ [0,10) 10→20 (150); [10,20) 20 (200) ⇒ TimeAverage = 17.5. + // Raw: [-5,10) Bad; [10,20) 20 Good (200) ⇒ TimeAverage2 = 20. Good 10 s of 20 ⇒ Uncertain. + StoredSample[] s = + [ + S(-10, HistorianValue.FromDouble(0), Good), S(-5, HistorianValue.Empty, BadNoCommunication), + S(10, HistorianValue.FromDouble(20), Good), S(30, HistorianValue.FromDouble(20), Good), + ]; + ((double)Run(HistoryAggregateType.TimeAverage, 0, 20, 0, s).Single().Value!).ShouldBe(17.5d); + var ta2 = Run(HistoryAggregateType.TimeAverage2, 0, 20, 0, s).Single(); + ((double)ta2.Value!).ShouldBe(20d); + ta2.StatusCode.ShouldBe(UncertainCalc); + var i = Run(HistoryAggregateType.Interpolative, 0, 20, 0, s).Single(); + ((double)i.Value!).ShouldBe(10d); // inside the outage, bridged -10..10 at 0 + i.StatusCode.ShouldBe(UncertainInterpolated); + } + + // ---------------------------------------------------------------- stepped Boolean, Uncertain, invalid inputs + + private static readonly StoredSample[] Switch = + [ + S(0, HistorianValue.FromBoolean(true), Good), + S(5, HistorianValue.FromBoolean(false), Good), + S(15, HistorianValue.FromBoolean(true), Good), + S(25, HistorianValue.FromBoolean(true), Good), // bound after the window + ]; + + [Fact] + public void Booleans_are_stepped_not_sloped() + { + // Window [0,20), 10 s. B0: 1 over [0,5), 0 over [5,10) ⇒ 0.5 (sloped would give 0.375). B1: 0 over [10,15), + // 1 over [15,20) ⇒ 0.5. All Good. Total: 5 and 5. + var s = Run(HistoryAggregateType.TimeAverage, 0, 20, 10, Switch); + Values(s, 0.5, 0.5); + s.ShouldAllBe(x => x.StatusCode == GoodCalc); + Values(Run(HistoryAggregateType.Total, 0, 20, 10, Switch), 5, 5); + // Interpolative at +10: stepped from +5 (false) ⇒ 0, Good. + var i = Run(HistoryAggregateType.Interpolative, 0, 20, 10, Switch); + Values(i, 1, 0); + i.Select(x => x.StatusCode).ShouldBe([GoodRaw, GoodInterpolated]); + } + + [Fact] + public void An_uncertain_sample_makes_its_segment_uncertain() + { + // 0 = 4 Uncertain, 10 = 6 Good, 20 = 6 Good: [0,10) sloped 4→6, quality Uncertain (worse end) ⇒ mean 5; + // [10,20) 6 Good. Window [0,20), 10 s. + StoredSample[] s = + [ + S(0, HistorianValue.FromDouble(4), Uncertain), S(10, HistorianValue.FromDouble(6), Good), + S(20, HistorianValue.FromDouble(6), Good), + ]; + var r = Run(HistoryAggregateType.TimeAverage2, 0, 20, 10, s); + Values(r, 5, 6); + r.Select(x => x.StatusCode).ShouldBe([UncertainCalc, GoodCalc]); + } + + [Fact] + public void Int64_samples_are_sloped_like_doubles() + { + StoredSample[] s = [S(0, HistorianValue.FromInt64(0), Good), S(10, HistorianValue.FromInt64(10), Good)]; + Values(Run(HistoryAggregateType.TimeAverage2, 0, 10, 0, s), 5); + } + + [Theory] + [InlineData(HistoryAggregateType.TimeAverage)] + [InlineData(HistoryAggregateType.TimeAverage2)] + [InlineData(HistoryAggregateType.Total)] + [InlineData(HistoryAggregateType.Total2)] + public void A_usable_string_sample_makes_the_interval_invalid_inputs(HistoryAggregateType agg) + { + StoredSample[] s = [S(0, HistorianValue.FromString("a"), Good), S(10, HistorianValue.FromString("b"), Good)]; + var r = Run(agg, 0, 20, 0, s).Single(); + r.StatusCode.ShouldBe(InvalidInputsCalc); + r.Value.ShouldBeNull(); + } + + [Fact] + public void Interpolative_on_a_string_is_invalid_inputs() + { + StoredSample[] s = [S(0, HistorianValue.FromString("a"), Good), S(10, HistorianValue.FromString("b"), Good)]; + Run(HistoryAggregateType.Interpolative, 0, 20, 10, s).ShouldAllBe(x => x.StatusCode == InvalidInputs); + } + + // ---------------------------------------------------------------- plan shape + + [Fact] + public void A_reverse_window_is_computed_forward_and_returned_descending() + { + var s = TimeWeightedAggregator.Evaluate(HistoryAggregateType.TimeAverage2, + BucketPlan.Create(At(80), At(0), TimeSpan.FromSeconds(20)), A, Now); + s.Select(x => x.SourceTimestampUtc).ShouldBe([At(60), At(40), At(20), At(0)]); + Values(s, 50, 50, 45, 17.5); + } + + [Fact] + public void A_truncated_last_bucket_uses_its_own_duration_for_Total() + { + // Window [0,50), 20 s: B2 = [40,50) is 10 s of the extrapolated 50 ⇒ TimeAverage 50, Total 500. + Values(Run(HistoryAggregateType.Total, 0, 50, 20, A), 400, 800, 500); + } + + [Fact] + public void A_duplicate_bound_is_ignored() + { + // The same row fetched as both "latest usable" and "latest any" bound. + var dup = new[] { Bounded[0], Bounded[0], Bounded[1], Bounded[2], Bounded[2] }; + Values(Run(HistoryAggregateType.TimeAverage2, 0, 20, 10, dup), 15, 20); + } + + [Fact] + public void A_non_time_weighted_aggregate_is_refused() + { + Should.Throw(() => + new TimeWeightedAggregator(HistoryAggregateType.Average, BucketPlan.Create(At(0), At(10), TimeSpan.Zero))); + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs index a2950a7d..9d3394d7 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadTests.cs @@ -292,6 +292,34 @@ public sealed class NodeManagerHistoryReadTests : IDisposable await host.DisposeAsync(); } + /// #671: the Part 13 TimeAverage2 / Total2 aggregate NodeIds map to their own members. + [Theory] + [InlineData("TimeAverage2")] + [InlineData("Total2")] + public async Task Processed_TimeAverage2_and_Total2_dispatch_to_the_source(string name) + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new RecordingHistorianDataSource { ProcessedResult = Empty() }; + nm.HistorianDataSource = fake; + nm.EnsureVariable("eq-1/tw", parentFolderNodeId: null, displayName: "T", dataType: "Float", + writable: false, historianTagname: "WW.Tw", realm: AddressSpaceRealm.Uns); + var nodeId = nm.TryGetVariable("eq-1/tw")!.NodeId; + var (aggregateId, expected) = name == "Total2" + ? (ObjectIds.AggregateFunction_Total2, HistoryAggregateType.Total2) + : (ObjectIds.AggregateFunction_TimeAverage2, HistoryAggregateType.TimeAverage2); + + InvokeHistoryRead(server, nm, new ReadProcessedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), EndTime = DateTime.UtcNow, ProcessingInterval = 10_000.0, + AggregateType = new NodeIdCollection { aggregateId }, + }, nodeId); + + fake.LastCall.ShouldBe("Processed"); + fake.LastAggregate.ShouldBe(expected); + await host.DisposeAsync(); + } + /// #285: a backend that cannot serve a mapped aggregate answers BadAggregateNotSupported /// (not the generic BadHistoryOperationUnsupported a plain throw produces). [Fact]