wisetop/controller-wisetopdataserver
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[update] meta_id & mini_id update

Browse Source
This commit is contained in:
peterlu14
2022-03-28 16:11:08 +08:00
parent 952403cbf4
commit b253aca73f
3 changed files with 133 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/biopro/recording/file.py
+50 -15
View File
@@ -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
python/biopro/server/database.py
+70 -21
View File
@@ -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)
python/biopro/util/logger.py
+13
View File
@@ -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