//******************************************************************************************************
//  OneSecondDataWindowAdapterBase.cs - Gbtc
//
//  Copyright © 2020, 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://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:
//  ----------------------------------------------------------------------------------------------------
//  02/11/2020 - J. Ritchie Carroll
//       Generated original version of source code.
//
//******************************************************************************************************

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using GSF.Diagnostics;
using GSF.Threading;
using GSF.TimeSeries.Data;
using GSF.Units.EE;
using ConnectionStringParser = GSF.Configuration.ConnectionStringParser<GSF.TimeSeries.Adapters.ConnectionStringParameterAttribute>;

namespace GSF.TimeSeries.Adapters
{
    /// <summary>
    /// Represents an adapter base class that provides functionality to operate on a 1-second window of data.
    /// </summary>
    public abstract class OneSecondDataWindowAdapterBase : ActionAdapterBase
    {
        #region [ Members ]

        // Constants

        /// <summary>
        /// Defines the default value for the <see cref="SourceMeasurementTable"/>.
        /// </summary>
        public const string DefaultSourceMeasurementTable = "ActiveMeasurements";

        // Fields
        private readonly ConcurrentDictionary<long, IMeasurement[,]> m_dataWindows = new ConcurrentDictionary<long, IMeasurement[,]>();
        private readonly Dictionary<MeasurementKey, int> m_keyIndexes = new Dictionary<MeasurementKey, int>();
        private ShortSynchronizedOperation m_processDataWindows;
        private long m_lastFrameTimestamp;
        private long m_processedDataWindows;
        private bool m_supportsTemporalProcessing;

        #endregion

        #region [ Properties ]

        /// <summary>
        /// Gets or sets the expected number of input measurements for a <see cref="OneSecondDataWindowAdapterBase"/> instance.
        /// </summary>
        public int InputCount { get; protected set; }

        /// <summary>
        /// Gets or sets the expected number of output measurements for a <see cref="OneSecondDataWindowAdapterBase"/> instance.
        /// </summary>
        public int OutputCount { get; protected set; }

        /// <summary>
        /// Gets sub-second offsets based on current <see cref="ActionAdapterBase.FramesPerSecond"/>.
        /// </summary>
        public Ticks[] SubsecondOffsets { get; private set; }

        /// <summary>
        /// Gets or sets primary keys of input measurements the action adapter expects.
        /// </summary>
        /// <remarks>
        /// If your adapter needs to receive all measurements, you must explicitly set InputMeasurementKeys to null.
        /// </remarks>
        [ConnectionStringParameter]
        [Description("Defines primary keys of input measurements the action adapter expects; can be one of a filter expression, measurement key, point tag or Guid.")]
        [DefaultValue(null)]
        [CustomConfigurationEditor("GSF.TimeSeries.UI.WPF.dll", "GSF.TimeSeries.UI.Editors.MeasurementEditor")]
        public override MeasurementKey[] InputMeasurementKeys
        {
            get => base.InputMeasurementKeys;
            set
            {
                base.InputMeasurementKeys = value;
                InputMeasurementKeyTypes = DataSource.GetSignalTypes(value, SourceMeasurementTable);

                for (int i = 0; i < value?.Length; i++)
                    m_keyIndexes[value[i]] = i;
            }
        }

        /// <summary>
        /// Gets or sets output measurements that the action adapter will produce, if any.
        /// </summary>
        [ConnectionStringParameter]
        [Description("Defines primary keys of output measurements the action adapter expects; can be one of a filter expression, measurement key, point tag or Guid.")]
        [DefaultValue(null)]
        [CustomConfigurationEditor("GSF.TimeSeries.UI.WPF.dll", "GSF.TimeSeries.UI.Editors.MeasurementEditor")]
        public override IMeasurement[] OutputMeasurements
        {
            get => base.OutputMeasurements;
            set
            {
                base.OutputMeasurements = value;
                OutputMeasurementTypes = DataSource.GetSignalTypes(value, SourceMeasurementTable);
            }
        }

        /// <summary>
        /// Gets or sets the source measurement table to use for configuration.
        /// </summary>
        [ConnectionStringParameter]
        [Description("Defines the source measurement table to use for configuration.")]
        [DefaultValue(DefaultSourceMeasurementTable)]
        public virtual string SourceMeasurementTable { get; set; } = DefaultSourceMeasurementTable;

