trajectories-to-sql.js

#!/usr/bin/env node
'use strict'

const {join: pathJoin} = require('path')
const readCsv = require('gtfs-utils/read-csv')
const readServices = require('gtfs-utils/read-services-and-exceptions')
const computeTrajectories = require('gtfs-utils/compute-trajectories')
const resolveTime = require('gtfs-utils/lib/resolve-time')
const {Stringifier} = require('csv-stringify')
const pkg = require('./package.json')

const TIMEZONE = process.env.TIMEZONE
if (!TIMEZONE) {
	console.error('Missing/empty TIMEZONE environment variable.')
	process.exit(1)
}

const src = process.env.GTFS_DIR || 'gtfs'
const readFile = async (name) => {
	return await readCsv(pathJoin(src, name + '.txt'))
}

const herrenberg779Bus = ['31-779-j21-2']
const herrenberg782Bus = ['31-782-j21-1']
const filters = {
	trip: t => [...herrenberg779Bus, ...herrenberg782Bus].includes(t.route_id),
}

const withAbsoluteTime = (tr, date) => {
	try {
		const withAbsTime = ([lon, lat, alt, arr, dep]) => [
			lon, lat, alt,
			resolveTime(TIMEZONE, date, arr), // arr
			resolveTime(TIMEZONE, date, dep), // dep
		]
		return {
			type: 'Feature',
			properties: {
				...tr.properties,
				id: tr.properties.id + '-' + date,
				date,
			},
			geometry: {
				type: 'LineString',
				coordinates: tr.geometry.coordinates.map(withAbsTime)
			},
		}
	} catch (err) {
		err.trajectory = tr
		err.date = date
		throw err
	}
}

;(async () => {
	const svcDates = new Map()
	for await (const [serviceId, dates] of readServices(readFile, TIMEZONE)) {
		svcDates.set(serviceId, dates)
	}

	process.stdout.write(`\
-- SQL dump generated by ${pkg.name}
-- ${pkg.homepage}
\\set ON_ERROR_STOP True
CREATE EXTENSION IF NOT EXISTS postgis;
BEGIN;

CREATE TABLE trajectories (
	id TEXT PRIMARY KEY,
	trip_id TEXT NOT NULL,
	shape_id TEXT NOT NULL,
	service_id TEXT NOT NULL,
	date TEXT NOT NULL,
	trajectory geography NOT NULL,
	-- todo: CONSTRAINT valid_tr CHECK (ST_IsValidTrajectory(trajectory))
);
-- CREATE INDEX ON trajectories (shape_id, schedule_id);
-- CREATE INDEX ON trajectories USING GIST (trajectory);

COPY trajectories (
	id,
	trip_id,
	shape_id,
	service_id,
	date,
	trajectory
) FROM STDIN csv;
`)

	const csv = new Stringifier({
		quoted: true,
		header: true,
	})
	csv.pipe(process.stdout)

	const trajectories = await computeTrajectories(readFile, filters)
	for await (const tr of trajectories) {
		const {id, tripId, serviceId} = tr.properties
		if (!svcDates.has(serviceId)) {
			console.error('invalid service_id', tr.properties)
			continue
		}
		const dates = svcDates.get(serviceId)
		if (dates.length === 0) {
			console.error('0 service dates', serviceId, dates)
			continue
		}

		console.error('processing', id, 'serviceId', serviceId, 'with', dates.length, 'service dates')
		for (const date of dates) {
			const absTr = withAbsoluteTime(tr, date)

			let coordsAsSql = 'ST_MakeLine(ARRAY['
			let first = true
			for (const [lon, lat, _, arr, __] of absTr.geometry.coordinates) {
				if (first) first = false
				else coordsAsSql += ','
				// todo: import altitude & departure time
				coordsAsSql += `ST_MakePointM(${lon}, ${lat}, ${arr})`
			}
			coordsAsSql += '])'

			csv.write({
				id: absTr.properties.id,
				trip_id: absTr.properties.tripId,
				shape_id: absTr.properties.shapeId,
				service_id: absTr.properties.serviceId,
				date: absTr.properties.date,
				trajectory: coordsAsSql,
			})
		}
	}

	process.stdout.write(`\
\\.

COMMIT;`)
})()
.catch((err) => {
	console.error(err)
	process.exit(1)
})