var redisLib = require('redis');
var sprintf = require('sprintf').vsprintf;
var request = require('request');
var fs = require('fs');
var spawn = require('child_process').spawn;
var readline = require('readline');
var async = require('async');
var _ = require('underscore');
var MongoClient = require('mongodb').MongoClient;

var url = require('url');
var redisUrl = new URL(process.env.REDIS_URL || 'redis://127.0.0.1:6379');

var redis = redisLib.createClient(redisUrl.port, redisUrl.hostname);
redis.auth(redisUrl.password);

var UPDATE_INTERVAL = 6 * 60 * 60 * 1000; // ms between source updates
var UPDATE_OFFSET = 4 * 60 * 60 * 1000; // ms delay before starting update
var BASE_URL = 'http://nomads.ncep.noaa.gov/pub/data/nccf/com/gfs/prod/';
var HOURS = [
  0, 3, 6, 9, 12, 15, 18, 21,
  24, 27, 30, 33, 36, 39, 42, 45,
  48, 51, 54, 57, 60, 63, 66, 69,
  72, 75, 78, 81, 84, 87, 90, 93,
  96, 99, 102, 105, 108, 111, 114, 117,
  120, 123, 126, 129, 132, 135, 138, 141,
  144, 147, 150, 153, 156, 159, 162, 165,
  168, 171, 174, 177, 180, 183, 186, 189,
  192, 204,
  216, 228,
  240, 252,
  264, 276,
  288, 300,
  312, 324,
  336, 348,
  360, 372,
  384
];
var LEVELS = [
  1, 2, 3, 5, 7,
  125, 175, 225, 275, 325,
  375, 425, 475, 525, 575,
  625, 675, 725, 775, 825,
  875
];

var MongoUrl = process.env.MONGODB_URI || 'mongodb://localhost:27017/stratocast';

/**
 * Generate the URL for the specified forecast.
 *
 * @param  {Number} analysisTimestamp Timestamp the analysis was performed
 * @param  {Integer} forecastHours Number of hours out from the analysis
 * @return {String} The URL of the sounding
 */
var dataUrl = function(analysisTimestamp, forecastHours) {
  var analysisTime = new Date(analysisTimestamp);
  // Example folder name
  // /gfs.20130524/00/
  var folder = sprintf(
    'gfs.%04d%02d%02d/%02d/', [
      analysisTime.getUTCFullYear(),
      analysisTime.getUTCMonth() + 1,
      analysisTime.getUTCDate(),
      analysisTime.getUTCHours()
    ]);

  // Example file name
  // gfs.t18z.pgrb2b.1p00.f384
  var file = sprintf(
    'gfs.t%02dz.pgrb2b.1p00.f%03d', [
      analysisTime.getUTCHours(),
      forecastHours
    ]);

  return BASE_URL + folder + file;
};

/**
 * Get the expected timestamp of the latest sounding.
 *
 * @return {Number} Unix timestamp
 */
latestTimestamp = function() {
  var now = Date.now();
  return Math.floor(now / UPDATE_INTERVAL) * UPDATE_INTERVAL;
};

/**
 * Get the expected timestamp of the next sounding
 *
 * @return {Number} Unix timestamp
 */
nextTimestamp = function() {
  return latestTimestamp() + UPDATE_INTERVAL;
};

/**
 * Make an error handling function for the wgrib2 parser.
 * 
 * @param  {Readline}   rl   Readline instance to close after error
 * @param  {Function} next Callback(error)
 */
make_errorHandler = function(rl, next) {
  return function(data) {
    rl.close();
    setImmediate(function() {next(data);});
  };
};

/**
 * Make a stream closed handling function for the wgrib2 parser.
 * 
 * @param  {Readline}   rl     Readline instance to close after error
 * @param  {Object}   parsed Object the parsed data is stored in
 * @param  {Function} next   Callback(error, data)
 */
make_closeHandler = function(rl, parsed, next) {
  return function(code) {
    rl.close();
    if(code !== 0) {
      setImmediate(function() {next('Error ' + code);});
    } else {
      next(null, parsed);
    }
  };
};

make_dataHandler = function(parsed) {
  var obs = {};
  return function(data) {
    var tokens = data.split(':');
    var line = parseInt(tokens[0], 10);
    var val = parseFloat(tokens[2].split('=', 10)[1]);
    var level = Math.floor((line - 1) / 4);
    var item = (line - 1) % 4;
    
    if(level < 25) {
      switch(item) {
        case 0: // Height
          obs.altitude = val;
          break;
        case 1: // Temp
          obs.temperature = val;
          break;
        case 2: // UGRD
          obs.windU = val;
          break;
        case 3: // VGRD
          obs.windV = val;
          obs.pressure = LEVELS[level] * 100;
          parsed.push(obs);
          obs = {};
      }
    } else {
      switch(item) {
        case 0: // Temp
          obs.temperature = val;
          break;
        case 1: // UGRD
          obs.windU = val;
          break;
        case 2: // VGRD
          obs.windV = val;
          break;
        case 3: // Height
          obs.altitude = val;
          obs.pressure = LEVELS[level] * 100;
          parsed.push(obs);
          obs = {};
      }
    }
  };
};

