spark/src/DataStreamSuite/DataStream/Primitives/Time/DateStandardizationStream.cs

using System;
using System.Collections.Generic;
using System.Text;


namespace BCG {
   /// <summary>
   /// If one adds annother value to this enum, be sure to update _date_conversion_to_type_map below as well
   /// </summary>
   public enum DateConversion {
      ConvertDoubleToDateTime = 0,
      ChopTime,
      UTCToCurrent,
      ConvertStringToDateTime,
      ConvertDateTimeToMilliseconds,
      SecondsOfDay,
      ISO8601,
      EpochToDatetime
   }

   [DataStreamUserRespresentation("dateStandardize")]
   public class DateStandardizationStream : ViewStream, IStreamMapper {
      private IBlueTableSchema mySchema;
      private DateConversion myConversion;
      private string myConversionColName;
      private readonly Dictionary<DateConversion, Type> _date_conversion_to_type_map = new Dictionary<DateConversion, Type>()
      {
         {DateConversion.ConvertDoubleToDateTime      , typeof(DateTime)},
         {DateConversion.ChopTime                     , typeof(DateTime)},
         {DateConversion.UTCToCurrent                 , typeof(DateTime)},
         {DateConversion.ConvertStringToDateTime      , typeof(DateTime)},
         {DateConversion.ConvertDateTimeToMilliseconds, typeof(Double)},
         {DateConversion.SecondsOfDay                 , typeof(Double)},
         {DateConversion.ISO8601                      , typeof(DateTime)},
         {DateConversion.EpochToDatetime              , typeof(DateTime)}
      };


      private static DateConversion _parse(string s) {
         DateConversion rv = DateConversion.ChopTime;
         Enum.TryParse<DateConversion>(s, false, out rv);
         return rv;
      }
      /// <summary>
      /// for DSQA
      /// </summary>
      public DateStandardizationStream(ViewStream vs, string colName, string conversion)
         : this(vs, colName, _parse(conversion)) { }

      public DateStandardizationStream(ViewStream vs, string colName, DateConversion conversion) {
         myConversionColName = colName;
         myConversion = conversion;
         mySetRealSource(new MapStream(vs, this));
      }

      #region IStreamMapper Members

      public IBlueTableSchema getOutputSchema(IBlueTableSchema inputSchema) {
         if (mySchema == null) {
            List<ColumnInfo> cis = new List<ColumnInfo>();
            foreach (ColumnInfo c in inputSchema.columnInfo) {
               if (c.colName == myConversionColName) {
                  Type convType;
                  if (!_date_conversion_to_type_map.TryGetValue(myConversion, out convType)) {
                     //BCG.Utilities.BlueEventLog.logEvent("Please update the _date_conversion_to_type_map in DateStandardizationStream.cs.", BCG.Utilities.BlueEventLevel.Error);
                     convType = typeof(DateTime);
                  }
                  cis.Add(new ColumnInfo(c.colName, c.colNo, convType));
               } else
                  cis.Add(new ColumnInfo(c.colName, c.colNo, c.colType));
            }
            mySchema = new BlueTableSchema(inputSchema.tableName, cis.ToArray());
         }
         return mySchema;
      }

      #endregion

      #region IDynamicStreamMapper Members

      public IBlueRow map(IBlueRow row) {
         MutableBlueRow rv = new MutableBlueRow(getOutputSchema(row.schema), row);
         rv.setValue(myConversionColName, mapDateTime(row[myConversionColName], myConversion));
         return rv;
         //return BlueDataUtility.getNewReplacedRow(row, mapDateTime(row[myConversionColName], myConversion), myConversionColName);
      }

      #endregion

      public static object mapDateTime(object obj_in, DateConversion conv) {
         if (obj_in == DBNull.Value || obj_in == null) {
            return DBNull.Value;
         } else {
            switch (conv) {
               case DateConversion.ChopTime: {
                     DateTime dtIn = (DateTime)obj_in;
                     return dtIn.Date;
                  }
               case DateConversion.ConvertDoubleToDateTime: {
                     double dtIn = (double)obj_in;
                     return DateTime.FromOADate(dtIn);
                  }
               case DateConversion.UTCToCurrent: {
                     DateTime dtIn = (DateTime)obj_in;
                     return dtIn.ToLocalTime();
                  }
               case DateConversion.ConvertStringToDateTime: {
                     return DateTime.Parse((string)obj_in);
                  }
               case DateConversion.ConvertDateTimeToMilliseconds: {
                     DateTime dtIn = (DateTime)obj_in;
                     return dtIn.TimeOfDay.TotalMilliseconds;
                  }
               case DateConversion.SecondsOfDay: {
                     DateTime dtIn = (DateTime)obj_in;
                     DateTime start_of_day = dtIn.Date;
                     TimeSpan day_period = dtIn - start_of_day;
                     return day_period.TotalSeconds;
                  }
               case DateConversion.ISO8601: {
                     string iso8601Date = obj_in.ToString();
                     return DateTime.Parse(iso8601Date, null, System.Globalization.DateTimeStyles.RoundtripKind);
                  }
               case DateConversion.EpochToDatetime: {
                     double d = Convert.ToDouble(obj_in);
                     long unix_time_seconds = (long)d;
                     DateTime epoch = new DateTime(1970, 1, 1, 0, 0, 0, DateTimeKind.Utc);
                     return epoch.AddSeconds(unix_time_seconds);
                  }
               default: {
                     bts.Log.Gen.info("incorrect type");
                     return DBNull.Value;
                  }
            }
         }
      }
   }
}