merge wt/wp7 into v3/batch2-raw-ui (Wave C — Calculation driver)

2026-07-16 04:30:27 -04:00
parent c07bbf7fd3 53d222e0f7
commit dc80a3b4f6
24 changed files with 1558 additions and 3 deletions
@@ -16,6 +16,7 @@ using TwinCATProbe   = Driver.TwinCAT.TwinCATDriverProbe;
 using FocasProbe     = Driver.FOCAS.FocasDriverProbe;
 using OpcUaProbe     = Driver.OpcUaClient.OpcUaClientDriverProbe;
 using GalaxyProbe    = Driver.Galaxy.GalaxyDriverProbe;
+using CalculationProbe = Driver.Calculation.CalculationDriverProbe;

 /// <summary>
 ///     Wires every cross-platform driver assembly's <c>Register(registry, loggerFactory)</c>
@@ -44,7 +45,10 @@ public static class DriverFactoryBootstrap
        {
            var registry = new DriverFactoryRegistry();
            var loggerFactory = sp.GetService<ILoggerFactory>();
-            Register(registry, loggerFactory);
+            // The calc driver needs the root script logger so a script failure fans out onto the
+            // script-logs topic; resolve it here (null on nodes without the script pipeline wired).
+            var scriptRoot = sp.GetService<ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger>();
+            Register(registry, loggerFactory, scriptRoot);
            return registry;
        });
        services.AddSingleton<IDriverFactory>(sp =>
@@ -113,6 +117,7 @@ public static class DriverFactoryBootstrap
        services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, FocasProbe>());
        services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, OpcUaProbe>());
        services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, GalaxyProbe>());
+        services.TryAddEnumerable(ServiceDescriptor.Singleton<IDriverProbe, CalculationProbe>());

        return services;
    }
@@ -123,10 +128,14 @@ public static class DriverFactoryBootstrap
    ///     handles platform/role-dependent stubbing (e.g. Galaxy on macOS), so registering a
    ///     factory here doesn't mean it always runs in production.
    /// </summary>
