Merge b647-mtc-dataset: MTConnect DATA_SET/TABLE as JSON strings; EVENT enumerations parked (#650)

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-08 19:21:34 -04:00
12 changed files with 647 additions and 39 deletions
+38 -14
View File
@@ -103,8 +103,8 @@ request/response handshake — is a different mechanism and remains deferred; se
| `TIME_SERIES` arrays (#627) | **Supported** as variable-length `double[]` (Float64 tags only) | [TIME_SERIES arrays](#time_series-arrays-627) |
| One Agent client per driver, one `/raw` Device (#628) | **Enforced** (`MTConnectMultipleDevices` deploy error); multi-device Agents use several driver instances | [Known limitations](#known-limitations) |
| Write-back (`IWritable`) | **Supported, off by default** (`Write.Enabled` + Agent `AllowPut`) | [Write-back](#write-back--observation-injection-off-by-default) |
| EVENT controlled vocabularies as OPC UA enumerations | **Not supported** (stay String); not scheduled | See [Deferred](#deferred-not-in-this-build) |
| `DATA_SET` / `TABLE` representations | **Not supported** - read as `BadNotSupported` | [Quality mapping](#quality-mapping) |
| EVENT controlled vocabularies as OPC UA enumerations | **Parked / won't fix** (stay String) - see rationale | [EVENT enumerations](#event-enumerations-650-parked--wont-fix) |
| `DATA_SET` / `TABLE` representations (#650) | **Supported** as a String carrying compact JSON | [DATA_SET / TABLE](#data_set--table-650) |
| MTConnect Interfaces, SHDR ingest | Not built | [Deferred](#deferred-not-in-this-build) |
**`IReadable.ReadAsync` is never called by the running server.** The
@@ -245,7 +245,7 @@ Other status codes an observation can surface:
| `BadTypeMismatch` (`0x80740000`) | The Agent's text isn't a value of the tag's coerced type at all |
| `BadOutOfRange` (`0x803C0000`) | The Agent reported a number the coerced type can't represent |
| `BadDecodingError` (`0x80070000`) | A `TIME_SERIES` vector whose element count disagrees with the observation's own `sampleCount` (or the tag's fixed `ArrayDim`) — see [TIME_SERIES](#time_series-arrays-627) |
| `BadNotSupported` (`0x803D0000`) | A shape this build doesn't materialize — a structured `DATA_SET`/`TABLE` observation (decided: stays unsupported, #627), or an array tag declared as anything other than Float64 |
| `BadNotSupported` (`0x803D0000`) | A shape this build doesn't materialize — a `DATA_SET`/`TABLE` observation whose children are not `<Entry>` elements (a well-formed one is served as JSON since #650), or an array tag declared as anything other than Float64 |
| `BadWaitingForInitialData` | Tag is authored but the Agent hasn't reported it yet |
| `BadNodeIdUnknown` | DataItem id is neither authored nor ever observed |
@@ -272,8 +272,30 @@ in the Tag modal's array section).
but the OPC UA node has no description/property slot through `DriverAttributeInfo`,
and a sibling property node would need a new address-space mechanism. Decision:
document it — the rate is a property of the Agent's DataItem (`sampleRate` in
`/probe`), read it there. **`DATA_SET` / `TABLE` stay `BadNotSupported`** (decision, #627): they are
key/value maps with no scalar or array OPC UA shape that preserves the keys.
`/probe`), read it there.
### DATA_SET / TABLE (#650)
`DATA_SET` and `TABLE` observations carry their content in `<Entry key="…">` children (a `TABLE` entry holds
`<Cell key="…">` children). They are served as a **String carrying compact, deterministic JSON** (OPC UA type
String, never an array or structure - no new node shape, history/ACL/alarm paths unchanged; discovery and
type inference already type these items String):
- `DATA_SET` -> `{"offsetX":1.25,"toolNumber":7}`; `TABLE` -> `{"G54":{"X":1.0,"Y":2},"G55":{"X":3}}`.
- Keys are ordinal-sorted at every level, no whitespace - the same content always serialises to the same
bytes, so value-dedup works. Values that are *strict JSON numbers* are emitted as numbers with their original
token (lossless: `1.0` stays `1.0`); anything else (`007`, `+5`, `on`) is a string.
- **Merge semantics follow MTConnect 2.x.** A `/current` document states the *whole* set and **replaces** it;
a `/sample` chunk carries *deltas* that **merge** by key (a `TABLE` row is replaced as a unit by a later entry
for the same row key). An observation with `resetTriggered` replaces the accumulated set, and a
`resetTriggered` observation with no entries is the empty set `{}`. Several observations of one item in one
document fold in document order. The accumulated set is dropped on an Agent restart (`instanceId` change).
- `removed="true"` deletes the key (or row): it is **absent** from the JSON, never `null`.
- **Empty set is Good `{}`**: a data-set-shaped observation (it carries the `count` attribute, which only DATA_SET/TABLE have) with no entries - `count="0"` in a `/current`, or `resetTriggered` - is the empty set, `{}`, never an absence; an all-removed set renders the same. `resetTriggered` on any other observation (e.g. an accumulating `PartCount`) is ignored.
- `UNAVAILABLE` is `BadNoCommunication` as for any observation, and also drops the accumulated set, so a
later delta cannot resurrect pre-outage entries.
- An observation whose children are not `<Entry>` elements stays `BadNotSupported`. Writes to these items stay
refused (`WriteAsync` excludes them).
### Subscribe — the `/sample` pump
@@ -483,9 +505,13 @@ and records a state only when an alarm handler was attached to receive it. Conse
unchanged triggers no re-subscribe, and needs none — the server-side carry-over covers it.
- An alarm-only driver (no value tags) still runs the `/sample` pump.
**EVENT controlled vocabularies as OPC UA enumerations — not built (parked, low priority).** EVENT values (`Execution`,
`ControllerMode`, …) stay Strings carrying the Agent's own word; an enumeration would need a per-type
vocabulary table that vendors extend freely.
### EVENT enumerations (#650): parked / won't fix
EVENT values (`Execution`, `ControllerMode`, …) stay Strings carrying the Agent's own word. Modelling them as
OPC UA enumerations was considered and **declined**: the vocabulary is vendor-extensible, so it needs a per-type
table plus an unknown-value fallback, a per-type enum node shape through discovery and raw-tag type inference,
and it would break every consumer that reads these items as String (ScadaBridge) for little gain. Reopen only
with an opt-in, tag-level `enumValues` design.
## Known limitations
@@ -563,9 +589,8 @@ for anything beyond values-and-conditions.
pin the parsing rule itself.
5. **A CONDITION alarm shows one condition per DataItem** — the most severe active `nativeCode`, not one
Part 9 condition per code (see [CONDITION alarms](#condition-alarms-626)).
6. **Structured `DATA_SET`/`TABLE` observations surface as
`BadNotSupported`**, not an approximation — see the Quality mapping table.
`TIME_SERIES` is served as `double[]` (#627, see
6. **`DATA_SET`/`TABLE` observations are JSON Strings** (#650, see
[DATA_SET / TABLE](#data_set--table-650)); `TIME_SERIES` is served as `double[]` (#627, see
[TIME_SERIES arrays](#time_series-arrays-627)); the sample rate is not
surfaced on the node.
@@ -585,9 +610,8 @@ for anything beyond values-and-conditions.
- **P1.5 fast-follow** (design
[§9](../plans/2026-07-15-mtconnect-driver-design.md#9-phasing--effort)): CONDITION → native Part 9
alarms **shipped** (#626, see [CONDITION alarms](#condition-alarms-626)); EVENT
controlled-vocabulary values → OPC UA enumerations is **not built** (EVENTs stay Strings), and
structured `DATA_SET`/`TABLE` observations stay `BadNotSupported`; both are tracked as a low-priority
follow-up, not part of P1.5.
controlled-vocabulary values → OPC UA enumerations is **parked / won't fix** (EVENTs stay Strings, #650);
`DATA_SET`/`TABLE` are served as JSON Strings (#650).
- **P2 (on demand): SHDR adapter ingest.** A `SourceMode: "Agent" | "Shdr"`
switch that opens the raw pipe-delimited SHDR TCP socket directly. Loses
auto-discovery (no device model without an Agent in front, so
@@ -747,7 +747,7 @@ public sealed partial class MTConnectDriver
// Built from the SAME authored tags as the shared index, so the authored-vs-unknown
// distinction and every coercion behave identically — it is a snapshot, not a variant.
var snapshot = new MTConnectObservationIndex(binding.IndexTags);
snapshot.Apply(current);
snapshot.Apply(current, baseline: true);
var results = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < results.Length; i++)
@@ -2337,7 +2337,7 @@ public sealed partial class MTConnectDriver
private void ApplyAndFanOut(MTConnectStreamsResult document, bool baseline = false)
{
var index = ObservationIndex;
index.Apply(document);
index.Apply(document, baseline);
// #626: CONDITION observations also feed the alarm path, which keeps its own multi-condition state
// (a /sample Normal for one nativeCode is not a return to normal while another code is active).
@@ -2801,7 +2801,7 @@ public sealed partial class MTConnectDriver
}
var index = new MTConnectObservationIndex(binding.IndexTags);
index.Apply(current);
index.Apply(current, baseline: true);
Volatile.Write(ref _index, index);
// #626: the CONDITION tracker is per-session state like the index. Nothing is RAISED here — the
@@ -167,6 +167,23 @@ public sealed record MTConnectStreamsResult(
/// observation. Carried for the parser contract; not yet surfaced on the served node (see
/// docs/drivers/MTConnect.md). <c>null</c> when absent.
/// </param>
/// <param name="Entries">
/// The parsed <c>&lt;Entry key="…"&gt;</c> children of a <c>DATA_SET</c> / <c>TABLE</c> observation (#650), in
/// document order, or <c>null</c> when the observation had none. A <c>TABLE</c> entry carries
/// <see cref="MTConnectDataSetEntry.Cells"/>. The observation index folds these into a JSON String — see
/// <see cref="MTConnectObservationIndex"/>. <see cref="IsStructured"/> stays <c>true</c> beside them: it is the
/// wire-shape fact, this is the content.
/// </param>
/// <param name="ResetTriggered">
/// <c>true</c> when the observation carried a non-empty <c>resetTriggered</c> attribute — the Agent's statement that
/// the set was reset, so this observation <b>replaces</b> the accumulated set rather than merging into it.
/// </param>
/// <param name="IsDataSet">
/// <c>true</c> when the element is data-set shaped: it carries the <c>count</c> attribute every DATA_SET / TABLE
/// observation has (and no SAMPLE or ordinary EVENT does) or has <c>&lt;Entry&gt;</c> children. This — not
/// <see cref="ResetTriggered"/> — is what lets the index serve an entry-less set as <c>{}</c> rather than as
/// an absent value; <c>resetTriggered</c> also appears on ordinary accumulating SAMPLEs and must not matter there.
/// </param>
/// <param name="Condition">
/// The CONDITION-only detail (#626) — set exactly when the observation came from a
/// <c>&lt;Condition&gt;</c> container, <c>null</c> for every Sample/Event. Its presence is what marks an
@@ -179,7 +196,26 @@ public sealed record MTConnectObservation(
bool IsStructured = false,
int? SampleCount = null,
double? SampleRate = null,
MTConnectConditionDetail? Condition = null);
MTConnectConditionDetail? Condition = null,
IReadOnlyList<MTConnectDataSetEntry>? Entries = null,
bool ResetTriggered = false,
bool IsDataSet = false);
/// <summary>One <c>&lt;Cell key="…"&gt;</c> of a <c>TABLE</c> row (#650).</summary>
public sealed record MTConnectDataSetCell(string Key, string Value);
/// <summary>
/// One <c>&lt;Entry key="…"&gt;</c> of a <c>DATA_SET</c> / <c>TABLE</c> observation (#650).
/// </summary>
/// <param name="Key">The entry's <c>key</c> attribute, verbatim.</param>
/// <param name="Value">A <c>DATA_SET</c> entry's trimmed text; <c>null</c> for a <c>TABLE</c> row or a removed entry.</param>
/// <param name="Removed"><c>true</c> when the entry carried <c>removed="true"</c>: the key is deleted from the set.</param>
/// <param name="Cells">A <c>TABLE</c> row's cells in document order; <c>null</c> for a <c>DATA_SET</c> entry.</param>
public sealed record MTConnectDataSetEntry(
string Key,
string? Value,
bool Removed,
IReadOnlyList<MTConnectDataSetCell>? Cells = null);
/// <summary>
/// What an MTConnect CONDITION observation says beyond its state word (#626). Every field is the
@@ -1,6 +1,9 @@
using System.Collections.Concurrent;
using System.Collections.Frozen;
using System.Globalization;
using System.Text.Encodings.Web;
using System.Text.Json;
using System.Text.RegularExpressions;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
@@ -50,7 +53,7 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect;
/// <para>
/// <b>Shapes this build does not represent</b> are coded <c>BadNotSupported</c> with a null
/// value rather than approximated: a non-Float64 array tag and a <c>DATA_SET</c> / <c>TABLE</c>
/// observation (a Float64 <c>TIME_SERIES</c> vector is served as <c>double[]</c>, #627). The latter carries its
/// observation whose children are not <c>&lt;Entry&gt;</c> elements (a Float64 <c>TIME_SERIES</c> vector is served as <c>double[]</c>, #627). The latter carries its
/// real content in <c>&lt;Entry key=…&gt;</c> child elements, which reach a text-valued
/// observation concatenated with the keys discarded — a two-entry set reads as the single
/// token <c>"12"</c> — and because the type inference correctly demotes those
@@ -111,6 +114,19 @@ internal sealed class MTConnectObservationIndex
}.ToFrozenDictionary(StringComparer.OrdinalIgnoreCase);
private readonly ConcurrentDictionary<string, DataValueSnapshot> _snapshots = new(StringComparer.Ordinal);
// #650: the accumulated DATA_SET / TABLE content per dataItemId. A /sample chunk carries only the entries
// that changed, so the served JSON is the merge of the baseline and every delta since. Row value is the
// entry's text (DATA_SET) or its cell map (TABLE). Only the single pump (or a one-shot snapshot) calls
// Apply, but the lock keeps a concurrent one-shot read-through from corrupting a shared index.
private readonly Dictionary<string, SortedDictionary<string, SetEntry>> _sets = new(StringComparer.Ordinal);
private readonly object _setsLock = new();
private bool _anySets;
private sealed record SetEntry(string? Value, SortedDictionary<string, string>? Cells);
private static readonly Regex JsonNumber = new(
@"^-?(0|[1-9][0-9]*)(\.[0-9]+)?([eE][+-]?[0-9]+)?$", RegexOptions.Compiled | RegexOptions.CultureInvariant);
private readonly FrozenDictionary<string, MTConnectTagDefinition> _tags;
private readonly TimeProvider _timeProvider;
@@ -159,7 +175,11 @@ internal sealed class MTConnectObservationIndex
/// <param name="result">The parsed streams document. An observation-free result is a legal
/// keep-alive and leaves the index untouched.</param>
/// <exception cref="ArgumentNullException"><paramref name="result"/> is <c>null</c>.</exception>
public void Apply(MTConnectStreamsResult result)
/// <param name="baseline">
/// <c>true</c> for a <c>/current</c> document, which states the <b>whole</b> DATA_SET / TABLE and so replaces
/// the accumulated set; <c>false</c> (default) for a <c>/sample</c> chunk, whose entries are deltas merged into it.
/// </param>
public void Apply(MTConnectStreamsResult result, bool baseline = false)
{
ArgumentNullException.ThrowIfNull(result);
@@ -170,9 +190,12 @@ internal sealed class MTConnectObservationIndex
// — writing every observation and merging snapshots in place — would need a read-modify-write
// per key and could expose a half-merged state to a concurrent reader.
var resolved = new Dictionary<string, MTConnectObservation>(StringComparer.Ordinal);
foreach (var observation in result.Observations)
var seenSets = new HashSet<string>(StringComparer.Ordinal);
foreach (var raw in result.Observations)
{
if (observation is null || string.IsNullOrWhiteSpace(observation.DataItemId)) continue;
if (raw is null || string.IsNullOrWhiteSpace(raw.DataItemId)) continue;
var observation = FoldDataSet(raw, baseline, seenSets);
resolved[observation.DataItemId] =
resolved.TryGetValue(observation.DataItemId, out var incumbent)
@@ -217,7 +240,150 @@ internal sealed class MTConnectObservationIndex
/// <c>instanceId</c> changes, every value the index holds describes a device model that no
/// longer exists, and serving it would be worse than serving nothing.
/// </summary>
public void Clear() => _snapshots.Clear();
public void Clear()
{
_snapshots.Clear();
lock (_setsLock) _sets.Clear();
}
/// <summary>
/// DATA_SET / TABLE (#650). Folds an observation's <c>&lt;Entry&gt;</c> children into the accumulated set for
/// its dataItemId and returns a plain String observation carrying the merged JSON; any other observation is
/// returned untouched, except that an <c>UNAVAILABLE</c> one drops the accumulated set (a later delta must not
/// resurrect pre-outage entries).
/// <para>
/// Semantics (MTConnect 2.x): a <c>/current</c> document (<paramref name="baseline"/>) or an observation
/// with <c>resetTriggered</c> <b>replaces</b> the set; otherwise entries merge by key. <c>removed="true"</c>
/// deletes the key (the key is absent from the JSON, never <c>null</c>). A TABLE row is replaced as a unit
/// by a later entry for the same row key; a removed row key disappears. Several observations of one item in
/// a single document fold in document order — only the first replaces on a baseline.
/// </para>
/// </summary>
private MTConnectObservation FoldDataSet(MTConnectObservation observation, bool baseline, HashSet<string> seenSets)
{
var id = observation.DataItemId;
if (observation.Entries is not { Count: > 0 } entries)
{
var unavailable = string.IsNullOrWhiteSpace(observation.Value) && !observation.IsDataSet ||
string.Equals(observation.Value, UnavailableSentinel, StringComparison.Ordinal);
if (unavailable)
{
// Hot path for every ordinary UNAVAILABLE observation: skip the lock until a set exists at all.
if (Volatile.Read(ref _anySets)) lock (_setsLock) _sets.Remove(id);
return observation;
}
if (observation.IsDataSet && !observation.IsStructured)
{
// An entry-less data-set-shaped observation (count="0", or resetTriggered) is the EMPTY set - a
// legitimate value, Good "{}", not an absence. A reset or a /current baseline discards what
// was accumulated; a bare empty delta leaves the set as it is.
string emptyJson;
lock (_setsLock)
{
var first = seenSets.Add(id);
if (!_sets.TryGetValue(id, out var set) || observation.ResetTriggered || (baseline && first))
{
set = new SortedDictionary<string, SetEntry>(StringComparer.Ordinal);
_sets[id] = set;
Volatile.Write(ref _anySets, true);
}
emptyJson = RenderSet(set);
}
return observation with { Value = emptyJson, ResetTriggered = false, IsDataSet = false };
}
return observation;
}
string json;
lock (_setsLock)
{
var first = seenSets.Add(id);
if (!_sets.TryGetValue(id, out var set) || observation.ResetTriggered || (baseline && first))
{
set = new SortedDictionary<string, SetEntry>(StringComparer.Ordinal);
_sets[id] = set;
Volatile.Write(ref _anySets, true);
}
foreach (var entry in entries)
{
if (entry.Removed)
{
set.Remove(entry.Key);
}
else if (entry.Cells is not null)
{
var cells = new SortedDictionary<string, string>(StringComparer.Ordinal);
foreach (var cell in entry.Cells) cells[cell.Key] = cell.Value;
set[entry.Key] = new SetEntry(null, cells);
}
else
{
set[entry.Key] = new SetEntry(entry.Value ?? string.Empty, null);
}
}
json = RenderSet(set);
}
return observation with { Value = json, IsStructured = false, Entries = null, ResetTriggered = false, IsDataSet = false };
}
/// <summary>
/// Compact, deterministic JSON: keys ordinal-sorted, no whitespace. Values that are strict JSON numbers are
/// emitted as numbers with their original token (lossless — <c>007</c> and <c>+5</c> stay strings); everything
/// else is a string.
/// </summary>
private static string RenderSet(SortedDictionary<string, SetEntry> set)
{
using var stream = new MemoryStream();
using (var w = new Utf8JsonWriter(stream, new JsonWriterOptions { Encoder = JavaScriptEncoder.UnsafeRelaxedJsonEscaping }))
{
w.WriteStartObject();
foreach (var (key, entry) in set)
{
w.WritePropertyName(key);
if (entry.Cells is null)
{
WriteScalar(w, entry.Value ?? string.Empty);
continue;
}
w.WriteStartObject();
foreach (var (cellKey, cellValue) in entry.Cells)
{
w.WritePropertyName(cellKey);
WriteScalar(w, cellValue);
}
w.WriteEndObject();
}
w.WriteEndObject();
}
return System.Text.Encoding.UTF8.GetString(stream.ToArray());
}
private static void WriteScalar(Utf8JsonWriter w, string text)
{
if (JsonNumber.IsMatch(text) &&
double.TryParse(text, NumberStyles.Float, CultureInfo.InvariantCulture, out var d) && double.IsFinite(d))
{
w.WriteRawValue(text);
}
else
{
w.WriteStringValue(text);
}
}
private DateTime Now => _timeProvider.GetUtcNow().UtcDateTime;
@@ -363,11 +363,13 @@ internal static class MTConnectStreamsParser
/// <c>null</c> when there is none. Text found anywhere below an observation belongs to it —
/// that is what <c>XElement.Value</c> meant.
/// </param>
/// <param name="entryCollector">The enclosing observation's DATA_SET / TABLE entry list (#650), or <c>null</c> outside one.</param>
private static void ReadElement(
XmlReader reader,
ContainerKind parent,
List<MTConnectObservation> observations,
ValueSink? textSink)
ValueSink? textSink,
List<MTConnectDataSetEntry>? entryCollector = null)
{
var isObservation = parent != ContainerKind.None;
var isCondition = parent == ContainerKind.Condition;
@@ -379,6 +381,9 @@ internal static class MTConnectStreamsParser
string? conditionState = null;
int? sampleCount = null;
double? sampleRate = null;
var resetTriggered = false;
var hasCount = false;
List<MTConnectDataSetEntry>? entries = null;
(string? NativeCode, string? NativeSeverity, string? Qualifier, string? Type) conditionAttributes = default;
if (isObservation)
@@ -405,6 +410,9 @@ internal static class MTConnectStreamsParser
// against the declared sampleCount only when one parses).
if (!isCondition)
{
resetTriggered = !string.IsNullOrWhiteSpace(MTConnectXml.OptionalAttribute(reader, "resetTriggered"));
hasCount = MTConnectXml.OptionalAttribute(reader, "count") is not null;
entries = [];
sampleCount = int.TryParse(
MTConnectXml.OptionalAttribute(reader, "sampleCount"),
NumberStyles.Integer,
@@ -431,7 +439,7 @@ internal static class MTConnectStreamsParser
// enclosing observation's sink: a condition is never nested inside a text-valued observation.
var sink = isObservation ? new ValueSink() : textSink;
var hasElements = !reader.IsEmptyElement && ReadChildren(reader, kind, observations, sink);
var hasElements = !reader.IsEmptyElement && ReadChildren(reader, kind, observations, sink, entries ?? entryCollector);
if (!isObservation)
{
@@ -475,7 +483,13 @@ internal static class MTConnectStreamsParser
IsStructured: !isCondition && hasElements,
SampleCount: sampleCount,
SampleRate: sampleRate,
Condition: condition);
Condition: condition,
Entries: entries is { Count: > 0 } ? entries : null,
// resetTriggered only means "replace the set" on a data-set-shaped observation; on an ordinary
// accumulating SAMPLE/EVENT (PartCount resetTriggered="DAILY") it is not our concern.
ResetTriggered: resetTriggered && (hasCount || entries is { Count: > 0 }),
IsDataSet: !isCondition && (hasCount || entries is { Count: > 0 }));
}
/// <summary>
@@ -486,7 +500,8 @@ internal static class MTConnectStreamsParser
XmlReader reader,
ContainerKind kind,
List<MTConnectObservation> observations,
ValueSink? sink)
ValueSink? sink,
List<MTConnectDataSetEntry>? entries)
{
var hasElements = false;
@@ -496,7 +511,17 @@ internal static class MTConnectStreamsParser
{
case XmlNodeType.Element:
hasElements = true;
ReadElement(reader, kind, observations, sink);
// #650: an <Entry> directly under a non-condition observation is DATA_SET / TABLE
// content, captured structurally rather than as concatenated text.
if (entries is not null && kind == ContainerKind.None && MTConnectXml.IsNamed(reader, "Entry"))
{
ReadEntry(reader, entries, sink);
}
else
{
ReadElement(reader, kind, observations, sink, entries);
}
break;
@@ -520,6 +545,60 @@ internal static class MTConnectStreamsParser
return hasElements;
}
/// <summary>
/// Consumes one <c>&lt;Entry&gt;</c> (and its <c>&lt;Cell&gt;</c> children, for a TABLE row), honouring the
/// <see cref="ReadChildren"/> cursor contract: returns on the entry's end tag, or on the entry itself when empty.
/// An entry with no <c>key</c> is dropped — it cannot be addressed.
/// </summary>
private static void ReadEntry(XmlReader reader, List<MTConnectDataSetEntry> entries, ValueSink? sink)
{
var key = MTConnectXml.OptionalAttribute(reader, "key");
var removed = string.Equals(MTConnectXml.OptionalAttribute(reader, "removed")?.Trim(), "true", StringComparison.OrdinalIgnoreCase);
var empty = reader.IsEmptyElement;
var text = new System.Text.StringBuilder();
List<MTConnectDataSetCell>? cells = null;
while (!empty && reader.Read())
{
if (reader.NodeType == XmlNodeType.EndElement) break;
if (reader.NodeType == XmlNodeType.Element && MTConnectXml.IsNamed(reader, "Cell"))
{
cells ??= [];
var cellKey = MTConnectXml.OptionalAttribute(reader, "key");
var cellText = new System.Text.StringBuilder();
if (!reader.IsEmptyElement)
{
while (reader.Read() && reader.NodeType != XmlNodeType.EndElement)
{
if (reader.NodeType is XmlNodeType.Text or XmlNodeType.CDATA or XmlNodeType.SignificantWhitespace)
{ cellText.Append(reader.Value); sink?.Append(reader.Value); }
else if (reader.NodeType == XmlNodeType.Element)
reader.Skip();
}
}
if (!string.IsNullOrEmpty(cellKey)) cells.Add(new MTConnectDataSetCell(cellKey, cellText.ToString().Trim()));
}
else if (reader.NodeType is XmlNodeType.Text or XmlNodeType.CDATA or XmlNodeType.SignificantWhitespace)
{
{ text.Append(reader.Value); sink?.Append(reader.Value); }
}
else if (reader.NodeType == XmlNodeType.Element)
{
reader.Skip();
}
}
if (string.IsNullOrEmpty(key)) return;
entries.Add(removed
? new MTConnectDataSetEntry(key, null, true)
: cells is not null
? new MTConnectDataSetEntry(key, null, false, cells)
: new MTConnectDataSetEntry(key, text.ToString().Trim(), false));
}
/// <summary>
/// A condition's state is its element name. <c>Unavailable</c> is normalized to the
/// <see cref="UnavailableSentinel"/> spelling so downstream quality mapping has exactly one
@@ -27,7 +27,8 @@
EVENT, controlled vocabulary -> String fixture_execution, fixture_avail, fixture_mode
EVENT, free text -> String fixture_program, fixture_block (never fed),
fixture_writeback (never fed; write target)
EVENT, DATA_SET representation -> String fixture_varset
EVENT, DATA_SET representation -> String fixture_varset (JSON object, #650)
EVENT, TABLE representation -> String fixture_worktable (JSON object of objects, #650)
CONDITION (any type) -> String + IsAlarm fixture_x_travel, fixture_logic
fixture_block is deliberately NEVER fed by the adapter, so the Agent reports it
@@ -113,8 +114,10 @@
legal value, and never fed by the adapter, so nothing races the assertion.
-->
<DataItem category="EVENT" id="fixture_writeback" name="Pwriteback" type="PROGRAM_COMMENT"/>
<!-- DATA_SET representation demotes to String even though the Agent sends key=value entries. -->
<!-- DATA_SET representation is served as a String carrying JSON (#650); the Agent sends key=value entries. -->
<DataItem category="EVENT" id="fixture_varset" name="Pvars" type="VARIABLE" representation="DATA_SET"/>
<!-- TABLE representation (#650): rows of cells, served as nested JSON. -->
<DataItem category="EVENT" id="fixture_worktable" name="Ptab" type="WORK_OFFSET" representation="TABLE"/>
<DataItem category="CONDITION" id="fixture_logic" name="Plogic" type="LOGIC_PROGRAM"/>
</DataItems>
</Path>
@@ -83,7 +83,8 @@ observation correlation key is visible.
| `fixture_linenumber` | `Pline` | EVENT | LINE_NUMBER | | Int64 |
| `fixture_program` | `Pprogram` | EVENT | PROGRAM | | String |
| `fixture_block` | `Pblock` | EVENT | BLOCK | | String — **never fed ⇒ UNAVAILABLE** |
| `fixture_varset` | `Pvars` | EVENT | VARIABLE | DATA_SET | String (structured ⇒ BadNotSupported) |
| `fixture_varset` | `Pvars` | EVENT | VARIABLE | DATA_SET | String — JSON object (#650); `toolNumber`/`offsetX`/`offsetZ` stable, `cycle` moves every ~5 s |
| `fixture_worktable` | `Ptab` | EVENT | WORK_OFFSET | TABLE | String — JSON object of objects (#650); `G54`/`G55` stable, `G56` row moves every ~5 s |
| `fixture_logic` | `Plogic` | CONDITION | LOGIC_PROGRAM | | String + IsAlarm |
| `fixture_asset_changed` / `fixture_asset_removed` | — | EVENT | ASSET_CHANGED / ASSET_REMOVED | | String |
@@ -89,9 +89,12 @@ class Feed:
[
("Favail", "AVAILABLE"),
("Pprogram", "OTOPCUA-FIXTURE.NC"),
# DATA_SET representation: the Agent renders these as <Entry key=...> children,
# which is precisely the wire shape the driver demotes to String / BadNotSupported.
("Pvars", "toolNumber=7 offsetX=1.25 offsetZ=-0.5"),
# DATA_SET / TABLE representations: the Agent renders these as <Entry key=...>
# children, which the driver serves as a JSON String (#650). The three Pvars keys
# are STABLE for the life of the fixture; the periodic feed only touches `cycle`
# (Pvars) and the G56 row (Ptab).
("Pvars", "toolNumber=7 offsetX=1.25 offsetZ=-0.5 cycle=0"),
("Ptab", "G54={X=1.0 Y=2.0} G55={X=3.0} G56={X=0 Z=0.5}"),
]
)
self.send_conditions()
@@ -142,6 +145,17 @@ class Feed:
[("Cmode", CONTROLLER_MODES[(self._ticks // 44) % len(CONTROLLER_MODES)])]
)
# DATA_SET / TABLE churn (#650): every ~5 s change one key / one row so a live gate can
# observe a /sample delta merge. The other keys / rows never change.
if self._ticks % 20 == 0:
n = self._ticks // 20
self.send_observations(
[
("Pvars", f"cycle={n}"),
("Ptab", f"G56={{X={n} Z=0.5}}"),
]
)
# Re-assert conditions occasionally so a late-attaching driver still sees them.
if self._ticks % 60 == 0:
self.send_conditions()
@@ -448,6 +448,40 @@ public sealed class MTConnectAgentIntegrationTests(MTConnectAgentFixture agent)
results.ShouldAllBe(r => r.Value == null);
}
/// <summary>
/// #650: the fixture's DATA_SET (<c>fixture_varset</c>) and TABLE (<c>fixture_worktable</c>) read as Good
/// Strings carrying JSON. Only the keys the adapter never mutates are asserted exactly; the periodic
/// <c>cycle</c> key / <c>G56</c> row are asserted present, because they move every ~5 s.
/// </summary>
[Fact(Timeout = TestTimeoutMs)]
public async Task Read_serves_data_set_and_table_as_json_strings()
{
SkipIfDown();
await using var live = await LiveDriverAsync(
[Tag("fixture_varset", DriverDataType.String), Tag("fixture_worktable", DriverDataType.String)]);
var results = await live.Driver.ReadAsync(["fixture_varset", "fixture_worktable"], Ct);
if (results.Any(r => r.StatusCode == BadNoCommunication))
{
Assert.Skip("The fixture's adapter is not feeding Pvars/Ptab (UNAVAILABLE). Is the adapter service running?");
}
results.ShouldAllBe(r => r.StatusCode == Good);
using var set = JsonDocument.Parse(results[0].Value.ShouldBeOfType<string>());
set.RootElement.GetProperty("toolNumber").GetInt32().ShouldBe(7);
set.RootElement.GetProperty("offsetX").GetDouble().ShouldBe(1.25);
set.RootElement.GetProperty("offsetZ").GetDouble().ShouldBe(-0.5);
set.RootElement.TryGetProperty("cycle", out _).ShouldBeTrue();
using var table = JsonDocument.Parse(results[1].Value.ShouldBeOfType<string>());
table.RootElement.GetProperty("G54").GetProperty("Y").GetDouble().ShouldBe(2.0);
table.RootElement.GetProperty("G55").GetProperty("X").GetDouble().ShouldBe(3.0);
table.RootElement.GetProperty("G56").GetProperty("Z").GetDouble().ShouldBe(0.5);
}
// ---- subscriptions: the /sample long poll ----
/// <summary>
@@ -0,0 +1,214 @@
using Shouldly;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
namespace ZB.MOM.WW.OtOpcUa.Driver.MTConnect.Tests;
/// <summary>
/// #650 — DATA_SET / TABLE observations are served as a String carrying compact, deterministic JSON.
/// Pins the parse (entries, cells, removed, resetTriggered) and the index's merge semantics
/// (/current replaces, /sample merges, UNAVAILABLE drops the accumulated set).
/// </summary>
public sealed class MTConnectDataSetTests
{
private const uint Good = 0x00000000u;
private const uint BadNoCommunication = 0x80310000u;
private static string Doc(string events, string attrs = "") =>
$"""
<MTConnectStreams>
<Header instanceId="1" nextSequence="2" firstSequence="1"/>
<Streams><DeviceStream><ComponentStream><Events>{events}</Events></ComponentStream></DeviceStream></Streams>
</MTConnectStreams>
""";
private static string DataSet(string body, string attrs = "") =>
$"""<VariableDataSet dataItemId="ds" timestamp="2026-07-24T12:00:00Z" count="0"{attrs}>{body}</VariableDataSet>""";
private static string Table(string body, string attrs = "") =>
$"""<WorkOffsetTable dataItemId="tb" timestamp="2026-07-24T12:00:00Z" count="0"{attrs}>{body}</WorkOffsetTable>""";
private static MTConnectObservationIndex NewIndex() =>
new([new MTConnectTagDefinition("ds", DriverDataType.String), new MTConnectTagDefinition("tb", DriverDataType.String)]);
private static void Apply(MTConnectObservationIndex idx, string events, bool baseline = false) =>
idx.Apply(MTConnectStreamsParser.Parse(Doc(events)), baseline);
[Fact]
public void Parser_captures_entries_cells_removed_and_reset()
{
var result = MTConnectStreamsParser.Parse(Doc(
DataSet("""<Entry key="a">1</Entry><Entry key="b" removed="true"/>""", " resetTriggered=\"MANUAL\"") +
Table("""<Entry key="G54"><Cell key="X">1.5</Cell><Cell key="Y">2</Cell></Entry>""")));
var ds = result.Observations[0];
ds.ResetTriggered.ShouldBeTrue();
ds.Entries!.Count.ShouldBe(2);
ds.Entries[0].ShouldBe(new MTConnectDataSetEntry("a", "1", false));
ds.Entries[1].Removed.ShouldBeTrue();
var tb = result.Observations[1];
tb.Entries!.ShouldHaveSingleItem().Cells!.Select(c => (c.Key, c.Value))
.ShouldBe([("X", "1.5"), ("Y", "2")]);
}
[Fact]
public void Data_set_is_served_as_sorted_compact_json_with_numbers_and_strings()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="toolNumber">7</Entry><Entry key="offsetX">1.25</Entry><Entry key="name">A &amp; "B"</Entry><Entry key="part">007</Entry>"""), baseline: true);
var snap = idx.Get("ds");
snap.StatusCode.ShouldBe(Good);
snap.Value.ShouldBe("""{"name":"A & \"B\"","offsetX":1.25,"part":"007","toolNumber":7}""");
}
[Fact]
public void Table_is_served_as_nested_json()
{
var idx = NewIndex();
Apply(idx, Table("""<Entry key="G55"><Cell key="X">3</Cell></Entry><Entry key="G54"><Cell key="Y">2</Cell><Cell key="X">1.0</Cell></Entry>"""), baseline: true);
idx.Get("tb").Value.ShouldBe("""{"G54":{"X":1.0,"Y":2},"G55":{"X":3}}""");
}
[Fact]
public void Sample_delta_merges_and_removed_entry_disappears()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry><Entry key="b">2</Entry>"""), baseline: true);
Apply(idx, DataSet("""<Entry key="b">3</Entry><Entry key="c">x</Entry>"""));
idx.Get("ds").Value.ShouldBe("""{"a":1,"b":3,"c":"x"}""");
Apply(idx, DataSet("""<Entry key="a" removed="true"/>"""));
idx.Get("ds").Value.ShouldBe("""{"b":3,"c":"x"}""");
}
[Fact]
public void Table_row_delta_replaces_the_row_and_removed_row_disappears()
{
var idx = NewIndex();
Apply(idx, Table("""<Entry key="G54"><Cell key="X">1</Cell><Cell key="Y">2</Cell></Entry><Entry key="G55"><Cell key="X">3</Cell></Entry>"""), baseline: true);
Apply(idx, Table("""<Entry key="G54"><Cell key="X">9</Cell></Entry><Entry key="G55" removed="true"/>"""));
idx.Get("tb").Value.ShouldBe("""{"G54":{"X":9}}""");
}
[Fact]
public void Current_baseline_replaces_rather_than_merges()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry><Entry key="b">2</Entry>"""), baseline: true);
Apply(idx, DataSet("""<Entry key="c">3</Entry>"""), baseline: true);
idx.Get("ds").Value.ShouldBe("""{"c":3}""");
}
[Fact]
public void ResetTriggered_replaces_the_accumulated_set_on_a_sample()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry>"""), baseline: true);
Apply(idx, DataSet("""<Entry key="z">9</Entry>""", " resetTriggered=\"MANUAL\""));
idx.Get("ds").Value.ShouldBe("""{"z":9}""");
}
[Fact]
public void ResetTriggered_with_no_entries_is_an_empty_set_not_an_absence()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry>"""), baseline: true);
Apply(idx, DataSet("", " resetTriggered=\"MANUAL\""));
var snap = idx.Get("ds");
snap.StatusCode.ShouldBe(Good);
snap.Value.ShouldBe("{}");
}
[Fact]
public void Unavailable_is_BadNoCommunication_and_drops_the_accumulated_set()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry>"""), baseline: true);
Apply(idx, DataSet("UNAVAILABLE"));
var snap = idx.Get("ds");
snap.StatusCode.ShouldBe(BadNoCommunication);
snap.Value.ShouldBeNull();
// A delta after the outage must not resurrect the pre-outage entry.
Apply(idx, DataSet("""<Entry key="b">2</Entry>"""));
idx.Get("ds").Value.ShouldBe("""{"b":2}""");
}
[Fact]
public void Several_observations_in_one_document_fold_in_order()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry>""") + DataSet("""<Entry key="b">2</Entry>"""), baseline: true);
idx.Get("ds").Value.ShouldBe("""{"a":1,"b":2}""");
}
[Fact]
public void Clear_forgets_the_accumulated_set()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry>"""), baseline: true);
idx.Clear();
Apply(idx, DataSet("""<Entry key="b">2</Entry>"""));
idx.Get("ds").Value.ShouldBe("""{"b":2}""");
}
[Fact]
public void Structured_content_that_is_not_entries_stays_BadNotSupported()
{
var idx = NewIndex();
Apply(idx, DataSet("<Other>1</Other>"), baseline: true);
idx.Get("ds").StatusCode.ShouldBe(0x803D0000u);
}
[Fact]
public void ResetTriggered_on_an_ordinary_sample_keeps_its_numeric_value()
{
// Regression: resetTriggered is also a standard attribute on accumulating SAMPLEs / counters. It must
// never turn one into an empty set.
var idx = new MTConnectObservationIndex(
[new MTConnectTagDefinition("pc", DriverDataType.Int64), new MTConnectTagDefinition("amt", DriverDataType.Float64)]);
idx.Apply(MTConnectStreamsParser.Parse(Doc(
"""<PartCount dataItemId="pc" timestamp="2026-07-24T12:00:00Z" resetTriggered="DAILY">5</PartCount>""" +
"""<Amount dataItemId="amt" timestamp="2026-07-24T12:00:00Z" resetTriggered="DAILY">2.5</Amount>""")));
idx.Get("pc").StatusCode.ShouldBe(Good);
idx.Get("pc").Value.ShouldBe(5L);
idx.Get("amt").Value.ShouldBe(2.5d);
}
[Fact]
public void An_empty_current_data_set_is_Good_empty_json_like_a_reset_one()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry>"""), baseline: true);
Apply(idx, DataSet(""), baseline: true);
var snap = idx.Get("ds");
snap.StatusCode.ShouldBe(Good);
snap.Value.ShouldBe("{}");
}
[Fact]
public void An_all_removed_set_is_the_same_empty_json()
{
var idx = NewIndex();
Apply(idx, DataSet("""<Entry key="a">1</Entry>"""), baseline: true);
Apply(idx, DataSet("""<Entry key="a" removed="true"/>"""));
idx.Get("ds").Value.ShouldBe("{}");
}
}
@@ -589,11 +589,12 @@ public sealed class MTConnectObservationIndexTests
/// <summary>
/// End-to-end through the real parser: the entry list that concatenates to the single token
/// "12" is caught by the parser's structured flag and never surfaces as a Good value. This is
/// the leg that would regress silently if the parser stopped setting the flag.
/// "12" must never surface as that noise. Since #650 it is served as JSON carrying the keys
/// (full semantics in <c>MTConnectDataSetTests</c>); the structured-without-entries shape above
/// stays BadNotSupported.
/// </summary>
[Fact]
public void A_parsed_data_set_observation_never_surfaces_as_a_good_value()
public void A_parsed_data_set_observation_never_surfaces_as_concatenated_text()
{
const string xml = """
<MTConnectStreams>
@@ -612,8 +613,8 @@ public sealed class MTConnectObservationIndexTests
var snap = idx.Get("ds1");
snap.StatusCode.ShouldBe(BadNotSupported);
snap.Value.ShouldBeNull();
snap.StatusCode.ShouldBe(Good);
snap.Value.ShouldBe("""{"V1":1,"V2":2}""");
}
/// <summary>An ordinary text observation is not structured — the flag must not be over-applied.</summary>
@@ -377,6 +377,42 @@ public sealed class MTConnectSubscribeTests
For(seen, "dev1_pos").Last().Snapshot.Value.ShouldBe(77.5d);
}
/// <summary>
/// #650: a re-baseline <c>/current</c> states the WHOLE data set, so it must REPLACE what the deltas
/// accumulated - a key removed while the stream was down must not linger. Reverting the
/// <c>baseline: true</c> argument at the re-baseline call site leaves {a,b} merged and fails this.
/// (The init and one-shot-read call sites build a fresh index, so their flag is not observable.)
/// </summary>
[Fact]
public async Task Rebaseline_replaces_the_accumulated_data_set_rather_than_merging()
{
var (driver, client) = await InitializedDriverAsync();
static MTConnectStreamsResult SetChunk(long first, long next, params (string Key, string Value)[] entries) =>
new(FixtureInstanceId, next, first,
[
new MTConnectObservation(
"ds", "", ObservedAt, IsStructured: true, IsDataSet: true,
Entries: [.. entries.Select(e => new MTConnectDataSetEntry(e.Key, e.Value, false))]),
]);
const string Oor = "MTConnect /sample answered an MTConnectError document (OUT_OF_RANGE) under HTTP 200";
// First re-baseline seeds {a,b}; the stream then resumes from its nextSequence (500).
client.SampleFailures.Enqueue(new InvalidDataException(Oor));
client.CurrentAnswers.Enqueue(SetChunk(490, 500, ("a", "1"), ("b", "2")));
await driver.SubscribeAsync(["dev1_pos"], TimeSpan.FromMilliseconds(50), Ct);
await client.PumpAsync(Chunk(500, 505, ("dev1_pos", "1.5"))).WaitAsync(Watchdog, Ct);
driver.ObservationIndex.Get("ds").Value.ShouldBe("""{"a":1,"b":2}""");
client.SampleFailures.Enqueue(new InvalidDataException(Oor));
client.CurrentAnswers.Enqueue(SetChunk(600, 610, ("a", "1")));
await client.PumpAsync(Chunk(610, 615, ("dev1_pos", "77.5"))).WaitAsync(Watchdog, Ct);
driver.ObservationIndex.Get("ds").Value.ShouldBe("""{"a":1}""");
}
// ---- agent restart ----
/// <summary>