diff --git a/python/biopro/recording/file.py b/python/biopro/recording/file.py index 19d1d66..d170e82 100644 --- a/python/biopro/recording/file.py +++ b/python/biopro/recording/file.py @@ -8,6 +8,7 @@ import gc from biopro.util.json import JSON from biopro.util.stack import print_exception +from biopro.util.logger import calculate_time from .data import RecordingData from .loader import * @@ -965,7 +966,8 @@ 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', '_id_db_save') + '_data_time_ch_for_rl', '_device_id', '_send_data', '_data_mqtt_ch', '_id_db_save', '_raw_create_not_done', + '_mini_create_not_done') def __init__(self, meta: RecordingMetaFile, device_id, database = None): self._meta = meta @@ -1046,6 +1048,9 @@ class RecordingFileWriter: self._splitting_size = None + self._raw_create_not_done = True + self._mini_create_not_done = True + @property def meta_file(self) -> RecordingMetaFile: @@ -1259,7 +1264,8 @@ class RecordingFileWriter: self._data_db[c].append(str(v)) self._time_now = int(t) return - + + @calculate_time(1) def write(self, data: Union[bytes, RecordingData, List[bytes], List[RecordingData]], mqtt_thread) -> int: # check size ths = self.splitting_threshold_size @@ -1318,18 +1324,43 @@ class RecordingFileWriter: self.get_data_iter(d, mqtt_thread) - for ch in self._recording_file_dict: - if self._recording_file_dict[ch]._id_db == 0: - return None - - data_save = False - mini_save = False if len(self._recording_file_dict) > 0: for ch in self._data_db.keys(): if self._time_now - self._time_real_time[ch] > 1000000: self._send_data[ch] = True self._time_real_time[ch] = self._time_now - if self._time_now - self._time[ch] > 2000000: + + for ch in self._recording_file_dict: + if self._recording_file_dict[ch]._id_db == 0: + return None + + for ch in self._data_db.keys(): + for scale in self._mini_scale_list: + if self._recording_mini_dict[str(scale)][ch]._id_db == 0: + return None + + if self._raw_create_not_done : + self._raw_create_not_done = False + raw_id_dict = {} + for ch in self._data_db.keys(): + raw_id_dict[ch] = self._recording_file_dict[ch]._id_db + self._database.put_queue(['data_meta_update_all_channel_raw_id', self._meta._id_db, raw_id_dict]) + + if self._mini_create_not_done : + self._mini_create_not_done = False + mini_id_dict = {} + for ch in self._data_db.keys(): + mini_id_dict[ch] = {} + for scale in self._mini_scale_list: + mini_id_dict[ch][scale] = self._recording_mini_dict[str(scale)][ch]._id_db + print('self._meta._id_db, mini_id_dict', self._meta._id_db, mini_id_dict) + self._database.put_queue(['data_meta_update_all_channel_mini_id', self._meta._id_db, mini_id_dict]) + + data_save = False + mini_save = False + if len(self._recording_file_dict) > 0: + for ch in self._data_db.keys(): + if self._time_now - self._time[ch] > 5000000: 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) @@ -1342,12 +1373,14 @@ class RecordingFileWriter: self._meta._last_time = self._time_now self._splitting_size += write_sz data_save = True - if len(self._data_mini_ch[ch]['1000']['mean']) >= 10: + if len(self._data_mini_ch[ch]['1000']['mean']) >= 1: mini_save = True for scale in self._mini_scale_list: + str_mean = [str(int) for int in self._data_mini_ch[ch][str(scale)]['mean']] + data_mean = str(self._data_mini_ch[ch][str(scale)]['start_time']) + ' ' + ' '.join(str_mean) + '"***"' self._mini_save[str(scale)]['id'][ch] = self._recording_mini_dict[str(scale)][ch]._id_db self._mini_save[str(scale)]['start_time'][ch] = str(self._data_mini_ch[ch][str(scale)]['start_time']) - self._mini_save[str(scale)]['data_mean'][ch] = self._data_mini_ch[ch][str(scale)]['mean'].copy() + self._mini_save[str(scale)]['data_mean'][ch] = data_mean # self._mini_save[str(scale)]['data_random'][ch] = self._data_mini_ch[ch][str(scale)]['random'].copy() # self._mini_save[str(scale)]['data_bar'][ch] = self._data_mini_ch[ch][str(scale)]['bar'].copy() self._recording_mini_dict[str(scale)][ch].write(self._mini_save[str(scale)]['data_mean'][ch]) @@ -1364,15 +1397,16 @@ class RecordingFileWriter: if mini_save is True: if self._database is not None: for scale in self._mini_scale_list: - self._database.put_queue(['data_mini_recording', self._mini_save[str(scale)]['id'], self._channel_list, self._mini_save[str(scale)]['start_time'], self._mini_save[str(scale)]['data_mean']]) + self._database.put_queue(['data_mini_recording_new', copy(self._mini_save[str(scale)]['id']), self._channel_list, copy(self._mini_save[str(scale)]['data_mean'])]) self._meta.update_subfile(database = self._database) - del data return None def _switch_recording_file(self): + self._raw_create_not_done = True + 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 @@ -1400,6 +1434,7 @@ class RecordingFileWriter: return None def _switch_recording_mini(self, scale): + self._mini_create_not_done = True if len(self._recording_mini_dict[str(scale)]) > 0: for ch in self._recording_mini_dict[str(scale)].keys(): @@ -1428,10 +1463,10 @@ class RecordingFileWriter: def update_raw_dict_id(self, _channel, _id): self._recording_file_dict[int(_channel)]._id_db = _id - self._database.put_queue(['data_meta_update_channel_raw_id', self._meta._id_db, _channel, _id]) + # self._database.put_queue(['data_meta_update_channel_raw_id', self._meta._id_db, _channel, _id]) return None def update_mini_dict_id(self, _scale, _channel, _id): self._recording_mini_dict[str(_scale)][int(_channel)]._id_db = _id - self._database.put_queue(['data_meta_update_channel_mini_id', self._meta._id_db, _channel, _scale, _id]) + # self._database.put_queue(['data_meta_update_channel_mini_id', self._meta._id_db, _channel, _scale, _id]) return None \ No newline at end of file diff --git a/python/biopro/server/database.py b/python/biopro/server/database.py index bc64f46..6a6d20b 100644 --- a/python/biopro/server/database.py +++ b/python/biopro/server/database.py @@ -23,6 +23,7 @@ import biopro.server._identify from time import time from concurrent.futures import ThreadPoolExecutor, as_completed +from biopro.util.logger import calculate_time _RUNTIME_COMPILE = False @@ -69,6 +70,7 @@ class DataBaseProcess(Process): self._data_raw_create_sql_str = None self._data_raw_update_sql_str = None self._data_raw_recording_sql_str = 'UPDATE "public"."%s_recording_data_raws" SET data = concat(data, %s), end_time=%s, size=%s where id = %s' + self._data_mini_recording_sql_str = 'UPDATE "public"."%s_recording_data_minis" SET data_mean = concat(data_mean, %s) where id = %s' @property def db_host(self) -> str: @@ -85,13 +87,6 @@ class DataBaseProcess(Process): @property def api_token(self) -> str: return self._api_token - - def calculate_time(func): - def warp(*args, **kwargs): - now = time() - func(*args, **kwargs) - print(func.__name__,'cost:', time() - now) - return warp def run(self) -> None: self.setup() @@ -165,7 +160,7 @@ class DataBaseProcess(Process): getattr(self, do)(*args) return None - # @calculate_time + @calculate_time def data_meta_write(self, _data: JSON_OBJECT, device_id, _uuid = None, _id = None): if _id > 0: sql_str = 'UPDATE "public"."recording_data_metas" SET ' @@ -268,6 +263,7 @@ class DataBaseProcess(Process): # _conn.close() return None + # @calculate_time def data_meta_get(self, _path = None, _id = None): if _id is not None: sql_str = 'SELECT * FROM "public"."recording_data_metas" WHERE id = %s' @@ -302,6 +298,7 @@ class DataBaseProcess(Process): # _conn.close() return None + # @calculate_time def data_meta_id_get_by_path(self, _path = None): ret = None if _path is not None: @@ -382,16 +379,17 @@ class DataBaseProcess(Process): # @calculate_time def data_raw_recording_new(self, _id_dict, _channel_list, _data_dict, _end_time_dict, _size_dict): try: + para_list = [] + for _channel in _channel_list: + para_list.append([_channel, _data_dict[_channel] + '"***"', _end_time_dict[_channel], _size_dict[_channel], _id_dict[_channel]]) with self._psql_conn as conn: with conn.cursor() as sql_cursor: - para_list = [] - for _channel in _channel_list: - para_list.append([_channel, _data_dict[_channel] + '"***"', _end_time_dict[_channel], _size_dict[_channel], _id_dict[_channel]]) execute_batch(sql_cursor, self._data_raw_recording_sql_str, para_list) except psycopg2.Error as e: print('recording error', e) return None - + + # @calculate_time def data_mini_create(self, _data_dict, _channel_list, _scale, device_id): for _channel in _channel_list: sql_str = 'INSERT INTO "public"."' + str(_channel) + '_recording_data_minis" (' @@ -425,6 +423,7 @@ class DataBaseProcess(Process): # _conn.close() return True + # @calculate_time def data_mini_recording(self, _id_dict, _channel_list, _start_time_dict, _data_mean_dict): with self._psql_conn as conn: with self._psql_conn.cursor() as sql_cursor: @@ -444,9 +443,23 @@ class DataBaseProcess(Process): for _channel in _channel_list: del _start_time_dict[_channel], _data_mean_dict[_channel] gc.collect() - + return True + # @calculate_time + def data_mini_recording_new(self, _id_dict, _channel_list, _data_mean_dict): + try: + para_list = [] + for _channel in _channel_list: + para_list.append([_channel, _data_mean_dict[_channel], _id_dict[_channel]]) + with self._psql_conn as conn: + with conn.cursor() as sql_cursor: + execute_batch(sql_cursor, self._data_mini_recording_sql_str, para_list) + except psycopg2.Error as e: + print('recording error', e) + return None + + # @calculate_time def data_mini_update(self, _id, _channel, _data: JSON_OBJECT): sql_str = 'UPDATE "public"."' + str(_channel) + '_recording_data_minis" SET ' sql_set = [] @@ -481,26 +494,62 @@ class DataBaseProcess(Process): print('e', e) return True - + + # @calculate_time def data_meta_update_channel_raw_id(self, meta_id, channel, raw_id): try: + with self._psql_conn as conn: + with conn.cursor() as sql_cursor: + sql_str = "UPDATE recording_data_metas SET raw_data = jsonb_insert(raw_data, '{%s, -1}', '%s', TRUE) WHERE id = %s;" + sql_set = [channel, raw_id, 571] + sql_cursor.execute(sql_str, sql_set) + except psycopg2.Error as e: + print('e', e) + + return True + + # @calculate_time + def data_meta_update_all_channel_raw_id(self, meta_id:str, raw_id_dict:dict): + try: + sql_str = "UPDATE recording_data_metas SET raw_data = " + sql_set = [] + for ch in raw_id_dict.keys(): + sql_str += "jsonb_insert(raw_data -> %s, '{-1}', '%s', TRUE) ||" + sql_set.extend([ch, raw_id_dict[ch]]) + sql_str[:-2] + "WHERE id = %s;" + sql_set.append(meta_id) + print(sql_str, sql_set) with self._psql_conn as conn: with conn.cursor() as sql_cursor: - sql_str = "UPDATE recording_data_metas SET raw_data = jsonb_insert(raw_data, '{%s, -1}', '%s', TRUE) WHERE id = %s;" - sql_set = [channel, raw_id, meta_id] + execute_batch(sql_cursor, sql_str, sql_set) + except psycopg2.Error as e: + print('e', e) + + return True + + def data_meta_update_channel_mini_id(self, meta_id, channel, scale, mini_id): + try: + with self._psql_conn as conn: + with conn.cursor() as sql_cursor: + sql_str = "UPDATE recording_data_metas SET mini_data = jsonb_insert(mini_data, '{%s, %s, -1}', '%s', TRUE) WHERE id = %s;" + sql_set = [channel, scale, mini_id, meta_id] sql_cursor.execute(sql_str, sql_set) except psycopg2.Error as e: print('e', e) return True - def data_meta_update_channel_mini_id(self, meta_id, channel, scale, raw_id): + # @calculate_time + def data_meta_update_all_channel_mini_id(self, meta_id, mini_id_dict): try: + sql_str = "UPDATE recording_data_metas SET mini_data = jsonb_insert(mini_data, '{%s, %s, -1}', '%s', TRUE) WHERE id = %s;" + sql_set = [] + for ch in mini_id_dict.keys(): + for scale in mini_id_dict[ch].keys(): + sql_set.append([ch, scale, mini_id_dict[ch][scale], 571]) with self._psql_conn as conn: - with conn.cursor() as sql_cursor: - sql_str = "UPDATE recording_data_metas SET mini_data = jsonb_insert(mini_data, '{%s, %s, -1}', '%s', TRUE) WHERE id = %s;" - sql_set = [channel, scale, raw_id, meta_id] - sql_cursor.execute(sql_str, sql_set) + with conn.cursor() as sql_cursor: + execute_batch(sql_cursor, sql_str, sql_set) except psycopg2.Error as e: print('e', e) diff --git a/python/biopro/util/logger.py b/python/biopro/util/logger.py index 96c6d31..eb33799 100644 --- a/python/biopro/util/logger.py +++ b/python/biopro/util/logger.py @@ -3,6 +3,7 @@ import inspect import os import sys import threading as _threading +from time import time from datetime import datetime from functools import wraps from typing import Any, Optional, Tuple, Union @@ -520,3 +521,15 @@ def logging_error(f): return f(self, *args, **kwargs) return _logger + + +def calculate_time(func, time_limit=None): + def warp(*args, **kwargs): + now = time() + func(*args, **kwargs) + if time_limit == None: + print(func.__name__,'cost:', time() - now) + else: + if time() - now > time_limit: + print(func.__name__,'cost:', time() - now) + return warp