[update] need to try execute_batch for optimize

src/controllers/file/download_functional.py

@@ -1,27 +1,34 @@
 import psycopg2
 import pandas as pd
 import numpy as np
+from os.path import exists
+import sys
+import os
+import time
 
 def log(df, channel):
-    # TODO: not sure need use 32 or 64?
-    df[f'{channel}_log'] = np.log10(np.absolute(df[channel].astype(np.int32)))
+    # TODO: not sure need use 32 or 64? => 64, since large integral will cause overflow
+    df[f'{channel}_log'] = np.log10(np.absolute(df[channel]))
+    # df[f'{channel}_log'] = np.log10(np.absolute(df[channel].astype(np.int64)))
     return df
 def integral(df, x_channel, y_channel, pre_last_x, pre_last_y, integral_result):
     offset = df[x_channel]
     offset = offset.shift(periods=1)
     offset[0] = pre_last_x
-    x_diff = df[x_channel].astype(np.int32) - offset.astype(np.int32)
+    x_diff = df[x_channel] - offset
+    # x_diff = df[x_channel].astype(np.int64) - offset.astype(np.int64)
     # print('x_diff\n', x_diff)
     offset = df[y_channel]
     offset = offset.shift(periods=1)
     offset[0] = pre_last_y
-    y_avg = (offset.astype(np.int32) + df[y_channel].astype(np.int32)) / 2
+    y_avg = (offset + df[y_channel]) / 2
+    # y_avg = (offset.astype(np.int64) + df[y_channel].astype(np.int64)) / 2
     # print('y_avg:', y_avg)
     # print('y_avg: ', y_avg)
     local_integral = x_diff * y_avg
-    local_integral[0] += integral_result
     # TODO: if user don't need, then can remove it
     df[f'{x_channel}_{y_channel}_integral_delta'] = local_integral
+    local_integral[0] += integral_result
     df[f'{x_channel}_{y_channel}_integral'] = local_integral.cumsum()
     # print(df)
     pre_last_x = df[x_channel].iat[-1]
@@ -33,73 +40,103 @@ def differential(df, x_channel, y_channel, pre_last_x, pre_last_y):
     offset = df[x_channel]
     offset = offset.shift(periods=1)
     offset[0] = pre_last_x
-    x_diff = df[x_channel].astype(np.int32) - offset.astype(np.int32)
+    x_diff = df[x_channel] - offset
+    # x_diff = df[x_channel].astype(np.int64) - offset.astype(np.int64)
     offset = df[y_channel]
     offset = offset.shift(periods=1)
     offset[0] = pre_last_y
-    y_diff = df[y_channel].astype(np.int32) - offset.astype(np.int32)
+    y_diff = df[y_channel] - offset
+    # y_diff = df[y_channel].astype(np.int64) - offset.astype(np.int64)
     local_diff = y_diff / x_diff
-    print(local_diff)
     # TODO: if user don't need, then can remove it
     df[f'{x_channel}_{y_channel}_diff'] = local_diff
     pre_last_x = df[x_channel].iat[-1]
     pre_last_y = df[y_channel].iat[-1]
     return df, pre_last_x, pre_last_y
-    
 
-if __name__ == '__main__':
-    # TODO: function parameters: id, mode, x_channel, y_channel
-    conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="5432")
+def download(id, mode, x_channel, y_channel):
+    # TODO: can change different chunck size
+    pages_chunck_size = 250
+    sql_total = 0
+    conn_start = time.time()
+    # TODO: need the check port id
+    conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="54321")
+    print('conn time: ', time.time() - conn_start)
+    # conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="5432")
     cur = conn.cursor()
