[update] 256K -> 64K, 5s -> 2s, save id_db

python/biopro/recording/file.py

@@ -996,7 +996,7 @@ class RecordingFileWriter:
                  '_data_value_ch', '_close', '_data_mini_ch',
                  '_mini_scale_list', '_time_real_time', '_data_rl', '_data_db',
                  '_raw_save', '_mini_save', '_data_time_ch', '_data_value_ch_for_rl',
-                 '_data_time_ch_for_rl', '_device_id', '_send_data', '_data_mqtt_ch')
+                 '_data_time_ch_for_rl', '_device_id', '_send_data', '_data_mqtt_ch', '_id_db_save')
 
     def __init__(self, meta: RecordingMetaFile, device_id, database = None):
         self._meta = meta
@@ -1052,6 +1052,7 @@ class RecordingFileWriter:
         self._data_value_ch_for_rl = {}
         self._data_time_ch_for_rl = {}
         self._data_time_ch = {}
+        self._id_db_save = {}
 
         # mini data
         self._data_mini_ch = {}
@@ -1068,7 +1069,7 @@ class RecordingFileWriter:
 
         # splitting
         self.splitting_threshold_time = 30 * 60 * 1000  # one minute
-        self.splitting_threshold_size = 16 * 16 * 1024  #  16 * 16KB
+        self.splitting_threshold_size = 4 * 16 * 1024  #  16 * 16KB
 
         self._writer_batch_size = 8192 
 
@@ -1123,7 +1124,7 @@ class RecordingFileWriter:
                     self._meta._size += self._recording_file_dict[ch]._size
             # self._data_db.clear()
             if self._database is not None:
-                self._database.put_queue(['data_raw_recording', self._raw_save['id'], self._channel_list, self._raw_save['data']])
+                self._database.put_queue(['data_raw_recording', self._raw_save['id'], self._channel_list, self._raw_save['data'], self._id_db_save])
                 # self._database.put_queue(['data_raw_recording', self._raw_save['id'], self._channel_list, self._raw_save['data']])
             self._recording_file_dict.clear()
         for scale in self._mini_scale_list:
@@ -1365,7 +1366,7 @@ class RecordingFileWriter:
                     # mqtt_thread[ch].on_message(mes)
                     # self._data_rl[ch].clear()
                     # # self._time_real_time[ch] = self._time_now
-                    if self._time_now - self._time[ch] > 5000000:
+                    if self._time_now - self._time[ch] > 2000000:
                         if self._recording_file_dict[ch]._status:
                             _data = ' '.join(self._data_db[ch])
                             write_sz = self._recording_file_dict[ch].write(_data, self._channel_list)
@@ -1394,7 +1395,9 @@ class RecordingFileWriter:
 
                 if data_save is True:
                     if self._database is not None:
-                        self._database.put_queue(['data_raw_recording', self._raw_save['id'], self._channel_list, self._raw_save['data']])
+                        # print('raw_save', self._raw_save['id'])
+                        # print('id_save', self._id_db_save)
+                        self._database.put_queue(['data_raw_recording', self._raw_save['id'], self._channel_list, self._raw_save['data'], self._id_db_save])
                         self._meta.update_subfile(database = self._database)
                 if mini_save is True:
                     if self._database is not None:
@@ -1414,6 +1417,7 @@ class RecordingFileWriter:
         
         if len(self._recording_file_dict) > 0:
             for ch in self._recording_file_dict.keys():
+                self._id_db_save[ch] = self._recording_file_dict[ch]._id_db
                 self._recording_file_dict[ch].close(self._time_now)
                 self._meta._size += self._recording_file_dict[ch]._size
                 

python/biopro/server/database.py

@@ -15,11 +15,12 @@ from biopro.util.json import JSON_OBJECT
 from .socket import ServerThread
 
 import psycopg2
-# from psycopg2.pool import ThreadedConnectionPool
+from psycopg2.extras import execute_batch
 from multiprocessing import Process, Queue
 
 import biopro.server._identify
 
+from time import time
 
 from concurrent.futures import ThreadPoolExecutor, as_completed
 
@@ -308,22 +309,16 @@ class DataBaseProcess(Process):
         return None
 
     def data_raw_create(self, _data_dict, _channel_list, device_id):
-
         for _channel in _channel_list:
             sql_str = 'INSERT INTO "public"."' + str(_channel) + '_recording_data_raws" ('
             sql_set = []
-            for item in _data_dict[_channel].keys():
-                sql_str = sql_str + str(item) + ', '
-                if isinstance(_data_dict[_channel][item], dict):
-                    sql_set.append(json_stringify(_data_dict[_channel][item]))
-                else:
-                    sql_set.append(str(_data_dict[_channel][item]))
 
-            sql_str = sql_str[0:-2] + ') VALUES ('
-            for item in _data_dict[_channel].keys():
-                sql_str = sql_str + '%s, '
+            key_list = _data_dict[_channel].keys()
+            key_str = ','.join(key_list)
+            sql_set = list(map(lambda x: str(x), _data_dict[_channel].values()))
+            values = ','.join(['%s'] * len(key_list))
 
-            sql_str = sql_str[0:-2] + ') RETURNING id'
+            sql_str = sql_str + key_str + ') VALUES (' + values + ') RETURNING id'
             sql_cursor = self._psql_conn.cursor()
 
             try:
