dohertj2/lmxopcua
1
0
Fork 0
Code Issues 2 Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(driver-calc): B2-WP7 Calculation driver — IDependencyConsumer + triggers + deploy cycle gate

Browse Source 
Adds the Calculation pseudo-driver (tags computed by C# scripts over other tags' live
values) plus the host seam that feeds it:

- IDependencyConsumer capability interface (Core.Abstractions) + DriverTypeNames.Calculation.
- CalculationDriver : IDriver, ISubscribable, IReadable, IDependencyConsumer — change-gate +
  dedupe (VirtualTagActor parity), timer-group scheduler, Bad-transition + script-log emission,
  Good recovery; reuses the T0-4 CalculationEvaluator.
- DriverHostActor spawns a DependencyConsumerMuxAdapter per dependency-consuming driver,
  registering its refs on the per-node DependencyMuxActor and forwarding DependencyValueChanged
  → OnDependencyValue; re-registers on every apply, torn down with the driver.
- Factory + probe registered in DriverFactoryBootstrap (keeps the T0-3 guard green); guard test
  csproj references the driver so the bin-scan discovers the factory.
- DeploymentArtifact injects the resolved scriptSource (scriptId → SourceCode) into a calc tag's
  delivered TagConfig so the host-blind driver can compile it.
- DraftValidator deploy gates: CalculationScriptMissing / CalculationScriptNotFound (scriptId
  existence) + CalculationDependencyCycle (DependencyGraph.DetectCycles over calc→calc edges).
- Tests: driver units (dep-ref extraction, change-gate, dedupe, Bad+script-log+recovery, timer,
  read), tag-config parsing, DraftValidator gates (self/2-cycle/cross-driver-terminal), and a
  Runtime integration test proving values FLOW mux → adapter → driver → calc-of-calc end-to-end.

