const fs = require('fs'); const parse = require('csv-parse'); const GtfsRealtimeBindings = require('gtfs-realtime-bindings'); const request = require('request'); const dotEnv = require('dotenv').config(); let stops; let lineToFeedId; function buildStopData() { return new Promise((resolve) => { buildStops() .then(function(data) { stops = data; resolve(data); }); }); } function buildFeedData() { lineToFeedId = buildFeedIds(); return lineToFeedId; } async function getNextBusTimes() { // TODO add B54 line? // TODO add walk_time to each station let BUS_STOPS = [ // {'line': 'B25', 'station': 'FULTON ST/JAY ST', 'direction': 'W', 'stop_id': '302433'}, // {'line': 'B26', 'station': 'FULTON ST/JAY ST', 'direction': 'W', 'stop_id': '302433'}, // {'line': 'B38', 'station': 'FULTON ST/JAY ST', 'direction': 'W', 'stop_id': '302433'}, // {'line': 'B52', 'station': 'FULTON ST/JAY ST', 'direction': 'W', 'stop_id': '302433'}, {'line': 'B25', 'station': 'FULTON ST/HOYT ST', 'direction': 'E', 'stop_id': '307491', 'walk_time': 1}, // {'line': 'B26', 'station': 'FULTON ST/HOYT ST', 'direction': 'E', 'stop_id': '307491'}, {'line': 'B38', 'station': 'FULTON ST/HOYT ST', 'direction': 'E', 'stop_id': '307491', 'walk_time':1} // {'line': 'B52', 'station': 'FULTON ST/HOYT ST', 'direction': 'E', 'stop_id': '307491'}, // {'line': 'B57', 'station': 'JAY ST/FULTON ST', 'direction': 'S', 'stop_id': '305211'}, // {'line': 'B57', 'station': 'JAY ST/FULTON ST', 'direction': 'NE', 'stop_id': '307907'} ] let results = []; for (let i = 0; i < BUS_STOPS.length; i++) { let bus_stop = BUS_STOPS[i]; let stopId = bus_stop["stop_id"]; let busLine = bus_stop["line"] let requestSettings = { method: 'GET', headers: { Accept: 'application/json' }, url: `http://bustime.mta.info/api/siri/stop-monitoring.json?key=${process.env.MTA_BUS_KEY}&MonitoringRef=${stopId}&LineRef=MTA NYCT_${busLine}`, encoding: null }; const data = await new Promise(function(resolve, reject) { request(requestSettings, (error, response, body) => { if (error) { return reject(error) } if (response.statusCode != 200) { return reject(new Error('unexpected status code ' + response.statusCode)) } if (!/json/.test(response.headers['content-type'])) { return reject(new Error('unexpected content type')) } resolve(JSON.parse(body.toString())); }); }); const departTimes = data.Siri.ServiceDelivery.StopMonitoringDelivery.map((s) => { return s.MonitoredStopVisit.map((m) => { return m.MonitoredVehicleJourney.MonitoredCall.ExpectedDepartureTime }) }).reduce((a, b) => [...a, ...b], []).filter(Boolean) //let deltaTimes = formatArrivalTimes(departTimes); let result = departTimes.map(function(t) {return {time: t, line:busLine, station:bus_stop.station, direction:bus_stop.direction};}); results.push(...result); }; return results } async function getNextTrainTimes() { // TODO (optional) reduce number of API calls with feedID // TODO figure out rate limit? // TODO error handling // TODO filter the results to contain relevant ones only // TODO sort the results // TODO add "updated at ..." time for frontend? // TODO check where the direction "N"/"S" is coming from; // optionally, edit the returned-json-parsing to get the actual bound name let TRAIN_STOPS = [ {line:"2",stop_id:"233"}, // {line:"3",stop_id:"233"}, {line:"Q",stop_id:"R30"}, // {line:"B",stop_id:"R30"}, // {line:"D",stop_id:"R30"}, {line:"N",stop_id:"R30"}, {line:"R",stop_id:"R29"}, // {line:"W",stop_id:"R29"}, {line:"A",stop_id:"A41"}, {line:"C",stop_id:"A41"}, {line:"F",stop_id:"A41"}, {line:"G",stop_id:"A42"} // {line:"4",stop_id:"232"}, // {line:"5",stop_id:"232"} ] let STATION_INFO = { "233": {station:"Hoyt St", walk_time:1}, "R30": {station:"Dekalb Ave", walk_time:6}, "R29": {station:"Jay St - MetroTech", walk_time:1}, "A41": {station:"Jay St - MetroTech", walk_time:4}, "A42": {station:"Hoyt - Schermerhorn", walk_time:5}, "232": {station:"Borough Hall", walk_time:4}, } let results = []; for (let i = 0; i < TRAIN_STOPS.length; i++) { let train_stop = TRAIN_STOPS[i]; let trainLine = train_stop.line; let stopId = train_stop.stop_id; let feed = await retryMakeRequest(trainLine); for (const direction of ["N","S"]) { let arrivalTimes = parseArrivalTimes(feed, trainLine, stopId, direction); //let deltaTimes = formatArrivalTimes(arrivalTimes); let result = arrivalTimes.map(function(t) {return {time: new Date(t), line:trainLine, station:STATION_INFO[stopId].station, direction:direction};}); results.push(...result); }; }; return results; } const ATTEMPTS = 5 async function retryMakeRequest (trainLine, attempts = 0) { let body = await makeRequest(trainLine); try { return GtfsRealtimeBindings.FeedMessage.decode(body); } catch (error) { if (attempts < ATTEMPTS) { console.log(`retrying request for ${trainLine}`) return retryMakeRequest(trainLine, attempts + 1) } else { throw new Error(`failed to parse successful response after ${ATTEMPTS} attempts`) } } } async function getNextTransitTimes() { [bustime, traintime] = await Promise.all([getNextBusTimes(), getNextTrainTimes()]); return [...bustime, ...traintime]; } function parseArrivalTimes(feed, trainLine, stopId, direction) { let arrivalTimes = []; feed.entity.forEach((entity) => { let tripUpdate = entity.trip_update; if (tripUpdate && tripUpdate.trip.route_id === trainLine){ tripUpdate.stop_time_update.forEach((update) => { if(update.stop_id === stopId + direction) { let time = update.arrival.time.low*1000; arrivalTimes.push(time); } }) } }); arrivalTimes.sort(); return arrivalTimes; } function makeRequest(trainLine) { let feedId = lineToFeedId[trainLine]; let requestSettings = { method: 'GET', url: `http://datamine.mta.info/mta_esi.php?key=${process.env.MTA_KEY}&feed_id=${feedId}`, encoding: null }; console.log(`GET ${requestSettings.url}`) return new Promise(function(resolve, reject) { request(requestSettings, (error, response, body) => { if (!/gtfs/.test(response.headers['content-disposition'])) { console.error('rejecting headers', response.headers) console.error('body for rejected headers', body) } else if (!error && response.statusCode == 200) { resolve(body); } else { reject(error); } }); }); } function validateInputs(trainLine, stopId, direction) { let feedId = lineToFeedId[trainLine]; if(!feedId) { return "Invalid train line provided." ; } if(!stops[stopId]) { return "Invalid stop id provided."; } if(direction !== "N" && direction !== "S" && !direction) { return "Invalid direction."; } return ""; } function formatArrivalTimes(arrivalTimes) { let deltaTimes = []; arrivalTimes.forEach((time) => { let arrival = new Date(time); let deltaTime = Math.floor((arrival - Date.now())/60000); //console.log((arrival - Date.now())/60000) if(deltaTime > 0) deltaTimes.push(deltaTime); }); return deltaTimes; } function buildStops() { return new Promise(function(resolve, reject) { let stops = {}; fs.createReadStream("stops.csv") .pipe(parse({delimiter: ','})) .on('data', function(csvrow) { let stop = { stationId: csvrow[0], complexId: csvrow[1], stopId: csvrow[2], division: csvrow[3], line: csvrow[4], stopName: csvrow[5], borough: csvrow[6], } stops[stop.stopId] = stop; }) .on('end',function() { resolve(stops); return; }); }); } function buildFeedIds() { //NOTE S LINE IS ALSO INCLUDED IN ACEH DATA, include S in that array to see S data let lineToFeedId = {}; ["1","2","3","4","5","6","S"].forEach((line) => { lineToFeedId[line] = 1; }); ["A","C","E","H"].forEach((line) => { lineToFeedId[line] = 26; }); ["N","Q","R","W"].forEach((line) => { lineToFeedId[line] = 16; }); ["B","D","F","M"].forEach((line) => { lineToFeedId[line] = 21; }); ["J", "M"].forEach((line) => { lineToFeedId[line] = 36; }); lineToFeedId["L"] = 2; lineToFeedId["SIR"] = 11; lineToFeedId["G"] = 31; lineToFeedId["7"] = 51; return lineToFeedId; } module.exports = { getNextTrainTimes: getNextTrainTimes, getNextBusTimes: getNextBusTimes, getNextTransitTimes: getNextTransitTimes, parseArrivalTimes: parseArrivalTimes, makeRequest: makeRequest, validateInputs: validateInputs, formatArrivalTimes: formatArrivalTimes, buildStops: buildStops, buildFeedIds: buildFeedIds, buildStopData: buildStopData, buildFeedData: buildFeedData }