Draft PR for SrCnn batch detection API interface

src/Microsoft.ML.Data/DataView/BatchDataViewMapperBase.cs

@@ -0,0 +1,173 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using Microsoft.ML.Runtime;
+
+namespace Microsoft.ML.Data.DataView
+{
+    internal abstract class BatchDataViewMapperBase<TInput, TBatch> : IDataView
+    {
+        public bool CanShuffle => false;
+
+        public DataViewSchema Schema => SchemaBindings.AsSchema;
+
+        private readonly IDataView _source;
+        private readonly IHost _host;
+        protected readonly ColumnBindingsBase SchemaBindings;
+
+        protected BatchDataViewMapperBase(IHostEnvironment env, string registrationName, IDataView input, ColumnBindingsBase schemaBindings)
+        {
+            _host = env.Register(registrationName);
+            _source = input;
+            SchemaBindings = schemaBindings;
+        }
+
+        public long? GetRowCount() => _source.GetRowCount();
+
+        public DataViewRowCursor GetRowCursor(IEnumerable<DataViewSchema.Column> columnsNeeded, Random rand = null)
+        {
+            _host.CheckValue(columnsNeeded, nameof(columnsNeeded));
+            _host.CheckValueOrNull(rand);
+
+            var predicate = RowCursorUtils.FromColumnsToPredicate(columnsNeeded, SchemaBindings.AsSchema);
+
+            // If we aren't selecting any of the output columns, don't construct our cursor.
+            // Note that because we cannot support random due to the inherently
+            // stratified nature, neither can we allow the base data to be shuffled,
+            // even if it supports shuffling.
+            if (!SchemaBindings.AnyNewColumnsActive(predicate))
+            {
+                var activeInput = SchemaBindings.GetActiveInput(predicate);
+                var inputCursor = _source.GetRowCursor(_source.Schema.Where(c => activeInput[c.Index]), null);
+                return new BindingsWrappedRowCursor(_host, inputCursor, SchemaBindings);
+            }
+            var active = SchemaBindings.GetActive(predicate);
+            Contracts.Assert(active.Length == SchemaBindings.ColumnCount);
+
+            // REVIEW: We can get a different input predicate for the input cursor and for the lookahead cursor. The lookahead
+            // cursor is only used for getting the values from the input column, so it only needs that column activated. The
+            // other cursor is used to get source columns, so it needs the rest of them activated.
+            var predInput = GetSchemaBindingDependencies(predicate);
+            var inputCols = _source.Schema.Where(c => predInput(c.Index));
+            return new Cursor(this, _source.GetRowCursor(inputCols), _source.GetRowCursor(inputCols), active);
+        }
+
+        public DataViewRowCursor[] GetRowCursorSet(IEnumerable<DataViewSchema.Column> columnsNeeded, int n, Random rand = null)
+        {
+            return new[] { GetRowCursor(columnsNeeded, rand) };
+        }
+
+        protected abstract TBatch InitializeBatch(DataViewRowCursor input);
+        protected abstract void ProcessBatch(TBatch currentBatch);
+        protected abstract void ProcessExample(TBatch currentBatch, TInput currentInput);
+        protected abstract Func<bool> GetLastInBatchDelegate(DataViewRowCursor lookAheadCursor);
+        protected abstract Func<bool> GetIsNewBatchDelegate(DataViewRowCursor lookAheadCursor);
+        protected abstract ValueGetter<TInput> GetLookAheadGetter(DataViewRowCursor lookAheadCursor);
+        protected abstract Delegate[] CreateGetters(DataViewRowCursor input, TBatch currentBatch, bool[] active);
+        protected abstract Func<int, bool> GetSchemaBindingDependencies(Func<int, bool> predicate);
+
+        private sealed class Cursor : RootCursorBase
+        {
+            private readonly BatchDataViewMapperBase<TInput, TBatch> _parent;
+            private readonly DataViewRowCursor _lookAheadCursor;
+            private readonly DataViewRowCursor _input;
+
+            private readonly bool[] _active;
+            private readonly Delegate[] _getters;
+
+            private TBatch _currentBatch;
+            private readonly Func<bool> _lastInBatchInLookAheadCursorDel;
+            private readonly Func<bool> _firstInBatchInInputCursorDel;
+            private readonly ValueGetter<TInput> _inputGetterInLookAheadCursor;
+            private TInput _currentInput;
+
+            public override long Batch => 0;
+
+            public override DataViewSchema Schema => _parent.Schema;
+
+            public Cursor(BatchDataViewMapperBase<TInput, TBatch> parent, DataViewRowCursor input, DataViewRowCursor lookAheadCursor, bool[] active)
+                : base(parent._host)
+            {
+                _parent = parent;
+                _input = input;
+                _lookAheadCursor = lookAheadCursor;
+                _active = active;
+
+                _currentBatch = _parent.InitializeBatch(_input);
+
+                _getters = _parent.CreateGetters(_input, _currentBatch, _active);
+
+                _lastInBatchInLookAheadCursorDel = _parent.GetLastInBatchDelegate(_lookAheadCursor);
+                _firstInBatchInInputCursorDel = _parent.GetIsNewBatchDelegate(_input);
+                _inputGetterInLookAheadCursor = _parent.GetLookAheadGetter(_lookAheadCursor);
+            }
+
+            public override ValueGetter<TValue> GetGetter<TValue>(DataViewSchema.Column column)
+            {
+                Contracts.CheckParam(IsColumnActive(column), nameof(column), "requested column is not active");
+
+                var col = _parent.SchemaBindings.MapColumnIndex(out bool isSrc, column.Index);
+                if (isSrc)
+                {
+                    Contracts.AssertValue(_input);
+                    return _input.GetGetter<TValue>(_input.Schema[col]);
+                }
+
+                Ch.AssertValue(_getters);
+                var getter = _getters[col];
+                Ch.Assert(getter != null);
+                var fn = getter as ValueGetter<TValue>;
+                if (fn == null)
+                    throw Ch.Except("Invalid TValue in GetGetter: '{0}'", typeof(TValue));
+                return fn;
+            }
+
+            public override ValueGetter<DataViewRowId> GetIdGetter()
+            {
+                return
+                    (ref DataViewRowId val) =>
+                    {
+                        Ch.Check(IsGood, "Cannot call ID getter in current state");
+                        val = new DataViewRowId((ulong)Position, 0);
+                    };
+            }
+
+            public override bool IsColumnActive(DataViewSchema.Column column)
+            {
+                Ch.Check(column.Index < _parent.SchemaBindings.AsSchema.Count);
+                return _active[column.Index];
+            }
+
+            protected override bool MoveNextCore()
+            {
+                if (!_input.MoveNext())
+                    return false;
+                if (!_firstInBatchInInputCursorDel())
+                    return true;
+
+                // If we are here, this means that _input.MoveNext() has gotten us to the beginning of the next batch,
+                // so now we need to look ahead at the entire next batch in the _lookAheadCursor.
+                // The _lookAheadCursor's position should be on the last row of the previous batch (or -1).
+                Ch.Assert(_lastInBatchInLookAheadCursorDel());
+
+                var good = _lookAheadCursor.MoveNext();
+                // The two cursors should have the same number of elements, so if _input.MoveNext() returned true,
+                // then it must return true here too.
+                Ch.Assert(good);
+
+                do
+                {
+                    _inputGetterInLookAheadCursor(ref _currentInput);
+                    _parent.ProcessExample(_currentBatch, _currentInput);
+                } while (!_lastInBatchInLookAheadCursorDel() && _lookAheadCursor.MoveNext());
+
+                _parent.ProcessBatch(_currentBatch);
+                return true;
+            }
+        }
+    }
+}

