Skip to content

Stream geojsonl #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Naitik333 merged 4 commits into from
Nov 1, 2018
Merged

Stream geojsonl #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6d4613e
streaming support for CSV and geojsonl
dhaneeshtnair Oct 24, 2018
86599a5
changed type from geojsonl to csv
dhaneeshtnair Oct 24, 2018
59ed432
Backpressure implementation while streaming to XYZ
dhaneeshtnair Oct 26, 2018
c78186d
minor updates
dhaneeshtnair Oct 26, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
297 changes: 192 additions & 105 deletions bin/here-xyz.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,8 @@ import * as transform from "./transformutil";
import * as fs from "fs";
import * as tmp from "tmp";
import * as summary from "./summary";
let cq = require("block-queue");
const gsv = require("geojson-validation");

const request = require("request");
let choiceList: { name: string, value: string}[] = [];
Expand Down Expand Up @@ -566,10 +568,48 @@ program
"option to enforce uniqueness to the id by creating a hash of feature and use that as id"
)
.option("-o, --override", "override the data even if it share same id")
.option("-s, --stream", "streaming data support for large file uploads")
.action(function (id, options) {
uploadToXyzSpace(id, options);
});

function collate(result:Array<any>){
return result.reduce((features: any, feature: any) => {
if (feature.type === "Feature") {
features.push(feature);
} else if (feature.type === "FeatureCollection") {
features = features.concat(feature.features);
} else {
console.log("Unknown type" + feature.type);
}
return features
}, []);
}

function streamingQueue(){
let queue = cq(10,function (task:any,done:Function) {
uploadData(task.id, task.options, task.tags, task.fc,
true, task.options.ptag, task.options.file, task.options.id)
.then(x=>{
queue.uploadCount += task.fc.features.length;
console.log("uploaded feature count :"+queue.uploadCount);
queue.chunksize--;
done();
});
});
queue.uploadCount=0;
queue.chunksize=0;
queue.send= async function(obj:any){
while(this.chunksize>25){
await new Promise(done => setTimeout(done, 1000));
}
this.push(obj);
this.chunksize++;
}
return queue;
}


