[update] babel src build copy python file to dist & port fix

package.json

@@ -7,7 +7,7 @@
   "scripts": {
     "start": "gulp nodemon",
     "dev": "gulp",
-    "build": "babel src -d dist",
+    "build": "babel src -d dist --copy-files",
     "production": "node dist/app.js",
     "test": "jest",
     "link": "eslint .",

src/controllers/analysis/index.js

@@ -0,0 +1,57 @@
+import RecordingDataMeta from '../../models/file/recording_data_meta'
+import childProcess from 'child_process'
+import path from 'path'
+
+const recordingDataMeta = new RecordingDataMeta()
+
+export const spikeDetect = async (ctx, next) => {
+  const request = ctx.request.body
+
+  const id = request.id
+  const channel = request.channel
+  const cutOffFreq = request.cut_off_freq
+  const threshold = request.threshold
+  const waveForm = request.waveForm
+  const retDataType = request.ret_data_type
+  // console.log(id, channel, cutOffFreq, threshold, waveForm)
+
+  const result = await recordingDataMeta.getByID(id)
+  const channels = JSON.parse(result[0].channels)
+  const sampleRate = 8e5 / result[0].parameter_set.ADC_CLOCK / channels.length
+  const timeDuration = result[0].time_duration / 1e6
+  // console.log(sampleRate, timeDuration)
+
+  const process = childProcess.spawnSync('python3', [
+    path.join(__dirname, '/spikeDetect.py'),
+    'meta_id=' + id,
+    'channel=' + channel,
+    'cut_off_freq=' + cutOffFreq,
+    'threshold=' + threshold,
+    'waveform=' + waveForm,
+    'sample_rate=' + sampleRate,
+    'time_duration=' + timeDuration,
+    'ret_data_type=' + retDataType
+  ], { maxBuffer: 2 * 1024 * 1024 * 1024 })
+
+  // console.log('out', process.stdout.toString())
+  console.log('err', process.stderr.toString())
+
+  ctx.body = process.stdout.toString()
+
+  // const process = child_process.execSync('python3 /Users/lubokai/Desktop/bioproapiserver/src/controllers/analysis/test.py 1089 0')
+  // console.log(process)
+  // process.stdout.on('data', (data) => {
+  //   _data += data.toString()
+  // })
+
+  // process.stderr.on('data', (data) => {
+  //   console.error(data.toString())
+  // })
+
+  // process.on('exit', (code) => {
+  //   console.log(_data)
+  //   console.log(`Child exited with code ${code}`)
+  // })
+
+  next()
+}

src/controllers/analysis/spikeDetect.py

@@ -0,0 +1,110 @@
+import sys
+import ast
+import psycopg2
+import numpy as np
+import pandas as pd
+
+from math import ceil, floor
+from json import loads, dumps
+from scipy.fft import fft, fftfreq, rfft, rfftfreq, irfft, ifft
+
+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]
+
+  meta_id = params['meta_id']
+  channel = params['channel']
+  cut_off_freq = int(params['cut_off_freq'])
+  threshold = ast.literal_eval('[' + params['threshold'] + ']')
+  waveform = ast.literal_eval('[' + params['waveform'] + ']')
+  SAMPLE_RATE = int(params['sample_rate'])
+  DURATION = ceil(float(params['time_duration']))
+  DELTA_TIME = 1e6 / SAMPLE_RATE
+  N = SAMPLE_RATE * DURATION
+  ret_data_type = params['ret_data_type']
+
+  # print(meta_id, channel, cut_off_freq, threshold, waveForm, SAMPLE_RATE, DURATION, N)
+
+  conn = psycopg2.connect(database="postgres", user="biopro", password="BioProControlBox", host="127.0.0.1", port="5432")
+  cur = conn.cursor()
+  sql_str = 'SELECT data FROM "public"."' + str(channel) + '_recording_data_raws" WHERE parent = ' + meta_id + 'ORDER BY id ASC'
+  cur.execute(sql_str)
+  ret = None
+  try:
+      ret = cur.fetchall()
+  except BaseException as e:
+    print(e)
+    exit
+  finally:
+      time = np.array([], dtype=int)
+      data = np.array([])
+
+      for sub_data in ret:
+        for split_data in sub_data[0].split('"***"')[:-1]:
+          value = np.array(split_data.split(' '))
+          time = np.append(time, value[::2].astype(np.int))
+          data = np.append(data, value[1::2].astype(np.int))
+
+      # print(len(time), len(data))
+      
+      xf = rfftfreq(N, 1 / SAMPLE_RATE)
+      points_per_freq = len(xf) / (SAMPLE_RATE / 2)
+      target_idx = int(points_per_freq * cut_off_freq)
+
+      # print(xf[target_idx])
+
+      yf = rfft(data)
+      yf[:target_idx] = 0
+      new_sig = np.round(irfft(yf, len(data)),3)
+
+      # print(time[-1])
+      # print(len(datas))
+      # print(len(yf),yf)
+
+      if threshold[0] == 'below':
+        th = new_sig < threshold[1]
+      elif threshold[0] == 'over':
+        th = new_sig > threshold[1]
+      elif threshold[0] == 'auto':
+        th = new_sig
+            
+      threshold_edges = np.convolve([1, -1], th, mode='same')
+      thresholded_edge_indices = np.where(threshold_edges==1)[0]
+
+      filter_array = []
+      for idx, point in enumerate(thresholded_edge_indices):
+        if idx > 0:
+          if time[thresholded_edge_indices[idx]] - int(waveform[2]) >= time[thresholded_edge_indices[idx-1]]:
+            filter_array.append(True)
+          elif time[thresholded_edge_indices[idx]] - int(waveform[1]) >= time[thresholded_edge_indices[idx-1]]:
+            filter_array.append(True)
+          else:
+            filter_array.append(False)
+        else:
+          if time[thresholded_edge_indices[idx]] - int(waveform[1]) < 0:
+            filter_array.append(False)
+          else:
+            filter_array.append(True)
+      
+      timemark_list = thresholded_edge_indices[filter_array]
+
+      if ret_data_type == 'all':
+        print(' '.join(str(x) for x in time))
+        print(' '.join(str(x) for x in new_sig))
+        print(' '.join(str(x) for x in timemark_list))
+      elif ret_data_type == 'partial':
+        return_sub_signal_lists = {}
+        for timemark in timemark_list:
+          start_index = timemark - floor(int(waveform[1] / DELTA_TIME))
+          end_index = timemark + ceil(int(waveform[2] / DELTA_TIME))
+
+          subSignal = new_sig[start_index: end_index + 2]
+          return_sub_signal_lists[str(timemark)] = ' '.join(str(x) for x in subSignal)
+        print(dumps(return_sub_signal_lists))
+
+  conn.commit()
+  conn.close()
+  exit