src/Microsoft.ML.TimeSeries/ExtensionsCatalog.cs

@@ -3,6 +3,8 @@
 // See the LICENSE file in the project root for more information.
 
 using Microsoft.ML.Data;
+using Microsoft.ML.Data.DataView;
+using Microsoft.ML.TimeSeries;
 using Microsoft.ML.Transforms.TimeSeries;
 
 namespace Microsoft.ML
@@ -146,6 +148,29 @@ public static SrCnnAnomalyEstimator DetectAnomalyBySrCnn(this TransformsCatalog
             int windowSize=64, int backAddWindowSize=5, int lookaheadWindowSize=5, int averageingWindowSize=3, int judgementWindowSize=21, double threshold=0.3)
             => new SrCnnAnomalyEstimator(CatalogUtils.GetEnvironment(catalog), outputColumnName, windowSize, backAddWindowSize, lookaheadWindowSize, averageingWindowSize, judgementWindowSize, threshold, inputColumnName);
 
+        /// <summary>
+        /// Create <see cref="SrCnnAnomalyEstimator"/>, which detects timeseries anomalies using SRCNN algorithm.
+        /// </summary>
+        /// <param name="catalog">The transform's catalog.</param>
+        /// <param name="input">...</param>
+        /// <param name="outputColumnName">Name of the column resulting from the transformation of <paramref name="inputColumnName"/>.
+        /// The column data is a vector of <see cref="System.Double"/>. The vector contains 3 elements: alert (1 means anomaly while 0 means normal), raw score, and magnitude of spectual residual.</param>
+        /// <param name="inputColumnName">Name of column to transform. The column data must be <see cref="System.Single"/>.</param>
+        /// <param name="threshold">The threshold to determine anomaly, score larger than the threshold is considered as anomaly. Should be in (0,1)</param>
+        /// <param name="batchSize">.Divide the input data into batches to fit SrCnn model. Must be -1 or a positive integer no less than 12. Default value is 1024.</param>
+        /// <param name="sensitivity">The sensitivity of boundaries. Must be in the interval (0, 100).</param>
+        /// <param name="detectMode">The detect mode of the SrCnn model.</param>
+        /// <example>
+        /// <format type="text/markdown">
+        /// <![CDATA[
+        /// [!code-csharp[DetectAnomalyBySrCnn](~/../docs/samples/docs/samples/Microsoft.ML.Samples/Dynamic/Transforms/TimeSeries/DetectAnomalyBySrCnn.cs)]
+        /// ]]>
+        /// </format>
+        /// </example>
+        public static IDataView BatchDetectAnomalyBySrCnn(this TransformsCatalog catalog, IDataView input, string outputColumnName, string inputColumnName,
+            double threshold = 0.3, int batchSize = 1024, double sensitivity = 99, SrCnnDetectMode detectMode = SrCnnDetectMode.AnomalyAndMargin)
+            => new SrCnnBatchAnomalyDetector(CatalogUtils.GetEnvironment(catalog), input, inputColumnName, outputColumnName, threshold, batchSize, sensitivity, detectMode);
+
         /// <summary>
         /// Singular Spectrum Analysis (SSA) model for univariate time-series forecasting.
         /// For the details of the model, refer to http://arxiv.org/pdf/1206.6910.pdf.