@@ -360,25 +355,32 @@ class DataBaseProcess(Process):
         
         return None
 
-    def data_raw_recording(self, _id_dict, _channel_list, _data_dict):
-        for _channel in _channel_list:
-            compressed_value = _data_dict[_channel] + '"***"'
-            # compressed_value = bytes(_data_dict[_channel],'utf-8')
-            # compressed_value = zlib.compress(compressed_value)
-            # compressed_value = base64.b64encode(compressed_value).decode() + '"***"'
-            # compressed_value = _data + '"***"'
-            sql_str = 'UPDATE "public"."' + str(_channel) + '_recording_data_raws" SET data = concat(data, %s) where id = %s'
-            sql_set = [compressed_value, _id_dict[_channel]]
-
-            sql_cursor = self._psql_conn.cursor()
-            sql_cursor.execute(sql_str, sql_set)
-            self._psql_conn.commit()
-            sql_cursor.close()
-
-            del _data_dict[_channel]
-            del compressed_value
-            gc.collect()
-            # _conn.close()
+    def data_raw_recording(self, _id_dict, _channel_list, _data_dict,  _id_db_save):
+        para_list = []
+        try:
+            ### execute way
+            # with self._psql_conn as conn:
+            #     with conn.cursor() as sql_cursor:
+            #         for _channel in _channel_list:
+            #             temp_id = _id_dict[_channel]
+            #             if temp_id == 0:
+            #                 temp_id = _id_db_save[_channel]
+            #             # print(_channel, temp_id)
+            #             sql_str = 'UPDATE "public"."%s_recording_data_raws" SET data = concat(data, %s) where id = %s'
+            #             sql_set = [_channel, _data_dict[_channel] + '"***"', temp_id]
+            #             sql_cursor.execute(sql_str, sql_set)
+            ### execute_batch
+            with self._psql_conn as conn:
+                with conn.cursor() as sql_cursor:
+                    sql_str = 'UPDATE "public"."%s_recording_data_raws" SET data = concat(data, %s) where id = %s'
+                    for _channel in _channel_list:
+                        temp_id = _id_dict[_channel]
+                        if temp_id == 0:
+                            temp_id = _id_db_save[_channel]
+                        para_list.append((_channel, _data_dict[_channel] + '"***"', temp_id))
+                    execute_batch(sql_cursor,sql_str, para_list)
+        except psycopg2.Error as e:
+            print('recording error', e)
         return None
 
     def data_mini_create(self, _data_dict, _channel_list, _scale, device_id):
@@ -415,38 +417,24 @@ class DataBaseProcess(Process):
         return True
 
     def data_mini_recording(self, _id_dict, _channel_list, _start_time_dict, _data_mean_dict):
-        for _channel in _channel_list:
-            data = _data_mean_dict.get(_channel)
-            if data is not None:
-                if len(data) != 0:
-                    str_mean = [str(int) for int in _data_mean_dict[_channel]]
-                    data_mean = _start_time_dict[_channel] + ' ' + ' '.join(str_mean)
-                    # data_random = _start_time_dict[_channel] + ' ' + ' '.join(_data_random_dict[_channel])
-                    # data_bar = _start_time_dict[_channel] + ' ' + ' '.join(_data_bar_dict[_channel])
+        with self._psql_conn as conn:
+            with self._psql_conn.cursor() as sql_cursor:
+                for _channel in _channel_list:
+                    data = _data_mean_dict.get(_channel)
+                    if data is not None:
+                        if len(data) != 0:
+                            str_mean = [str(int) for int in _data_mean_dict[_channel]]
+                            data_mean = _start_time_dict[_channel] + ' ' + ' '.join(str_mean)
 
-                    compressed_mean = data_mean + '"***"'
-
-                    # compressed_mean = bytes(data_mean,'utf-8')
-                    # compressed_mean = zlib.compress(compressed_mean)
-                    # compressed_mean = base64.b64encode(compressed_mean).decode() + '"***"'
-
-                    # compressed_random = bytes(data_bar,'utf-8')
-                    # compressed_random = zlib.compress(compressed_random)
-                    # compressed_random = base64.b64encode(compressed_random).decode() + '"***"'
-
-                    # compressed_random = bytes(data_random,'utf-8')
-                    # compressed_random = zlib.compress(compressed_random)
-                    # compressed_random = base64.b64encode(compressed_random).decode() + '"***"'
                     sql_str = 'UPDATE "public"."' + str(_channel) + '_recording_data_minis" SET data_mean = concat(data_mean, %s) where id = %s'
-                    sql_set = [compressed_mean, _id_dict[_channel]]
-
-                    sql_cursor = self._psql_conn.cursor()
+                    sql_set = [data_mean + '"***"', _id_dict[_channel]]
                     sql_cursor.execute(sql_str, sql_set)
-                    self._psql_conn.commit()
-                    sql_cursor.close()
 
-                    del str_mean, _start_time_dict[_channel], _data_mean_dict[_channel], data_mean, compressed_mean
-                    gc.collect()
+                    del str_mean, data_mean
+        
+        for _channel in _channel_list:
+            del _start_time_dict[_channel], _data_mean_dict[_channel]
+        gc.collect()
         
         return True