-    private static void Register(DriverFactoryRegistry registry, ILoggerFactory? loggerFactory)
+    private static void Register(
+        DriverFactoryRegistry registry,
+        ILoggerFactory? loggerFactory,
+        ZB.MOM.WW.OtOpcUa.Core.Scripting.ScriptRootLogger? scriptRoot = null)
    {
        Driver.AbCip.AbCipDriverFactoryExtensions.Register(registry);
        Driver.AbLegacy.AbLegacyDriverFactoryExtensions.Register(registry, loggerFactory);
+        Driver.Calculation.CalculationDriverFactoryExtensions.Register(registry, loggerFactory, scriptRoot);
        Driver.FOCAS.FocasDriverFactoryExtensions.Register(registry);
        Driver.Galaxy.GalaxyDriverFactoryExtensions.Register(registry, loggerFactory);
        Driver.Modbus.ModbusDriverFactoryExtensions.Register(registry, loggerFactory);
@@ -59,6 +59,7 @@
    <ItemGroup>
        <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.AbCip\ZB.MOM.WW.OtOpcUa.Driver.AbCip.csproj"/>
        <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.AbLegacy\ZB.MOM.WW.OtOpcUa.Driver.AbLegacy.csproj"/>
+        <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Calculation\ZB.MOM.WW.OtOpcUa.Driver.Calculation.csproj"/>
        <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.FOCAS\ZB.MOM.WW.OtOpcUa.Driver.FOCAS.csproj"/>
        <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Galaxy\ZB.MOM.WW.OtOpcUa.Driver.Galaxy.csproj"/>
        <ProjectReference Include="..\..\Drivers\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway\ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.csproj"/>
@@ -1,4 +1,5 @@
 using System.Text.Json;
+using System.Text.Json.Nodes;
 using ZB.MOM.WW.OtOpcUa.Commons.Types;
 using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
@@ -141,6 +142,13 @@ public static class DeploymentArtifact

        var resolver = new RawPathResolver(folders, drivers, devices, groups);

+        // scriptId → SourceCode, from the artifact's Scripts array — the same join the VirtualTag plan
+        // builder uses. A calc tag's TagConfig carries a scriptId (not the source); the calc driver is
+        // host-blind and can't resolve it, so we inject the resolved `scriptSource` into the delivered
+        // TagConfig here (the deploy-side counterpart of the Calculation driver). Tags with no scriptId
+        // are untouched.
+        var scriptSourceById = BuildScriptSourceMap(root);
+
        foreach (var el in EnumerateArray(root, "Tags"))
        {
            var deviceId = ReadString(el, "DeviceId");
@@ -152,6 +160,7 @@ public static class DeploymentArtifact
            if (rawPath is null) continue; // broken chain — dropped (deploy gate rejects invalid names)
            var tagConfig = el.TryGetProperty("TagConfig", out var tcEl) && tcEl.ValueKind == JsonValueKind.String
                ? tcEl.GetString() ?? "{}" : "{}";
+            tagConfig = InjectScriptSource(tagConfig, scriptSourceById);
            // WriteIdempotent (bool; non-bool/absent ⇒ false — the safe R2 default: writes are non-idempotent).
            var writeIdempotent = el.TryGetProperty("WriteIdempotent", out var wiEl)
                && wiEl.ValueKind is JsonValueKind.True or JsonValueKind.False && wiEl.GetBoolean();
@@ -168,6 +177,47 @@ public static class DeploymentArtifact
        return byDriver;
    }

+    /// <summary>Build the <c>scriptId → SourceCode</c> map from the artifact's <c>Scripts</c> array (mirrors
+    /// the VirtualTag plan builder's join). Empty when the artifact carries no Scripts.</summary>
+    private static Dictionary<string, string> BuildScriptSourceMap(JsonElement root)
+    {
+        var byId = new Dictionary<string, string>(StringComparer.Ordinal);
+        foreach (var el in EnumerateArray(root, "Scripts"))
+        {
+            var sid = ReadString(el, "ScriptId");
+            if (string.IsNullOrWhiteSpace(sid)) continue;
+            byId[sid!] = el.TryGetProperty("SourceCode", out var srcEl) && srcEl.ValueKind == JsonValueKind.String
+                ? srcEl.GetString() ?? string.Empty : string.Empty;
+        }
+        return byId;
+    }
+
+    /// <summary>
+    ///     When <paramref name="tagConfig"/> carries a <c>scriptId</c> (a calc tag), inject the resolved
+    ///     <c>scriptSource</c> so the host-blind Calculation driver can compile + evaluate it. Identity for
+    ///     tags with no <c>scriptId</c> (every other driver's tags), an unresolvable id, or malformed JSON —
+    ///     so no existing tag's blob changes. The pre-existing <c>scriptId</c>/trigger fields are preserved.
+    /// </summary>
+    /// <param name="tagConfig">The tag's raw <c>TagConfig</c> JSON.</param>
+    /// <param name="scriptSourceById">The <c>scriptId → SourceCode</c> map from the artifact.</param>
+    /// <returns>The TagConfig with <c>scriptSource</c> injected, or the original when nothing to inject.</returns>
+    private static string InjectScriptSource(string tagConfig, IReadOnlyDictionary<string, string> scriptSourceById)
+    {
+        if (scriptSourceById.Count == 0 || string.IsNullOrWhiteSpace(tagConfig)) return tagConfig;
+        JsonNode? node;
+        try { node = JsonNode.Parse(tagConfig); }
+        catch (JsonException) { return tagConfig; }
+        if (node is not JsonObject obj) return tagConfig;
+
+        var scriptId = obj.TryGetPropertyValue("scriptId", out var sidNode) && sidNode is JsonValue sv
+            && sv.TryGetValue<string>(out var sid) ? sid : null;
+        if (string.IsNullOrWhiteSpace(scriptId) || !scriptSourceById.TryGetValue(scriptId!, out var source))
+            return tagConfig;
+
+        obj["scriptSource"] = source;
+        return obj.ToJsonString();
+    }
+
    /// <summary>Group the artifact's Devices into per-driver <see cref="DriverDeviceConfigMerger.DeviceRow"/>
    /// lists (entity Name + schemaless DeviceConfig), ordered by DeviceId for a stable merge.</summary>
    /// <param name="root">The artifact root element.</param>
@@ -98,6 +98,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers

    private readonly Dictionary<string, ChildEntry> _children = new(StringComparer.Ordinal);

+    /// <summary>Per-driver <see cref="DependencyConsumerMuxAdapter"/> children — one for each spawned driver
+    /// that implements <see cref="IDependencyConsumer"/> (e.g. the Calculation pseudo-driver). Each registers
+    /// the driver's declared dependency refs on the per-node <see cref="_dependencyMux"/> and forwards value
+    /// changes into the driver. Spawned in <see cref="SpawnChild"/>, torn down with the driver in
+    /// <see cref="StopDependencyAdapter"/>, and re-registered on every apply (refs change when tags/scripts
+    /// change). Empty when no mux is wired (the dev/None path) or no dependency-consuming driver is hosted.</summary>
+    private readonly Dictionary<string, IActorRef> _depConsumerAdapters = new(StringComparer.Ordinal);
+
    // Monotonic counter feeding the child actor-name suffix (see ActorNameFor / SpawnChild). Single-
    // threaded actor, so a plain increment is safe; it only ever grows, guaranteeing a unique name per
    // spawn so a restart's respawn never collides with the still-terminating old child.
@@ -1372,6 +1380,12 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
        foreach (var id in plan.ToStop) StopChild(id);
        foreach (var spec in plan.ToApplyDelta) ApplyChildDelta(spec);
        foreach (var spec in plan.ToSpawn) SpawnChild(spec);
+
+        // v3 (WP7): re-register every dependency-consumer adapter's interest on each apply — a driver's
+        // declared dependency refs change when its tags/scripts change. Newly-spawned adapters already
+        // registered in PreStart; this refreshes the ones an ApplyDelta kept in place. A Tell (no blocking).
+        foreach (var adapter in _depConsumerAdapters.Values)
+            adapter.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags.DependencyConsumerMuxAdapter.ReRegister());
    }

    /// <summary>
