graphy-dev/main.js

#!/usr/bin/env node



	// ttl package
const ttl = {
		// read ttl output
	get read() {
			// memoize
		delete ttl.read;
		return ttl.read = require('@graphy-dev/content.ttl.read');  // eslint-disable-line global-require
	},
	// scan ttl output
	get scan() {
			// memoize
		delete ttl.scan;
		return ttl.scan = require('@graphy-dev/content.ttl.scan');  // eslint-disable-line global-require
	},
	// write ttl input
	get write() {
			// memoize
		delete ttl.write;
		return ttl.write = require('@graphy-dev/content.ttl.write');  // eslint-disable-line global-require
	},
};
	// trig package
const trig = {
		// read trig output
	get read() {
			// memoize
		delete trig.read;
		return trig.read = require('@graphy-dev/content.trig.read');  // eslint-disable-line global-require
	},
	// scan trig output
	get scan() {
			// memoize
		delete trig.scan;
		return trig.scan = require('@graphy-dev/content.trig.scan');  // eslint-disable-line global-require
	},
	// write trig input
	get write() {
			// memoize
		delete trig.write;
		return trig.write = require('@graphy-dev/content.trig.write');  // eslint-disable-line global-require
	},
};
	// nt package
const nt = {
		// read nt output
	get read() {
			// memoize
		delete nt.read;
		return nt.read = require('@graphy-dev/content.nt.read');  // eslint-disable-line global-require
	},
	// scan nt output
	get scan() {
			// memoize
		delete nt.scan;
		return nt.scan = require('@graphy-dev/content.nt.scan');  // eslint-disable-line global-require
	},
	// write nt input
	get write() {
			// memoize
		delete nt.write;
		return nt.write = require('@graphy-dev/content.nt.write');  // eslint-disable-line global-require
	},
};
	// nq package
const nq = {
		// read nq output
	get read() {
			// memoize
		delete nq.read;
		return nq.read = require('@graphy-dev/content.nq.read');  // eslint-disable-line global-require
	},
	// scan nq output
	get scan() {
			// memoize
		delete nq.scan;
		return nq.scan = require('@graphy-dev/content.nq.scan');  // eslint-disable-line global-require
	},
	// write nq input
	get write() {
			// memoize
		delete nq.write;
		return nq.write = require('@graphy-dev/content.nq.write');  // eslint-disable-line global-require
	},
};


// // SPARQL Results package
// const sparql_results = {
// 	// deserialize sparql_results input
// 	get deserializer() {
// 		// memoize
// 		delete sparql_results.deserializer;
// 		return (sparql_results.deserializer = require('../sparql-results/deserializer.js'));
// 	},
// };


const H_CONTENT_MIMES = {
	'text/turtle': ttl,
	'application/trig': trig,
	'application/n-triples': nt,
	'application/n-quads': nq,
	// 'application/sparql-results+json': sparql_results,
};

const H_CONTENT_TAGS = {
	ttl,
	trig,
	nt,
	nq,
	// 'application/sparql-results+json': sparql_results,
};



