added uploading dataset and drag & drop support

modules/demo/client-server/components/server/cache.js

@@ -93,11 +93,11 @@ readMortgageData() {
     sources: [path.resolve('./public', 'data/mortgage.csv')],
     dataTypes: {
       index: new Int16,
-      zip: new Int32,
+      zip: new Uint32,
       dti: new Float32,
       current_actual_upb: new Float32,
       borrower_credit_score: new Int16,
-      load_id: new Int32,
+      load_id: new Uint32,
       delinquency_12_prediction: new Float32,
       seller_name: new Int16
     }

modules/demo/ssr/graph/index.js

@@ -19,6 +19,10 @@ const fastify = require('fastify')();
 fastify  //
   .register(require('./plugins/webrtc'), require('./plugins/graph')(fastify))
   .register(require('fastify-static'), {root: require('path').join(__dirname, 'public')})
-  .get('/', (req, reply) => reply.sendFile('video.html'));
+  .register(require('fastify-multipart'))
+  .register(require('fastify-cors'), {})
+  .register((require('fastify-arrow')))
+  .register(require('./plugins/api'))
+  .get('/', (req, reply) => reply.sendFile('video.html'))
 
 fastify.listen(8080).then(() => console.log('server ready'));

modules/demo/ssr/graph/package.json

@@ -21,11 +21,15 @@
     "fastify-socket.io": "2.0.0",
     "fastify-static": "4.2.3",
     "fastify": "3.20.2",
+    "fastify-multipart": "5.0.2",
+    "fastify-cors": "6.0.2",
+    "fastify-arrow": "0.1.0",
     "nanoid": "3.1.25",
     "rxjs": "6.6.7",
     "shm-typed-array": "0.0.13",
     "simple-peer": "9.11.0",
-    "socket.io": "4.1.3"
+    "socket.io": "4.1.3",
+    "glob": "7.2.0"
   },
   "files": [
     "render",

modules/demo/ssr/graph/plugins/api/index.js

@@ -0,0 +1,249 @@
+// Copyright (c) 2021, NVIDIA CORPORATION.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+const {graphs, clients}                           = require('../graph');
+const fs                                          = require('fs')
+const util                                        = require('util')
+const {pipeline}                                  = require('stream')
+const pump                                        = util.promisify(pipeline)
+var glob                                          = require('glob');
+const {Float32Buffer}                             = require('@rapidsai/cuda');
+const {GraphCOO}                                  = require('@rapidsai/cugraph');
+const {DataFrame, Series, Int32, Uint64, Float64} = require('@rapidsai/cudf');
+const {loadEdges, loadNodes}                      = require('../graph/loader');
+const {RecordBatchStreamWriter}                   = require('apache-arrow');
+
+function readDataFrame(path) {
+  if (path.indexOf('.csv', path.length - 4) !== -1) {
+    // csv file
+    return DataFrame.readCSV({sources: [path], header: 0, sourceType: 'files'});
+
+  } else if (path.indexOf('.parquet', path.length - 8) !== -1) {
+    // csv file
+    return DataFrame.readParquet({sources: [path]});
+  }
+  // if (df.names.includes('Unnamed: 0')) { df = df.cast({'Unnamed: 0': new Uint32}); }
+  return new DataFrame({});
+}
+
+async function getNodesForGraph(asDeviceMemory, nodes, numNodes) {
+  let nodesRes = {};
+  const pos    = new Float32Buffer(Array.from(
+    {length: numNodes * 2},
+    () => Math.random() * 1000 * (Math.random() < 0.5 ? -1 : 1),
+    ));
+
+  if (nodes.x in nodes.dataframe.names) {
+    nodesRes.nodeXPositions = asDeviceMemory(nodes.dataframe.get(node.x).data);
+  } else {
+    nodesRes.nodeXPositions = pos.subarray(0, pos.length / 2);
+  }
+  if (nodes.y in nodes.dataframe.names) {
+    nodesRes.nodeYPositions = asDeviceMemory(nodes.dataframe.get(node.y).data);
+  } else {
+    nodesRes.nodeYPositions = pos.subarray(pos.length / 2);
+  }
+  if (nodes.dataframe.names.includes(nodes.size)) {
+    nodesRes.nodeRadius = asDeviceMemory(nodes.dataframe.get(nodes.size).cast(new Float64).data);
+  }
+  if (nodes.dataframe.names.includes(nodes.color)) {
+    nodesRes.nodeFillColors =
+      asDeviceMemory(nodes.dataframe.get(nodes.color).cast(new Float64).data);
+  }
+  if (nodes.dataframe.names.includes(nodes.id)) {
+    nodesRes.nodeElementIndices =
+      asDeviceMemory(nodes.dataframe.get(nodes.id).cast(new Uint64).data);
+  }
+  return nodesRes;
+}
+
+async function getEdgesForGraph(asDeviceMemory, edges) {
+  let edgesRes = {};
+
+  if (edges.dataframe.names.includes(edges.color)) {
+    edgesRes.edgeColors = asDeviceMemory(edges.dataframe.get(edges.color).data);
+  } else {
+    edgesRes.edgeColors = asDeviceMemory(
+      Series
+        .sequence(
+          {type: new Uint64, size: edges.dataframe.numRows, init: 18443486512814075489n, step: 0})
+        .data);
+  }
+  if (edges.dataframe.names.includes(edges.id)) {
+    edgesRes.edgeList = asDeviceMemory(edges.dataframe.get(edges.id).cast(new Uint64).data);
+  }
+  if (edges.dataframe.names.includes(edges.bundle)) {
+    edgesRes.edgeBundles = asDeviceMemory(edges.dataframe.get(edges.bundle).data);
+  }
+  return edgesRes;
+}
+
+async function getPaginatedRows(df, pageIndex = 1, pageSize = 400, selected = []) {
+  console.log(selected);
+  if (selected.length == 0) {
+    return [df.head(pageIndex * pageSize).tail(pageSize).toArrow(), df.numRows];
+  }
+  const selectedSeries = Series.new({type: new Int32, data: selected});
+  let dfUpdated        = df.gather(selectedSeries);
+  return [dfUpdated.head(pageIndex * pageSize).tail(pageSize).toArrow(), dfUpdated.numRows];
+}
+
+module.exports = function(fastify, opts, done) {
+  // fastify.addHook('preValidation', (request, reply, done) => {
+  //   console.log('this is executed', request);
+  //   done()
+  // });
+
+  async function loadGraph(id, data) {
+    if (!(id in fastify[graphs])) {
+      const asDeviceMemory = (buf) => new (buf[Symbol.species])(buf);
+      const src                    = data.edges.dataframe.get(data.edges.src);
+      const dst                    = data.edges.dataframe.get(data.edges.dst);
+      const graph                  = new GraphCOO(src._col, dst._col, {directedEdges: true});
+      fastify[graphs][id]          = {
+        refCount: 0,
+        nodes: await getNodesForGraph(asDeviceMemory, data.nodes, graph.numNodes()),
+        edges: await getEdgesForGraph(asDeviceMemory, data.edges),
+        graph: graph,
+      };
+    }
+
+    ++fastify[graphs][id].refCount;
+
+    return {
+      gravity: 0.0,
+      linLogMode: false,
+      scalingRatio: 5.0,
+      barnesHutTheta: 0.0,
+      jitterTolerance: 0.05,
+      strongGravityMode: false,
+      outboundAttraction: false,
+      graph: fastify[graphs][id].graph,
+      nodes: {
+        ...fastify[graphs][id].nodes,
+        length: fastify[graphs][id].graph.numNodes(),
+      },
+      edges: {
+        ...fastify[graphs][id].edges,
+        length: fastify[graphs][id].graph.numEdges(),
+      },
+    };
+  }
+
+  fastify.get('/getIDValue', async (request, reply) => {
+    console.log(fastify[clients][request.query.id + ':video']);
+    reply.send(fastify[clients][request.query.id + ':video'].graph.dataframes[0].numRows);
+  });
+
+  fastify.post('/uploadFile', async function(req, reply) {
+    const data     = await req.file();
+    const basePath = `${__dirname}/../../data/`;
+    if (!fs.existsSync(basePath)) { fs.mkdirSync(basePath); }
+    const filepath = `${basePath}${data.filename}`;
+    const target = fs.createWriteStream(filepath);
+    try {
+      await pump(data.file, target);
+      console.log('success');
+    } catch (err) { console.log(err); }
+    reply.send()
+  });
+
+  fastify.get('/getFileNames', async (request, reply) => {
+    if (`${request.query.id}:video` in fastify[clients]) {
+      glob(`*.{csv,parquet}`,
+           {cwd: `${__dirname}/../../data/`},
+           (er, files) => { reply.send(JSON.stringify(files.concat(['defaultExample']))); });
+    } else {
+      reply.code(500).send('client handshake not established');
+    }
+  });
+
+  fastify.get('/loadOnGPU', async (request, reply) => {
+    const id       = `${request.query.id}:video`;
+    const filePath = `${__dirname}/../../data/`
+    if (id in fastify[clients]) {
+      if (fs.existsSync(`${filePath}${request.query.nodes}`) &&
+          fs.existsSync(`${filePath}${request.query.edges}`)) {
+        fastify[clients][id].data.nodes.dataframe =
+          await readDataFrame(`${filePath}${request.query.nodes}`);
+
+        fastify[clients][id].data.edges.dataframe =
+          await readDataFrame(`${filePath}${request.query.edges}`);
+      } else {
+        fastify[clients][id].data.nodes.dataframe = await loadNodes();
+        fastify[clients][id].data.edges.dataframe = await loadEdges();
+      }
+      if (fastify[clients][id].data.nodes.dataframe.numRows == 0) {
+        reply.code(500).send('no dataframe loaded');
+      }
+      reply.send(JSON.stringify({
+        'nodes': fastify[clients][id].data.nodes.dataframe.numRows,
+        'edges': fastify[clients][id].data.edges.dataframe.numRows
+      }));
+    }
+    else {
+      reply.code(500).send('client handshake not established');
+    }
+  })
+
+  fastify.get('/fetchDFParameters', async (request, reply) => {
+    const id = `${request.query.id}:video`;
+    if (id in fastify[clients]) {
+      reply.send(JSON.stringify({
+        nodesParams: fastify[clients][id].data.nodes.dataframe.names.concat([null]),
+        edgesParams: fastify[clients][id].data.edges.dataframe.names.concat([null])
+      }));
+    } else {
+      reply.code(500).send('client handshake not established');
+    }
+  });
+
+  fastify.post('/updateRenderColumns', async (request, reply) => {
+    const id = `${request.body.id}:video`;
+    if (id in fastify[clients]) {
+      Object.assign(fastify[clients][id].data.nodes, request.body.nodes);
+      Object.assign(fastify[clients][id].data.edges, request.body.edges);
+      fastify[clients][id].graph = await loadGraph('default', fastify[clients][id].data);
+      reply.code(200).send();
+    } else {
+      reply.code(500).send('client handshake not established');
+    }
+  });
+
+  fastify.post('/fetchPaginatedData', async (request, reply) => {
+    const id = `${request.body.id}:video`;
+    if (id in fastify[clients]) {
+      const pageIndex = request.body.pageIndex;
+      const pageSize  = request.body.pageSize;
+      const dataframe = request.body.dataframe;  //{'nodes', 'edges'}
+      const [res, numRows] =
+        await getPaginatedRows(fastify[clients][id].data[dataframe].dataframe,
+                               pageIndex,
+                               pageSize,
+                               fastify[clients][id].state.selectedInfo[dataframe]);
+
+      reply.send(JSON.stringify({page: res.toArray(), numRows: numRows}));
+      // try {
+      //   // RecordBatchStreamWriter.writeAll(res).pipe(reply.stream());
+      // } catch (err) {
+      //   request.log.error({err}, '/run_query error');
+      //   reply.code(500).send(err);
+      // }
+    } else {
+      reply.code(500).send('client handshake not established');
+    }
+  });
+
+  done();
+}