@@ -1736,10 +1750,33 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
        }

        _children[spec.DriverInstanceId] = new ChildEntry(child, spec, stub);
+
+        // v3 (WP7): if the (real, non-stub) driver consumes other tags' live values, spawn a mux-adapter
+        // child that registers its dependency refs on the per-node dependency mux and forwards changes
+        // into the driver. Requires a wired mux (dev/None deployments have none — then calc inputs simply
+        // can't flow, matching the VirtualTag dev path).
+        if (!stub && driver is IDependencyConsumer consumer && _dependencyMux is not null)
+        {
+            var adapter = Context.ActorOf(
+                ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags.DependencyConsumerMuxAdapter.Props(consumer, _dependencyMux),
+                "depmux-" + actorName);
+            _depConsumerAdapters[spec.DriverInstanceId] = adapter;
+            _log.Info("DriverHost {Node}: wired dependency-consumer mux adapter for {Id} ({Count} ref(s))",
+                _localNode, spec.DriverInstanceId, consumer.DependencyRefs.Count);
+        }
+
        _log.Info("DriverHost {Node}: spawned {Type} driver {Id} (stub={Stub})",
            _localNode, spec.DriverType, spec.DriverInstanceId, stub);
    }

+    /// <summary>Stops + forgets the dependency-consumer mux adapter for a driver (if any). Called wherever a
+    /// driver child is stopped/respawned so a stale adapter never keeps a dead driver's refs registered.</summary>
+    private void StopDependencyAdapter(string driverInstanceId)
+    {
+        if (_depConsumerAdapters.Remove(driverInstanceId, out var adapter))
+            Context.Stop(adapter);
+    }
+
    private void ApplyChildDelta(DriverInstanceSpec spec)
    {
        if (!_children.TryGetValue(spec.DriverInstanceId, out var entry)) return;
@@ -1752,6 +1789,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
    private void StopChild(string driverInstanceId)
    {
        if (!_children.TryGetValue(driverInstanceId, out var entry)) return;
+        StopDependencyAdapter(driverInstanceId);
        Context.Stop(entry.Actor);
        _children.Remove(driverInstanceId);
        _log.Info("DriverHost {Node}: stopped driver child {Id}", _localNode, driverInstanceId);
@@ -1805,6 +1843,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers
            _localNode, msg.DriverInstanceId, msg.ActorByUserName);

        // Stop the existing child actor — DriverInstanceActor.PostStop calls ShutdownAsync.
+        StopDependencyAdapter(msg.DriverInstanceId);
        Context.Stop(entry.Actor);
        _children.Remove(msg.DriverInstanceId);

@@ -0,0 +1,71 @@
+using Akka.Actor;
+using Akka.Event;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.Runtime.VirtualTags;
+
+/// <summary>
+///     Bridges a driver that implements <see cref="IDependencyConsumer"/> onto the per-node
+///     <see cref="DependencyMuxActor"/>. Spawned as a child of <c>DriverHostActor</c> for each such
+///     driver: on start (and on every apply via <see cref="ReRegister"/>) it registers the driver's
+///     declared <see cref="IDependencyConsumer.DependencyRefs"/> with the mux, and forwards each
+///     <see cref="VirtualTagActor.DependencyValueChanged"/> the mux fans out into
+///     <see cref="IDependencyConsumer.OnDependencyValue"/>.
+///
+///     <para>
+///         This is the exact seam a <c>VirtualTagActor</c> uses to receive its inputs, generalised so a
+///         host-blind driver (the <c>Calculation</c> pseudo-driver) can read other tags' live values with
+///         no cross-driver plumbing. The mux fans out value + timestamp only (quality is stripped at the
+///         mux boundary, same as for virtual tags), so the forwarded status code is <c>Good</c> (0).
+///     </para>
+/// </summary>
+public sealed class DependencyConsumerMuxAdapter : ReceiveActor
+{
+    /// <summary>Sent by the host after an apply so the adapter re-reads the driver's (possibly changed)
+    /// dependency refs and re-registers them with the mux.</summary>
+    public sealed record ReRegister;
+
+    private readonly IDependencyConsumer _consumer;
+    private readonly IActorRef _mux;
+    private readonly ILoggingAdapter _log = Context.GetLogger();
+
+    /// <summary>Creates props for a <see cref="DependencyConsumerMuxAdapter"/>.</summary>
+    /// <param name="consumer">The dependency-consuming driver to feed.</param>
+    /// <param name="mux">The per-node dependency mux to register interest with.</param>
+    /// <returns>Props for the adapter.</returns>
+    public static Props Props(IDependencyConsumer consumer, IActorRef mux) =>
+        Akka.Actor.Props.Create(() => new DependencyConsumerMuxAdapter(consumer, mux));
+
+    /// <summary>Initializes a new <see cref="DependencyConsumerMuxAdapter"/>.</summary>
+    /// <param name="consumer">The dependency-consuming driver to feed.</param>
+    /// <param name="mux">The per-node dependency mux to register interest with.</param>
+    public DependencyConsumerMuxAdapter(IDependencyConsumer consumer, IActorRef mux)
+    {
+        _consumer = consumer ?? throw new ArgumentNullException(nameof(consumer));
+        _mux = mux ?? throw new ArgumentNullException(nameof(mux));
+
+        Receive<VirtualTagActor.DependencyValueChanged>(OnDependencyValueChanged);
+        Receive<ReRegister>(_ => RegisterInterest());
+    }
+
+    /// <inheritdoc />
+    protected override void PreStart() => RegisterInterest();
+
+    /// <inheritdoc />
+    protected override void PostStop() => _mux.Tell(new DependencyMuxActor.UnregisterInterest(Self));
+
+    private void RegisterInterest()
+    {
+        var refs = _consumer.DependencyRefs.ToList();
+        // A RegisterInterest replaces the prior interest set for this subscriber (mux is idempotent on
+        // re-register), so re-registering after an apply cleanly adopts an added/removed ref.
+        _mux.Tell(new DependencyMuxActor.RegisterInterest(refs, Self));
+        _log.Debug("DependencyConsumerMuxAdapter: registered {Count} dependency ref(s)", refs.Count);
+    }
+
+    private void OnDependencyValueChanged(VirtualTagActor.DependencyValueChanged msg)
+    {
+        // Quality is stripped at the mux boundary (value + timestamp only), so forward Good (0).
+        _consumer.OnDependencyValue(msg.TagId, msg.Value, statusCode: 0u, msg.TimestampUtc);
+    }
+}