/* ========================================================================
* Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved.
*
* OPC Foundation MIT License 1.00
*
* Permission is hereby granted, free of charge, to any person
* obtaining a copy of this software and associated documentation
* files (the "Software"), to deal in the Software without
* restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the
* Software is furnished to do so, subject to the following
* conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
* HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
* WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
* OTHER DEALINGS IN THE SOFTWARE.
*
* The complete license agreement can be found here:
* http://opcfoundation.org/License/MIT/1.00/
* ======================================================================*/
using System;
using System.Threading;
using System.Threading.Tasks;
namespace Opc.Ua.Server
{
///
/// An object that manages aggregate factories supported by the server.
///
public class AggregateManager : IDisposable
{
///
/// Initilizes the manager.
///
public AggregateManager(IServerInternal server)
{
m_server = server;
m_factories = [];
m_minimumProcessingInterval = 1000;
}
///
/// Frees any unmanaged resources.
///
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
///
/// An overrideable version of the Dispose.
///
protected virtual void Dispose(bool disposing)
{
if (disposing)
{
// TBD
}
}
///
/// Checks if the aggregate is supported by the server.
///
/// The id of the aggregate function.
/// True if the aggregate is supported.
public bool IsSupported(NodeId aggregateId)
{
if (aggregateId.IsNull)
{
return false;
}
lock (m_lock)
{
return m_factories.ContainsKey(aggregateId);
}
}
///
/// The minimum processing interval for any aggregate calculation.
///
public double MinimumProcessingInterval
{
get
{
lock (m_lock)
{
return m_minimumProcessingInterval;
}
}
set
{
lock (m_lock)
{
m_minimumProcessingInterval = value;
}
}
}
///
/// Returns the default configuration for the specified variable id.
///
/// The id of history data node.
/// The configuration.
public AggregateConfiguration GetDefaultConfiguration(NodeId variableId)
{
lock (m_lock)
{
m_defaultConfiguration ??= new AggregateConfiguration
{
PercentDataBad = 100,
PercentDataGood = 100,
// Part 13 v1.05.07 ยง4.2.1.2: the TreatUncertainAsBad default is True.
TreatUncertainAsBad = true,
UseSlopedExtrapolation = false,
UseServerCapabilitiesDefaults = false
};
return m_defaultConfiguration;
}
}
///
/// Sets the default aggregate configuration.
///
/// The default aggregate configuration..
public void SetDefaultConfiguration(AggregateConfiguration configuration)
{
lock (m_lock)
{
m_defaultConfiguration = configuration;
}
}
///
/// Creates a new aggregate calculator.
///
/// The id of the aggregate function.
/// When to start processing.
/// When to stop processing.
/// The processing interval.
/// Whether stepped interpolation should be used.
/// The configuration to use.
public IAggregateCalculator? CreateCalculator(
NodeId aggregateId,
DateTimeUtc startTime,
DateTimeUtc endTime,
double processingInterval,
bool stepped,
AggregateConfiguration configuration)
{
if (aggregateId.IsNull)
{
return null;
}
AggregatorFactory? factory;
lock (m_lock)
{
if (!m_factories.TryGetValue(aggregateId, out factory))
{
return null;
}
}
if (configuration.UseServerCapabilitiesDefaults)
{
// ensure the configuration is initialized
configuration = GetDefaultConfiguration(default);
}
return factory!(
aggregateId,
startTime,
endTime,
processingInterval,
stepped,
configuration,
m_server.Telemetry);
}
///
/// Registers an aggregate factory.
///
/// The id of the aggregate function.
/// The id of the aggregate name.
/// The factory used to create calculators.
/// A cancellation token.
public async ValueTask RegisterFactoryAsync(
NodeId aggregateId,
string aggregateName,
AggregatorFactory factory,
CancellationToken cancellationToken = default)
{
lock (m_lock)
{
m_factories[aggregateId] = factory;
}
await m_server.DiagnosticsNodeManager.AddAggregateFunctionAsync(aggregateId, aggregateName, true, cancellationToken).ConfigureAwait(false);
}
///
/// Unregisters an aggregate factory.
///
/// The id of the aggregate function.
public void RegisterFactory(NodeId aggregateId)
{
lock (m_lock)
{
m_factories.Remove(aggregateId);
}
}
private readonly Lock m_lock = new();
private readonly IServerInternal m_server;
private AggregateConfiguration? m_defaultConfiguration;
private readonly NodeIdDictionary m_factories;
private double m_minimumProcessingInterval;
}
}