const R_CONTENT_TYPE = /^((?:application|text)\/[^\0-\x20()<>@,;:\\"\/[\]?.=]+)(;.+)*$/i;

const graphy = module.exports = Object.assign({

	content: Object.assign((s_query) => {
		let m_content_type = R_CONTENT_TYPE.exec(s_query);
		if(!m_content_type) throw new Error(`invalid content-type string: "${s_query}"`);
		let [, s_content_type, s_parameters] = m_content_type;
		let s_content_type_normal = s_content_type.toLowerCase();

		if(s_content_type_normal in H_CONTENT_TAGS) {
			return H_CONTENT_TAGS[s_content_type_normal];
		}
		else if(s_content_type_normal in H_CONTENT_MIMES) {
			return H_CONTENT_MIMES[s_content_type_normal];
		}
		else {
			throw new Error(`no content handlers matched query for "${s_content_type_normal}"`);
		}
	}, {
		ttl,
		trig,
		nt,
		nq,

	}),

	util: {
		dataset: {
			get tree() {
	// memoize
				delete graphy.util.dataset.tree;
				return (graphy.util.dataset.tree = require('@graphy-dev/util.dataset.tree'));
			},
		},
	},


}, require('@graphy-dev/core.data.factory'));


// export graphy to window object if in main thread of browser
if('undefined' !== typeof window) window.graphy = graphy;

// cli
if(module === require.main) {
	const fs =require('fs');
	const mk_yargs = require('yargs/yargs');

	const exit = (s_exit) => {
		console.error(s_exit);
		process.exit(1);
	};

	const command = s_command => mk_yargs()
		.strict()
		.usage(`Usage: $0 ${s_command} [OPTIONS] [--pipe COMMAND]`);

	const reader = f_reader => (a_args, g_context) => new Promise((fk_resolve) => {
		let g_argv = command(g_context.command)
			.boolean('v')
			.alias('v', 'validate')
			.describe('v', 'validate all tokens within the RDF document')
			.string('b')
			.alias('b', ['base', 'base-uri'])
			.describe('b', 'set a base URI on the document')
			.help()
			.version(false)
			.parse(a_args);

		let gc_read = {
			validate: g_argv.validate || false,
		};

		if(g_argv['base-uri']) {
			gc_read.baseUri = g_argv['base-uri'];
		}

		fk_resolve(g_context.inputs.map((ds_input) => {
			let ds_reader = f_reader({...gc_read});
			return ds_input.pipe(ds_reader);
		}));
	});

	const writer = f_writer => (a_args, g_context) => {
		let g_argv = command(g_context.command)
			.help()
			.parse(a_args);

		let gc_write = {};

		return g_context.inputs.map((ds_input) => {
			let ds_writer = f_writer({...gc_write});
			return ds_input.pipe(ds_writer);
		});
	};

	// commands
	let h_commands = {
		'content.nt.read': reader(graphy.content.nt.read),
		'content.nq.read': reader(graphy.content.nq.read),
		'content.ttl.read': reader(graphy.content.ttl.read),
		'content.trig.read': reader(graphy.content.trig.read),

		'content.nt.write': writer(graphy.content.nt.write),
		'content.nq.write': writer(graphy.content.nq.write),
		'content.ttl.write': writer(graphy.content.ttl.write),
		'content.trig.write': writer(graphy.content.trig.write),

		'util.dataset.tree': async(a_args, g_context) => {
			const dataset_tree = graphy.util.dataset.tree;

			let a_dual_input_keys = ['u', 'i', 'd', 'm'];

			let s_group_multi_input = 'With multiple inputs:';
			let s_group_dual_input = 'With exactly two inputs:';

			let g_argv = command(g_context.command)
				.options({
					u: {
						type: 'boolean',
						alias: 'union',
						group: s_group_multi_input,
						describe: 'perform the union of multiple inputs',
						conflicts: a_dual_input_keys.filter(s => 'u' !== s),
					},
					i: {
						type: 'boolean',
						alias: ['intersect', 'intersection'],
						group: s_group_multi_input,
						describe: 'perform the intersection of multiple inputs',
						conflicts: a_dual_input_keys.filter(s => 'i' !== s),
					},
					d: {
						type: 'boolean',
						alias: ['diff', 'difference'],
						group: s_group_dual_input,
						describe: 'perform a difference between two inputs',
						conflicts: a_dual_input_keys.filter(s => 'd' !== s),
					},
					m: {
						type: 'boolean',
						alias: ['minus', 'subtract', 'subtraction'],
						group: s_group_dual_input,
						describe: 'perform a subtraction by removing input-B from input-A',
						conflicts: a_dual_input_keys.filter(s => 'm' !== s),
					},
				})
				.help()
				.version(false)
				.parse(a_args);

			// ref inputs; cache length
			let a_inputs = g_context.inputs;
			let n_inputs = a_inputs.length;

			// multi-input stream-output operation
			if(g_argv.union || g_argv.intersection) {
				let s_operation = g_argv.union
					? 'union'
					: 'intersection';

				// less than 2 inputs; no-op
				if(n_inputs < 2) return a_inputs;

				// create trees
				let a_trees = a_inputs.map(() => dataset_tree());

				// initial tree
				let k_tree_out = a_trees[0];

				// pairwise readiness
				for(let i_input=0; i_input<n_inputs; i_input++) {
					let k_tree_b = a_trees[i_input];

					// pipe input stream to tree b
					a_inputs[i_input].pipe(k_tree_b);

					// wait for input stream to finish writing to b
					await k_tree_b.until('finish');

					// non-first input
					if(i_input) {
						// perform pairwise operation
						k_tree_out = k_tree_out[s_operation](k_tree_b);
					}
				}

				// return readable tree
				return [k_tree_out];
			}
			// dual-input stream-output operation
			else if(g_argv.difference || g_argv.subtraction) {
				let s_operation = g_argv.difference
					? 'difference'
					: 'minus';

				// not two inputs
				if(2 !== n_inputs) {
					exit(`dual input operation expects two inputs but found ${n_inputs}`);
				}

				// async
				return new Promise((fk_resolve) => {
					let operate = () => [k_tree_a[s_operation](k_tree_b)];

					// wait for a
					let k_tree_a = dataset_tree();
					let b_finished_a = false;
					k_tree_a.on('finish', () => {
						b_finished_a = true;
						if(b_finished_b) fk_resolve(operate());
					});

					// wait for b
					let k_tree_b = dataset_tree();
					let b_finished_b = false;
					k_tree_b.on('finish', () => {
						b_finished_b = true;
						if(b_finished_a) fk_resolve(operate());
					});

					// ref both input streams
					let [ds_input_a, ds_input_b] = a_inputs;

					// pipe each to its tree
					ds_input_a.pipe(k_tree_a);
					ds_input_b.pipe(k_tree_b);
				});
			}
			// reduction; n-to-n
			else {
				return g_context.inputs.map(ds_input => ds_input.pipe(dataset_tree()));
			}
		},
	};

	let a_argv = process.argv.slice(2);
	let n_args = a_argv.length;

	// inputs
	let a_inputs = [];
	{
		for(let i_argv=0; i_argv<n_args; i_argv++) {
			let s_arg = a_argv[i_argv];

			// input
			if(s_arg.startsWith('--input')) {
				let m_input = /^--input=(.+)$/.exec(s_arg);
				if(!m_input) {
					exit(`invalid input argument format: ${s_arg}`);
				}
				a_argv.shift();
				i_argv -= 1;
				n_args -= 1;
				a_inputs.push(fs.createReadStream(m_input[1]));
				continue;
			}

			break;
		}
	}

	// pipeline
	let a_pipeline = [];
	{
		let a_series = [a_argv[0]];

		for(let i_argv=1; i_argv<n_args; i_argv++) {
			let s_arg = a_argv[i_argv];

			if('--pipe' === s_arg) {
				a_pipeline.push(a_series);
				if(i_argv === n_args) {
					exit(`was expecting pipe destination after --pipe: ${a_argv}`);
				}
				a_series = [];
				continue;
			}

			a_series.push(s_arg);
		}

		a_pipeline.push(a_series);
	}

	// inspect last arguments
	{
		let a_series = a_pipeline[a_pipeline.length-1];
		let n_series = a_series.length;

		// backwards
		for(let i_end=n_series-1; i_end>=0; i_end--) {
			// arg is path
			if(a_series[i_end].startsWith('/')) {
				// append as input
				a_inputs.push(fs.createReadStream(a_series.pop()));

				// keep going
				continue;
			}

			// stop eating args
			break;
		}

		// series now empty; pop from pipeline
		if(!a_series.length) a_pipeline.pop();
	}

	// empty command list
	if(!a_pipeline.length) {
		exit('no commands given');
	}

	// const parse_series = a_args => mk_yargs()
	// 	.command('content.ttl.read', 'read Turtle document', (yargs_read) => {
	// 		return yargs_read
	// 			.options({
	// 				u: {
	// 					type: 'boolean',
	// 					default: false,
	// 					alias: 'union',
	// 					group: s_group_multi_input,
	// 					describe: 'perform the union of multiple inputs',
	// 					conflicts: a_dual_input_keys.filter(s => 'u' !== s),
	// 				},
	// 				i: {
	// 					type: 'boolean',
	// 					default: false,
	// 					alias: ['intersect', 'intersection'],
	// 					group: s_group_multi_input,
	// 					describe: 'perform the intersection of multiple inputs',
	// 					conflicts: a_dual_input_keys.filter(s => 'i' !== s),
	// 				},
	// 				d: {
	// 					type: 'boolean',
	// 					alias: ['diff', 'difference'],
	// 					group: s_group_dual_input,
	// 					describe: 'perform a difference between two inputs',
	// 					conflicts: a_dual_input_keys.filter(s => 'd' !== s),
	// 				},
	// 				m: {
	// 					type: 'boolean',
	// 					alias: ['minus', 'subtract', 'subtraction'],
	// 					group: s_group_dual_input,
	// 					describe: 'perform a subtraction by removing input-B from input-A',
	// 					conflicts: a_dual_input_keys.filter(s => 'm' !== s),
	// 				},
	// 			})
	// 			.help()
	// 			.parse();
	// 	})
	// 	.command('util.dataset.tree')
	// 	.help()
	// 	.parse(a_args);

	(async() => {
		// starting inputs default to stdin if no explicit inputs given
		let a_prev = a_inputs.length? a_inputs: [process.stdin];

		// each series in pipeline
		for(let a_series of a_pipeline) {
			// start with command string
			let s_command = a_series[0];

			// no such command
			if(!(s_command in h_commands)) {
				exit(`no such command '${s_command}'`);
			}

			try {
				// eval command with its args
				let a_curr = await h_commands[s_command](a_series.slice(1), {
					command: s_command,
					inputs: a_prev,
				});

				// advance inputs
				a_prev = a_curr;
			}
			catch(e_parse) {
				// console.warn(e_parse.stack);
				return;
			}
		}

		// expect single output
		if(1 !== a_prev.length) {
			exit(`expected a single output stream but last command produces ${a_prev.length} streams`);
		}

		// pipe output to stdout
		a_prev[0].pipe(process.stdout);
	})();
}