Created functions to SQLITE

tago-io · Nov 13, 2024 · 962db64 · 962db64
1 parent d79ff08
commit 962db64
Show file tree

Hide file tree

Showing 10 changed files with 1,262 additions and 124 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/plugins/mysql/src/Providers/DeviceData/importDeviceData.ts b/plugins/mysql/src/Providers/DeviceData/importDeviceData.ts
@@ -89,15 +89,6 @@ async function importDeviceData(
     throw new Error("Device not found");
   }
   return new Promise((resolve, reject) => {
-    // TODO - Fix to use copy from mysql
-    // const sqlCommand = `COPY "${deviceID}" FROM '${fileName}' DELIMITER ',' CSV HEADER`;
-    // deviceDB.write.raw(sqlCommand).then(() => {
-    //   resolve("Data imported successfully");
-    // }).catch((error) => {
-    //   console.log(error);
-    //   reject(error);
-    // });
-
     const data: any[] = [];
     fs.createReadStream(fileName)
       .pipe(parse({ columns: true, encoding: "utf8" }))

diff --git a/plugins/postgres/src/Providers/DeviceData/exportDeviceData.ts b/plugins/postgres/src/Providers/DeviceData/exportDeviceData.ts
@@ -13,30 +13,44 @@ async function exportDeviceData(
   folder: string,
 ): Promise<string> {
   const client = await getDeviceConnection(deviceID, type);
-  return new Promise((resolve, reject) => {
-    const filename = `${folder}/${csvNameGenerator(`data_${deviceID}`)}`;
-    const file = fs.createWriteStream(filename);
-    file.setDefaultEncoding("utf8");
-
-    // TODO - Fix to use copy from mysql
-    // const sqlCommand = `COPY "${deviceID}" TO STDOUT WITH CSV HEADER`;
-    // const deviceDataStream = deviceDB.read.raw(sqlCommand).stream();
-    const deviceDataStream = client.read("data")?.select("*").stream();
-    if (!deviceDataStream) {
-      return reject("No data to export");
-    }
-
-    deviceDataStream
-      .pipe(stringify({ header: true }))
-      .pipe(file)
-      .on("finish", () => {
-        resolve(filename);
-      })
-      .on("error", (error) => {
-        console.log(error);
-        reject(error);
-      });
+
+  const filename = `${folder}/${csvNameGenerator(`data_${deviceID}`)}`;
+  const file = fs.createWriteStream(filename);
+  file.setDefaultEncoding("utf8");
+  const columns = [
+    "id",
+    "variable",
+    "type",
+    "value",
+    "unit",
+    "group",
+    "location",
+    "metadata",
+    "time",
+    "created_at",
+  ];
+
+  const stringifier = stringify({
+    header: true,
+    columns,
+    cast: {
+      date: (value: Date) => value.toISOString(),
+    },
   });
+
+  stringifier.pipe(file);
+
+  const deviceDataStream = client.read("data")?.select(columns).stream();
+
+  if (!deviceDataStream) {
+    return Promise.reject("No data to export");
+  }
+
+  for await (const row of deviceDataStream) {
+    stringifier.write(row);
+  }
+
+  return Promise.resolve(filename);
 }
 
 export default exportDeviceData;
diff --git a/plugins/postgres/src/Providers/DeviceData/importDeviceData.ts b/plugins/postgres/src/Providers/DeviceData/importDeviceData.ts
@@ -5,6 +5,7 @@ import {
   type TGenericID,
   generateResourceID,
   zDeviceData,
+  zDeviceImmutableData,
 } from "@tago-io/tcore-sdk/types";
 import { parse } from "csv";
 import type { Knex } from "knex";
@@ -37,28 +38,30 @@ function _parseData(row: any, device: IDevice) {
     id: generateResourceID(),
     variable: row.variable,
     type: row.type,
-    value: !row.value ? null : row.value,
-    unit: !row.unit ? null : row.unit,
+    value: !row.value ? undefined : row.value,
+    unit: !row.unit ? undefined : row.unit,
     group: row.group,
-    location: !row.location ? null : row.unit,
-    metadata: !row.metadata ? null : JSON.parse(row.location),
-    time: row.time,
+    location: !row.location ? undefined : JSON.parse(row.location),
+    metadata: !row.metadata ? undefined : JSON.parse(row.metadata),
+    time: new Date(row.time),
     created_at: new Date(),
-    chunk_timestamp_start: null,
-    chunk_timestamp_end: null,
-    serie: !row.serie ? null : row.unit,
+    chunk_timestamp_start: undefined,
+    chunk_timestamp_end: undefined,
+    serie: !row.serie ? undefined : row.serie,
   };
 
+  let zodData: any;
   if (device.chunk_period) {
     const chunkTimestamp = _getChunkTimestamp(new Date(row.time), device);
     if (chunkTimestamp) {
       parsedData.chunk_timestamp_start = chunkTimestamp.startDate;
       parsedData.chunk_timestamp_end = chunkTimestamp.endDate;
     }
+    zodData = zDeviceImmutableData.parse(parsedData);
+  } else {
+    zodData = zDeviceData.parse(parsedData);
   }
 
-  const zodData = zDeviceData.parse(parsedData);
-
   return zodData;
 }
 
@@ -86,15 +89,6 @@ async function importDeviceData(
     throw new Error("Device not found");
   }
   return new Promise((resolve, reject) => {
-    // TODO - Fix to use copy from postgres
-    // const sqlCommand = `COPY "${deviceID}" FROM '${fileName}' DELIMITER ',' CSV HEADER`;
-    // deviceDB.write.raw(sqlCommand).then(() => {
-    //   resolve("Data imported successfully");
-    // }).catch((error) => {
-    //   console.log(error);
-    //   reject(error);
-    // });
-
     const data: any[] = [];
     fs.createReadStream(fileName)
       .pipe(parse({ columns: true, encoding: "utf8" }))

diff --git a/plugins/sqlite/package.json b/plugins/sqlite/package.json
@@ -20,7 +20,7 @@
   },
   "dependencies": {
     "knex": "3.1.0",
-    "better-sqlite3": "11.5.0",
+    "sqlite3": "5.1.7",
     "csv": "6.3.10"
   },
   "devDependencies": {}

diff --git a/plugins/sqlite/src/Helpers/DeviceDatabase.ts b/plugins/sqlite/src/Helpers/DeviceDatabase.ts
@@ -47,7 +47,7 @@ export async function createDeviceConnection(
   const filename = await helpers.getFileURI(path.join("devices", `${id}.db`));
 
   const connection = knex({
-    client: "better-sqlite3",
+    client: "sqlite3",
     connection: { filename },
     useNullAsDefault: true,
   });

diff --git a/plugins/sqlite/src/Helpers/PluginConnection.ts b/plugins/sqlite/src/Helpers/PluginConnection.ts
@@ -30,7 +30,7 @@ async function createPluginConnection(id: TGenericID) {
   const filename = await helpers.getFileURI(path.join("plugins", `${id}.db`));
 
   const connection = knex({
-    client: "better-sqlite3",
+    client: "sqlite3",
     connection: { filename },
     useNullAsDefault: true,
   });

diff --git a/plugins/sqlite/src/Providers/DeviceData/exportDeviceData.ts b/plugins/sqlite/src/Providers/DeviceData/exportDeviceData.ts
@@ -13,22 +13,43 @@ async function exportDeviceData(
   folder: string,
 ): Promise<string> {
   const client = await getDeviceConnection(deviceID, type);
-  return new Promise((resolve, reject) => {
-    const deviceDataStream = client.select("*").from("data").stream();
-    const filename = `${folder}/${csvNameGenerator(`data_${deviceID}`)}`;
-    const file = fs.createWriteStream(filename);
-    file.setDefaultEncoding("utf8");
-
-    deviceDataStream
-      .pipe(stringify({ header: true }))
-      .pipe(file)
-      .on("finish", () => {
-        resolve(filename);
-      })
-      .on("error", (error) => {
-        reject(error);
-      });
+
+  const filename = `${folder}/${csvNameGenerator(`data_${deviceID}`)}`;
+  const file = fs.createWriteStream(filename);
+  file.setDefaultEncoding("utf8");
+  const columns = [
+    "id",
+    "variable",
+    "type",
+    "value",
+    "unit",
+    "group",
+    "location",
+    "metadata",
+    "time",
+    "created_at",
+  ];
+
+  const stringifier = stringify({
+    header: true,
+    columns,
   });
+
+  stringifier.pipe(file);
+
+  const deviceDataStream = client.select(columns).from("data").stream();
+
+  if (!deviceDataStream) {
+    return Promise.reject("No data to export");
+  }
+
+  for await (const row of deviceDataStream) {
+    row.time = new Date(row.time).toISOString();
+    row.created_at = new Date(row.created_at).toISOString();
+    stringifier.write(row);
+  }
+
+  return Promise.resolve(filename);
 }
 
 export default exportDeviceData;
diff --git a/plugins/sqlite/src/Providers/DeviceData/importDeviceData.ts b/plugins/sqlite/src/Providers/DeviceData/importDeviceData.ts
@@ -1,12 +1,17 @@
 import fs from "node:fs";
 import {
+  type IDevice,
   type TDeviceType,
   type TGenericID,
   generateResourceID,
+  zDeviceData,
+  zDeviceImmutableData,
 } from "@tago-io/tcore-sdk/types";
 import { parse } from "csv";
 import type { Knex } from "knex";
 import { getDeviceConnection } from "../../Helpers/DeviceDatabase.ts";
+import { DateTime } from "luxon";
+import getDeviceInfo from "../Device/getDeviceInfo.ts";
 
 async function _insertData(client: Knex, data: any[]) {
   client
@@ -19,6 +24,67 @@ async function _insertData(client: Knex, data: any[]) {
     });
 }
 
+/**
+ * Gets the chunk timestamps for a date.
+ */
+function _getChunkTimestamp(date: Date, device: IDevice) {
+  const dateJS = DateTime.fromJSDate(date).toUTC();
+
+  if (!device?.chunk_period) {
+    return null;
+  }
+
+  if (!dateJS.isValid) {
+    throw "Invalid Database Chunk Address (date)";
+  }
+
+  const startDate = dateJS.startOf(device.chunk_period).toJSDate();
+  const endDate = dateJS.endOf(device.chunk_period).toJSDate();
+
+  return { startDate, endDate };
+}
+
+function _parseData(row: any, device: IDevice) {
+  const parsedData: any = {
+    id: generateResourceID(),
+    variable: row.variable,
+    type: row.type,
+    value: !row.value ? undefined : row.value,
+    unit: !row.unit ? undefined : row.unit,
+    group: row.group,
+    location: !row.location ? undefined : JSON.parse(row.location),
+    metadata: !row.metadata ? undefined : JSON.parse(row.metadata),
+    time: new Date(row.time),
+    created_at: new Date(),
+    chunk_timestamp_start: undefined,
+    chunk_timestamp_end: undefined,
+    serie: !row.serie ? undefined : row.serie,
+  };
+
+  let zodData: any;
+  if (device.chunk_period) {
+    const chunkTimestamp = _getChunkTimestamp(new Date(row.time), device);
+    if (chunkTimestamp) {
+      parsedData.chunk_timestamp_start = chunkTimestamp.startDate;
+      parsedData.chunk_timestamp_end = chunkTimestamp.endDate;
+    }
+    zodData = zDeviceImmutableData.parse(parsedData);
+  } else {
+    zodData = zDeviceData.parse(parsedData);
+  }
+
+  // This is to fix json type in sqlite
+  if (zodData.location){
+    zodData.location = JSON.stringify(zodData.location);
+  }
+
+  if (zodData.metadata){
+    zodData.metadata = JSON.stringify(zodData.metadata);
+  }
+
+  return zodData;
+}
+
 /**
  * Import a device data from CSV file.
  */
@@ -28,34 +94,18 @@ async function importDeviceData(
   fileName: string,
 ): Promise<string> {
   const client = await getDeviceConnection(deviceID, type);
+  const device = await getDeviceInfo(deviceID);
+  if (!device) {
+    throw new Error("Device not found");
+  }
   return new Promise((resolve, reject) => {
     const data: any[] = [];
     fs.createReadStream(fileName)
       .pipe(parse({ columns: true, encoding: "utf8" }))
       .on("data", (row) => {
-        row.id = generateResourceID();
-
-        if (!row.location) {
-          row.location = null;
-        }
-
-        if (!row.metadata) {
-          row.metadata = null;
-        }
-
-        if (!row.value) {
-          row.value = null;
-        }
-
-        if (!row.unit) {
-          row.unit = null;
-        }
-
-        if (!row.serie) {
-          row.serie = null;
-        }
+        const parsedData = _parseData(row, device);
+        data.push(parsedData);
 
-        data.push(row);
         if (data.length === 1000) {
           _insertData(client, data);
           data.length = 0;

diff --git a/plugins/sqlite/src/knex.ts b/plugins/sqlite/src/knex.ts
@@ -20,7 +20,7 @@ export async function setupKnex(data: any) {
   const filename = data.file;
 
   knexClient = knex({
-    client: "better-sqlite3",
+    client: "sqlite3",
     connection: { filename },
     useNullAsDefault: true,
   });