/**
 * Put a sounding into the cache
 *
 * Passes any error as the first parameter to the callback.
 * Null error indicates success.
 * 
 * @param  {Date}   time     UTC timestamp of the sounding
 * @param  {Number}   lat      Latitude of the sounding
 * @param  {Number}   lng      Longitude of the sounding
 * @param  {Object}   sounding Sounding object
 * @param  {Function} next     Callback(err)
 */
putCache = function(time, lat, lng, sounding, next) {

    MongoClient.connect(MongoUrl, function(err, db) {

        db.collection('soundingCache').update({time: time, loc:[lng, lat]},
                { $set: { sounding } },
                {upsert: true},
                function(err, res) {
                    next(err);
        });

        db.close();
    });
};

/**
 * Get a sounding from the cache
 *
 * Passes any error as the first parameter to the callback.
 * Sounding object is passed as the second parameter.
 * 
 * @param  {Date}   time UTC timestamp of the sounding
 * @param  {Number}   lat  Latitude of the sounding
 * @param  {Number}   lng  Longitude of the sounding
 * @param  {Function} next Callback(err, sounding)
 */
getCache = function(time, lat, lng, next) {

    MongoClient.connect(MongoUrl, function(err, db) {

        db.collection('soundingCache').findOne({
            time: time,
            loc:[lng, lat]
            },
            function(err, doc){
                if(err || !doc) {
                    next(err);
                } else {
                    next(null, doc.sounding);
                }
        });

        db.close();
    });
};

/**
 * Remove all soundings for the given time from the cache
 *
 * Passes any error as the first parameter to the callback.
 * 
 * @param  {Date}   time UTC timestamp to be removed
 * @param  {Function} next Callback(err)
 */
clearCache = function(time, next) {

    MongoClient.connect(MongoUrl, function(err, db) {

        if(err) {
            console.log("MongoDB error in clearCache: " + err);
        } else {
            db.collection('soundingCache').remove({
                time: time
            });

            db.close();
        }
    });
    next();
};

/**
 * Extract a sounding from a compressed data file
 *
 * Passes any error as the first parameter to the callback.
 * The sounding object is passed as the second parameter if successful.
 * 
 * @param  {Date}   time UTC time of the sounding
 * @param  {Number}   lat  Latitude of the sounding
 * @param  {Number}   lng  Longitude of the sounding
 * @param  {Function} next Callback(error, sounding)
 */
extractSounding = function(time, lat, lng, next) {

  // Get filename
  getFilename(time, function(err, fileName) {
    if(err) {
      next(err);
    } else {
      // Convert lat/lng to i/j
      var i = lng + 1;
      var j = lat + 91;

      var parsed = [];
      var parser = spawn('grib2/wgrib2/wgrib2', [
        fileName,
        '-ij',
        i,
        j
      ]);
      var rl = readline.createInterface({
        input: parser.stdout,
        output: parser.stdin
      });
      parser.stderr.on('data', make_errorHandler(rl, next));
      parser.on('close', make_closeHandler(rl, parsed, next));
      rl.on('line', make_dataHandler(parsed));
    }
  });
};

/**
 * Get the name of the file representing a sounding for the provided time
 * 
 * @param  {Date}   time Date of the sounding
 * @param  {Function} next Callback(err, fileName)
 */
getFilename = function(time, next) {

    MongoClient.connect(MongoUrl, function(err, db) {

        db.collection('soundingFiles').findOne({time: time}, function(localerr, res) {
            if(localerr || !res) {
                next(localerr);
            } else {
                next(null, res.record.fileName);
            }
        });

        db.close();
    });
};

/**
 * Set the name of the file representing a sounding for the provided time
 * 
 * @param {Date}   time     Date of the sounding
 * @param {String}   fileName Name of the file
 */
setFilename = function(time, fileName, next) {

  var record = {
    time: time,
    fileName: fileName
  };

  MongoClient.connect(MongoUrl, function(err, db) {

        db.collection('soundingFiles').update({
            time: time
        },
        {
            $set: { record }
        },
        {
            upsert: true
        },
            function(err, res) {
                next(err);
        });

        db.close();
    });
};

/**
 * Get a sounding from the cache or from the file store.
 * 
 * @param  {Date}   time UTC time of the sounding
 * @param  {Number}   lat  Latitude of the sounding
 * @param  {Number}   lng  Longitude of the sounding
 * @param  {Function} next Callback(err, sounding)
 */
