diff --git a/docs/drivers/Mqtt.md b/docs/drivers/Mqtt.md index 76be2763..c01aeb83 100644 --- a/docs/drivers/Mqtt.md +++ b/docs/drivers/Mqtt.md @@ -204,7 +204,7 @@ survive a reconnect; a message fans out to *every* matching tag, and unauthored | `groupId` | string | — | **Required.** Must match the driver's `Sparkplug.GroupId` to ever receive a message | | `edgeNodeId` | string | — | **Required.** | | `deviceId` | string | *absent* | **Omit** for a metric published by the edge node itself (NBIRTH/NDATA). Absent is meaningfully different from `""` | -| `metricName` | string | — | **Required.** The tag's stable binding key across rebirths. **May contain `/`** — `Node Control/Rebirth` and `Properties/Serial Number` are canonical Sparkplug names, and the whole name is one value, never split | +| `metricName` | string | — | **Required.** The tag's stable binding key across rebirths. **May contain `/`** — `Node Control/Rebirth` and `Properties/Serial Number` are canonical Sparkplug names, and the whole name is one value, never split. Also how a flattened structured member is addressed: `Motor/Speed`, `Recipe/Target/0` (see [Structured metrics](#structured-metrics-template--dataset--propertyset)) | | `dataType` | `DriverDataType` | *from the birth* | **Optional override.** Absent ⇒ take whatever type the birth certificate declared. The AdminUI's "(from birth certificate)" option removes the key entirely | | `isHistorized` / `historianTagname` | — | — | Server-side historization, unchanged from Plain | @@ -222,6 +222,11 @@ bind. `metricName` is deliberately **exempt**: it is not a topic segment. // A node-level metric — note deviceId is ABSENT, not blank. {"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","metricName":"Temperature"} + +// A flattened Template member, and a flattened DataSet cell. `Motor` and `Recipe` are what the wire +// carries; nothing on the wire is ever called by these names. +{"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"Motor/Stator/Temp"} +{"groupId":"OtOpcUaSim","edgeNodeId":"EdgeA","deviceId":"Filler1","metricName":"Recipe/Target/1"} ``` ## Sparkplug B Ingest @@ -240,9 +245,11 @@ One subscription (`spBv1.0/{GroupId}/#`) feeds a per-edge-node state machine - **Death → STALE.** An NDEATH whose `bdSeq` matches the live session (the bdSeq tie is what stops a stale death from killing a fresh session) marks every metric under that edge node `Bad` / `BadNoCommunication`. A DDEATH does the same for one device. The next birth restores them. -- **Unsupported datatypes are skipped with a warning, never guessed**: `DataSet`, `Template`, - `PropertySet`, `PropertySetList`, `Unknown`. `Int8`→`Int16` and `UInt8`→`UInt16` widen; `*Array` - variants map to the element type with the array bit set. +- **Structured metrics flatten to sibling tags** — `Template` and `DataSet`; see + [Structured metrics](#structured-metrics-template--dataset--propertyset). +- **Unsupported datatypes are skipped with a warning, never guessed**: `PropertySetList`, `Unknown`, + extension values, and any `Template`/`DataSet` that cannot be flattened honestly. `Int8`→`Int16` and + `UInt8`→`UInt16` widen; `*Array` variants map to the element type with the array bit set. - **Ingest state is reset at every session/authoring seam** — a reconnect or a `ReinitializeAsync` must not carry a previous session's alias table into a new one. @@ -260,6 +267,53 @@ re-births freely — on its own schedule, on every reconnect, and once per rebir > through the `/raw` browse-commit flow, and grafting nodes at runtime would materialise ones nobody > approved and no deployment artifact records. +## Structured metrics (`Template` / `DataSet` / `PropertySet`) + +Sparkplug metrics are not all scalars. `SparkplugFlattener` +([#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515)) turns the structured ones into +addressable tags: + +| Metric kind | Mapping | Example | +|---|---|---| +| `Template` (UDT instance) | members become **sibling metrics**, `/`-joined, recursively | `Motor` → `Motor/Speed`, `Motor/Stator/Temp` | +| `DataSet` | **one metric per cell**, `{column}/{rowIndex}` | `Recipe` → `Recipe/Step/0`, `Recipe/Target/1` | +| `PropertySet` (a metric's `properties`) | node **metadata**, never tags | `engUnit`, `Range/Low` | + +Flattened members are authored exactly like any other Sparkplug metric — the composed name goes in +`metricName`. The browse picker offers them directly (the structured **parent** is deliberately *not* +offered: it has no value and no servable datatype, so a tag committed against it could never read). + +**Why flattening rather than OPC UA Structures.** Owner decision, 2026-07-31. Structures are truer to +Sparkplug's model but much harder to consume — flattened scalars are readable by any OPC UA client +with no type dictionary, and the result mirrors how the `/raw` tree already works. + +### Things worth knowing before you touch this + +- **The row index is always present, including on row 0.** `Recipe/Target/0`, never `Recipe/Target`. + Emitting a bare column name for a single-row table and an indexed one otherwise would make a tag's + name depend on how many rows the plant published — so a table growing from one row to two would + rename every tag already authored against it. +- **A DATA metric carries the parent's alias and no name**, so member names cannot be composed until + the alias table has resolved the parent. Flattening at decode time alone is not enough; the + expansion happens in `SparkplugIngestor.PublishStructuredMembers`, after resolution. +- **Member datatype falls back to the birth's.** A DDATA template may omit its members' datatypes. + Guessing from the CLR type of the wire value would read a signed `Int32` of −42 out of its unsigned + field as 4294967254 — plausible, `Good`, and wrong. +- **Skip-and-warn survives, and refusal is whole.** A template **definition** (`is_definition`), an + unnamed member, an extension value, a column/type count mismatch, a short row, or a payload past + `MaxDepth` (5) / `MaxMembers` (512) refuses the **entire** metric rather than publishing part of it. + A silently half-flattened UDT looks exactly like a complete one in the picker. +- **Member paths contain `/`**, which `RawBrowseCommitMapper.ToRawSegment` + ([#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509)) already handles: the slashed name + travels intact as `metricName` while the derived tag **Name** is coerced to a legal RawPath segment, + with a collision guard for two members that would sanitise onto one name. +- **The rediscovery signature tracks members, not the parent.** A UDT gaining or losing a member leaves + the parent metric's name identical, so a gate built on raw wire names would never move. + +`PropertySet` is projected as metadata rather than as sibling tags because it *describes* a metric +(units, ranges, quality hints) — surfacing it as tags would invent plant signals out of documentation. +`PropertySetList` and property extension values are dropped rather than stringified. + ## Primary host (`actAsPrimaryHost`) A Sparkplug **Primary Host Application** is the thing edge nodes watch to decide whether anyone is @@ -515,7 +569,7 @@ identical words. See [TestConnectProbes.md](TestConnectProbes.md). reconnect/rebuild branches, CONNACK classification, and the Sparkplug half — golden-payload decode, topic parse/format, datatype map, alias/birth-cache rebuild, sequence + bdSeq-death ties, the rediscovery change gate, and the browse session's "publishes nothing" assertion. -- **Live (env-gated)** — `tests/Drivers/…Driver.Mqtt.IntegrationTests` (**26 tests**) against the +- **Live (env-gated)** — `tests/Drivers/…Driver.Mqtt.IntegrationTests` (**29 tests**) against the Mosquitto broker + the C# Sparkplug edge-node simulator; skips cleanly when `MQTT_FIXTURE_ENDPOINT` is unset. See [Mqtt-Test-Fixture.md](Mqtt-Test-Fixture.md) and `infra/README.md` §3.
The primary-host legs (`MqttPrimaryHostLiveTests`) include the one the ordinary lifecycle cannot @@ -541,7 +595,8 @@ identical words. See [TestConnectProbes.md](TestConnectProbes.md). Every row is tracked. None is a silent failure — each either refuses in words or logs. **Closed 2026-07-31:** write-through ([#508](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/508), live-gated), primary-host STATE publishing -([#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511), live-gated), the slashed-metric browse-commit +([#511](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/511), live-gated), structured +`Template`/`DataSet`/`PropertySet` metrics ([#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515), live-gated), the slashed-metric browse-commit ([#509](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/509)), the stale rebirth affordance ([#510](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/510)), the `.Browser` layering exception ([#512](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/512)), the stray Sparkplug `payloadFormat` ([#513](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/513)) and rebirth on a not-yet-birthed plant ([#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/514)). @@ -549,4 +604,3 @@ not-yet-birthed plant ([#514](https://gitea.dohertylan.com/dohertj2/lmxopcua/iss | Gap | Issue | Effect | Notes | |---|:---:|---|---| | **Rediscovery is advisory, not a rebuild** | [#507](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/507) (closed, superseded) | A DBIRTH introducing a metric does not change the served OPC UA tree; an operator must re-browse the device and commit | **No longer "inert".** `DriverInstanceActor` now subscribes to `OnRediscoveryNeeded` and the raise rides the driver-health snapshot to the AdminUI `/hosts` page as a **re-browse chip**. It deliberately does not inject nodes: v3 authors raw tags through the `/raw` browse-commit flow, and grafting nodes at runtime would materialise ones nobody approved and no deployment artifact records. #507's injection path was deleted rather than revived | -| **`DataSet` / `Template` / `PropertySet` metrics** | [#515](https://gitea.dohertylan.com/dohertj2/lmxopcua/issues/515) | Skipped with a warning | v1-unsupported; never coerced into a guessed type | diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs index cb240b42..2e37d579 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Browser/MqttBrowseSession.cs @@ -585,20 +585,77 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession continue; } - var metricPath = ownerPath + MetricSeparator + name; - if (metricPath.Length > MaxTopicChars) + // Gitea #515: a Template/DataSet metric contributes its flattened MEMBERS as leaves, never + // itself. The parent has no value and no servable datatype, so a leaf for it would offer + // the operator an address that can only ever commit a tag which never reads. + if (metric.ValueKind == SparkplugValueKind.Structured) { - Volatile.Write(ref _oversized, 1); + foreach (var member in metric.MemberList) + { + var memberName = $"{name}{SparkplugFlattener.MemberSeparator}{member.Path}"; + if (!IsUsableLabel(memberName)) + { + Volatile.Write(ref _oversized, 1); + continue; + } + + if (!AddMetricLeaf(owner, ownerPath, ownerScope, memberName, member.DataType, parsed.Type, alias: null)) + { + return; // node cap tripped + } + } + continue; } - var node = GetOrCreate(owner.Children, metricPath, name, BrowseNodeKind.Leaf, ownerScope); - if (node is null) return; // node cap tripped - - node.RecordMetric(new SparkplugMetricInfo(name, metric.DataType, parsed.Type, metric.Alias)); + if (!AddMetricLeaf(owner, ownerPath, ownerScope, name, metric.DataType, parsed.Type, metric.Alias)) + { + return; // node cap tripped + } } } + /// + /// Places one metric leaf under . + /// + /// The owning edge-node / device folder. + /// That folder's path. + /// The scope stamped on the leaf. + /// + /// The metric's Sparkplug name — for a flattened structured member, the composed + /// {parent}/{memberPath}. It travels intact: it is the binding key, and only the + /// display name is later coerced to a legal RawPath segment (Gitea #509). + /// + /// The declared datatype, or null. + /// The birth message type this leaf was seen in. + /// The metric's alias, or null — flattened members carry none of their own. + /// when the node cap tripped and the caller must stop. + private bool AddMetricLeaf( + TopicNode owner, + string ownerPath, + SparkplugScope ownerScope, + string metricName, + Org.Eclipse.Tahu.Protobuf.DataType? dataType, + SparkplugMessageType messageType, + ulong? alias) + { + var metricPath = ownerPath + MetricSeparator + metricName; + if (metricPath.Length > MaxTopicChars) + { + Volatile.Write(ref _oversized, 1); + return true; + } + + var node = GetOrCreate(owner.Children, metricPath, metricName, BrowseNodeKind.Leaf, ownerScope); + if (node is null) + { + return false; + } + + node.RecordMetric(new SparkplugMetricInfo(metricName, dataType, messageType, alias)); + return true; + } + /// /// Finds or creates one tree node under , honouring the node cap. /// @@ -698,6 +755,40 @@ internal sealed class MqttBrowseSession : IRebirthCapableBrowseSession ObserveBirth(groupId, edgeNodeId, device, metric); } + /// + /// Test seam feeding a birth whose metric is a structured Template (Gitea #515). + /// + /// The Sparkplug group id. + /// The Sparkplug edge-node id. + /// The device id, or for an NBIRTH. + /// The template metric's own name. + /// The template's members, as (name, datatype). + internal void ObserveTemplateBirthForTest( + string groupId, + string edgeNodeId, + string? device, + string metricName, + params (string Name, SparkplugDataType DataType)[] members) + { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + foreach (var (name, dataType) in members) + { + var member = new Payload.Types.Metric { Name = name, Datatype = (uint)dataType }; + + // A value is required: a member carrying only a name and a datatype is indistinguishable + // from a template DEFINITION's member, which the flattener deliberately refuses. + member.DoubleValue = 0; + template.Metrics.Add(member); + } + + ObserveBirth(groupId, edgeNodeId, device, new Payload.Types.Metric + { + Name = metricName, + Datatype = (uint)SparkplugDataType.Template, + TemplateValue = template, + }); + } + /// Test seam feeding a birth certificate that declares no metrics at all. /// The Sparkplug group id. /// The Sparkplug edge-node id. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs index b428ee19..309aa3c1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugCodec.cs @@ -45,10 +45,19 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; /// wire, not of the tag. /// /// -/// DataSet and Template metrics are out of scope for v1. They decode to +/// Template and DataSet metrics decode to +/// with their members flattened into +/// (Gitea #515) — a template member Speed of metric +/// Motor becomes the addressable Motor/Speed. See +/// for the mapping and its bounds. Anything that cannot be flattened honestly stays /// with a null value — deliberately visible, so a -/// consumer can warn about a metric it cannot serve instead of silently treating it as one -/// that was never published. +/// consumer warns about a metric it cannot serve instead of silently treating it as one that was +/// never published, and never coerces it into a guessed scalar. +/// +/// +/// A metric's PropertySet is projected to +/// as node metadata, never as sibling values: a PropertySet describes a metric (units, ranges, +/// quality hints); it is not something the metric is. /// /// public static class SparkplugCodec @@ -161,7 +170,7 @@ public static class SparkplugCodec /// The projection. private static SparkplugMetric ProjectMetric(Payload.Types.Metric metric) { - var (kind, value) = ProjectValue(metric); + var (kind, value, members) = ProjectValue(metric); return new SparkplugMetric( Name: metric.HasName ? metric.Name : null, @@ -173,7 +182,22 @@ public static class SparkplugCodec DataType: metric.HasDatatype ? (TahuDataType)metric.Datatype : null, TimestampMs: metric.HasTimestamp ? metric.Timestamp : null, ValueKind: kind, - Value: value); + Value: value, + Members: members, + Properties: SparkplugFlattener.ProjectProperties(metric.Properties)); + } + + /// + /// Resolves a template member's value oneof. Internal to the assembly: shared with + /// , which walks nested templates and needs the identical + /// projection so a member and a top-level metric of the same type never decode differently. + /// + /// The template member, itself a generated Metric. + /// The value kind and the projected raw value. + internal static (SparkplugValueKind Kind, object? Value) ProjectMemberValue(Payload.Types.Metric metric) + { + var (kind, value, _) = ProjectValue(metric); + return (kind, value); } /// Resolves a metric's value oneof to a kind plus a raw CLR value. @@ -184,34 +208,52 @@ public static class SparkplugCodec /// that some datatypes have no spare sentinel, so a publisher that sets it means "null" even if /// a value field is also populated. /// - private static (SparkplugValueKind Kind, object? Value) ProjectValue(Payload.Types.Metric metric) + private static (SparkplugValueKind Kind, object? Value, IReadOnlyList Members) ProjectValue( + Payload.Types.Metric metric) { if (metric.HasIsNull && metric.IsNull) { - return (SparkplugValueKind.Null, null); + return (SparkplugValueKind.Null, null, []); + } + + // Gitea #515: structured values flatten to addressable members. A refusal to flatten falls + // through to Unsupported — the pre-#515 behaviour — rather than to a guessed scalar. + switch (metric.ValueCase) + { + case Payload.Types.Metric.ValueOneofCase.TemplateValue: + return SparkplugFlattener.TryFlattenTemplate(metric.TemplateValue, out var templateMembers) + ? (SparkplugValueKind.Structured, null, templateMembers) + : (SparkplugValueKind.Unsupported, null, []); + + case Payload.Types.Metric.ValueOneofCase.DatasetValue: + return SparkplugFlattener.TryFlattenDataSet(metric.DatasetValue, out var cells) + ? (SparkplugValueKind.Structured, null, cells) + : (SparkplugValueKind.Unsupported, null, []); } return metric.ValueCase switch { - Payload.Types.Metric.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, metric.IntValue), - Payload.Types.Metric.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, metric.LongValue), - Payload.Types.Metric.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, metric.FloatValue), - Payload.Types.Metric.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, metric.DoubleValue), - Payload.Types.Metric.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, metric.BooleanValue), - Payload.Types.Metric.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, metric.StringValue), + Payload.Types.Metric.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, metric.IntValue, Empty), + Payload.Types.Metric.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, metric.LongValue, Empty), + Payload.Types.Metric.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, metric.FloatValue, Empty), + Payload.Types.Metric.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, metric.DoubleValue, Empty), + Payload.Types.Metric.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, metric.BooleanValue, Empty), + Payload.Types.Metric.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, metric.StringValue, Empty), // Copied out of the ByteString: the projection must outlive the parsed message. Payload.Types.Metric.ValueOneofCase.BytesValue => - (SparkplugValueKind.Scalar, metric.BytesValue.ToByteArray()), + (SparkplugValueKind.Scalar, metric.BytesValue.ToByteArray(), Empty), - // v1 scope boundary — decoded as a visible refusal, not as an exception or a silent drop. - Payload.Types.Metric.ValueOneofCase.DatasetValue => (SparkplugValueKind.Unsupported, null), - Payload.Types.Metric.ValueOneofCase.TemplateValue => (SparkplugValueKind.Unsupported, null), - Payload.Types.Metric.ValueOneofCase.ExtensionValue => (SparkplugValueKind.Unsupported, null), + // An extension value is a third-party escape hatch with no defined semantics — there is + // nothing honest to flatten it into. + Payload.Types.Metric.ValueOneofCase.ExtensionValue => (SparkplugValueKind.Unsupported, null, Empty), - _ => (SparkplugValueKind.Absent, null), + _ => (SparkplugValueKind.Absent, null, Empty), }; } + + /// The shared empty member list — every scalar metric would otherwise allocate one. + private static readonly IReadOnlyList Empty = []; } /// @@ -279,13 +321,31 @@ public sealed record SparkplugPayload( /// switch over a dozen arms. /// /// +/// +/// The flattened members of a metric — a +/// Template or DataSet broken into addressable scalars (Gitea #515). Empty for every +/// other kind. Paths are relative to this metric; see . +/// +/// +/// The metric's PropertySet, projected as node metadata (Gitea #515). Empty when the metric +/// carried none. Never a value — a PropertySet describes the metric rather than being one. +/// public readonly record struct SparkplugMetric( string? Name, ulong? Alias, TahuDataType? DataType, ulong? TimestampMs, SparkplugValueKind ValueKind, - object? Value); + object? Value, + IReadOnlyList? Members = null, + IReadOnlyList? Properties = null) +{ + /// The flattened members, never null. + public IReadOnlyList MemberList => Members ?? []; + + /// The projected properties, never null. + public IReadOnlyList PropertyList => Properties ?? []; +} /// What a decoded metric's represents. /// @@ -305,9 +365,17 @@ public enum SparkplugValueKind Scalar, /// - /// A DataSet, Template or extension value — decoded and reported, but not - /// supported in v1. The consumer should warn and skip the metric rather than treat it as - /// missing. + /// An extension value, or a Template/DataSet that could not be flattened + /// honestly (see ). Decoded and reported, but not servable: + /// the consumer warns and skips the metric rather than treating it as missing — and never + /// coerces it into a guessed scalar. /// Unsupported, + + /// + /// A Template or DataSet whose members were flattened into + /// (Gitea #515). is + /// null — the parent has no scalar value of its own; its members carry the values. + /// + Structured, } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugFlattener.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugFlattener.cs new file mode 100644 index 00000000..7cb81bb8 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Contracts/Sparkplug/SparkplugFlattener.cs @@ -0,0 +1,368 @@ +using Org.Eclipse.Tahu.Protobuf; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; + +/// +/// Flattens Sparkplug's structured metric values — Template (UDT instances) and +/// DataSet — into addressable scalar members, and projects a metric's PropertySet +/// into node metadata (Gitea #515). +/// +/// +/// +/// Flatten to siblings, by owner decision (2026-07-31). The alternative was mapping a +/// Template to an OPC UA Structure/UDT, which is truer to Sparkplug's model but far harder to +/// consume: flattened scalars are readable by any OPC UA client (ScadaBridge included) with no +/// type dictionary, and the result mirrors how the /raw tree already works. A member of +/// template metric Motor called Speed becomes the metric Motor/Speed. +/// +/// +/// Member paths contain /, which is already solved. That interaction is what +/// #515 flagged against #509, and #509 shipped first: +/// RawBrowseCommitMapper.ToRawSegment coerces the tag Name to a legal RawPath +/// segment while carrying the original slashed string through as metricName, plus a +/// collision guard that refuses two members which would sanitise onto one name. So the +/// flattened name travels intact as the binding key and only the display name is coerced. +/// +/// +/// Skip-and-warn survives, and that is a requirement rather than a leftover. Anything +/// this cannot flatten honestly — an extension value, a nested member with no name, a payload +/// past the bounds below — stays and is skipped +/// with a warning. A structured value coerced into a guessed scalar would bind a tag to data +/// that is not what its name says, which is worse than not binding it at all. +/// +/// +/// Bounded, because this walks an attacker-shaped object graph. A Template may contain +/// Templates, so the natural recursion is unbounded in a format anyone on the broker can +/// publish. and cap it; exceeding either +/// abandons the whole metric to Unsupported rather than publishing a truncated member +/// set, because a silently half-flattened UDT looks exactly like a complete one. +/// +/// +/// Nothing here throws — same contract as , for the same +/// reason: this runs on MQTTnet's shared dispatcher thread behind an unauthenticated firehose. +/// +/// +public static class SparkplugFlattener +{ + /// + /// How deeply nested a Template may be before the whole metric is refused. Five is far past + /// anything a real UDT library produces and still bounds the walk. + /// + public const int MaxDepth = 5; + + /// + /// Ceiling on the flattened members one metric may produce. A DataSet is columns × rows, so a + /// modest-looking table is the realistic way to reach this. + /// + public const int MaxMembers = 512; + + /// + /// Separator between a structured metric's name and its member path. The same character + /// Sparkplug itself uses for metric-name hierarchy, so Motor/Speed is what an edge node + /// would have called a flat metric of the same meaning. + /// + public const char MemberSeparator = '/'; + + /// + /// Flattens a Template into member paths relative to the owning metric. + /// + /// The template value. + /// The flattened members on success; empty otherwise. + /// + /// when the whole template flattened. means the + /// caller must treat the metric as . + /// + public static bool TryFlattenTemplate(Payload.Types.Template template, out IReadOnlyList members) + { + var collected = new List(); + var ok = template is not null && AppendTemplate(template, prefix: "", depth: 1, collected); + members = ok ? collected : []; + return ok; + } + + /// + /// Flattens a DataSet into one member per cell, named {column}/{rowIndex}. + /// + /// The dataset value. + /// The flattened members on success; empty otherwise. + /// + /// when the whole dataset flattened. means the + /// caller must treat the metric as . + /// + /// + /// + /// Per cell, not per column — and the row index is always present. The recorded + /// decision says "per-column tags, not one array-valued variable", and a column of N rows + /// is not a scalar, so per-column can only mean per-cell once N > 1. Emitting a bare + /// {column} for the single-row case and {column}/{row} otherwise would make + /// a tag's own name depend on how many rows the plant happened to publish — so a table + /// that grew from one row to two would rename every existing tag out from under its + /// authored binding. Uniform naming costs a /0 on the common single-row "current + /// values" shape and buys a stable address. + /// + /// + /// A row that disappears simply stops publishing, and its tag holds its last value — the + /// same as any metric an edge node stops sending. + /// + /// + public static bool TryFlattenDataSet(Payload.Types.DataSet dataSet, out IReadOnlyList members) + { + members = []; + if (dataSet is null) + { + return false; + } + + // Columns and types are parallel arrays. A publisher that disagrees with itself has produced a + // dataset whose cells cannot be typed, and guessing which array to trust would type a column + // by its neighbour — refuse instead. + if (dataSet.Columns.Count == 0 || dataSet.Types_.Count != dataSet.Columns.Count) + { + return false; + } + + if ((long)dataSet.Columns.Count * dataSet.Rows.Count > MaxMembers) + { + return false; + } + + var collected = new List(); + for (var row = 0; row < dataSet.Rows.Count; row++) + { + var elements = dataSet.Rows[row].Elements; + + // A short row is a malformed dataset, not a row of nulls: emitting the cells that ARE + // present would publish a table silently missing its right-hand columns. + if (elements.Count != dataSet.Columns.Count) + { + return false; + } + + for (var col = 0; col < dataSet.Columns.Count; col++) + { + var column = dataSet.Columns[col]; + if (string.IsNullOrEmpty(column)) + { + return false; // An unnamed column has no address. + } + + var (kind, value) = ProjectDataSetValue(elements[col]); + collected.Add(new SparkplugMember( + Path: $"{column}{MemberSeparator}{row.ToString(System.Globalization.CultureInfo.InvariantCulture)}", + DataType: (TahuDataType)dataSet.Types_[col], + ValueKind: kind, + Value: value)); + } + } + + members = collected; + return true; + } + + /// + /// Projects a metric's PropertySet into flat name/value metadata. + /// + /// The property set, or . + /// The projected properties; empty when there are none or the set is malformed. + /// + /// + /// Metadata, never tags — the recorded decision, and the honest reading of the + /// spec: a PropertySet describes a metric (engineering units, ranges, quality hints), it is + /// not a value the metric has. Surfacing it as sibling tags would invent plant signals out + /// of documentation. + /// + /// + /// Nested PropertySet/PropertySetList values are flattened one level with a + /// /-joined key, matching the metric-member convention; anything deeper, or an + /// extension value, is dropped rather than stringified. + /// + /// + public static IReadOnlyList ProjectProperties(Payload.Types.PropertySet? properties) + { + if (properties is null || properties.Keys.Count == 0 || properties.Keys.Count != properties.Values.Count) + { + return []; + } + + var collected = new List(); + AppendProperties(properties, prefix: "", depth: 1, collected); + return collected; + } + + private static void AppendProperties( + Payload.Types.PropertySet set, string prefix, int depth, List into) + { + if (set.Keys.Count != set.Values.Count || depth > 2 || into.Count >= MaxMembers) + { + return; + } + + for (var i = 0; i < set.Keys.Count && into.Count < MaxMembers; i++) + { + var key = set.Keys[i]; + if (string.IsNullOrEmpty(key)) + { + continue; + } + + var name = prefix.Length == 0 ? key : $"{prefix}{MemberSeparator}{key}"; + var value = set.Values[i]; + + if (value.HasIsNull && value.IsNull) + { + into.Add(new SparkplugProperty(name, null)); + continue; + } + + switch (value.ValueCase) + { + case Payload.Types.PropertyValue.ValueOneofCase.IntValue: + into.Add(new SparkplugProperty(name, value.IntValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.LongValue: + into.Add(new SparkplugProperty(name, value.LongValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.FloatValue: + into.Add(new SparkplugProperty(name, value.FloatValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.DoubleValue: + into.Add(new SparkplugProperty(name, value.DoubleValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.BooleanValue: + into.Add(new SparkplugProperty(name, value.BooleanValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.StringValue: + into.Add(new SparkplugProperty(name, value.StringValue)); + break; + case Payload.Types.PropertyValue.ValueOneofCase.PropertysetValue: + AppendProperties(value.PropertysetValue, name, depth + 1, into); + break; + default: + // PropertySetList and extension values are dropped rather than stringified — a + // property whose displayed value is not what its type says is worse than an absent + // one, and these are documentation, not signal. + break; + } + } + } + + /// Walks one template level, appending members under . + /// as soon as anything cannot be flattened honestly. + private static bool AppendTemplate( + Payload.Types.Template template, string prefix, int depth, List into) + { + if (depth > MaxDepth) + { + return false; + } + + // A template DEFINITION carries the UDT's shape with no instance values. It is metadata an + // edge node publishes so consumers can understand instances — flattening it would materialise + // a parallel set of tags that never carry a value. + if (template.HasIsDefinition && template.IsDefinition) + { + return false; + } + + foreach (var member in template.Metrics) + { + if (into.Count >= MaxMembers) + { + return false; + } + + if (!member.HasName || string.IsNullOrEmpty(member.Name)) + { + // A template member is addressed by name; there is no alias fallback inside a template. + return false; + } + + var path = prefix.Length == 0 ? member.Name : $"{prefix}{MemberSeparator}{member.Name}"; + + if (member.ValueCase == Payload.Types.Metric.ValueOneofCase.TemplateValue) + { + if (!AppendTemplate(member.TemplateValue, path, depth + 1, into)) + { + return false; + } + + continue; + } + + if (member.ValueCase == Payload.Types.Metric.ValueOneofCase.DatasetValue) + { + if (!TryFlattenDataSet(member.DatasetValue, out var cells) + || into.Count + cells.Count > MaxMembers) + { + return false; + } + + foreach (var cell in cells) + { + into.Add(cell with { Path = $"{path}{MemberSeparator}{cell.Path}" }); + } + + continue; + } + + var (kind, value) = SparkplugCodec.ProjectMemberValue(member); + if (kind == SparkplugValueKind.Unsupported) + { + return false; // An extension value inside a template: refuse the whole metric. + } + + into.Add(new SparkplugMember( + Path: path, + DataType: member.HasDatatype ? (TahuDataType)member.Datatype : null, + ValueKind: kind, + Value: value)); + } + + return true; + } + + private static (SparkplugValueKind Kind, object? Value) ProjectDataSetValue( + Payload.Types.DataSet.Types.DataSetValue value) => value.ValueCase switch + { + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.IntValue => (SparkplugValueKind.Scalar, value.IntValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.LongValue => (SparkplugValueKind.Scalar, value.LongValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.FloatValue => (SparkplugValueKind.Scalar, value.FloatValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.DoubleValue => (SparkplugValueKind.Scalar, value.DoubleValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.BooleanValue => (SparkplugValueKind.Scalar, value.BooleanValue), + Payload.Types.DataSet.Types.DataSetValue.ValueOneofCase.StringValue => (SparkplugValueKind.Scalar, value.StringValue), + + // A DataSetValue has no is_null field, so an empty oneof IS the null cell — unlike a Metric, + // where Absent and explicit-null are genuinely different states. + _ => (SparkplugValueKind.Null, null), + }; +} + +/// +/// One flattened member of a structured Sparkplug metric (Gitea #515). +/// +/// +/// The member path relative to the owning metric, /-joined — Speed for a +/// top-level template member, Motor/Speed for a nested one, Temp/0 for a dataset +/// cell. The consumer composes the full metric name as {parentName}/{Path}; it is kept +/// relative here because a DATA metric carries only the parent's alias, so the parent's name is +/// not known until the alias table has resolved it. +/// +/// +/// The member's declared datatype, or when it carried none — in which case +/// the consumer falls back to the datatype the birth installed for this member path. +/// +/// What means; never Unsupported here. +/// The raw wire value, with the same signed-integer caveat as a scalar metric. +public readonly record struct SparkplugMember( + string Path, + TahuDataType? DataType, + SparkplugValueKind ValueKind, + object? Value); + +/// One entry of a metric's PropertySet, projected as node metadata (Gitea #515). +/// The property key, /-joined for a nested set. +/// +/// The boxed scalar value, or for an explicitly-null property. +/// +public readonly record struct SparkplugProperty(string Name, object? Value); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs index 45d66b0b..e5dc9030 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/AliasTable.cs @@ -128,6 +128,36 @@ public sealed class AliasTable aliasCollisions++; (collidedAliases ??= []).Add(alias); } + + // Gitea #515: a Template/DataSet metric also installs one binding per flattened member, + // named `{parent}/{memberPath}`. The parent's own binding above is kept — a DATA metric + // carries only the parent's ALIAS, so resolving that alias to the parent's name is the only + // way the member names can be composed at all. But the parent is not itself servable (its + // Sparkplug datatype maps to no DriverDataType), so it is the member bindings that browse + // enumerates and that authored tags bind to. + // + // Members carry NO alias, deliberately: aliases are per-metric on the wire, so inventing + // one here would create a number the edge node never assigned, which a later birth could + // legitimately hand to a different metric. + foreach (var member in metric.MemberList) + { + if (string.IsNullOrWhiteSpace(member.Path)) + { + rejectedUnnamed++; + continue; + } + + var memberBinding = new SparkplugMetricBinding( + $"{binding.Name}{SparkplugFlattener.MemberSeparator}{member.Path}", + Alias: null, + member.DataType ?? TahuDataType.Unknown); + + if (!byName.TryAdd(memberBinding.Name, memberBinding)) + { + duplicateNames++; + byName[memberBinding.Name] = memberBinding; + } + } } if (collidedAliases is not null) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs index 91b0836d..156d356e 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt/Sparkplug/SparkplugIngestor.cs @@ -1177,6 +1177,15 @@ public sealed class SparkplugIngestor SparkplugMetric metric, DateTime? sourceUtc) { + // Gitea #515: a Template/DataSet has no value of its own — its flattened members do. Expand + // here, at the single funnel both the DATA and the birth-value paths already pass through, so + // neither can end up supporting structured metrics while the other silently does not. + if (metric.ValueKind == SparkplugValueKind.Structured) + { + PublishStructuredMembers(scope, binding, metric, sourceUtc); + return; + } + // Bound by NAME. The alias got us here; it is never the key. var key = new SparkplugMetricKey(scope.GroupId, scope.EdgeNodeId, scope.DeviceId, binding.Name); if (!_authored.ByMetric.TryGetValue(key, out var defs)) @@ -1287,6 +1296,62 @@ public sealed class SparkplugIngestor : BadTypeMismatch(def, binding, target, stamp, serverUtc); } + /// + /// Publishes each flattened member of a structured (Template / DataSet) metric as + /// the sibling metric {parent}/{memberPath} (Gitea #515). + /// + /// + /// + /// Each member is re-entered through as a scalar, rather + /// than duplicating the coercion, array-gate and authored-tag fan-out here. A second copy of + /// that logic would be free to drift — and the drift would be silent, since both paths + /// publish values that look plausible. + /// + /// + /// The member's datatype falls back to the BIRTH's. A DDATA template usually + /// re-declares its members' datatypes, but it is not obliged to; the binding installed at + /// birth is the authority when it does not. Guessing from the CLR type of the wire value + /// would type a signed Int32 of -42 as a UInt32 of 4294967254 — the exact + /// plausible-and-wrong failure ReinterpretSigned exists to prevent. + /// + /// + private void PublishStructuredMembers( + SparkplugScope scope, + SparkplugMetricBinding parent, + SparkplugMetric metric, + DateTime? sourceUtc) + { + var table = Births.Find(scope); + + foreach (var member in metric.MemberList) + { + if (string.IsNullOrWhiteSpace(member.Path)) + { + continue; + } + + var name = $"{parent.Name}{SparkplugFlattener.MemberSeparator}{member.Path}"; + + // Wire datatype first, birth binding second. Unknown last — which the typed layer refuses + // once, loudly, rather than coercing. + var dataType = member.DataType + ?? table?.ResolveByName(name)?.DataType + ?? Org.Eclipse.Tahu.Protobuf.DataType.Unknown; + + PublishMetric( + scope, + new SparkplugMetricBinding(name, Alias: null, dataType), + new SparkplugMetric( + Name: name, + Alias: null, + DataType: dataType, + TimestampMs: metric.TimestampMs, + ValueKind: member.ValueKind, + Value: member.Value), + sourceUtc); + } + } + /// /// The metric decoded, but does not fit the tag's effective type. Warned once per tag; the value /// itself is deliberately never logged — a payload value is plant data, not diagnostics. @@ -1370,9 +1435,7 @@ public sealed class SparkplugIngestor try { - subscribers(this, new SparkplugBirthObservedEventArgs( - scope, - [.. payload.Metrics.Where(m => m.Name is not null).Select(m => m.Name!)])); + subscribers(this, new SparkplugBirthObservedEventArgs(scope, [.. DiscoverableNames(payload)])); } catch (Exception ex) { @@ -1380,6 +1443,38 @@ public sealed class SparkplugIngestor } } + /// + /// The metric names a birth contributes to the discoverable address space — which is + /// what the rediscovery change gate must compare, not the raw wire names. + /// + /// + /// Gitea #515: a structured metric contributes its flattened MEMBERS and not itself. Comparing + /// raw wire names instead would make the gate blind to the change that matters most for a UDT: + /// a template gaining or losing a member leaves the parent metric's name identical, so the + /// signature would not move and the operator would never be told the tree changed. + /// + private static IEnumerable DiscoverableNames(SparkplugPayload payload) + { + foreach (var metric in payload.Metrics) + { + if (metric.Name is not { } name) + { + continue; + } + + if (metric.ValueKind != SparkplugValueKind.Structured) + { + yield return name; + continue; + } + + foreach (var member in metric.MemberList) + { + yield return $"{name}{SparkplugFlattener.MemberSeparator}{member.Path}"; + } + } + } + // ---- sequence + rebirth ---- private SequenceTracker TrackerFor(SparkplugNodeKey node) => diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs index 29037bfe..4943dee6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Drivers/DriverFactoryBootstrap.cs @@ -5,6 +5,7 @@ using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Core.Hosting; using ZB.MOM.WW.OtOpcUa.Core.Resilience; using ZB.MOM.WW.Secrets.Abstractions; +using ZB.MOM.WW.OtOpcUa.Runtime.Redundancy; namespace ZB.MOM.WW.OtOpcUa.Host.Drivers; @@ -45,6 +46,19 @@ public static class DriverFactoryBootstrap /// The same service collection, for chaining. public static IServiceCollection AddOtOpcUaDriverFactories(this IServiceCollection services) { + // Gitea #511: the Primary data-plane verdict the MQTT driver's Sparkplug host claim is gated on. + // + // Registered HERE as well as in AddOtOpcUaRuntime, and both sides TryAdd the same concrete + // RedundancyRoleView, so whichever extension runs first creates it and the other resolves the + // SAME instance. That is deliberately not a defensive duplicate: the resolve below is + // GetRequiredService (a null would silently fall the driver back to "claim unconditionally", + // which on a redundant pair is the dual-claim this mechanism exists to prevent), so without + // this line the registration becomes order-dependent on two extensions a caller is free to + // invoke in either order — and any graph that wires driver factories without the full runtime + // fails at resolution. Two Host integration suites do exactly that, and caught it. + services.TryAddSingleton(); + services.TryAddSingleton(sp => sp.GetRequiredService()); + services.AddSingleton(sp => { var registry = new DriverFactoryRegistry(); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs index db045291..87b81759 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/Program.cs @@ -81,7 +81,14 @@ foreach (var node in nodes) "Filler1", new SparkplugMetricSeed("Temperature", 10, DataType.Float, 55.5f), new SparkplugMetricSeed("FillCount", 11, DataType.Int64, 1000L), - new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false)); + new SparkplugMetricSeed("Jammed", 12, DataType.Boolean, false), + + // Structured metrics for the Gitea #515 gate. They live on the DEVICE rather than the node + // deliberately: a DDATA template is the case a naive implementation gets wrong, because its + // metric carries the parent's ALIAS and no name, so the member names cannot be composed until + // the alias table has resolved the parent. + new SparkplugMetricSeed("Motor", 13, DataType.Template, BuildMotorTemplate()), + new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, BuildRecipeDataSet())); edges.Add(edge); } @@ -112,3 +119,55 @@ return 0; static string Env(string name, string fallback) => Environment.GetEnvironmentVariable(name) is { Length: > 0 } value ? value : fallback; + +// A UDT instance: `Motor` with a `Speed` double, a `Running` bool and a nested `Stator` template. +// Flattens to `Motor/Speed`, `Motor/Running` and `Motor/Stator/Temp` (Gitea #515). +static Payload.Types.Template BuildMotorTemplate() +{ + var stator = new Payload.Types.Template { Version = "1", IsDefinition = false }; + stator.Metrics.Add(new Payload.Types.Metric + { + Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = 71.5, + }); + + var motor = new Payload.Types.Template { Version = "1", IsDefinition = false }; + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 1500, + }); + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true, + }); + + // Nested one level: proves the recursion, and proves the `/`-joined path composes correctly rather + // than only working for flat templates. + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator, + }); + + return motor; +} + +// A two-column, two-row table. Flattens to `Recipe/Step/0`, `Recipe/Target/0`, `Recipe/Step/1`, +// `Recipe/Target/1` — the row index is always present, so a table that grows a row never renames the +// tags already authored against it. +static Payload.Types.DataSet BuildRecipeDataSet() +{ + var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL }; + dataSet.Columns.Add("Step"); + dataSet.Columns.Add("Target"); + dataSet.Types_.Add((uint)DataType.Int32); + dataSet.Types_.Add((uint)DataType.Double); + + for (var row = 0; row < 2; row++) + { + var r = new Payload.Types.DataSet.Types.Row(); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row }); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = 20d + row }); + dataSet.Rows.Add(r); + } + + return dataSet; +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs index 0140af0d..425cafdc 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugSimulator/SparkplugEdgeNode.cs @@ -756,6 +756,18 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable case DataType.File: metric.BytesValue = ByteString.CopyFrom((byte[])value); break; + + // Structured values (Gitea #515). The seed carries the already-built protobuf message + // rather than a scalar, because a Template's shape IS its value — there is nothing simpler + // to convert from, and inventing a mini-DSL here would put a second, divergent notion of + // "what a template is" between the simulator and the wire it is supposed to reproduce. + case DataType.Template: + metric.TemplateValue = (Payload.Types.Template)value; + break; + case DataType.DataSet: + metric.DatasetValue = (Payload.Types.DataSet)value; + break; + default: throw new NotSupportedException( $"The simulator does not encode Sparkplug datatype {dataType}; v1 of the driver does " @@ -824,9 +836,83 @@ public sealed class SparkplugEdgeNode : IAsyncDisposable DataType.Uint8 or DataType.Uint16 or DataType.Uint32 or DataType.Uint64 => (uint)tick, DataType.Boolean => tick % 2 == 0, DataType.String or DataType.Text or DataType.Uuid => $"tick-{tick}", + + // Structured values move too, or the live gate could not tell a flattened member that is being + // fed from one that was merely published once at birth and then went quiet. + DataType.Template => AdvanceTemplate((Payload.Types.Template)seed.Value!, tick), + DataType.DataSet => AdvanceDataSet((Payload.Types.DataSet)seed.Value!, tick), + _ => seed.Value, }; + /// + /// Rebuilds a template with its numeric members nudged, preserving names, datatypes and order. + /// + private static Payload.Types.Template AdvanceTemplate(Payload.Types.Template seed, int tick) + { + var next = new Payload.Types.Template { Version = seed.Version, IsDefinition = false }; + foreach (var member in seed.Metrics) + { + var clone = member.Clone(); + switch ((DataType)clone.Datatype) + { + case DataType.Double: + clone.DoubleValue = 1500d + (tick % 100); + break; + case DataType.Float: + clone.FloatValue = 40f + (tick % 20) / 10f; + break; + case DataType.Int32: + clone.IntValue = unchecked((uint)(-tick)); + break; + case DataType.Boolean: + clone.BooleanValue = tick % 2 == 0; + break; + default: + break; // leave anything else at its seeded value + } + + clone.Timestamp = NowMs(); + next.Metrics.Add(clone); + } + + return next; + } + + /// Rebuilds a dataset with its numeric cells nudged, preserving columns, types and shape. + private static Payload.Types.DataSet AdvanceDataSet(Payload.Types.DataSet seed, int tick) + { + var next = new Payload.Types.DataSet { NumOfColumns = seed.NumOfColumns }; + next.Columns.AddRange(seed.Columns); + next.Types_.AddRange(seed.Types_); + + for (var row = 0; row < seed.Rows.Count; row++) + { + var cloned = new Payload.Types.DataSet.Types.Row(); + for (var col = 0; col < seed.Rows[row].Elements.Count; col++) + { + var cell = seed.Rows[row].Elements[col].Clone(); + switch ((DataType)seed.Types_[col]) + { + case DataType.Double: + cell.DoubleValue = 20d + row + (tick % 10) / 10d; + break; + case DataType.Int32: + cell.IntValue = (uint)(tick + row); + break; + default: + break; + } + + cloned.Elements.Add(cell); + } + + next.Rows.Add(cloned); + } + + return next; + } + private void ConfigureTls(MqttClientTlsOptionsBuilder tls) { if (!_options.UseTls) diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugStructuredLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugStructuredLiveTests.cs new file mode 100644 index 00000000..1ccb00c5 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests/SparkplugStructuredLiveTests.cs @@ -0,0 +1,332 @@ +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.SparkplugSimulator; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.IntegrationTests; + +/// +/// Live gate for structured Sparkplug metrics (Gitea #515): a Template and a +/// DataSet published by a real edge node over the Mosquitto fixture, flattened into +/// addressable OPC UA tags. +/// +/// +/// +/// What this adds over the offline suite. The offline tests build decoded payloads by +/// hand, so they prove the mapping but not the wire: that a nested template really does +/// round-trip through protobuf with its member names intact, and — the case a naive +/// implementation gets wrong — that a DDATA structured metric arrives carrying the parent's +/// alias and nothing else, so the member names cannot be composed until the alias table +/// has resolved the parent. +/// +/// +/// The issue's acceptance criterion, verbatim: "a simulator publishing a Template and a +/// DataSet metric produces addressable, correctly-typed OPC UA nodes, live-verified; +/// every still-unsupported kind continues to skip with a warning." +/// +/// +[Trait("Category", "Integration")] +[Collection("MqttFixture")] +public sealed class SparkplugStructuredLiveTests(MqttFixture fixture) +{ + private const uint Good = 0x00000000u; + private static readonly TimeSpan Timeout = TimeSpan.FromSeconds(20); + + private void SkipIfUnconfigured() => Assert.SkipWhen(fixture.SkipReason is not null, fixture.SkipReason ?? ""); + + [Fact] + public async Task A_template_metric_flattens_to_addressable_tags_across_birth_and_data() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("tmpl"); + const string node = "EdgeA"; + const string device = "Filler1"; + + const string speedPath = "Fixture/Sparkplug/Motor/Speed"; + const string runningPath = "Fixture/Sparkplug/Motor/Running"; + const string statorPath = "Fixture/Sparkplug/Motor/Stator/Temp"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Motor", 13, DataType.Template, MotorTemplate(speed: 1500, running: true, statorTemp: 71.5))); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + Tag(speedPath, group, node, device, "Motor/Speed"), + Tag(runningPath, group, node, device, "Motor/Running"), + Tag(statorPath, group, node, device, "Motor/Stator/Temp")), + "mqtt-live-sp-template"); + var recorder = new Recorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([speedPath, runningPath, statorPath], TimeSpan.FromSeconds(1), ct); + + // The DBIRTH is a full snapshot, so these are the birth's own member values — including the + // NESTED template's, which is what proves the `/`-joined path composes past one level. + var birth = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1500d), Timeout, ct); + birth.Snapshot.StatusCode.ShouldBe(Good); + await recorder.WaitAsync(runningPath, v => Equals(v.Value, true), Timeout, ct); + await recorder.WaitAsync(statorPath, v => Equals(v.Value, 71.5d), Timeout, ct); + + // …and now DDATA. The metric on the wire carries the PARENT's alias and no name at all, so a + // driver that flattened only at decode time — without resolving the parent alias to its name + // first — would have nothing to compose `Motor/Speed` from and would feed none of these. + await sim.PublishDeviceDataAsync( + device, [("Motor", MotorTemplate(speed: 1750, running: false, statorTemp: 68.25))], ct); + + var data = await recorder.WaitAsync(speedPath, v => Equals(v.Value, 1750d), Timeout, ct); + data.Snapshot.StatusCode.ShouldBe(Good); + data.FullReference.ShouldBe(speedPath, "values publish under the tag's RawPath, never the metric name"); + + await recorder.WaitAsync(runningPath, v => Equals(v.Value, false), Timeout, ct); + await recorder.WaitAsync(statorPath, v => Equals(v.Value, 68.25d), Timeout, ct); + + // The read path serves the same last values under the same keys. + var read = await driver.ReadAsync([speedPath, runningPath, statorPath], ct); + read.Select(r => r.StatusCode).ShouldAllBe(s => s == Good); + read[0].Value.ShouldBe(1750d); + } + + [Fact] + public async Task A_dataset_metric_flattens_to_one_tag_per_cell() + { + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("dset"); + const string node = "EdgeA"; + const string device = "Filler1"; + + const string step0Path = "Fixture/Sparkplug/Recipe/Step0"; + const string target1Path = "Fixture/Sparkplug/Recipe/Target1"; + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetDeviceMetrics(device, new SparkplugMetricSeed("Recipe", 14, DataType.DataSet, RecipeDataSet(baseTarget: 20d))); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + // The row index is always present, including on row 0 — the naming is uniform so a + // table that grows a row never renames the tags already authored against it. + Tag(step0Path, group, node, device, "Recipe/Step/0"), + Tag(target1Path, group, node, device, "Recipe/Target/1")), + "mqtt-live-sp-dataset"); + var recorder = new Recorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([step0Path, target1Path], TimeSpan.FromSeconds(1), ct); + + // Cells are typed per COLUMN: Step is Int32, Target is Double. A dataset whose columns and + // types disagree is refused outright, so a wrong type here is not a near-miss — it is the + // difference between a typed tag and no tag. + await recorder.WaitAsync(step0Path, v => Equals(v.Value, 0), Timeout, ct); + await recorder.WaitAsync(target1Path, v => Equals(v.Value, 21d), Timeout, ct); + + await sim.PublishDeviceDataAsync(device, [("Recipe", RecipeDataSet(baseTarget: 30d))], ct); + + var target = await recorder.WaitAsync(target1Path, v => Equals(v.Value, 31d), Timeout, ct); + target.Snapshot.StatusCode.ShouldBe(Good); + } + + [Fact] + public async Task A_template_definition_is_still_skipped_rather_than_flattened() + { + // Skip-and-warn survives — the requirement the issue is explicit about. A definition names its + // members and carries no values, so flattening it would materialise tags that can never read, + // indistinguishable in the picker from an instance's. + SkipIfUnconfigured(); + + var ct = TestContext.Current.CancellationToken; + var group = NewGroup("tdef"); + const string node = "EdgeA"; + const string device = "Filler1"; + + const string speedPath = "Fixture/Sparkplug/Proto/Speed"; + const string livePath = "Fixture/Sparkplug/Proto/Live"; + + var definition = new Payload.Types.Template { Version = "1", IsDefinition = true }; + definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)DataType.Double }); + + await using var sim = new SparkplugEdgeNode(SimOptions(group, node)); + sim.SetDeviceMetrics( + device, + new SparkplugMetricSeed("MotorType", 15, DataType.Template, definition), + + // The positive control, and it is load-bearing: without it "Speed never arrived" is + // satisfied by any failure to connect, birth or subscribe, and the test would pass against + // a driver that ingests nothing at all. + new SparkplugMetricSeed("Live", 16, DataType.Double, 42.5d)); + + await sim.ConnectAsync(ct); + await sim.BirthAsync(ct); + + await using var driver = new MqttDriver( + SparkplugOptions( + group, + Tag(speedPath, group, node, device, "MotorType/Speed"), + Tag(livePath, group, node, device, "Live")), + "mqtt-live-sp-tdef"); + var recorder = new Recorder(driver); + + await driver.InitializeAsync(string.Empty, ct); + await driver.SubscribeAsync([speedPath, livePath], TimeSpan.FromSeconds(1), ct); + + // Wait for the control FIRST, so the absence assertion below is made at a moment the birth is + // known to have been processed. + await recorder.WaitAsync(livePath, v => Equals(v.Value, 42.5d), Timeout, ct); + + recorder.Count(speedPath).ShouldBe(0, "a template definition carries no values to publish"); + + var read = await driver.ReadAsync([speedPath], ct); + read[0].StatusCode.ShouldNotBe(Good, "a tag bound to an unflattened member has never had a value"); + } + + // ---- helpers ----------------------------------------------------------------------------------- + + private static string NewGroup(string label) => $"OtOpcUaLive-{label}-{Guid.NewGuid():N}"[..40]; + + private static Payload.Types.Template MotorTemplate(double speed, bool running, double statorTemp) + { + var stator = new Payload.Types.Template { Version = "1", IsDefinition = false }; + stator.Metrics.Add(new Payload.Types.Metric + { + Name = "Temp", Datatype = (uint)DataType.Double, DoubleValue = statorTemp, + }); + + var motor = new Payload.Types.Template { Version = "1", IsDefinition = false }; + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = speed, + }); + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = running, + }); + motor.Metrics.Add(new Payload.Types.Metric + { + Name = "Stator", Datatype = (uint)DataType.Template, TemplateValue = stator, + }); + return motor; + } + + private static Payload.Types.DataSet RecipeDataSet(double baseTarget) + { + var dataSet = new Payload.Types.DataSet { NumOfColumns = 2UL }; + dataSet.Columns.Add("Step"); + dataSet.Columns.Add("Target"); + dataSet.Types_.Add((uint)DataType.Int32); + dataSet.Types_.Add((uint)DataType.Double); + + for (var row = 0; row < 2; row++) + { + var r = new Payload.Types.DataSet.Types.Row(); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { IntValue = (uint)row }); + r.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = baseTarget + row }); + dataSet.Rows.Add(r); + } + + return dataSet; + } + + private MqttDriverOptions SparkplugOptions(string groupId, params RawTagEntry[] rawTags) => new() + { + Host = fixture.TlsHost, + Port = fixture.TlsPort, + UseTls = true, + CaCertificatePath = fixture.CaCertificatePath, + AllowUntrustedServerCertificate = fixture.CaCertificatePath is null, + Username = fixture.Username, + Password = fixture.Password, + ClientId = $"otopcua-live-spx-{Guid.NewGuid():N}", + ConnectTimeoutSeconds = 15, + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions { GroupId = groupId, RequestRebirthOnGap = true }, + RawTags = rawTags, + }; + + private SparkplugEdgeNodeOptions SimOptions(string groupId, string edgeNodeId) => new() + { + Host = fixture.TlsHost, + Port = fixture.TlsPort, + UseTls = true, + CaCertificatePath = fixture.CaCertificatePath, + AllowUntrustedServerCertificate = fixture.CaCertificatePath is null, + Username = fixture.Username, + Password = fixture.Password, + GroupId = groupId, + EdgeNodeId = edgeNodeId, + }; + + private static RawTagEntry Tag( + string rawPath, string groupId, string edgeNodeId, string? deviceId, string metricName) + { + var device = deviceId is null ? "" : $"\"deviceId\":\"{deviceId}\","; + return new RawTagEntry( + rawPath, + $$"""{"groupId":"{{groupId}}","edgeNodeId":"{{edgeNodeId}}",{{device}}"metricName":"{{metricName}}"}""", + WriteIdempotent: false); + } + + /// Records every published data change, so a test can wait on one and count others. + private sealed class Recorder + { + private readonly List _events = []; + private readonly Lock _gate = new(); + + public Recorder(MqttDriver driver) => driver.OnDataChange += (_, e) => + { + lock (_gate) + { + _events.Add(e); + } + }; + + public int Count(string rawPath) + { + lock (_gate) + { + return _events.Count(e => e.FullReference == rawPath); + } + } + + public async Task WaitAsync( + string rawPath, Func predicate, TimeSpan timeout, CancellationToken ct) + { + var deadline = DateTime.UtcNow + timeout; + while (true) + { + lock (_gate) + { + foreach (var candidate in _events) + { + if (candidate.FullReference == rawPath && predicate(candidate.Snapshot)) + { + return candidate; + } + } + } + + if (DateTime.UtcNow >= deadline) + { + lock (_gate) + { + throw new TimeoutException( + $"No matching value for '{rawPath}' within {timeout}. Seen: " + + string.Join(", ", _events.Select(e => $"{e.FullReference}={e.Snapshot.Value}"))); + } + } + + await Task.Delay(100, ct); + } + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs index 39cffaa9..5427e985 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/MqttBrowseSessionTests.cs @@ -486,6 +486,31 @@ public sealed class MqttBrowseSessionTests s.PublishCountForTest.ShouldBe(0); // passive — browsing a live plant broker publishes nothing } + [Fact] + public async Task SparkplugTree_TemplateMetric_OffersItsMembersAndNotItself() + { + // Gitea #515. The parent has no value and no servable datatype, so a leaf for `Motor` would + // offer the operator an address that can only ever commit a tag which never reads — and the + // picker gives no hint that one leaf is real and the other is not. + await using var s = new MqttBrowseSession(MqttMode.SparkplugB); + s.ObserveTemplateBirthForTest( + "Plant1", "EdgeA", "Filler1", "Motor", + ("Speed", SparkplugDataType.Double), + ("Torque", SparkplugDataType.Double)); + + var metrics = await s.ExpandAsync("Plant1/EdgeA/Filler1", TestContext.Current.CancellationToken); + + metrics.Select(n => n.DisplayName).ShouldBe(["Motor/Speed", "Motor/Torque"]); + metrics.ShouldNotContain(n => n.DisplayName == "Motor"); + metrics[0].Kind.ShouldBe(BrowseNodeKind.Leaf); + + // The metric name travels INTACT as the node id — it is the binding key. Only the derived tag + // Name is later coerced to a legal RawPath segment, by RawBrowseCommitMapper (Gitea #509). + metrics[0].NodeId.ShouldBe("Plant1/EdgeA/Filler1::Motor/Speed"); + + s.PublishCountForTest.ShouldBe(0); + } + [Fact] public async Task SparkplugTree_NodeLevelMetrics_HangDirectlyUnderTheEdgeNode() { diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs index 77cbff89..aefaca6a 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugCodecTests.cs @@ -186,35 +186,82 @@ public sealed class SparkplugCodecTests } /// - /// DataSet and Template metrics are out of scope for v1. They must decode to an - /// explicit "unsupported" kind — never an exception on the MQTT dispatcher thread, and never a - /// silently-dropped metric that looks identical to a metric that was never published. + /// A DataSet metric decodes to with its cells + /// flattened (Gitea #515) — never an exception on the MQTT dispatcher thread, and never a + /// silently-dropped metric that looks identical to one that was never published. /// [Fact] - public void Decode_GoldenNbirth_DataSetMetric_IsUnsupported_NotAThrow() + public void Decode_GoldenNbirth_DataSetMetric_FlattensToCells() { var payload = SparkplugCodec.Decode(SparkplugGoldenPayloads.Read(SparkplugGoldenPayloads.NbirthFile)); var config = payload.Metrics.Single(m => m.Name == "Config"); - config.ValueKind.ShouldBe(SparkplugValueKind.Unsupported); + config.ValueKind.ShouldBe(SparkplugValueKind.Structured); + + // The PARENT carries no value — its members do. A parent that reported a value would be a + // guess about which cell "the" value is. config.Value.ShouldBeNull(); config.DataType.ShouldBe(DataType.DataSet); + + var cell = config.MemberList.Single(); + cell.Path.ShouldBe("col0/0"); + cell.DataType.ShouldBe(DataType.Int32); + cell.Value.ShouldBe(1U); } - /// A Template metric takes the same unsupported path as DataSet. + /// A Template metric flattens its members to sibling paths. [Fact] - public void Decode_TemplateMetric_IsUnsupported() + public void Decode_TemplateMetric_FlattensMembersToSiblingPaths() { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)DataType.Double, DoubleValue = 12.5, + }); + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)DataType.Boolean, BooleanValue = true, + }); + var payload = new Payload(); payload.Metrics.Add(new Payload.Types.Metric { Name = "Motor", Datatype = (uint)DataType.Template, - TemplateValue = new Payload.Types.Template { Version = "1", IsDefinition = false }, + TemplateValue = template, }); - SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single().ValueKind - .ShouldBe(SparkplugValueKind.Unsupported); + var metric = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single(); + + metric.ValueKind.ShouldBe(SparkplugValueKind.Structured); + metric.MemberList.Select(m => m.Path).ShouldBe(["Speed", "Running"]); + metric.MemberList[0].Value.ShouldBe(12.5); + metric.MemberList[1].Value.ShouldBe(true); + } + + /// + /// A template definition is refused rather than flattened, and refusal is still the + /// visible kind. + /// + [Fact] + public void Decode_TemplateDefinition_StaysUnsupported() + { + // A definition is the UDT's SHAPE, published so consumers can understand instances — its member + // metrics carry no values. Flattening it would materialise a parallel set of tags that can + // never read, indistinguishable in the picker from the real instance's. + var definition = new Payload.Types.Template { Version = "1", IsDefinition = true }; + definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)DataType.Double }); + + var payload = new Payload(); + payload.Metrics.Add(new Payload.Types.Metric + { + Name = "MotorType", Datatype = (uint)DataType.Template, TemplateValue = definition, + }); + + var metric = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single(); + + metric.ValueKind.ShouldBe(SparkplugValueKind.Unsupported); + metric.MemberList.ShouldBeEmpty(); } /// Payload and per-metric timestamps survive as raw Sparkplug epoch milliseconds. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugFlattenerTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugFlattenerTests.cs new file mode 100644 index 00000000..5b63c931 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugFlattenerTests.cs @@ -0,0 +1,294 @@ +using Google.Protobuf; +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// Unit tests for — the Template/DataSet/PropertySet mapping +/// (Gitea #515). +/// +/// +/// Two properties dominate this file. First, what is flattened is addressable and correctly +/// typed. Second — and the one that would be easy to lose in a later edit — +/// skip-and-warn survives: anything that cannot be flattened honestly stays +/// rather than becoming a guessed scalar, because a +/// tag bound to data that is not what its name says is worse than a tag that never binds. +/// +[Trait("Category", "Unit")] +public sealed class SparkplugFlattenerTests +{ + // ---- Template ------------------------------------------------------------------------------- + + [Fact] + public void A_nested_template_flattens_to_slash_joined_paths() + { + // `/` is the separator Sparkplug itself uses for metric-name hierarchy, so `Motor/Stator/Temp` + // is exactly what an edge node would have called a flat metric of the same meaning. + var stator = Template( + Member("Temp", TahuDataType.Double, m => m.DoubleValue = 71.5), + Member("Current", TahuDataType.Float, m => m.FloatValue = 3.25f)); + + var motor = Template( + Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1500), + MemberTemplate("Stator", stator)); + + SparkplugFlattener.TryFlattenTemplate(motor, out var members).ShouldBeTrue(); + + members.Select(m => m.Path).ShouldBe(["Speed", "Stator/Temp", "Stator/Current"]); + members[1].DataType.ShouldBe(TahuDataType.Double); + members[2].Value.ShouldBe(3.25f); + } + + [Fact] + public void A_template_member_with_no_name_refuses_the_whole_metric() + { + // A template member is addressed by name; there is no alias fallback inside a template. Emitting + // the members that DO have names would publish a UDT silently missing part of itself, which + // reads in the picker exactly like a complete one. + var template = Template( + Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1), + new Payload.Types.Metric { Datatype = (uint)TahuDataType.Double, DoubleValue = 2 }); + + SparkplugFlattener.TryFlattenTemplate(template, out var members).ShouldBeFalse(); + members.ShouldBeEmpty(); + } + + [Fact] + public void An_extension_value_inside_a_template_refuses_the_whole_metric() + { + // An extension value is a third-party escape hatch with no defined semantics. There is nothing + // honest to flatten it into, and the sibling members cannot be trusted to be the whole story. + var template = Template( + Member("Speed", TahuDataType.Double, m => m.DoubleValue = 1), + Member("Odd", TahuDataType.Unknown, m => m.ExtensionValue = new Payload.Types.Metric.Types.MetricValueExtension())); + + SparkplugFlattener.TryFlattenTemplate(template, out _).ShouldBeFalse(); + } + + [Fact] + public void A_template_definition_is_refused() + { + // A definition carries the UDT's shape with no instance values — metadata for consumers, not + // signal. Flattening it would materialise tags that can never read. + var definition = Template(Member("Speed", TahuDataType.Double, _ => { })); + definition.IsDefinition = true; + + SparkplugFlattener.TryFlattenTemplate(definition, out _).ShouldBeFalse(); + } + + [Fact] + public void Nesting_past_the_depth_bound_refuses_the_whole_metric_rather_than_truncating() + { + // This walks an object graph anyone on the broker can publish, so the recursion must be bounded. + // Refusing whole beats truncating: a silently half-flattened UDT looks exactly like a complete + // one, and the operator would author tags against a subtree that quietly stops partway down. + var deepest = Template(Member("Leaf", TahuDataType.Double, m => m.DoubleValue = 1)); + var current = deepest; + for (var i = 0; i < SparkplugFlattener.MaxDepth + 1; i++) + { + current = Template(MemberTemplate($"L{i}", current)); + } + + SparkplugFlattener.TryFlattenTemplate(current, out var members).ShouldBeFalse(); + members.ShouldBeEmpty(); + + // Falsifying control: the same shape one level inside the bound flattens fine, so the refusal + // above is the bound doing its job rather than nesting being broken outright. + var shallow = Template(MemberTemplate("L0", deepest)); + SparkplugFlattener.TryFlattenTemplate(shallow, out var ok).ShouldBeTrue(); + ok.Single().Path.ShouldBe("L0/Leaf"); + } + + // ---- DataSet --------------------------------------------------------------------------------- + + [Fact] + public void A_dataset_flattens_to_one_member_per_cell_with_the_row_index_always_present() + { + // Uniform naming, deliberately. Emitting a bare `{column}` for a single row and + // `{column}/{row}` otherwise would make a tag's own name depend on how many rows the plant + // happened to publish, so a table growing from one row to two would rename every existing tag + // out from under its authored binding. + var dataSet = DataSet( + columns: ["Temp", "Pressure"], + types: [TahuDataType.Double, TahuDataType.Int32], + rows: + [ + [Cell(v => v.DoubleValue = 20.5), Cell(v => v.IntValue = 3)], + [Cell(v => v.DoubleValue = 21.0), Cell(v => v.IntValue = 4)], + ]); + + SparkplugFlattener.TryFlattenDataSet(dataSet, out var members).ShouldBeTrue(); + + members.Select(m => m.Path).ShouldBe(["Temp/0", "Pressure/0", "Temp/1", "Pressure/1"]); + members[0].DataType.ShouldBe(TahuDataType.Double); + members[1].DataType.ShouldBe(TahuDataType.Int32); + members[3].Value.ShouldBe(4U); + } + + [Fact] + public void A_single_row_dataset_still_carries_the_row_index() + => Flatten(DataSet(["Temp"], [TahuDataType.Double], [[Cell(v => v.DoubleValue = 1)]])) + .Single().Path.ShouldBe("Temp/0"); + + [Fact] + public void An_empty_cell_is_a_null_member_not_a_zero() + { + // A DataSetValue has no is_null field, so an empty oneof IS the null cell. Reading it as the + // proto default would publish 0 for a cell the plant left blank — plausible, Good, and wrong. + var members = Flatten(DataSet(["Temp"], [TahuDataType.Double], [[new Payload.Types.DataSet.Types.DataSetValue()]])); + + members.Single().ValueKind.ShouldBe(SparkplugValueKind.Null); + members.Single().Value.ShouldBeNull(); + } + + [Fact] + public void A_column_and_type_count_mismatch_refuses_the_dataset() + // Columns and types are parallel arrays. Guessing which to trust would type a column by its + // neighbour, and every cell in it would publish at Good quality as the wrong type. + => SparkplugFlattener.TryFlattenDataSet( + DataSet(["Temp", "Pressure"], [TahuDataType.Double], [] ), out _) + .ShouldBeFalse(); + + [Fact] + public void A_short_row_refuses_the_dataset() + // Not a row of nulls: emitting the cells that ARE present publishes a table silently missing + // its right-hand columns. + => SparkplugFlattener.TryFlattenDataSet( + DataSet(["Temp", "Pressure"], [TahuDataType.Double, TahuDataType.Int32], + [[Cell(v => v.DoubleValue = 1)]]), + out _) + .ShouldBeFalse(); + + [Fact] + public void An_unnamed_column_refuses_the_dataset() + => SparkplugFlattener.TryFlattenDataSet( + DataSet(["", "Pressure"], [TahuDataType.Double, TahuDataType.Int32], + [[Cell(v => v.DoubleValue = 1), Cell(v => v.IntValue = 2)]]), + out _) + .ShouldBeFalse(); + + [Fact] + public void A_dataset_past_the_member_bound_refuses_rather_than_truncating() + { + var columns = Enumerable.Range(0, 8).Select(i => $"c{i}").ToArray(); + var types = Enumerable.Repeat(TahuDataType.Int32, 8).ToArray(); + var rows = Enumerable.Range(0, 100) + .Select(_ => Enumerable.Range(0, 8).Select(_ => Cell(v => v.IntValue = 1)).ToArray()) + .ToArray(); + + // 800 cells, past MaxMembers (512). + SparkplugFlattener.TryFlattenDataSet(DataSet(columns, types, rows), out var members).ShouldBeFalse(); + members.ShouldBeEmpty(); + } + + // ---- PropertySet ------------------------------------------------------------------------------ + + [Fact] + public void Properties_project_to_flat_metadata_including_one_nested_level() + { + var nested = new Payload.Types.PropertySet(); + nested.Keys.Add("Low"); + nested.Values.Add(new Payload.Types.PropertyValue { DoubleValue = 0 }); + + var set = new Payload.Types.PropertySet(); + set.Keys.Add("engUnit"); + set.Values.Add(new Payload.Types.PropertyValue { StringValue = "degC" }); + set.Keys.Add("Range"); + set.Values.Add(new Payload.Types.PropertyValue { PropertysetValue = nested }); + + var projected = SparkplugFlattener.ProjectProperties(set); + + projected.Select(p => p.Name).ShouldBe(["engUnit", "Range/Low"]); + projected[0].Value.ShouldBe("degC"); + } + + [Fact] + public void Properties_never_become_members() + { + // The recorded decision, and the honest reading of the spec: a PropertySet describes a metric + // (units, ranges, quality hints) — it is not a value the metric HAS. Surfacing it as sibling + // tags would invent plant signals out of documentation. + var set = new Payload.Types.PropertySet(); + set.Keys.Add("engUnit"); + set.Values.Add(new Payload.Types.PropertyValue { StringValue = "degC" }); + + var payload = new Payload(); + var metric = new Payload.Types.Metric + { + Name = "Temperature", Datatype = (uint)TahuDataType.Double, DoubleValue = 20.5, Properties = set, + }; + payload.Metrics.Add(metric); + + var decoded = SparkplugCodec.Decode(payload.ToByteArray()).Metrics.Single(); + + decoded.ValueKind.ShouldBe(SparkplugValueKind.Scalar); + decoded.MemberList.ShouldBeEmpty(); + decoded.PropertyList.Single().Name.ShouldBe("engUnit"); + } + + [Fact] + public void A_malformed_property_set_projects_to_nothing_rather_than_throwing() + { + var set = new Payload.Types.PropertySet(); + set.Keys.Add("orphan"); // no matching value + + SparkplugFlattener.ProjectProperties(set).ShouldBeEmpty(); + SparkplugFlattener.ProjectProperties(null).ShouldBeEmpty(); + } + + // ---- helpers ---------------------------------------------------------------------------------- + + private static IReadOnlyList Flatten(Payload.Types.DataSet dataSet) + { + SparkplugFlattener.TryFlattenDataSet(dataSet, out var members).ShouldBeTrue(); + return members; + } + + private static Payload.Types.Template Template(params Payload.Types.Metric[] members) + { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + template.Metrics.AddRange(members); + return template; + } + + private static Payload.Types.Metric Member( + string name, TahuDataType dataType, Action setValue) + { + var metric = new Payload.Types.Metric { Name = name, Datatype = (uint)dataType }; + setValue(metric); + return metric; + } + + private static Payload.Types.Metric MemberTemplate(string name, Payload.Types.Template value) => + new() { Name = name, Datatype = (uint)TahuDataType.Template, TemplateValue = value }; + + private static Payload.Types.DataSet.Types.DataSetValue Cell( + Action setValue) + { + var cell = new Payload.Types.DataSet.Types.DataSetValue(); + setValue(cell); + return cell; + } + + private static Payload.Types.DataSet DataSet( + string[] columns, + TahuDataType[] types, + Payload.Types.DataSet.Types.DataSetValue[][] rows) + { + var dataSet = new Payload.Types.DataSet { NumOfColumns = (ulong)columns.Length }; + dataSet.Columns.AddRange(columns); + dataSet.Types_.AddRange(types.Select(t => (uint)t)); + foreach (var cells in rows) + { + var row = new Payload.Types.DataSet.Types.Row(); + row.Elements.AddRange(cells); + dataSet.Rows.Add(row); + } + + return dataSet; + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs index 29b0544f..e052f7e4 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugGoldenPayloads.cs @@ -132,8 +132,9 @@ internal static class SparkplugGoldenPayloads IntValue = unchecked((uint)CountBirthValue), }); - // DataSet is explicitly out of scope for v1 — present here so "unsupported" is a decoded, - // testable outcome rather than an exception on the MQTT dispatcher thread. + // A single-column, single-row DataSet. Gitea #515 flattens it to the addressable member + // `Config/col0/0`; before that it decoded as an explicit "unsupported" kind. Kept in the + // golden vector either way — it is the only structured metric on the wire-format path. var dataSet = new Payload.Types.DataSet { NumOfColumns = 1UL }; dataSet.Columns.Add("col0"); dataSet.Types_.Add((uint)DataType.Int32); diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugStructuredIngestTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugStructuredIngestTests.cs new file mode 100644 index 00000000..d99be5df --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests/SparkplugStructuredIngestTests.cs @@ -0,0 +1,283 @@ +using Google.Protobuf; +using Org.Eclipse.Tahu.Protobuf; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Sparkplug; +using TahuDataType = Org.Eclipse.Tahu.Protobuf.DataType; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Mqtt.Tests; + +/// +/// End-to-end ingest of flattened structured metrics (Gitea #515): a tag authored against +/// Motor/Speed is fed by a DBIRTH/DDATA carrying a Template metric named +/// Motor. +/// +/// +/// pins the mapping in isolation; this file pins that the +/// chain actually connects — flatten → alias-table binding → authored-tag lookup → publish. Those +/// are four separate places a member name is composed, and a mismatch in any one of them produces +/// a driver that decodes structured metrics perfectly and feeds nothing. +/// +[Trait("Category", "Unit")] +public sealed class SparkplugStructuredIngestTests +{ + private const string Group = "Plant1"; + private const string Node = "EdgeA"; + private const string Device = "Filler1"; + private const ulong MotorAlias = 20UL; + private const ulong TableAlias = 21UL; + private const uint StatusGood = 0x00000000u; + + private const string SpeedPath = "Plant/Mqtt/spb/MotorSpeed"; + private const string TempCellPath = "Plant/Mqtt/spb/TempCell0"; + + [Fact] + public async Task A_template_member_feeds_the_tag_authored_against_its_flattened_name() + { + // The whole point of #515, asserted end to end. `Motor` is what the wire carries; `Motor/Speed` + // is what an operator authors, and nothing on the wire is ever called that. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); + + fired.ShouldContainKey(SpeedPath); + fired[SpeedPath].Value.ShouldBe(1500d); + fired[SpeedPath].StatusCode.ShouldBe(StatusGood); + } + + [Fact] + public async Task A_ddata_template_carrying_only_the_parents_alias_still_reaches_the_member_tag() + { + // The load-bearing case, and the one a naive implementation gets wrong. A DDATA metric carries + // an ALIAS and no name — so the member names cannot be composed until the alias table has + // resolved the parent's name. Flattening at decode time alone is not enough. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); + fired.Clear(); + + // seq 2, alias only — no Name on the metric at all. + var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL }; + data.Metrics.Add(new Payload.Types.Metric + { + Alias = MotorAlias, + TemplateValue = TemplateOf(speed: 1750, running: false), + }); + + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data)); + + fired.ShouldContainKey(SpeedPath); + fired[SpeedPath].Value.ShouldBe(1750d); + } + + [Fact] + public async Task A_dataset_cell_feeds_the_tag_authored_against_its_indexed_name() + { + var (ing, fired) = await NewIngestorAsync(Tag(TempCellPath, Blob("Table/Temp/0", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, TableDataSetMetric(20.5, 21.5))); + + fired.ShouldContainKey(TempCellPath); + fired[TempCellPath].Value.ShouldBe(20.5d); + } + + [Fact] + public async Task A_member_that_carries_no_datatype_falls_back_to_the_birth_binding() + { + // A DDATA template is not obliged to re-declare its members' datatypes. Guessing from the CLR + // type of the wire value would read a signed Int32 of -42 out of its unsigned field as + // 4294967254 — plausible, Good quality, and wrong. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Offset", "Int32"))); + + var birthTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false }; + birthTemplate.Metrics.Add(new Payload.Types.Metric + { + Name = "Offset", Datatype = (uint)TahuDataType.Int32, IntValue = unchecked((uint)-42), + }); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric + { + Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = birthTemplate, + })); + + fired[SpeedPath].Value.ShouldBe(-42); + fired.Clear(); + + // DDATA member with NO datatype of its own. + var dataTemplate = new Payload.Types.Template { Version = "1", IsDefinition = false }; + dataTemplate.Metrics.Add(new Payload.Types.Metric { Name = "Offset", IntValue = unchecked((uint)-7) }); + + var data = new Payload { Seq = 2UL, Timestamp = 1721822405000UL }; + data.Metrics.Add(new Payload.Types.Metric { Alias = MotorAlias, TemplateValue = dataTemplate }); + ing.Dispatch(Topic(SparkplugMessageType.DDATA, Device), Decode(data)); + + // -7, not 4294967289: the birth's Int32 binding drove ReinterpretSigned. + fired[SpeedPath].Value.ShouldBe(-7); + } + + [Fact] + public async Task A_tag_authored_against_the_structured_parent_itself_is_never_fed() + { + // The parent has no value — its members do. Feeding it anything would require choosing which + // member "the" value is, which is exactly the guess #515 refuses to make. + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1500, running: true))); + + fired.ShouldNotContainKey(SpeedPath); + } + + [Fact] + public async Task An_unflattenable_structured_metric_still_feeds_nothing_rather_than_a_guess() + { + // Skip-and-warn survives — the explicit requirement on the issue. A template DEFINITION is the + // realistic instance of this: it names members and carries no values. + var definition = new Payload.Types.Template { Version = "1", IsDefinition = true }; + definition.Metrics.Add(new Payload.Types.Metric { Name = "Speed", Datatype = (uint)TahuDataType.Double }); + + var (ing, fired) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, new Payload.Types.Metric + { + Name = "Motor", Alias = MotorAlias, Datatype = (uint)TahuDataType.Template, TemplateValue = definition, + })); + + fired.ShouldNotContainKey(SpeedPath); + } + + [Fact] + public async Task The_rediscovery_signature_tracks_members_not_the_parent_metric_name() + { + // A UDT gaining or losing a member leaves the PARENT metric's name identical, so a gate built + // on raw wire names would never move — and the operator would never be told the tree changed, + // which for a UDT is the change that matters most. + var (ing, _) = await NewIngestorAsync(Tag(SpeedPath, Blob("Motor/Speed", "Float64"))); + + var observed = new List>(); + ing.BirthObserved += (_, e) => observed.Add(e.MetricNames); + + ing.Dispatch(Topic(SparkplugMessageType.NBIRTH), Nbirth()); + ing.Dispatch(Topic(SparkplugMessageType.DBIRTH, Device), Birth(seq: 1, MotorTemplateMetric(speed: 1, running: true))); + + var deviceBirth = observed[^1]; + deviceBirth.ShouldBe(["Motor/Speed", "Motor/Running"], ignoreOrder: true); + deviceBirth.ShouldNotContain("Motor"); + } + + // ---- helpers --------------------------------------------------------------------------------- + + /// + /// A registered and subscribed ingestor. The subscribe step is not optional scaffolding: + /// SparkplugIngestor.Raise only fires OnDataChange for a RawPath that holds a + /// subscription handle, so an unsubscribed ingestor caches values silently and every assertion + /// in this file would read as "no value" whether or not the flattening worked. + /// + private static async Task<(SparkplugIngestor Ingestor, Dictionary Fired)> NewIngestorAsync( + params RawTagEntry[] tags) + { + var options = new MqttDriverOptions + { + Mode = MqttMode.SparkplugB, + Sparkplug = new MqttSparkplugOptions { GroupId = Group, RequestRebirthOnGap = false }, + RawTags = tags, + }; + + var ingestor = new SparkplugIngestor( + options, + "drv-1", + subscribeTransport: new GrantingSubscribeTransport(), + publishTransport: null, + logger: null, + cache: new LastValueCache()); + ingestor.Register(tags); + await ingestor.SubscribeAsync( + [.. ingestor.AuthoredRawPaths], TimeSpan.Zero, TestContext.Current.CancellationToken); + + var fired = new Dictionary(StringComparer.Ordinal); + ingestor.OnDataChange += (_, e) => fired[e.FullReference] = e.Snapshot; + return (ingestor, fired); + } + + /// A SUBACK that grants everything; the tests here are about values, not subscribe policy. + private sealed class GrantingSubscribeTransport : IMqttSubscribeTransport + { + public Task> SubscribeAsync( + IReadOnlyList filters, CancellationToken cancellationToken) + => Task.FromResult>( + [.. filters.Select(f => new MqttSubscribeOutcome(f.Topic, true, "GrantedQoS1"))]); + } + + private static RawTagEntry Tag(string rawPath, string tagConfig) => new(rawPath, tagConfig, WriteIdempotent: false); + + private static string Blob(string metric, string dataType) => + $$"""{"groupId":"{{Group}}","edgeNodeId":"{{Node}}","deviceId":"{{Device}}","metricName":"{{metric}}","dataType":"{{dataType}}"}"""; + + private static SparkplugTopic Topic(SparkplugMessageType type, string? device = null) => + new(type, Group, Node, device, null); + + private static SparkplugPayload Decode(Payload payload) => SparkplugCodec.Decode(payload.ToByteArray()); + + private static SparkplugPayload Nbirth() + { + var payload = new Payload { Seq = 0UL, Timestamp = 1721822400000UL }; + payload.Metrics.Add(new Payload.Types.Metric + { + Name = "bdSeq", Datatype = (uint)TahuDataType.Uint64, LongValue = 1UL, + }); + return Decode(payload); + } + + private static SparkplugPayload Birth(ulong seq, params Payload.Types.Metric[] metrics) + { + var payload = new Payload { Seq = seq, Timestamp = 1721822400000UL }; + payload.Metrics.AddRange(metrics); + return Decode(payload); + } + + private static Payload.Types.Template TemplateOf(double speed, bool running) + { + var template = new Payload.Types.Template { Version = "1", IsDefinition = false }; + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Speed", Datatype = (uint)TahuDataType.Double, DoubleValue = speed, + }); + template.Metrics.Add(new Payload.Types.Metric + { + Name = "Running", Datatype = (uint)TahuDataType.Boolean, BooleanValue = running, + }); + return template; + } + + private static Payload.Types.Metric MotorTemplateMetric(double speed, bool running) => new() + { + Name = "Motor", + Alias = MotorAlias, + Datatype = (uint)TahuDataType.Template, + TemplateValue = TemplateOf(speed, running), + }; + + private static Payload.Types.Metric TableDataSetMetric(params double[] temps) + { + var dataSet = new Payload.Types.DataSet { NumOfColumns = 1UL }; + dataSet.Columns.Add("Temp"); + dataSet.Types_.Add((uint)TahuDataType.Double); + foreach (var temp in temps) + { + var row = new Payload.Types.DataSet.Types.Row(); + row.Elements.Add(new Payload.Types.DataSet.Types.DataSetValue { DoubleValue = temp }); + dataSet.Rows.Add(row); + } + + return new Payload.Types.Metric + { + Name = "Table", Alias = TableAlias, Datatype = (uint)TahuDataType.DataSet, DatasetValue = dataSet, + }; + } +}