-    id = 1262
-    # id = 407
-    # id = 1073
-    # id = 968
-    mode = 'differential'
-    # mode = 'integral'
-    # mode = 'log'
-    sql_str = f'select "raw_data" from "recording_data_metas" WHERE id = {id}'  # recording_data_metas: meta info to get the raw_data table
-    # sql_str = 'select "raw_data" from "recording_data_metas" WHERE id = 407'  # recording_data_metas: meta info to get the raw_data table
-    # sql_str = 'select * from "0_recording_data_raws" WHERE id = 33384'
+    # recording_data_metas: meta info to get the raw_data table
+    sql_str = f'select "raw_data" from "recording_data_metas" WHERE id = {id}'
+    first_sql = time.time()
     cur.execute(sql_str)
+    print('first sql time: ', time.time() - first_sql)
     raw_data_list = None
-    # x_
     try:
         raw_data_list = cur.fetchall()[0][0]
-        # print(ret[0][0]['0'])
         channels = list(raw_data_list.keys())
-        df = pd.DataFrame()
+        df_all = pd.DataFrame()
+        # df = pd.DataFrame()
         df_time = []
         pre_last_x = 0
         pre_last_y = 0
         integral_result = 0
-        # print('len(raw_data_list[channels[0]]): ', len(raw_data_list[channels[0]]))
-        for raw_data_pages in range(len(raw_data_list[channels[0]])):
+        for raw_data_pages in range(len(raw_data_list[channels[0]])):                
+            df = pd.DataFrame()
             for channel in channels:
                 sql_str = f'select "data" from "{channel}_recording_data_raws" WHERE id = {raw_data_list[channel][raw_data_pages]}'
+                sql_start = time.time()
                 cur.execute(sql_str)
+                sql_total += (time.time() - sql_start)
                 data_value = cur.fetchall()[0][0].replace('"***"', ' ')
                 data_value = data_value.split(' ')[:-1]
                 if df.empty:
                     df_time = data_value[::2]
                 df[channel] = data_value
             df = df.iloc[lambda x: x.index % 2 == 1].reset_index(drop=True)
-            df.insert(0, 'time', df_time)
-            if mode == 'log':
-                x_channel = '1'
+            df.insert(0, 'Time', df_time)
+            # transfer dataframe from string to number
+            cols = df.columns
+            df[cols] = df[cols].apply(pd.to_numeric, errors='coerce')
+            if mode == 'Log':
                 df = log(df, x_channel)
-            elif mode == 'integral':
-                # TODO: x_channel and y_channel will need to be set as parameters
-                x_channel = 'time'
-                y_channel = '0'
+            elif mode == 'Integral':
                 df, pre_last_x, pre_last_y, integral_result = integral(df, x_channel, y_channel, pre_last_x, pre_last_y, integral_result)
-            elif mode == 'differential':
-                x_channel = '1'
-                y_channel = '0'
+            elif mode == 'Differential':
                 df, pre_last_x, pre_last_y = differential(df, x_channel, y_channel, pre_last_x, pre_last_y)
-            # print(df)
-            df.to_csv(f"csv_test/{id}.csv", mode='a', index=False, header=False)
-            df = pd.DataFrame()
+            df_all = pd.concat([df_all, df], ignore_index=True, sort=False)
+            if ~raw_data_pages and (raw_data_pages % pages_chunck_size == 0 or raw_data_pages == len(raw_data_list[channels[0]]) - 1):
+                if not(exists(f"./src/csv/{id}.csv")):
+                    df.to_csv(f"./src/csv/{id}.csv", mode='a', index=False, header=True)
+                else:
+                    df.to_csv(f"./src/csv/{id}.csv", mode='a', index=False, header=False)
+                """ if not(exists(f"./{id}.csv")):
+                    df_all.to_csv(f"./{id}.csv", mode='a', index=False, header=True)
+                else:
+                    df_all.to_csv(f"./{id}.csv", mode='a', index=False, header=False) """
+                df_all = pd.DataFrame()
+        print('total sql: ', sql_total)
     except BaseException as e:
         print(e)