        /// <summary>
        /// Gets or sets input measurement <see cref="SignalType"/>'s for each of the <see cref="ActionAdapterBase.InputMeasurementKeys"/>, if any.
        /// </summary>
        public virtual SignalType[] InputMeasurementKeyTypes { get; private set; }

        /// <summary>
        /// Gets or sets output measurement <see cref="SignalType"/>'s for each of the <see cref="ActionAdapterBase.OutputMeasurements"/>, if any.
        /// </summary>
        public virtual SignalType[] OutputMeasurementTypes { get; private set; }

        /// <summary>
        /// Gets the flag indicating if this adapter supports temporal processing.
        /// </summary>
        public override bool SupportsTemporalProcessing => m_supportsTemporalProcessing;

        /// <summary>
        /// Returns the detailed status of the <see cref="OneSecondDataWindowAdapterBase"/>.
        /// </summary>
        public override string Status
        {
            get
            {
                StringBuilder status = new StringBuilder();

                status.AppendFormat("       Queued Data Windows: {0:N0}", m_dataWindows.Count);
                status.AppendLine();
                status.AppendFormat("      Last Frame Timestamp: {0:yyyy-MM-dd HH:mm:ss}", new DateTime(Interlocked.Read(ref m_lastFrameTimestamp)));
                status.AppendLine();
                status.AppendFormat("    Processed Data Windows: {0:N0}", Interlocked.Read(ref m_processedDataWindows));
                status.AppendLine();
                status.AppendFormat("  Source Measurement Table: {0}", SourceMeasurementTable);
                status.AppendLine();
                status.Append(base.Status);

                if (OutputMeasurements != null && OutputMeasurements.Length > 0)
                {
                    status.AppendLine();
                    status.AppendLine("Output measurements signal type summary:");
                    status.AppendLine();

                    foreach (SignalType signalType in Enum.GetValues(typeof(SignalType)))
                    {
                        int count = OutputMeasurements.Where((key, index) => OutputMeasurementTypes[index] == signalType).Count();

                        if (count <= 0)
                            continue;

                        status.AppendFormat("{0} {1} signal{2}", count.ToString().PadLeft(15), signalType.GetFormattedName(), count > 1 ? "s" : "");
                        status.AppendLine();
                    }
                }

                if (InputMeasurementKeys != null && InputMeasurementKeys.Length > 0)
                {
                    status.AppendLine();
                    status.AppendLine("Input measurement keys signal type summary:");
                    status.AppendLine();

                    foreach (SignalType signalType in Enum.GetValues(typeof(SignalType)))
                    {
                        int count = InputMeasurementKeys.Where((key, index) => InputMeasurementKeyTypes[index] == signalType).Count();

                        if (count <= 0)
                            continue;

                        status.AppendFormat("{0} {1} signal{2}", count.ToString().PadLeft(15), signalType.GetFormattedName(), count > 1 ? "s" : "");
                        status.AppendLine();
                    }
                }

                return status.ToString();
            }
        }

        #endregion

        #region [ Methods ]

        /// <summary>
        /// Initializes <see cref="OneSecondDataWindowAdapterBase" />.
        /// </summary>
        public override void Initialize()
        {
            string setting;

            // Parse all properties marked with ConnectionStringParameterAttribute from provided ConnectionString value
            ConnectionStringParser parser = new ConnectionStringParser();
            parser.ParseConnectionString(ConnectionString, this);

            base.Initialize();

            // Reparse configured inputs and outputs if a non-standard source measurement table is being used
            if (!SourceMeasurementTable.Equals(DefaultSourceMeasurementTable, StringComparison.OrdinalIgnoreCase))
            {
                Dictionary<string, string> settings = Settings;

                if (settings.TryGetValue(nameof(InputMeasurementKeys), out setting))
                    InputMeasurementKeys = AdapterBase.ParseInputMeasurementKeys(DataSource, true, setting, SourceMeasurementTable);

                if (settings.TryGetValue(nameof(OutputMeasurements), out setting))
                    OutputMeasurements = AdapterBase.ParseOutputMeasurements(DataSource, true, setting, SourceMeasurementTable);
            }

            int configuredInputCount = InputMeasurementKeys?.Length ?? 0;
            int configuredOutputCount = OutputMeasurements?.Length ?? 0;

            if (configuredInputCount == 0)
                throw new InvalidOperationException("No inputs specified. Cannot initialize adapter.");

            if (configuredInputCount != InputCount)
                throw new InvalidOperationException($"Expected {InputCount:N0} input measurements, there are {configuredInputCount:N0} defined. Cannot initialize adapter.");

            if (configuredOutputCount != OutputCount)
                throw new InvalidOperationException($"Expected {OutputCount:N0} output measurements, there are {configuredOutputCount:N0} defined. Cannot initialize adapter.");

            m_supportsTemporalProcessing = Settings.TryGetValue("supportsTemporalProcessing", out setting) && setting.ParseBoolean();

            SubsecondOffsets = Ticks.SubsecondDistribution(FramesPerSecond);

            // Define a synchronized operation to process completed data windows
            m_processDataWindows = new ShortSynchronizedOperation(ProcessDataWindows, ex => OnProcessException(MessageLevel.Warning, ex));
        }