function uploadToXyzSpace(id: string, options: any){
(async () => {
let tags = "";
Expand All @@ -590,22 +630,33 @@ function uploadToXyzSpace(id: string, options: any){
options.unique = true;
}

if(options.assign && options.stream){
console.log(
"conflicting options together. You cannot choose assign mode while selecting streaming option"
);
process.exit(1);
}

if (options.file) {
const fs = require("fs");
if (options.file.indexOf(".geojsonl") != -1) {
transform.readLineFromFile(options.file, 100).then((result: any) => {
const totalFeatures = result.reduce((features: any, feature: any) => {
if (feature.type === "Feature") {
features.push(feature);
} else if (feature.type === "FeatureCollection") {
features = features.concat(feature.features);
} else {
console.log("Unknown type" + feature.type);
}
return features
}, []);
uploadData(id, options, tags, { type: "FeatureCollection", features: totalFeatures }, true, options.ptag, options.file, options.id);
});
if(!options.stream){
transform.readLineFromFile(options.file, 100).then((result: any) => {
uploadData(id, options, tags, { type: "FeatureCollection", features: collate(result) }, true, options.ptag, options.file, options.id);
});
}else{
let queue = streamingQueue();
transform.readLineAsChunks(options.file, options.chunk?options.chunk:1000,function(result:any){
return new Promise((res,rej)=>{
( async()=>{
if(result.length>0){
await queue.send({id:id,options:options,tags:tags,fc:{ type: "FeatureCollection", features: collate(result) }});
}
res();
})();
});
});
}
} else if (options.file.indexOf(".shp") != -1) {
let result = await transform.readShapeFile(
options.file,
Expand All @@ -621,29 +672,53 @@ function uploadToXyzSpace(id: string, options: any){
options.id
);
} else if (options.file.indexOf(".csv") != -1) {
let result = await transform.read(
options.file,
true
);
const object = {
features: transform.transform(
result,
options.lat,
options.lon,
options.alt
),
type: "FeatureCollection"
};
uploadData(
id,
options,
tags,
object,
true,
options.ptag,
if(!options.stream){
let result = await transform.read(
options.file,
options.id
);
true
);
const object = {
features: transform.transform(
result,
options.lat,
options.lon,
options.alt
),
type: "FeatureCollection"
};
uploadData(
id,
options,
tags,
object,
true,
options.ptag,
options.file,
options.id
);
}else{
let queue = streamingQueue();
transform.readCSVAsChunks(options.file, options.chunk?options.chunk:1000,function(result:any){
return new Promise((res,rej)=>{
( async()=>{
if(result.length>0){
const fc = {
features: transform.transform(
result,
options.lat,
options.lon,
options.alt
),
type: "FeatureCollection"
};
await queue.send({id:id,options:options,tags:tags,fc:fc});
res();
}
})();
});

});
}
} else {
let result = await transform.read(
options.file,
Expand Down Expand Up @@ -718,24 +793,39 @@ function uploadData(
fileName: string | null,
uid: string
) {
if (object.type == "Feature") {
object = { features: [object], type: "FeatureCollection" };
}
if (options.assign) {
//console.log("assign mode on");
const questions = createQuestionsList(object);
inquirer.prompt(questions).then((answers: any) => {
if (options.ptag === undefined) {
options.ptag = "";
}
options.ptag = options.ptag + answers.tagChoices;
if (options.id === undefined) {
options.id = "";
}
options.id = options.id + answers.idChoice;
//console.log(options.ptag);
//console.log("unique key - " + options.id);
//Need to be inside if, else this will be executed before user choice is inserted as its async
return new Promise((resolve, reject) => {

if (object.type == "Feature") {
object = { features: [object], type: "FeatureCollection" };
}
if (options.assign) {
//console.log("assign mode on");
const questions = createQuestionsList(object);
inquirer.prompt(questions).then((answers: any) => {
if (options.ptag === undefined) {
options.ptag = "";
}
options.ptag = options.ptag + answers.tagChoices;
if (options.id === undefined) {
options.id = "";
}
options.id = options.id + answers.idChoice;
//console.log(options.ptag);
//console.log("unique key - " + options.id);
//Need to be inside if, else this will be executed before user choice is inserted as its async
uploadDataToSpaceWithTags(
id,
options,
tags,
object,
false,
options.ptag,
fileName,
options.id
).then(x=>resolve(x));

});
} else {
uploadDataToSpaceWithTags(
id,
options,
Expand All @@ -745,23 +835,14 @@ function uploadData(
options.ptag,
fileName,
options.id
);
});
} else {
uploadDataToSpaceWithTags(
id,
options,
tags,
object,
false,
options.ptag,
fileName,
options.id
);
}
).then(x=>resolve(x));
}

});

}

function uploadDataToSpaceWithTags(
async function uploadDataToSpaceWithTags(
id: string,
options: any,
tags: any,
Expand All @@ -771,46 +852,52 @@ function uploadDataToSpaceWithTags(
fileName: string | null,
uid: string
) {
const gsv = require("geojson-validation");
gsv.valid(object, async function (valid: boolean, errs: any) {
if (!valid) {
console.log(errs);
return;
}
const featureOut = await mergeAllTags(
object.features,
tags,
tagProperties,
fileName,
uid,
options
);

const chunks = options.chunk
? chunkify(featureOut, parseInt(options.chunk))
: [featureOut];
const chunkSize = chunks.length;
const index = 0;
await iterateChunks(
chunks,
"/hub/spaces/" + id + "/features",
index,
chunkSize,
);
if (isFile)
console.log(
"'" +
options.file +
"' uploaded to xyzspace '" +
id +
"' successfully"
return new Promise((resolve, reject) => {
gsv.valid(object, async function (valid: boolean, errs: any) {
if (!valid) {
console.log(errs);
reject(errs);
return;
}
const featureOut = await mergeAllTags(
object.features,
tags,
tagProperties,
fileName,
uid,
options
);
else
console.log(
"data upload to xyzspace '" + id + "' completed successfully"

const chunks = options.chunk
? chunkify(featureOut, parseInt(options.chunk))
: [featureOut];
const chunkSize = chunks.length;
const index = 0;
await iterateChunks(
chunks,
"/hub/spaces/" + id + "/features",
index,
chunkSize,
);
if(!options.stream){
if (isFile)
console.log(
"'" +
options.file +
"' uploaded to xyzspace '" +
id +
"' successfully"
);
else
console.log(
"data upload to xyzspace '" + id + "' completed successfully"
);

summary.summarize(featureOut, id, true);
summary.summarize(featureOut, id, true);

}
resolve(true);
});
});
}

Expand Down
Loading