BinaryYELTConverter.cs

Go to the documentation of this file.

using System;
using System.Globalization;
using System.Text;
using AnalyzeRe.Utilities;
 
namespace AnalyzeRe.LargeDataUpload
{
    [Obsolete("The binary YELT format and related options will be deprecated in the near future." +
        "We encourage you to upload your YELT data as human-readable CSV.")]
    public class BinaryYELTConverter : ProducerConsumerBuffer<BufferedBytes>
    {
        public static readonly Encoding DEFAULT_ENCODING = Encoding.UTF8;
 
        private const NumberStyles IntParseStyle = NumberStyles.None;
 
        private const NumberStyles DoubleParseStyle =
            NumberStyles.AllowDecimalPoint | NumberStyles.AllowLeadingSign;
 
        private readonly CultureInfo _parseCulture = CultureInfo.InvariantCulture;
        private static readonly char[] RowSeparators = { '\r', '\n' };
        private readonly StringSplitter _rowSplitter = new StringSplitter(1000);
        private readonly StringSplitter _binarySplitter = new StringSplitter(4);
 
        public Encoding Encoding { get; }
 
        private readonly IProducerConsumerBuffer<BufferedBytes> _source;
 
        private string _leftOver = null;
 
        public BinaryYELTConverter(IProducerConsumerBuffer<BufferedBytes> source,
            int max_queue_size = 3, Encoding encoding = null)
            : base(max_queue_size)
        {
            Encoding = encoding ?? DEFAULT_ENCODING;
            _source = source;
        }
 
        protected override void OnStart()
        {
            base.OnStart();
            // Start the source producer if it hasn't already started.
            if (!_source.IsRunning)
                _source.Start(Cancellation.Token);
        }
 
        protected override bool IsProducerFinished() =>
            !_source.CanTake && _leftOver == null;
 
        protected override bool TryProduceNext(out BufferedBytes next)
        {
            BufferedBytes buffer = _source.TryTake(out bool success, Cancellation.Token);
            next = new BufferedBytes();
            // If we've reached the end of the source file, take note
            bool end_of_file = !_source.CanTake;
            if (!success && (!end_of_file || _leftOver == null))
                return false;
            // Start with any characters left over from the last byte array, then add new stuff
            string decoded = _leftOver;
            if (success)
                decoded += Encoding.GetString(buffer.Bytes, 0, buffer.LengthFilled);
            // Get the number of rows.
            int row_count = _rowSplitter.SafeSplit(decoded, RowSeparators, StringSplitOptions.RemoveEmptyEntries);
            string[] rows = _rowSplitter.buffer;
            // If we aren't at the end of file, ignore the last row, it's probably incomplete.
            _leftOver = end_of_file ? null : rows[row_count - 1];
            // Need to check if the last char was a newline, since split would ignore it
            if (decoded[decoded.Length - 1] == '\n' && !end_of_file)
                _leftOver += '\n';
            if (!end_of_file) row_count -= 1;
 
            // Allocate enough bytes to hold all but the last row (unless we're at the end)
            next.Bytes = new byte[row_count * 32];
            int byte_offset = 0, rows_skipped = 0, i = 0;
            try
            {
                // Parse the rows as quick as we can, with no validation
                for (i = 0; i < row_count; i++)
                {
                    if (rows[i].Length == 0)
                    {
                        rows_skipped++;
                        continue;
                    }
                    _binarySplitter.Split(rows[i], ',');
                    string[] items = _binarySplitter.buffer;
 
                    CopyAsBytes(next.Bytes, ref byte_offset,
                        UInt64.Parse(items[1], IntParseStyle, _parseCulture), //Trial ID
                        UInt64.Parse(items[0], IntParseStyle, _parseCulture), //Event ID
                        Double.Parse(items[2], DoubleParseStyle, _parseCulture), //Day / Sequence
                        Double.Parse(items[3], DoubleParseStyle, _parseCulture)); //Loss
                }
            }
            catch (Exception ex)
            {
                throw new ArgumentException("There was an error parsing the contents of " +
                    "the YELT row " + (i + 1) + " containing \"" + rows[i] +
                    "\". Expected \"UInt64,UInt64,Double,Double\". Error: " + ex.Message, ex);
            }
            // We always fill the whole byte array, minus allocated rows that were skipped.
            next.LengthFilled = next.Bytes.Length - (rows_skipped * 32);
            next.TotalBytesRead = buffer.TotalBytesRead;
            return true;
        }
 
        public static void CopyAsBytes(byte[] destination, ref int offset, ulong trialId, ulong eventId,
            double sequence, double loss)
        {
            BitConverter.GetBytes(trialId).CopyTo(destination, offset);
            offset += 8;
            BitConverter.GetBytes(eventId).CopyTo(destination, offset);
            offset += 8;
            BitConverter.GetBytes(sequence).CopyTo(destination, offset);
            offset += 8;
            BitConverter.GetBytes(loss).CopyTo(destination, offset);
            offset += 8;
        }
    }
}