src/Microsoft.ML.TimeSeries/SrCnnBatchAnomalyDetection.cs

@@ -0,0 +1,186 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+// See the LICENSE file in the project root for more information.
+
+using System;
+using System.Collections.Generic;
+using Microsoft.ML.Data;
+using Microsoft.ML.Data.DataView;
+using Microsoft.ML.Internal.Utilities;
+using Microsoft.ML.Numeric;
+using Microsoft.ML.Runtime;
+
+namespace Microsoft.ML.TimeSeries
+{
+    /// <summary>
+    /// The detect modes of SrCnn models.
+    /// </summary>
+    public enum SrCnnDetectMode
+    {
+        /// <summary>
+        /// In this mode, output (IsAnomaly, RawScore, Mag).
+        /// </summary>
+        AnomalyOnly = 0,
+
+        /// <summary>
+        /// In this mode, output (IsAnomaly, AnomalyScore, Mag, ExpectedValue, BoundaryUnit, UpperBoundary, LowerBoundary).
+        /// </summary>
+        AnomalyAndMargin = 1,
+
+        /// <summary>
+        /// In this mode, output (IsAnomaly, RawScore, Mag, ExpectedValue).
+        /// </summary>
+        AnomalyAndExpectedValue = 2
+    }
+
+    // TODO: SrCnn
+    internal sealed class SrCnnBatchAnomalyDetector : BatchDataViewMapperBase<float, SrCnnBatchAnomalyDetector.Batch>
+    {
+        private readonly int _batchSize;
+        private const int _minBatchSize = 12;
+        private readonly string _inputColumnName;
+
+        private class Bindings : ColumnBindingsBase
+        {
+            private readonly DataViewType _outputColumnType;
+            private readonly int _inputColumnIndex;
+
+            public Bindings(DataViewSchema input, string inputColumnName, string outputColumnName, DataViewType outputColumnType)
+                : base(input, true, outputColumnName)
+            {
+                _outputColumnType = outputColumnType;
+                _inputColumnIndex = Input[inputColumnName].Index;
+            }
+
+            protected override DataViewType GetColumnTypeCore(int iinfo)
+            {
+                Contracts.Check(iinfo == 0);
+                return _outputColumnType;
+            }
+
+            // Get a predicate for the input columns.
+            public Func<int, bool> GetDependencies(Func<int, bool> predicate)
+            {
+                Contracts.AssertValue(predicate);
+
+                var active = new bool[Input.Count];
+                for (int col = 0; col < ColumnCount; col++)
+                {
+                    if (!predicate(col))
+                        continue;
+
+                    bool isSrc;
+                    int index = MapColumnIndex(out isSrc, col);
+                    if (isSrc)
+                        active[index] = true;
+                    else
+                        active[_inputColumnIndex] = true;
+                }
+
+                return col => 0 <= col && col < active.Length && active[col];
+            }
+        }
+
+        public SrCnnBatchAnomalyDetector(IHostEnvironment env, IDataView input, string inputColumnName, string outputColumnName, double threshold, int batchSize, double sensitivity, SrCnnDetectMode detectMode)
+            : base(env, "SrCnnBatchAnomalyDetector", input, new Bindings(input.Schema, inputColumnName, outputColumnName, NumberDataViewType.Single))
+        {
+            Contracts.CheckParam(batchSize >= _minBatchSize, nameof(batchSize), "batch size is too small");
+            _batchSize = batchSize;
+            _inputColumnName = inputColumnName;
+        }
+
+        protected override Delegate[] CreateGetters(DataViewRowCursor input, Batch currentBatch, bool[] active)
+        {
+            if (!SchemaBindings.AnyNewColumnsActive(x => active[x]))
+                return new Delegate[1];
+            return new[] { currentBatch.CreateGetter(input, _inputColumnName) };
+        }
+
+        protected override Batch InitializeBatch(DataViewRowCursor input) => new Batch(_batchSize);
+
+        protected override Func<bool> GetIsNewBatchDelegate(DataViewRowCursor input)
+        {
+            return () => input.Position % _batchSize == 0;
+        }
+
+        protected override Func<bool> GetLastInBatchDelegate(DataViewRowCursor input)
+        {
+            return () => (input.Position + 1) % _batchSize == 0;
+        }
+
+        protected override ValueGetter<float> GetLookAheadGetter(DataViewRowCursor input)
+        {
+            return input.GetGetter<float>(input.Schema[_inputColumnName]);
+        }
+
+        protected override Func<int, bool> GetSchemaBindingDependencies(Func<int, bool> predicate)
+        {
+            return (SchemaBindings as Bindings).GetDependencies(predicate);
+        }
+
+        protected override void ProcessExample(Batch currentBatch, float currentInput)
+        {
+            currentBatch.AddValue(currentInput);
+        }
+
+        protected override void ProcessBatch(Batch currentBatch)
+        {
+            currentBatch.Process();
+            currentBatch.Reset();
+        }
+
+        public sealed class Batch
+        {
+            private List<float> _previousBatch;
+            private List<float> _batch;
+            private float _cursor;
+            private readonly int _batchSize;
+
+            public Batch(int batchSize)
+            {
+                _batchSize = batchSize;
+                _previousBatch = new List<float>(batchSize);
+                _batch = new List<float>(batchSize);
+            }
+
+            public void AddValue(float value)
+            {
+                _batch.Add(value);
+            }
+
+            public int Count => _batch.Count;
+
+            public void Process()
+            {
+                // TODO: replace with SrCnn
+                _cursor = VectorUtils.NormSquared(new ReadOnlySpan<float>(_batch.ToArray()));
+                if (_batch.Count < _batchSize)
+                {
+                    _cursor += VectorUtils.NormSquared(new ReadOnlySpan<float>(
+                        _previousBatch.GetRange(_batch.Count, _batchSize - _batch.Count).ToArray()));
+                }
+            }
+
+            public void Reset()
+            {
+                var tempBatch = _previousBatch;
+                _previousBatch = _batch;
+                _batch = tempBatch;
+                _batch.Clear();
+            }
+
+            public ValueGetter<float> CreateGetter(DataViewRowCursor input, string inputCol)
+            {
+                ValueGetter<float> srcGetter = input.GetGetter<float>(input.Schema[inputCol]);
+                ValueGetter<float> getter =
+                    (ref float dst) =>
+                    {
+                        float src = default;
+                        srcGetter(ref src);
+                        dst = src * _cursor;
+                    };
+                return getter;
+            }
+        }
+    }
+}