[debug]

src/controllers/file/download_functional.py

@@ -1,27 +1,33 @@
 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]))
     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]
@@ -29,77 +35,109 @@ def integral(df, x_channel, y_channel, pre_last_x, pre_last_y, integral_result):
     integral_result = df[f'{x_channel}_{y_channel}_integral'].iat[-1]
     # print('next run info: ', pre_last_x, pre_last_y, integral_result)
     return df, pre_last_x, pre_last_y, integral_result
+
 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")
-    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'
-    cur.execute(sql_str)
+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)
+    cursor = conn.cursor()
+    # 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()
+    cursor.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()
+        raw_data_list = cursor.fetchall()[0][0]
+        channels = list(raw_data_list.keys()) # [0,1,2,3]
+        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 channel in channels:
-                sql_str = f'select "data" from "{channel}_recording_data_raws" WHERE id = {raw_data_list[channel][raw_data_pages]}'
-                cur.execute(sql_str)
-                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 = 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'
-                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'
-                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)
+        print('raw_data_list', raw_data_list)
+        print('channels', channels)
+        for page in range(len(raw_data_list[channels[0]])):
             df = pd.DataFrame()
+            print(raw_data_list[channels[0]][page])
+            for channel in channels:
+                raw_data_page = raw_data_list[channel][page]
+                sql_str = f'SELECT "data" FROM "{channel}_recording_data_raws" WHERE id = {raw_data_list[channel][page]}'
+                sql_start = time.time()
+                cursor.execute(sql_str)
+                sql_total += (time.time() - sql_start)
+                data_value = cursor.fetchall()[0][0].replace('"***"', ' ').split(' ')
+                if ~('Time' in df.columns):
+                    df['Time'] = data_value[:-1:2]
+                df[channel] = data_value[1:-1:2]
+            # df = df.loc[lambda x: x.index % 2 == 1].reset_index(drop=True)
+            # 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') # coerce force non-a-number string be converted to NaN
+            if mode == 'Log':
+                df = log(df, x_channel)
+            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':
+                df, pre_last_x, pre_last_y = differential(df, x_channel, y_channel, pre_last_x, pre_last_y)
+            df_all = pd.concat([df_all, df], ignore_index=True, sort=False)
+            print(df_all)
+            with open(f"../../csv/{id}.csv", mode = 'a') as export_file:
+                df_all.to_csv(export_file, header=(export_file.tell() == 0))
+            df_all = pd.DataFrame()
+        print('total sql: ', sql_total)
     except BaseException as e:
         print(e)
         exit
+    return
+
+if __name__ == '__main__':
+    params = {
+        'id': 0,
+        'mode': 0,
+        'x_channel': 0,
+        'y_channel': 0
+    }
+    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(849, '', '', '')
+      # 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)