Convert Green Button power data to InfluxDB LineProtocol

runme.js

// @ts-check
/* eslint-disable no-console */
import path from 'path';
import {readFile} from 'fs/promises';
import {createWriteStream} from 'fs';
import {argv, exit, cwd} from 'process';

let config;
let getPrice;

try {
	config = JSON.parse(await readFile('./config.json', 'utf8'));
	if (config.priceModule) {
		({getPrice} = await import(config.priceModule));
	}
} catch {}

let [, script, sourceFile, destinationFile] = argv;

function help() {
	console.error('Usage: %s source [destination]', path.relative(cwd(), script));
	exit(1);
}

if (!sourceFile) {
	console.error('Convert "Green Button Download" to InfluxDB Line Protocol');
	help();
}

destinationFile ??= sourceFile.endsWith('.csv') ? sourceFile.replace('.csv', '.lp') : `${sourceFile}.lp`;

/** @type {string} */
// @ts-expect-error
const sourceContents = await readFile(sourceFile, 'utf8').then(file => {
	if (!file) {
		throw new Error('No data');
	}

	return file;
}).catch(error => {
	if (error.code === 'EISDIR') {
		console.error('source is a directory');
		help();
	}

	if (error.code === 'ENOENT') {
		console.error('source does not exist');
		help();
	}

	console.error('Unable to read source:');
	console.error(error);
	help();
});

function easySplit(line) {
	const tokens = line.split(',');
	const response = [];
	for (let token of tokens) {
		if (token.startsWith('"')) {
			if (!token.endsWith('"')) {
				throw new Error('Special parsing not implemented');
			}

			token = token.slice(1, -1);
		}

		response.push(token);
	}

	return response;
}

function parseTime(time) {
	if (time.includes('AM')) {
		const [timeToken] = time.split(' ');
		let [h, m] = timeToken.split(':');
		if (h === '12') {
			h = 0;
		}

		return {h, m};
	}

	if (time.includes('PM')) {
		const [primaryToken] = time.split(' ');
		const [h, m] = primaryToken.split(':');
		return {h: String(Number(h) + 12), m};
	}

	throw new Error('Unable to parse time');
}

function getStream() {
	return new Promise((resolve, reject) => {
		const response = createWriteStream(destinationFile, {flags: 'w'});
		response.on('open', () => {
			resolve(response);
		});
		response.on('error', reject);
	});
}

const stream = await getStream().catch(error => {
	console.error('An unexpected error ocurred');
	console.error(error);
	exit(1);
});

/** @type {string[][]} */
const parsedContents = sourceContents.split(/\r?\n/)
	// Dependency free, very lazy way to parse a csv
	.map(line => easySplit(line))
	// Filter out key-value information
	.filter(columns => columns.length > 2);

if (parsedContents.length < 2) {
	console.error('Could not find at least 2 usable rows in the file');
	exit(1);
}

/** @type {string[]} */
// @ts-expect-error
const columns = parsedContents.shift().map(label => label.toLowerCase());
const dateColumn = columns.findIndex(column => column.toLowerCase().includes('date'));
const timeColumn = columns.findIndex(column => column.toLowerCase().includes('time'));
const meterColumn = columns.findIndex(column => column.toLowerCase().includes('meter'));
const ignoredIndexes = new Set([dateColumn, timeColumn, meterColumn]);

if (dateColumn === -1 || timeColumn === -1 || meterColumn === -1) {
	console.error('Unable to find date, time, or meter column');
	exit(1);
}

let first;
let last;

for (const row of parsedContents) {
	const date = new Date(`${row[dateColumn]}`);
	const {h, m} = parseTime(row[timeColumn]);
	date.setHours(h, m);
	date.setMinutes(date.getMinutes() - date.getTimezoneOffset());
	let formattedLine = `usage,meter=${row[meterColumn]} `;

	for (const [index, value] of row.entries()) {
		if (!ignoredIndexes.has(index)) {
			// The space at the end is required!
			formattedLine += `${columns[index]}=${value},`;
		}
	}

	if (getPrice) {
		const [firstTierPrice, secondTierPrice] = getPrice(date); 
		formattedLine += `first_tier_price=${firstTierPrice},second_tier_price=${secondTierPrice},`;
	}

	// Date.valueOf() returns a time in milliseconds. Add 6 0's to the end to convert to ns
	formattedLine = formattedLine.slice(0, -1) + ' ' + date.getTime();
	if (!first) {
		first = date;
	}

	last = date;
	stream.write(`${formattedLine}\n`);
}

console.log('Wrote to', destinationFile);
console.log('First point:', first);
console.log('Last point:', last);

console.log('To write to influx:');
console.log(`influx write --org %s --bucket %s --format lp -p ms -f ${destinationFile}`, config.org ?? '{org}', config.bucket ?? 'power');