        /// <summary>
        /// Processes 1-second data window.
        /// </summary>
        /// <param name="timestamp">Top of second window timestamp.</param>
        /// <param name="dataWindow">1-second data window.</param>
        protected abstract void ProcessDataWindow(Ticks timestamp, IMeasurement[,] dataWindow);

        /// <summary>
        /// Publish <see cref="IFrame" /> of time-aligned collection of <see cref="IMeasurement" />
        /// values that arrived within defined <see cref="ConcentratorBase.LagTime" />.
        /// </summary>
        /// <param name="frame">Collection of measurements with the same timestamp that arrived within <see cref="ConcentratorBase.LagTime" /> that are ready for processing.</param>
        /// <param name="index">Index of <see cref="IFrame" /> within a second ranging from zero to <c><see cref="ConcentratorBase.FramesPerSecond" /> - 1</c>.</param>
        /// <remarks>
        /// If user implemented publication function consistently exceeds available publishing time (i.e., <c>1 / <see cref="ConcentratorBase.FramesPerSecond" /></c> seconds),
        /// concentration will fall behind. A small amount of this time is required by the <see cref="ConcentratorBase" /> for processing overhead, so actual total time
        /// available for user function process will always be slightly less than <c>1 / <see cref="ConcentratorBase.FramesPerSecond" /></c> seconds.
        /// </remarks>
        protected override void PublishFrame(IFrame frame, int index)
        {
            // Baseline timestamp to the top of the second - this represents the window key
            Interlocked.Exchange(ref m_lastFrameTimestamp, frame.Timestamp.BaselinedTimestamp(BaselineTimeInterval.Second));

            // Get current data window, creating it if needed
            IMeasurement[,] dataWindow = m_dataWindows.GetOrAdd(Interlocked.Read(ref m_lastFrameTimestamp), CreateNewDataWindow);

            // At the top of each second, process any prior data windows
            if (index == 0)
                m_processDataWindows.RunOnceAsync();

            // Update current data window with provided measurements
            foreach (IMeasurement measurement in frame.Measurements.Values)
                dataWindow[m_keyIndexes[measurement.Key], index] = measurement;
        }

        private void ProcessDataWindows()
        {
            long lastFrameTimestamp = Interlocked.Read(ref m_lastFrameTimestamp);
            long[] timestamps = m_dataWindows.Keys.Where(key => key < lastFrameTimestamp).ToArray();

            // Process all completed data windows
            foreach (long timestamp in timestamps)
            {
                if (m_dataWindows.TryRemove(timestamp, out IMeasurement[,] dataWindow))
                {
                    Debug.Assert(dataWindow.GetLength(0) == InputCount, $"Unexpected data window input size: {dataWindow.GetLength(0)}, expected {InputCount}");
                    Debug.Assert(dataWindow.GetLength(1) == FramesPerSecond, $"Unexpected data window row size: {dataWindow.GetLength(1)}, expected {FramesPerSecond}");

                    ProcessDataWindow(timestamp, dataWindow);
                    
                    Interlocked.Increment(ref m_processedDataWindows);
                }
            }
        }

        private IMeasurement[,] CreateNewDataWindow(long timestamp)
        {
            IMeasurement[,] dataWindow = new IMeasurement[InputCount, FramesPerSecond];

            for (int i = 0; i < InputCount; i++)
            {
                MeasurementKey key = InputMeasurementKeys[i];

                for (int j = 0; j < FramesPerSecond; j++)
                {
                    dataWindow[i, j] = new Measurement
                    {
                        Metadata = key.Metadata,
                        Value = double.NaN,
                        Timestamp = timestamp + SubsecondOffsets[j],
                        StateFlags = MeasurementStateFlags.BadData
                    };
                }
            }

            return dataWindow;
        }

        #endregion
    }
}