Claude-Session: https://claude.ai/code/session_01LVneM3eh1UtJxEisFXgmox
This commit is contained in:
Joseph Doherty
2026-07-16 04:29:37 -04:00
parent ae9f906f52
commit 53d222e0f7
24 changed files with 1558 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Calculation/CalculationDriver.cs
+384
View File
@@ -0,0 +1,384 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Serilog;
using ZB.MOM.WW.OtOpcUa.Commons.Engines;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation;
/// <summary>
/// The <c>Calculation</c> pseudo-driver. Its tags are ordinary raw tags whose value is <b>computed</b>
/// by a C# script over other tags' live values — there is no backend, no endpoint, no discovery, no
/// writes. It implements:
/// <list type="bullet">
/// <item><see cref="IDriver"/> — lifecycle + always-Connected health.</item>
/// <item><see cref="IDependencyConsumer"/> — the host feeds it the other tags' live values.</item>
/// <item><see cref="ISubscribable"/> — computed values flow out via <see cref="OnDataChange"/>
/// exactly like any driver, so the ordinary node fan-out + the mux re-entry (calc-of-calc) apply.</item>
/// <item><see cref="IReadable"/> — on-demand read returns the last computed snapshot
/// (<c>BadWaitingForInitialData</c> before the first evaluation).</item>
/// </list>
///
/// <para>
/// <b>Triggers.</b> A tag re-evaluates on any of its declared dependencies changing (change-trigger,
/// gated exactly like <c>VirtualTagActor</c>: nothing publishes until every declared dep has arrived
/// at least once, and equal results are deduped) and/or on a timer (one timer per interval group).
/// </para>
/// <para>
/// <b>Error semantics (mini-design §6).</b> An evaluator <c>Failure</c> publishes <b>Bad</b> quality
/// carrying the last-known value, once per Good→Bad transition and only after all deps have arrived,
/// and emits a <c>ScriptLogEntry</c> on the <c>script-logs</c> topic (via the injected
/// <see cref="ScriptRootLogger"/>'s <c>ScriptLogTopicSink</c>). Recovery to Good is force-published.
/// A historized Bad records <c>BadInternalError</c>.
/// </para>
/// </summary>
public sealed class CalculationDriver : IDriver, IDependencyConsumer, ISubscribable, IReadable, IDisposable
{
// OPC UA status codes surfaced by the calc driver.
private const uint StatusGood = 0u;
private const uint StatusBadWaitingForInitialData = 0x80320000u;
private const uint StatusBadInternalError = 0x80020000u;
private readonly string _driverInstanceId;
private readonly CalculationDriverOptions _options;
private readonly ILogger<CalculationDriver> _logger;
private readonly ScriptRootLogger _scriptRoot;
private readonly CalculationEvaluator _evaluator;
// Authored calc tags, keyed by RawPath (identity + evaluation id). Built once at construction so
// DependencyRefs is available BEFORE InitializeAsync (the host reads it when spawning the mux-adapter).
private readonly Dictionary<string, CalculationTagDefinition> _tagsByRawPath = new(StringComparer.Ordinal);
// rawPath (a dependency) → the calc tags that read it AND are change-triggered. Drives OnDependencyValue.
private readonly Dictionary<string, List<string>> _changeDependents = new(StringComparer.Ordinal);
// Union of every tag's declared dependency refs — the interest set the host registers on the mux.
private readonly IReadOnlyCollection<string> _dependencyRefs;
// Live evaluation state. Guarded by _lock: OnDependencyValue (actor thread) and timer ticks
// (pool threads) can both drive an evaluation, so all state mutation is serialized.
private readonly object _lock = new();
private readonly Dictionary<string, object?> _depValues = new(StringComparer.Ordinal);
private readonly HashSet<string> _arrivedDeps = new(StringComparer.Ordinal);
private readonly Dictionary<string, TagState> _stateByRawPath = new(StringComparer.Ordinal);
private CalculationTimerScheduler? _timers;
private volatile bool _running;
private DateTime? _lastComputeUtc;
private bool _disposed;
/// <summary>Occurs when a calc tag's computed value changes (or transitions Good↔Bad).</summary>
public event EventHandler<DataChangeEventArgs>? OnDataChange;
/// <summary>Initializes a new <see cref="CalculationDriver"/>.</summary>
/// <param name="options">Bound options carrying the authored raw calc tags.</param>
/// <param name="driverInstanceId">Stable logical id of this driver instance.</param>
/// <param name="scriptRoot">Root script logger — user <c>ctx.Logger</c> output + failure entries fan out
/// to the <c>script-logs</c> topic through its <c>ScriptLogTopicSink</c>. When null a no-op logger is used.</param>
/// <param name="logger">Host-side diagnostics logger; defaults to the null logger.</param>
public CalculationDriver(
CalculationDriverOptions options,
string driverInstanceId,
ScriptRootLogger? scriptRoot = null,
ILogger<CalculationDriver>? logger = null)
{
ArgumentNullException.ThrowIfNull(options);
ArgumentException.ThrowIfNullOrWhiteSpace(driverInstanceId);
_options = options;
_driverInstanceId = driverInstanceId;
_logger = logger ?? NullLogger<CalculationDriver>.Instance;
_scriptRoot = scriptRoot ?? new ScriptRootLogger(new LoggerConfiguration().CreateLogger());
_evaluator = new CalculationEvaluator(
_logger is ILogger<CalculationEvaluator> el ? el : NullLogger<CalculationEvaluator>.Instance,
_scriptRoot,
_options.RunTimeout);
BuildTagTable();
_dependencyRefs = _tagsByRawPath.Values
.SelectMany(t => t.DependencyRefs)
.Distinct(StringComparer.Ordinal)
.ToArray();
}
private void BuildTagTable()
{
foreach (var entry in _options.RawTags)
{
if (!CalculationTagDefinition.TryFromTagConfig(entry.TagConfig, entry.RawPath, out var def))
{
_logger.LogWarning(
"Calculation {Driver}: raw tag {RawPath} has no scriptId; skipping (not a calc tag)",
_driverInstanceId, entry.RawPath);
continue;
}
if (string.IsNullOrWhiteSpace(def.ScriptSource))
{
_logger.LogWarning(
"Calculation {Driver}: calc tag {RawPath} (script {Script}) has no resolved scriptSource; " +
"it will not compute until the deploy artifact injects the script source",
_driverInstanceId, entry.RawPath, def.ScriptId);
}
_tagsByRawPath[entry.RawPath] = def;
_stateByRawPath[entry.RawPath] = new TagState();
if (def.ChangeTriggered)
{
foreach (var dep in def.DependencyRefs)
{
if (!_changeDependents.TryGetValue(dep, out var list))
_changeDependents[dep] = list = new List<string>();
if (!list.Contains(entry.RawPath, StringComparer.Ordinal)) list.Add(entry.RawPath);
}
}
}
}
// ---- IDriver ----
/// <inheritdoc />
public string DriverInstanceId => _driverInstanceId;
/// <inheritdoc />
public string DriverType => DriverTypeNames.Calculation;
/// <inheritdoc />
public Task InitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
_running = true;
StartTimers();
_logger.LogInformation(
"Calculation {Driver}: initialized with {TagCount} calc tag(s), {DepCount} distinct dependency ref(s)",
_driverInstanceId, _tagsByRawPath.Count, _dependencyRefs.Count);
return Task.CompletedTask;
}
/// <inheritdoc />
public Task ReinitializeAsync(string driverConfigJson, CancellationToken cancellationToken)
{
// The authored tag table is fixed at construction (a RawTags change respawns the driver, mirroring
// the fleet's factory-built-options pattern). Reinit drops compiled scripts (IScriptCacheOwner
// contract — stops collectible ALCs accreting) + re-arms timers so the calc set stays live.
StopTimers();
_evaluator.ClearCompiledScripts();
_running = true;
StartTimers();
return Task.CompletedTask;
}
/// <inheritdoc />
public Task ShutdownAsync(CancellationToken cancellationToken)
{
_running = false;
StopTimers();
return Task.CompletedTask;
}
/// <inheritdoc />
public DriverHealth GetHealth() => new(DriverState.Healthy, _lastComputeUtc, null);
/// <inheritdoc />
public long GetMemoryFootprint() => 0;
/// <inheritdoc />
public Task FlushOptionalCachesAsync(CancellationToken cancellationToken)
{
// The compile cache is the only optional cache; dropping it is safe (recompiled on next eval).
_evaluator.ClearCompiledScripts();
return Task.CompletedTask;
}
// ---- IDependencyConsumer ----
/// <inheritdoc />
public IReadOnlyCollection<string> DependencyRefs => _dependencyRefs;
/// <inheritdoc />
public void OnDependencyValue(string rawPath, object? value, uint statusCode, DateTime timestampUtc)
{
if (!_running) return;
List<string>? dependents;
lock (_lock)
{
_depValues[rawPath] = value;
_arrivedDeps.Add(rawPath);
if (!_changeDependents.TryGetValue(rawPath, out dependents) || dependents.Count == 0) return;
// Evaluate each change-triggered dependent inline under the lock (the compile cache makes this
// a bounded method invocation — the same "fast enough to run inline" contract as the VT evaluator).
foreach (var calcRawPath in dependents)
EvaluateTag(calcRawPath, timestampUtc);
}
}
// ---- ISubscribable ----
/// <inheritdoc />
public Task<ISubscriptionHandle> SubscribeAsync(
IReadOnlyList<string> fullReferences, TimeSpan publishingInterval, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var handle = new CalculationSubscriptionHandle($"calc-{Guid.NewGuid():N}");
// Initial-data callback (OPC UA convention): surface the current snapshot for each subscribed calc
// tag so a freshly-materialised node shows BadWaitingForInitialData until its first evaluation lands.
foreach (var reference in fullReferences)
{
var snapshot = ReadOne(reference);
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, reference, snapshot));
}
return Task.FromResult<ISubscriptionHandle>(handle);
}
/// <inheritdoc />
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
=> Task.CompletedTask;
// ---- IReadable ----
/// <inheritdoc />
public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(fullReferences);
var results = new DataValueSnapshot[fullReferences.Count];
for (var i = 0; i < fullReferences.Count; i++)
results[i] = ReadOne(fullReferences[i]);
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(results);
}
private DataValueSnapshot ReadOne(string reference)
{
var now = DateTime.UtcNow;
lock (_lock)
{
if (!_stateByRawPath.TryGetValue(reference, out var st) || !st.HasValue)
return new DataValueSnapshot(null, StatusBadWaitingForInitialData, null, now);
var status = st.LastPublishedBad ? StatusBadInternalError : StatusGood;
return new DataValueSnapshot(st.LastValue, status, st.LastTimestampUtc, st.LastTimestampUtc);
}
}
// ---- evaluation ----
/// <summary>Timer-trigger entry point — evaluates a single calc tag. Guards the lock itself (called
/// from a pool thread, unlike the change path which is already under the lock).</summary>
private void EvaluateTagLocked(string rawPath)
{
if (!_running) return;
lock (_lock) EvaluateTag(rawPath, DateTime.UtcNow);
}
/// <summary>Evaluate one calc tag and publish the outcome. MUST be called under <see cref="_lock"/>.</summary>
private void EvaluateTag(string rawPath, DateTime timestampUtc)
{
if (!_tagsByRawPath.TryGetValue(rawPath, out var def)) return;
if (string.IsNullOrWhiteSpace(def.ScriptSource)) return; // unresolved script — never computes (logged at init)
var st = _stateByRawPath[rawPath];
// Change-gate (mini-design §4): publish nothing until every declared dep has arrived at least once.
var allArrived = def.DependencyRefs.All(_arrivedDeps.Contains);
// Build the per-tag dependency snapshot (only its declared reads, current values).
var deps = new Dictionary<string, object?>(StringComparer.Ordinal);
foreach (var dep in def.DependencyRefs)
if (_depValues.TryGetValue(dep, out var v)) deps[dep] = v;
var result = _evaluator.Evaluate(rawPath, def.ScriptSource, deps);
if (!result.Success)
{
OnEvaluationFailed(def, st, result.Reason ?? "evaluator failure", allArrived, timestampUtc);
return;
}
if (!allArrived) return; // change-trigger warm-up: hold until inputs are ready
// Value dedup — but bypass it while recovering from Bad so a recovery whose value equals the
// pre-failure value still republishes Good (otherwise the node would stay Bad forever).
if (!st.LastPublishedBad && st.HasValue && Equals(st.LastValue, result.Value))
return;
st.HasValue = true;
st.LastValue = result.Value;
st.LastPublishedBad = false;
st.LastTimestampUtc = timestampUtc;
_lastComputeUtc = DateTime.UtcNow;
Publish(rawPath, result.Value, StatusGood, timestampUtc);
}
/// <summary>Failure tail (mirrors <c>VirtualTagActor</c> 02/S13): always emits the per-failure
/// script-log; additionally publishes a Bad snapshot — carrying the last-known value — once per
/// Good→Bad transition and only after every declared dependency has arrived.</summary>
private void OnEvaluationFailed(
CalculationTagDefinition def, TagState st, string reason, bool allArrived, DateTime timestampUtc)
{
PublishScriptLog(def, reason);
if (st.LastPublishedBad || !allArrived) return;
st.LastPublishedBad = true;
st.LastTimestampUtc = timestampUtc;
// Carry the last-known value (null if none) with Bad quality.
Publish(def.RawPath, st.HasValue ? st.LastValue : null, StatusBadInternalError, timestampUtc);
}
private void Publish(string rawPath, object? value, uint statusCode, DateTime timestampUtc)
{
var snapshot = new DataValueSnapshot(value, statusCode, timestampUtc, timestampUtc);
// Fired under _lock — the subscriber (DriverInstanceActor) marshals the event to its own thread,
// so there is no re-entrant call back into this driver.
OnDataChange?.Invoke(this, new DataChangeEventArgs(SharedHandle, rawPath, snapshot));
}
private void PublishScriptLog(CalculationTagDefinition def, string reason)
{
// Route through the script logger bound with ScriptId + the calc tag's RawPath (in the VirtualTagId
// slot the Script-log page attributes on). The ScriptRootLogger's ScriptLogTopicSink converts this
// Warning event into a ScriptLogEntry on the script-logs DPS topic — the driver needs no Akka access.
_scriptRoot.Logger
.ForContext(ScriptLoggerFactory.ScriptIdProperty, def.ScriptId)
.ForContext(ScriptLoggerFactory.VirtualTagIdProperty, def.RawPath)
.Warning("{Reason}", reason);
}
private void StartTimers()
{
var timed = _tagsByRawPath.Values.Where(t => t.TimerInterval is not null).ToArray();
if (timed.Length == 0) return;
_timers = new CalculationTimerScheduler(EvaluateTagLocked, _logger);
_timers.Start(timed);
}
private void StopTimers()
{
_timers?.Dispose();
_timers = null;
}
/// <summary>Test/diagnostic accessor: number of timer ticks skipped for in-flight groups.</summary>
internal long TimerSkippedTickCount => _timers?.SkippedTickCount ?? 0;
// A single shared handle for host-pushed publishes (change + timer). SubscribeAsync mints its own
// per-call handle for the initial-data callback; steady-state publishes reuse this so OnDataChange
// always carries a non-null handle.
private static readonly CalculationSubscriptionHandle SharedHandle = new("calc-shared");
/// <inheritdoc />
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_running = false;
StopTimers();
_evaluator.Dispose();
}
/// <summary>Per-tag published state (guarded by <see cref="_lock"/>).</summary>
private sealed class TagState
{
public bool HasValue;
public object? LastValue;
public bool LastPublishedBad;
public DateTime LastTimestampUtc;
}
private sealed record CalculationSubscriptionHandle(string DiagnosticId) : ISubscriptionHandle;
}