-        exit
+        exit
+    return
+
+if __name__ == '__main__':
+    params = {}
+    for input in sys.argv[1:]: #Now we're going to iterate over argv[1:] (argv[0] is the program name)
+        param  = input.split("=") #Get what's left of the '='
+        params[param[0]] = param[1]
+
+    id = params['id']
+    mode = params['mode']
+    x_channel = params['x_channel']
+    y_channel = params['y_channel']
+    # print('download_functional init at main: ', id, mode, x_channel, y_channel)
+    start_time = time.time()
+    download(id, mode, x_channel, y_channel)
+    # download(408, '', 'Time', '0')
+    # download(408, '', 'Time', '0')
+    # download(405, 'Integral', 'Time', '0')
+    # download(405, 'Integral', 'Time', '0')
+    # download(408, 'Integral', 'Time', '0')
+    print('total: ', time.time() - start_time)
+    # download(1262, 'differential', '1', '0')
+    # 1262/differential/1/0

src/controllers/file/recording_data_raw.js

@@ -1,7 +1,12 @@
 import RecordingDataRaw from '../../models/file/recording_data_raw'
+import RecordingDataMeta from '../../models/file/recording_data_meta'
 import * as auth from '../auth'
+import childProcess from 'child_process'
+import path from 'path'
+import { createReadStream } from 'fs'
 
 const recordingDataRaw = new RecordingDataRaw()
+const recordingDataMeta = new RecordingDataMeta()
 
 export const create = async (ctx, next) => {
   const index = ctx.params.channel
@@ -162,3 +167,37 @@ export const getAll = async (ctx, next) => {
 
   next()
 }
+
+export const getCSV = async (ctx, next) => {
+  const id = ctx.params.id
+  const mode = ctx.params.mode
+  const x_channel = ctx.params.x_channel
+  const y_channel = ctx.params.y_channel
+  // console.log('getCSV: ', id, mode, x_channel, y_channel)
+  // const result = await recordingDataRaw.getCSV(id, mode, x_channel, y_channel)
+  const process = childProcess.spawnSync('python', [
+    path.join(__dirname, '/download_functional.py'),
+    'id=' + id,
+    'mode=' + mode,
+    'x_channel=' + x_channel,
+    'y_channel=' + y_channel,
+  ], { maxBuffer: 2 * 1024 * 1024 * 1024 })
+  // console.log('process: ', process)
+  // process.stdout.on('data', (data) => {
+  //   conosle.log('data', data)
+  // })
+  const result = await recordingDataMeta.getByID(id)
+  // console.log('result: ', result)
+  // console.log('result: ', result[0]['dataValues']['name'])
+  var file_name = id + '.csv'
+  // ctx.body = fs.stdout.toString()
+  // console.log('process.stdout.toString(): ', process.stdout.toString())
+  // console.log('process.stderr.toString(): ', process.stderr.toString())
+  ctx.type = 'stream'
+  ctx.attachment(result[0]['dataValues']['name']+'.csv')
+  ctx.body = createReadStream(path.resolve(__dirname, '../../csv', file_name))
+  // need to remove the csv file after download, since using 'append' for set header one time
+  const fs = require('fs')
+  fs.unlinkSync(path.resolve(__dirname, '../../csv', file_name))
+  next()
+}

src/routes/main-routes.js

@@ -59,6 +59,7 @@ export default router
   .get('/api/file/raw/get_attr_by_id/:channel/:id/:attr', controllers.recordingDataRaw.getAttrByID)
   .get('/api/file/raw/get_attr_by_ids/:channel/:id/:attr', controllers.recordingDataRaw.getAttrByIDs)
   .get('/api/file/raw/get_attr_by_parent/:channel/:parent/:attr', controllers.recordingDataRaw.getAttrByParent)
+  .get('/api/file/raw/get_csv/:id/:mode/:x_channel/:y_channel', controllers.recordingDataRaw.getCSV)
 
   // mini
   .post('/api/file/mini/create/:channel', controllers.recordingDataMini.create)