index.js

const {
  BaseKonnector,
  log,
  addData,
  hydrateAndFilter,
  cozyClient
} = require('cozy-konnector-libs')
const moment = require('moment')
const rp = require('request-promise')
require('moment-timezone')

moment.locale('fr') // set the language
moment.tz.setDefault('Europe/Paris') // set the timezone

/*** Connector Constants ***/
const startDailyDate = moment()
  .subtract(32, 'month')
  .format('YYYY-MM-DD')
const startLoadDate = moment()
  .subtract(7, 'day')
  .format('YYYY-MM-DD')
const endDate = moment().format('YYYY-MM-DD')
const baseUrl = 'https://gw.hml.api.enedis.fr'

/**
 * The start function is run by the BaseKonnector instance only when it got all the account
 * information (fields). When you run this connector yourself in "standalone" mode or "dev" mode,
 * the account information come from ./konnector-dev-config.json file
 * cozyParameters are static parameters, independents from the account. Most often, it can be a
 * secret api key.
 */
async function start(fields) {
  try {
    const { access_token } = fields
    const { oauth_callback_results } = fields
    const usage_point_id = oauth_callback_results.usage_point_id

    log('info', 'Fetching enedis daily data')
    const fetchedDailyData = await getDailyData(access_token, usage_point_id)
    log('info', 'Process enedis daily data')
    const processedDailyData = await processData(
      fetchedDailyData,
      'io.enedis.day',
      ['year', 'month', 'day']
    )
    log('info', 'Agregate enedis daily data for month and year')
    await agregateMonthAndYearData(processedDailyData)

    log('info', 'Fetching enedis load data')
    const fetchedLoadData = await getLoadData(access_token, usage_point_id)
    if (fetchedLoadData && fetchedLoadData.length > 0) {
      log('info', 'Process enedis load data')
      const processedLoadData = await processData(
        fetchedLoadData,
        'io.enedis.minute',
        ['year', 'month', 'day', 'hour', 'minute']
      )
      log('info', 'Agregate enedis load data for hour')
      await agregateHourlyData(processedLoadData)
    } else {
      log('info', 'No consent or data for load curve')
    }
  } catch (err) {
    log('error', err.message)
  }
}

/**
 * Retrieve data from the API
 * Format: { value: "Wh", "date": "YYYY-MM-DD" }
 */
async function getDailyData(token, usagePointID) {
  const dataRequest = {
    method: 'GET',
    uri:
      baseUrl +
      '/v4/metering_data/daily_consumption?start=' +
      startDailyDate +
      '&end=' +
      endDate +
      '&usage_point_id=' +
      usagePointID,
    headers: {
      Accept: 'application/json',
      Authorization: 'Bearer ' + token
    }
  }
  try {
    const response = await rp(dataRequest)
    return response
  } catch (error) {
    throw error
  }
}

/**
 * Retrieve data from the API
 * Format: { value: "W", "date": "YYYY-MM-DD hh:mm:ss" }
 */
async function getLoadData(token, usagePointID) {
  const dataRequest = {
    method: 'GET',
    uri:
      baseUrl +
      '/v4/metering_data/consumption_load_curve?start=' +
      startLoadDate +
      '&end=' +
      endDate +
      '&usage_point_id=' +
      usagePointID,
    headers: {
      Accept: 'application/json',
      Authorization: 'Bearer ' + token
    }
  }
  try {
    const response = await rp(dataRequest)
    return response
  } catch (error) {
    throw error
  }
}

/**
 * Parse data
 * Remove existing data from DB using hydrateAndFilter
 * Store filtered data
 * Return the list of filtered data
 */
async function processData(data, doctype, filterKeys) {
  const parsedData = JSON.parse(data)
  const intervalData = parsedData.meter_reading.interval_reading
  const formatedData = await formateData(intervalData, doctype)
  // Remove data for existing days into the DB
  const filteredData = await hydrateAndFilter(formatedData, doctype, {
    keys: filterKeys
  })
  // Store new day data
  await storeData(filteredData, doctype, filterKeys)
  return filteredData
}

/**
 * Agregate data from daily data to monthly and yearly data
 */
async function agregateMonthAndYearData(data) {
  // Sum year and month values into object with year or year-month as keys
  if (data && data.length > 0) {
    let monthData = {}
    let yearData = {}
    data.forEach(element => {
      element.year + '-' + element.month in monthData
        ? (monthData[element.year + '-' + element.month] += element.load)
        : (monthData[element.year + '-' + element.month] = element.load)
      element.year in yearData
        ? (yearData[element.year] += element.load)
        : (yearData[element.year] = element.load)
    })
    // Agregation for Month data
    const agregatedMonthData = await buildAgregatedData(
      monthData,
      'io.enedis.month'
    )
    await storeData(agregatedMonthData, 'io.enedis.month', ['year', 'month'])
    // Agregation for Year data
    const agregatedYearData = await buildAgregatedData(
      yearData,
      'io.enedis.year'
    )
    await storeData(agregatedYearData, 'io.enedis.year', ['year'])
  }
}

