feat(mtconnect): serve DATA_SET/TABLE as compact JSON strings; park EVENT enumerations (#650)

Parser captures Entry/Cell content; the observation index merges /sample deltas into the /current baseline (resetTriggered replaces, removed deletes, UNAVAILABLE drops). Fixture gains a TABLE item and periodic DATA_SET/TABLE churn. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
2026-10-08 19:15:32 -04:00
parent 8801631739
commit f8f8eb9fb8
11 changed files with 537 additions and 39 deletions
@@ -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,18 @@ 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 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 +174,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 +189,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 +239,130 @@ 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)
+        {
+            if (observation.ResetTriggered && !observation.IsStructured)
+            {
+                // resetTriggered with no entries: an empty set, which is a legitimate value rather than an absence.
+                lock (_setsLock) _sets[id] = new SortedDictionary<string, SetEntry>(StringComparer.Ordinal);
+                seenSets.Add(id);
+                return observation with { Value = "{}", ResetTriggered = false };
+            }
+
+            if (string.IsNullOrWhiteSpace(observation.Value) ||
+                string.Equals(observation.Value, UnavailableSentinel, StringComparison.Ordinal))
+            {
+                lock (_setsLock) _sets.Remove(id);
+            }
+
+            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;
+            }
+
+            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 };
+    }
+
+    /// <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;