//******************************************************************************************************//  PIOutputAdapter.cs - Gbtc////  Copyright © 2010, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may//  not use this file except in compliance with the License. You may obtain a copy of the License at:////      http://www.opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  08/13/2012 - Ryan McCoy//       Generated original version of source code.//  06/18/2014 - J. Ritchie Carroll//       Optimized PI-SDK code performance with connection pooling//  12/17/2014 - J. Ritchie Carroll//       Updated code to use AF-SDK - using single PIConnection for now//  03/16/2021 - J. Ritchie Carroll//       Updated to add automated tag-removal operations during metadata synchronization.////****************************************************************************************************** using GSF;using GSF.Collections;using GSF.Data;using GSF.Diagnostics;using GSF.IO;using GSF.Threading;using GSF.TimeSeries;using GSF.TimeSeries.Adapters;using OSIsoft.AF;using OSIsoft.AF.Asset;using OSIsoft.AF.Data;using OSIsoft.AF.PI;using OSIsoft.AF.Time;using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.ComponentModel;using System.Data;using System.IO;using System.Linq;using System.Text;using System.Threading; // ReSharper disable InconsistentlySynchronizedFieldnamespace PIAdapters{    #region [ Enumerations ]     /// <summary>    /// Enumeration that defines the available operations for PI tag removal during metadata synchronization.    /// </summary>    public enum TagRemovalOperation    {        /// <summary>        /// Do not remove any existing PI tags. This is the default operation.        /// </summary>        [Description("Do not remove any existing PI tags. This is the default operation.")]        DoNotRemove,        /// <summary>        /// Remove any PI tags where the PI point source matches the setting defined in the PI output adapter and        /// the PI tag no longer exists in the local metadata. The is the recommended option when it is desirable        /// that any locally defined tags that get removed also get removed from PI.        /// </summary>        /// <remarks>        /// This is the least aggressive tag removal synchronization operation. Only PI tags that have a matching        /// point source name will be considered for removal. After point source match, then any PI tags where        /// existing measurement signal ID is not found in any extended descriptors will be removed. Note that        /// removal of any point tags in PI metadata will cause history for removed tags to be inaccessible.        /// </remarks>        [Description("Remove any PI tags where the PI point source matches the setting defined in the PI output adapter and the PI tag no longer exists in the local metadata. The is the recommended option when it is desirable that any locally defined tags that get removed also get removed from PI.\r\nThis is the least aggressive tag removal synchronization operation. Only PI tags that have a matching point source name will be considered for removal. After point source match, then any PI tags where existing measurement signal ID is not found in any extended descriptors will be removed. Note that removal of any point tags in PI metadata will cause history for removed tags to be inaccessible.")]        LocalOnly,        /// <summary>        /// Remove any PI tags where existing measurement signal ID is not found in any extended descriptors.        /// Use with caution as metadata in PI will be updated to exactly match local metadata.        /// </summary>        /// <remarks>        /// WARNING: Do not use this option if target PI instance is used for storage of any data other than this        /// local PI adapter. This is the most aggressive tag removal synchronization operation. Any PI tags that        /// do not exist in local metadata will be removed from PI. This option ensures that PI metadata will        /// exactly match local metadata. Note that removal of any point tags in PI metadata will cause history        /// for removed tags to be inaccessible.        /// </remarks>        [Description("Remove any PI tags where existing measurement signal ID is not found in any extended descriptors. Use with caution as metadata in PI will be updated to exactly match local metadata.\r\nWARNING: Do not use this option if target PI instance is used for storage of any data other than this local PI adapter. This is the most aggressive tag removal synchronization operation. Any PI tags that do not exist in local metadata will be removed from PI. This option ensures that PI metadata will exactly match local metadata. Note that removal of any point tags in PI metadata will cause history for removed tags to be inaccessible.")]        FullClone    }     #endregion     /// <summary>    /// Exports measurements to PI if the point tag or alternate tag corresponds to a PI point's tag name.    /// </summary>    [Description("OSI-PI: Archives measurements to an OSI-PI server using AF-SDK")]    public class PIOutputAdapter : OutputAdapterBase    {        #region [ Members ]         // Constants        private const string DefaultUserName = "";        private const string DefaultPassword = "";        private const bool DefaultRunMetadataSync = true;        private const bool DefaultAutoCreateTags = true;        private const bool DefaultAutoUpdateTags = true;        private const string DefaultAutoRemoveTags = nameof(TagRemovalOperation.DoNotRemove);        private const bool DefaultSyncAlternateTagOnly = false;        private const bool DefaultSkipDigitalAlternateTagSync = true;        private const int DefaultTagNamePrefixRemoveCount = 0;        private const string DefaultPIPointSource = nameof(GSF);        private const string DefaultPIPointClass = "classic";        private const bool DefaultUseCompression = true;        private const bool DefaultUpdateExistingTagCompressionState = false;        private const string DefaultTagMapCacheFileName = "";        private const double DefaultMaximumPointResolution = 0.0D;        private const bool DefaultEnableTimeReasonabilityCheck = false;        private const double DefaultPastTimeReasonabilityLimit = 43200.0D;        private const double DefaultFutureTimeReasonabilityLimit = 43200.0D;         // Fields         // Define cached mapping between GSFSchema measurements and PI points        private readonly ConcurrentDictionary<MeasurementKey, PIPoint> m_mappedPIPoints;         private readonly ProcessQueue<AFValue>[] m_archiveQueues;           // Collection of point archival queues        private readonly ProcessQueue<MeasurementKey> m_mapRequestQueue;    // Requested measurement to PI point mapping queue        private readonly ShortSynchronizedOperation m_restartConnection;    // Restart connection operation        private readonly ConcurrentDictionary<Guid, string> m_tagMap;       // Tag name to measurement Guid lookup        private readonly HashSet<MeasurementKey> m_pendingMappings;         // List of pending measurement mappings        private readonly LongSynchronizedOperation m_handleTagRemoval;      // Tag removal operation, if any        private Dictionary<Guid, Ticks> m_lastArchiveTimes;                 // Cache of last point archive times        private PIConnection m_connection;                                  // PI server connection for meta-data synchronization        private long m_pastTimeReasonabilityLimit;                          // Past-timestamp reasonability limit        private long m_futureTimeReasonabilityLimit;                        // Future-timestamp reasonability limit        private DateTime m_lastMetadataRefresh;                             // Tracks time of last meta-data refresh        private long m_processedMappings;                                   // Total number of mappings processed so far        private long m_processedMeasurements;                               // Total number of measurements processed so far        private long m_totalProcessingTime;                                 // Total point processing time         private volatile bool m_refreshingMetadata;                         // Flag that determines if meta-data is currently refreshing        private double m_metadataRefreshProgress;                           // Current meta-data refresh progress        private bool m_disposed;                                            // Flag that determines if class is disposed         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new <see cref="PIOutputAdapter"/>        /// </summary>        public PIOutputAdapter()        {            m_mappedPIPoints = new ConcurrentDictionary<MeasurementKey, PIPoint>();            m_archiveQueues = new ProcessQueue<AFValue>[Environment.ProcessorCount];             for (int i = 0; i < m_archiveQueues.Length; i++)                m_archiveQueues[i] = ProcessQueue<AFValue>.CreateRealTimeQueue(ArchiveAFValues, Timeout.Infinite, false, false);             m_mapRequestQueue = ProcessQueue<MeasurementKey>.CreateAsynchronousQueue(EstablishPIPointMappings, Environment.ProcessorCount);            m_restartConnection = new ShortSynchronizedOperation(Start);            m_tagMap = new ConcurrentDictionary<Guid, string>();            m_pendingMappings = new HashSet<MeasurementKey>();            m_handleTagRemoval = new LongSynchronizedOperation(HandleTagRemoval, ex => OnProcessException(MessageLevel.Error, ex)) { IsBackground = true };            m_lastMetadataRefresh = DateTime.MinValue;        }         #endregion         #region [ Properties ]         /// <summary>        /// Returns true to indicate that this <see cref="PIOutputAdapter"/> is sending measurements to a historian, OSIsoft PI.        /// </summary>        public override bool OutputIsForArchive => true;         /// <summary>        /// Returns false to indicate that this <see cref="PIOutputAdapter"/> will connect synchronously        /// </summary>        protected override bool UseAsyncConnect => false;         /// <summary>        /// Gets or sets the name of the PI server for the adapter's PI connection.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the name of the PI server for the adapter's PI connection.")]        public string ServerName { get; set; }         /// <summary>        /// Gets or sets the name of the PI user ID for the adapter's PI connection.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the name of the PI user ID for the adapter's PI connection.")]        [DefaultValue(DefaultUserName)]        public string UserName { get; set; } = DefaultUserName;         /// <summary>        /// Gets or sets the password used for the adapter's PI connection.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the password used for the adapter's PI connection.")]        [DefaultValue(DefaultPassword)]        public string Password { get; set; } = DefaultPassword;         /// <summary>        /// Gets or sets the timeout interval (in milliseconds) for the adapter's connection        /// </summary>        [ConnectionStringParameter]        [Description("Defines the timeout interval (in milliseconds) for the adapter's connection")]        [DefaultValue(PIConnection.DefaultConnectTimeout)]        public int ConnectTimeout { get; set; } = PIConnection.DefaultConnectTimeout;         /// <summary>        /// Gets or sets whether or not this adapter should automatically manage metadata for PI points.        /// </summary>        [ConnectionStringParameter]        [Description("Determines if this adapter should automatically manage metadata for PI points (recommended).")]        [DefaultValue(DefaultRunMetadataSync)]        public bool RunMetadataSync { get; set; } = DefaultRunMetadataSync;         /// <summary>        /// Gets or sets whether or not this adapter should automatically create new tags when managing metadata for PI points.        /// </summary>        [ConnectionStringParameter]        [Description("Determines if this adapter should automatically create new tags when managing metadata for PI points (recommended). Value will only be considered when RunMetadataSync is True.")]        [DefaultValue(DefaultAutoCreateTags)]        public bool AutoCreateTags { get; set; } = DefaultAutoCreateTags;         /// <summary>        /// Gets or sets whether or not this adapter should automatically update existing tags when managing metadata for PI points.        /// </summary>        [ConnectionStringParameter]        [Description("Determines if this adapter should automatically update existing tags when managing metadata for PI points (recommended). This will make openPDC the controller for maintaining PI tag metadata (like the tag name) for local metadata; otherwise, when False, PI tools will be required to maintain points, including local tag name updates. Value will only be considered when RunMetadataSync is True.")]        [DefaultValue(DefaultAutoUpdateTags)]        public bool AutoUpdateTags { get; set; } = DefaultAutoUpdateTags;         /// <summary>        /// Gets or sets whether or not this adapter should automatically remove PI tags that no longer exist locally in metadata.        /// </summary>        [ConnectionStringParameter]        [Description("Determines if this adapter should automatically remove PI tags that no longer exist locally in metadata (use with caution). This will make openPDC the controller - even for deletes - when maintaining PI tag metadata; otherwise, when value is set to DoNotRemove, PI tags will persist even when they no longer exist in local metadata. Value will only be considered when RunMetadataSync is True.")]        [DefaultValue(typeof(TagRemovalOperation), DefaultAutoRemoveTags)]        public TagRemovalOperation AutoRemoveTags { get; set; } = (TagRemovalOperation)Enum.Parse(typeof(TagRemovalOperation), DefaultAutoRemoveTags);         /// <summary>        /// Gets or sets flag that determines if tag synchronization should only use alternate tag fields.        /// </summary>        [ConnectionStringParameter]        [Description("Determines if tag synchronization should only use alternate tag fields. Only relevant when RunMetadataSync is True.")]        [DefaultValue(DefaultSyncAlternateTagOnly)]        public bool SyncAlternateTagOnly { get; set; } = DefaultSyncAlternateTagOnly;         /// <summary>        /// Gets or sets flag that determines if tag synchronization should skip digitals when alternate tag field is being used.        /// </summary>        [ConnectionStringParameter]        [Description("Determines if tag synchronization should skip digitals when alternate tag field is being used. AlternateTag field is commonly used to hold IEEE C37.118 16-bit digital labels.")]        [DefaultValue(DefaultSkipDigitalAlternateTagSync)]        public bool SkipDigitalAlternateTagSync { get; set; } = DefaultSkipDigitalAlternateTagSync;         /// <summary>        /// Gets or sets the number of tag name prefixes, e.g., "SOURCE!", applied by subscriptions to remove from PI tag names.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the number of tag name prefixes applied by subscriptions, e.g., \"SOURCE!\", to remove from PI tag names. Value will only be considered when RunMetadataSync is True.")]        [DefaultValue(DefaultTagNamePrefixRemoveCount)]        public int TagNamePrefixRemoveCount { get; set; } = DefaultTagNamePrefixRemoveCount;         /// <summary>        /// Gets or sets the point source string used when automatically creating new PI points during the metadata update        /// </summary>        [ConnectionStringParameter]        [Description("Defines the point source string used when automatically creating new PI points during the metadata update. Value will only be considered when RunMetadataSync is True.")]        [DefaultValue(DefaultPIPointSource)]        public string PIPointSource { get; set; } = DefaultPIPointSource;         /// <summary>        /// Gets or sets the point class string used when automatically creating new PI points during the metadata update. On the PI server, this class should inherit from classic.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the point class string used when automatically creating new PI points during the metadata update. On the PI server, this class should inherit from classic. Value will only be considered when RunMetadataSync is True.")]        [DefaultValue(DefaultPIPointClass)]        public string PIPointClass { get; set; } = DefaultPIPointClass;         /// <summary>        /// Gets or sets flag that determines if defined point compression will be used during archiving.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the flag that determines if defined point compression will be used during archiving.")]        [DefaultValue(DefaultUseCompression)]        public bool UseCompression { get; set; } = DefaultUseCompression;         /// <summary>        /// Gets or sets the flag that determines if the compression enabled state, per UseCompression flag, should be adjusted for existing tags.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the flag that determines if the compression enabled state, per UseCompression flag, should be adjusted for existing tags.")]        [DefaultValue(DefaultUpdateExistingTagCompressionState)]        public bool UpdateExistingTagCompressionState { get; set; } = DefaultUpdateExistingTagCompressionState;         /// <summary>        /// Gets or sets the filename to be used for tag map cache.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the filename to be used for tag map cache file name. Leave blank for cache name to be same as adapter name with a \".cache\" extension.")]        [DefaultValue(DefaultTagMapCacheFileName)]        public string TagMapCacheFileName { get; set; } = DefaultTagMapCacheFileName;         /// <summary>        /// Gets or sets the maximum time resolution, in seconds, for data points being archived, e.g., a value 1.0 would mean that data would be archived no more than once per second. A zero value indicates that all data should be archived.        /// </summary>        [ConnectionStringParameter]        [Description("Defines the maximum time resolution, in seconds, for data points being archived, e.g., a value 1.0 would mean that data would be archived no more than once per second. A zero value indicates that all data should be archived.")]        [DefaultValue(DefaultMaximumPointResolution)]        public double MaximumPointResolution { get; set; } = DefaultMaximumPointResolution;         /// <summary>        /// Gets or sets flag that indicates if incoming timestamps to the historian should be validated for reasonability.        /// </summary>        [ConnectionStringParameter]        [Description("Define the flag that indicates if incoming timestamps to the historian should be validated for reasonability.")]        [DefaultValue(DefaultEnableTimeReasonabilityCheck)]        public bool EnableTimeReasonabilityCheck { get; set; } = DefaultEnableTimeReasonabilityCheck;         /// <summary>        /// Gets or sets the maximum number of seconds that a past timestamp, as compared to local clock, will be considered valid.        /// </summary>        [ConnectionStringParameter]        [Description("Define the maximum number of seconds that a past timestamp, as compared to local clock, will be considered valid.")]        [DefaultValue(DefaultPastTimeReasonabilityLimit)]        public double PastTimeReasonabilityLimit        {            get => new Ticks(m_pastTimeReasonabilityLimit).ToSeconds();            set => m_pastTimeReasonabilityLimit = Ticks.FromSeconds(Math.Abs(value));        }         /// <summary>        /// Gets or sets the maximum number of seconds that a future timestamp, as compared to local clock, will be considered valid.        /// </summary>        [ConnectionStringParameter]        [Description("Define the maximum number of seconds that a future timestamp, as compared to local clock, will be considered valid.")]        [DefaultValue(DefaultFutureTimeReasonabilityLimit)]        public double FutureTimeReasonabilityLimit        {            get => new Ticks(m_futureTimeReasonabilityLimit).ToSeconds();            set => m_futureTimeReasonabilityLimit = Ticks.FromSeconds(Math.Abs(value));        }         /// <summary>        /// Returns the detailed status of the data output source.        /// </summary>        public override string Status        {            get            {                StringBuilder status = new();                 status.Append(base.Status);                status.AppendLine();                status.AppendLine($"        OSI-PI server name: {ServerName}");                status.AppendLine($"       Connected to server: {(m_connection?.Connected ?? false ? "Yes" : "No")}");                status.AppendLine($"         Using compression: {UseCompression}");                status.AppendLine($"  Update compression state: {UpdateExistingTagCompressionState}");                status.AppendLine($"  Maximum point resolution: {MaximumPointResolution:N3} seconds{(MaximumPointResolution <= 0.0D ? " - all data will be archived" : "")}");                status.AppendLine($"  Time reasonability check: {(EnableTimeReasonabilityCheck ? "Enabled" : "Not Enabled")}");                 if (EnableTimeReasonabilityCheck)                {                    status.AppendLine($"   Maximum past time limit: {PastTimeReasonabilityLimit:N4}s, i.e., {new Ticks(m_pastTimeReasonabilityLimit).ToElapsedTimeString(4)}");                    status.AppendLine($" Maximum future time limit: {FutureTimeReasonabilityLimit:N4}s, i.e., {new Ticks(m_futureTimeReasonabilityLimit).ToElapsedTimeString(4)}");                }                 status.AppendLine($"    Meta-data sync enabled: {RunMetadataSync}");                 if (RunMetadataSync)                {                    status.AppendLine($"          Auto-create tags: {AutoCreateTags}");                    status.AppendLine($"          Auto-update tags: {AutoUpdateTags}");                    status.AppendLine($"          Auto-remove tags: {AutoRemoveTags}");                    status.AppendLine($"   Sync alternate tag only: {SyncAlternateTagOnly}");                    status.AppendLine($" Skip digital alt-tag sync: {SkipDigitalAlternateTagSync}");                    status.AppendLine($"    Tag prefixes to remove: {TagNamePrefixRemoveCount}");                    status.AppendLine($"       OSI-PI point source: {PIPointSource}");                    status.AppendLine($"        OSI-PI point class: {PIPointClass}");                }                 if (m_mapRequestQueue is not null)                {                    status.AppendLine();                    status.AppendLine(">> Mapping Request Queue Status");                    status.AppendLine();                    status.Append(m_mapRequestQueue.Status);                    status.AppendLine();                }                 status.AppendLine($"   Tag-map cache file name: {FilePath.TrimFileName(TagMapCacheFileName.ToNonNullString("[undefined]"), 51)}");                status.AppendLine($" Active tag-map cache size: {m_mappedPIPoints.Count:N0} mappings");                status.AppendLine($"      Pending tag-mappings: {m_pendingMappings.Count:N0} mappings, {1.0D - m_pendingMappings.Count / (double)m_mappedPIPoints.Count:0.00%} complete");                 if (RunMetadataSync)                    status.AppendLine($"    Meta-data sync process: {(m_refreshingMetadata ? "Active" : "Idle")}, {m_metadataRefreshProgress:0.00%} complete");                 if (m_archiveQueues is not null && m_archiveQueues.Length > 0)                {                    status.AppendLine();                    status.AppendLine($">> Archive Queue Status (1 of {m_archiveQueues.Length:N0})");                    status.AppendLine();                    status.Append(m_archiveQueues[0].Status);                    status.AppendLine();                }                 status.AppendLine($"    Points archived/second: {Interlocked.Read(ref m_processedMeasurements) / (Interlocked.Read(ref m_totalProcessingTime) / (double)Ticks.PerSecond):#,##0.00}");                 return status.ToString();            }        }         #endregion         #region [ Methods ]         /// <summary>        /// Releases the unmanaged resources used by the <see cref="PIOutputAdapter"/> object and optionally releases the managed resources.        /// </summary>        /// <param name="disposing">true to release both managed and unmanaged resources; false to release only unmanaged resources.</param>        protected override void Dispose(bool disposing)        {            if (m_disposed)                return;             try            {                if (!disposing)                    return;                 m_mapRequestQueue?.Dispose();                 if (m_archiveQueues is not null)                {                    foreach (ProcessQueue<AFValue> archiveQueue in m_archiveQueues)                        archiveQueue.Dispose();                }                 if (m_connection is not null)                {                    m_connection.Disconnected -= Connection_Disconnected;                    m_connection.Dispose();                    m_connection = null;                }                 m_mappedPIPoints?.Clear();            }            finally            {                m_disposed = true;       // Prevent duplicate dispose.                base.Dispose(disposing); // Call base class Dispose().            }        }         /// <summary>        /// Gets full archive queue status.        /// </summary>        [AdapterCommand("Gets full archive queue status.", "Administrator", "Editor")]        public void GetArchiveQueueStatus()        {            if (m_archiveQueues is null)                return;             StringBuilder status = new();             for (int i = 0; i < m_archiveQueues.Length; i++)            {                status.AppendLine();                status.AppendLine($">> Archive Queue Status (1 of {i:N0})");                status.AppendLine();                status.Append(m_archiveQueues[i].Status);                status.AppendLine();            }             OnStatusMessage(MessageLevel.Info, status.ToString());        }         /// <summary>        /// Returns a brief status of this <see cref="PIOutputAdapter"/>        /// </summary>        /// <param name="maxLength">Maximum number of characters in the status string</param>        /// <returns>Status</returns>        public override string GetShortStatus(int maxLength) =>             $"Archived {m_processedMeasurements:N0} measurements to PI.".CenterText(maxLength);         /// <summary>        /// Initializes this <see cref="PIOutputAdapter"/>.        /// </summary>        public override void Initialize()        {            base.Initialize();

Lines 1–500 of 1388. Load more