src/controllers/file/recording_data_raw.js

@@ -25,6 +25,16 @@ export const update = async (ctx, next) => {
   next()
 }
 
+export const updateSdData = async (ctx, next) => {
+  const index = ctx.params.channel
+  const data = ctx.request.body
+  const id = ctx.params.id
+  const result = await recordingDataRaw.updateSdData(data, id, index)
+  ctx.body = result
+
+  next()
+}
+
 export const updateByIDs = async (ctx, next) => {
   const index = ctx.params.channel
   const data = ctx.request.body

src/controllers/index.js

@@ -8,6 +8,7 @@ import * as task from './controller/task'
 import * as recordingDataMeta from './file/recording_data_meta'
 import * as recordingDataRaw from './file/recording_data_raw'
 import * as recordingDataMini from './file/recording_data_mini'
+import * as analysis from './analysis'
 
 export default {
   upload,
@@ -19,5 +20,6 @@ export default {
   task,
   recordingDataMeta,
   recordingDataRaw,
-  recordingDataMini
+  recordingDataMini,
+  analysis
 }

src/db/schema/recording_data_raw.model.js

@@ -43,6 +43,10 @@ module.exports = (sequelize, Sequelize, i) => {
       type: Sequelize.TEXT,
       defaultValue: ''
     },
+    sd_data: {
+      type: Sequelize.TEXT,
+      defaultValue: ''
+    },
     compressed: {
       type: Sequelize.BOOLEAN,
       defaultValue: false

src/models/file/recording_data_meta.js

@@ -62,9 +62,15 @@ class RecordingDataMeta {
           recordingDataRaw.clearDeleted(channel)
         }
         if (meta.mini_data[channel] != null) {
-          await recordingDataMini.updateByIDs({ deleted: true }, meta.mini_data[channel]['10'], channel)
-          await recordingDataMini.updateByIDs({ deleted: true }, meta.mini_data[channel]['100'], channel)
-          await recordingDataMini.updateByIDs({ deleted: true }, meta.mini_data[channel]['100'], channel)
+          if (meta.mini_data[channel]['10'] !== undefined) {
+            await recordingDataMini.updateByIDs({ deleted: true }, meta.mini_data[channel]['10'], channel)
+          }
+          if (meta.mini_data[channel]['100'] !== undefined) {
+            await recordingDataMini.updateByIDs({ deleted: true }, meta.mini_data[channel]['100'], channel)
+          }
+          if (meta.mini_data[channel]['1000'] !== undefined) {
+            await recordingDataMini.updateByIDs({ deleted: true }, meta.mini_data[channel]['1000'], channel)
+          }
           recordingDataMini.clearDeleted(channel)
         }
       }

src/models/file/recording_data_raw.js

@@ -20,6 +20,11 @@ class RecordingDataRaw {
     return ret
   }
 
+  async updateSdData (req, id, index) {
+    const ret = await db.sequelize.query(`UPDATE "public"."${index}_recording_data_raws" SET sd_data = sd_data || '${req.data}' WHERE id = ${String(id)}`)
+    return ret
+  }
+
   async updateByIDs (req, id, index) {
     const ret = await db[index + '_recording_data_raws'].update(
       req,

src/routes/main-routes.js

@@ -49,6 +49,7 @@ export default router
   // raw
   .post('/api/file/raw/create/:channel', controllers.recordingDataRaw.create)
   .post('/api/file/raw/update/:channel/:id', controllers.recordingDataRaw.update)
+  .post('/api/file/raw/update_sd_data/:channel/:id', controllers.recordingDataRaw.updateSdData)
   .get('/api/file/raw/clear_deleted/:channel/:id', controllers.recordingDataRaw.clearDeleted)
   .get('/api/file/raw/get_by_id/:channel/:id', controllers.recordingDataRaw.getByID)
   .get('/api/file/raw/get_by_ids/:channel/:id', controllers.recordingDataRaw.getByIDs)
@@ -106,3 +107,5 @@ export default router
 
   .post('/api/task/update', controllers.task.update)
   .post('/api/task/update_by_id/:id', controllers.task.updateByID)
+  // analyze
+  .post('/api/analysis/spike_detect', controllers.analysis.spikeDetect)