exports.getSounding = function(time, lat, lng, next) {
  // Round inputs to nearest expected value
  var nearest = Math.round(time.getTime() / UPDATE_INTERVAL) * UPDATE_INTERVAL;
  time = new Date(nearest);
  lat = Math.round(lat);
  lng = Math.round(lng);
  if(lng < 0) lng += 360;
  
  // Check cache
  getCache(time, lat, lng, function(err, res) {
    if(!err && res) { // Cache hit
      next(null, res);
    } else { // Cache miss
      // Extract sounding
      extractSounding(time, lat, lng, function(err, res) {
        if(err) {
          next(err);
        } else {
          var sounding = {time: time, loc: [lng, lat]};
          sounding.data = res;
          // Add to cache
          putCache(time, lat, lng, sounding, function(err) {
            if(err) {
              next(err);
            } else {
              // Return sounding
              next(null, sounding);
            }
          });
        }
      });
    }
  });
};

/**
 * Schedule a download for the provided analysis time.
 *
 * @param  {Number} analysisTime Unix timestamp
 */
scheduleFetch = function(analysisTime) {
  // Compute the delay time
  var delay = analysisTime + UPDATE_OFFSET - Date.now();

  // Print the status update
  console.log('Scheduling ' + (new Date(analysisTime)).toUTCString() + ' in ' + delay / 1000 + ' seconds');
  // Schedule the download
  setTimeout(function() {
    startFetch(analysisTime);
    // Schedule the next fetch
    if (analysisTime != nextTimestamp()) scheduleFetch(nextTimestamp());
  }, delay);
};

/**
 * Start the fetch process after triggered
 *
 * @param  {Number} analysisTime Unix timestamp
 */
startFetch = function(analysisTime) {
  // Print the status update
  console.log('Queing ' + (new Date(analysisTime)).toUTCString());
  // Remove any stragglers from previous fetch
  clearQueue();
  // Add soundings to download queue
  for (var i in HOURS) {
    redis.rpush('soundingQueue', analysisTime + ':' + HOURS[i]);
  }
};

/**
 * Clear the download queue
 */
clearQueue = function() {
  redis.del('soundingQueue');
};

/**
 * Fetch a key off the download queue if available
 *
 * Callback returns any error and first parameter and reply as second.
 *
 * @param  {Function} next Callback function
 */
exports.popQueue = function(next) {
  redis.lpop('soundingQueue', function(err, reply) {
    if (err) {
      next(err, null);
    } else {
      next(null, reply);
    }
  });
};

/**
 * Preprocess a grib file for later use
 *
 * Callback returns any error as first paramter and returns the
 * new filename as the second parameter.
 *
 * @param  {String}   fileName File name to be processed
 * @param  {Function} next     Callback function
 */
exports.preprocessSounding = function(fileName, next) {

  var name = fileName.split('/')[1];
  var time = new Date(parseInt(name.split('.')[0], 10));
  var newfile = fileName + '.small';
  var processor = spawn('grib2/wgrib2/wgrib2', [
    fileName,
    '-match',
    'HGT|TMP|UGRD|VGRD',
    '-match',
    'mb:',
    '-set_grib_type',
    'c1',
    '-grib_out',
    newfile
  ]);
  processor.on('close', function(code) {
    try {
      fs.unlinkSync(fileName);
    } catch (e) {
      // Don't care
    }
    if (code !== 0) {
      next(code, null);
    } else {
      clearCache(time, function(err) {
        setFilename(time, newfile, function(err) {
          if(err) {
            next(err);
          } else {
            next(null, newfile);
          }
        });
      });
    }
  });
};

/**
 * Download the sounding specified by the provided key
 *
 * The callback passes any error as the first parameter and the
 * filename of the downloaded file as the second parameter.
 *
 * @param  {String}   key  Key from download queue
 * @param  {Function} next Callback called on completion of download
 */
exports.downloadSounding = function(key, next) {
  // Split the key
  var tokens = key.split(':');
  // Parse the key fields
  var analysisTime = parseInt(tokens[0], 10);
  var forecastHours = parseInt(tokens[1], 10);
  // Get url and filename
  var url = dataUrl(analysisTime, forecastHours);
  var fileName = 'data/' + (analysisTime + (forecastHours * 60*60*1000)) + '.grib2';
  // Download the sounding
  request(url, function(err, res) {
    if (err) {
      next(err, null);
      return;
    }
    if (res.statusCode !== 200) {
      next(res.statusCode, null);
      try {
        fs.unlinkSync(fileName);
      } catch (e) {
        console.log('Could not delete ' + fileName);
      }
      return;
    }
    // Call the provided callback
    next(null, fileName);
  }).pipe(fs.createWriteStream(fileName));
};

/**
 * Put a failed key back in the queue
 * 
 * @param  {String} key Key to be pushed
 */
exports.requeue = function(key) {
  redis.rpush('soundingQueue', key);
};

/**
 * Initialize the data-store
 */
exports.init = function() {
  // Initialize the scheduler
  // TODO test for latest sounding
  clearQueue();
  scheduleFetch(latestTimestamp());
};