/**
 * Agregate data from load data (every 30 min) to Hourly data
 */
async function agregateHourlyData(data) {
  // Sum year and month values into object with year or year-month as keys
  if (data && data.length > 0) {
    let hourData = {}
    data.forEach(element => {
      let key =
        element.year +
        '-' +
        element.month +
        '-' +
        element.day +
        '-' +
        element.hour
      key in hourData
        ? (hourData[key] += element.load)
        : (hourData[key] = element.load)
    })
    // Agregation for Month data
    const agregatedMonthData = await buildAgregatedData(
      hourData,
      'io.enedis.hour'
    )
    await storeData(agregatedMonthData, 'io.enedis.hour', [
      'year',
      'month',
      'day',
      'hour'
    ])
  }
}

/**
 * Save data in the right doctype db and prevent duplicated keys
 */
async function storeData(data, doctype, filterKeys) {
  log('debug', doctype, 'Store into')
  const filteredDocuments = await hydrateAndFilter(data, doctype, {
    keys: filterKeys
  })
  return await addData(filteredDocuments, doctype)
}

/**
 * Format data for DB storage
 * Remove bad data
 */
async function formateData(data, doctype) {
  log('info', 'Formating data')
  return data.map(record => {
    const date = moment(record.date, 'YYYY/MM/DD h:mm:ss')
    if (record.value != -2) {
      const load =
        doctype === 'io.enedis.minute' ? record.value / 2 : record.value
      return {
        load: parseFloat(load / 1000),
        year: parseInt(date.format('YYYY')),
        month: parseInt(date.format('M')),
        day: parseInt(date.format('D')),
        hour: parseInt(date.format('H')),
        minute: parseInt(date.format('m'))
      }
    }
  })
}

/**
 * Retrieve and remove old data for a specific doctype
 * Return an Array of agregated data
 */
async function buildAgregatedData(data, doctype) {
  let agregatedData = []
  for (let [key, value] of Object.entries(data)) {
    const data = await buildDataFromKey(doctype, key, value)
    const oldValue = await resetInProgressAggregatedData(data, doctype)
    data.load += oldValue
    agregatedData.push(data)
  }
  return agregatedData
}

/**
 * Format an entry for DB storage
 * using key and value
 * For year doctype: key = "YYYY"
 * For month doctype: key = "YYYY-MM"
 */
async function buildDataFromKey(doctype, key, value) {
  let year, month, day, hour
  if (doctype === 'io.enedis.year') {
    year = key
    month = 1
    day = 0
    hour = 0
  } else if (doctype === 'io.enedis.month') {
    const split = key.split('-')
    year = split[0]
    month = split[1]
    day = 0
    hour = 0
  } else {
    const split = key.split('-')
    year = split[0]
    month = split[1]
    day = split[2]
    hour = split[3]
  }
  return {
    load: Math.round(value * 10000) / 10000,
    year: parseInt(year),
    month: parseInt(month),
    day: parseInt(day),
    hour: parseInt(hour),
    minute: 0
  }
}

/**
 * Function handling special case.
 * The temporary aggregated data need to be remove in order for the most recent one te be saved.
 * ex for io.enedis.year :
 * { load: 76.712, year: 2020, ... } need to be replace by
 * { load: 82.212, year: 2020, ... } after enedis data reprocess
 */
async function resetInProgressAggregatedData(data, doctype) {
  // /!\ Warning: cannot use mongo queries because not supported for dev by cozy-konnectors-libs
  log('debug', doctype, 'Remove aggregated data for')
  const result = await cozyClient.data.findAll(doctype)
  if (result && result.length > 0) {
    // Filter data to remove
    var filtered = []
    if (doctype === 'io.enedis.year') {
      // Yearly case
      filtered = result.filter(function(el) {
        return el.year == data.year
      })
    } else if (doctype === 'io.enedis.month') {
      // Monthly case
      filtered = result.filter(function(el) {
        return el.year == data.year && el.month == data.month
      })
    } else {
      // Hourly case
      filtered = result.filter(function(el) {
        return (
          el.year == data.year &&
          el.month == data.month &&
          el.day == data.day &&
          el.hour == data.hour
        )
      })
    }
    // Remove data
    let sum = 0.0
    for (const doc of filtered) {
      sum += doc.load
      log('debug', doc, 'Removing this entry for ' + doctype)
      await cozyClient.data.delete(doctype, doc)
    }
    return sum
  }
  return 0.0
}

module.exports = new BaseKonnector(start)