peterlu14
583 lines
21 KiB
Python
583 lines
21 KiB
Python
import base64
|
|
import re
|
|
import zlib
|
|
import sys
|
|
import gc
|
|
|
|
from json import loads as json_parse, dumps as _json_stringify
|
|
from typing import Dict, Optional, Any
|
|
from datetime import datetime
|
|
import threading
|
|
|
|
from biopro.recording.file import RecordingMetaFile, RecordingFile, RecordingMini
|
|
|
|
from biopro.util.json import JSON_OBJECT
|
|
from .socket import ServerThread
|
|
|
|
import psycopg2
|
|
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
|
|
from biopro.util.logger import calculate_time
|
|
|
|
_RUNTIME_COMPILE = False
|
|
|
|
|
|
def json_stringify(o: Any) -> str:
|
|
return _json_stringify(o, separators=(',', ':'))
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class DataBaseProcess(Process):
|
|
"""
|
|
connect to DB and send sql command
|
|
"""
|
|
|
|
def __init__(self, client_id: str, log_verbose, queue_db, queue_ds, queue_error, name = 'Database-Process'):
|
|
super(DataBaseProcess, self).__init__(name = name)
|
|
# super().__init__('DATABASE-' + client_id, MAGENTA)
|
|
|
|
self._db_host = '127.0.0.1'
|
|
self._db_name = 'postgres'
|
|
self._db_user = 'biopro'
|
|
self._db_password = 'BioProControlBox'
|
|
|
|
self._api_host = biopro.server._identify.API_SERVER_HOST
|
|
self._api_port = biopro.server._identify.API_SERVER_PORT
|
|
self._api_token = None
|
|
self._api_header = None
|
|
self.log_verbose = log_verbose
|
|
self._executor = ThreadPoolExecutor(max_workers=8)
|
|
|
|
self._psql_multi_thread = None
|
|
self._psql_conn_pool = None
|
|
self._psql_conn = None
|
|
|
|
self._queue_db = queue_db
|
|
self._queue_ds = queue_ds
|
|
self._queue_error = queue_error
|
|
|
|
self._is_close = False
|
|
|
|
|
|
self._meta_write_sql_str = None
|
|
|
|
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) where id = %s'
|
|
self._new_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:
|
|
return self._db_host
|
|
|
|
@property
|
|
def api_host(self) -> str:
|
|
return self._api_host
|
|
|
|
@property
|
|
def api_port(self) -> int:
|
|
return self._api_port
|
|
|
|
@property
|
|
def api_token(self) -> str:
|
|
return self._api_token
|
|
|
|
def run(self) -> None:
|
|
self.setup()
|
|
self.routine()
|
|
|
|
def setup(self) -> None:
|
|
self.log_verbose('DataBaseProcess setup')
|
|
conn_info = "host={0} user={1} dbname={2} password={3}".format(self._db_host, self._db_user, self._db_name, self._db_password)
|
|
try:
|
|
self._psql_conn = psycopg2.connect(conn_info)
|
|
except:
|
|
self.log_verbose('can not connect to database')
|
|
else:
|
|
self.log_verbose('connect to DB ', self._db_host, self._db_name)
|
|
self._is_close = False
|
|
|
|
def routine(self) -> None:
|
|
while self._is_close is False:
|
|
# request = [ function_name, *args]
|
|
request = None
|
|
try:
|
|
request = self._queue_db.get()
|
|
except KeyboardInterrupt:
|
|
self._is_close = True
|
|
if request is not None:
|
|
func_name = request[0]
|
|
args = request[1:]
|
|
if request[0] != 'data_meta_write':
|
|
try:
|
|
self._executor.submit(self.solve_for, func_name, args)
|
|
except RuntimeError as e:
|
|
self.log_verbose('can not submit database sql:', func_name, e)
|
|
else:
|
|
try:
|
|
self.solve_for(func_name, args)
|
|
except RuntimeError as e:
|
|
self.log_verbose('can not submit database sql:', func_name, e)
|
|
del request
|
|
return None
|
|
|
|
def shutdown(self) -> None:
|
|
if self._psql_conn is not None:
|
|
try:
|
|
self._psql_conn.close()
|
|
except:
|
|
self.log_verbose('can not close database')
|
|
else:
|
|
self.log_verbose('closed DB ', self._db_host, self._db_name)
|
|
self._is_close = True
|
|
|
|
def put_queue(self, request: list) -> None:
|
|
self._queue_db.put(request)
|
|
del request
|
|
return
|
|
|
|
def get_queue(self, device_id) -> None:
|
|
task = None
|
|
try:
|
|
task = self._queue_ds[int(device_id)].get()
|
|
except:
|
|
# Handle empty queue here
|
|
pass
|
|
else:
|
|
# Handle task here and call q.task_done()
|
|
pass
|
|
return task
|
|
|
|
def solve_for(self, name: str, args: list):
|
|
do = f"{name}"
|
|
if callable(getattr(self, do)):
|
|
getattr(self, do)(*args)
|
|
return None
|
|
|
|
# @calculate_time(1)
|
|
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 '
|
|
sql_set = []
|
|
for item in _data.keys():
|
|
sql_str = sql_str + str(item) + ' = %s, '
|
|
if isinstance(_data[item], dict):
|
|
sql_set.append(json_stringify(_data[item]))
|
|
else:
|
|
sql_set.append(_data[item])
|
|
|
|
sql_str = sql_str[0:-2] + ' WHERE (id = %s)'
|
|
sql_set.append(_id)
|
|
sql_cursor = self._psql_conn.cursor()
|
|
try:
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
except:
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
self._queue_error.put(device_id)
|
|
finally:
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
return True
|
|
|
|
if _uuid is not None:
|
|
sql_str = 'SELECT id FROM "public"."recording_data_metas" WHERE uuid = %s'
|
|
sql_set = [_uuid, ]
|
|
sql_cursor = self._psql_conn.cursor()
|
|
sql_id = None
|
|
|
|
# try:
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
sql_id = sql_cursor.fetchone()
|
|
# except RuntimeError as e:
|
|
# print(e)
|
|
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
if sql_id is None:
|
|
sql_str = 'INSERT INTO "public"."recording_data_metas" ('
|
|
sql_set = []
|
|
for item in _data.keys():
|
|
sql_str = sql_str + str(item) + ', '
|
|
if isinstance(_data[item], dict):
|
|
sql_set.append(json_stringify(_data[item]))
|
|
else:
|
|
sql_set.append(_data[item])
|
|
|
|
sql_str = sql_str[0:-2] + ') VALUES ('
|
|
for item in _data.keys():
|
|
sql_str = sql_str + '%s, '
|
|
|
|
sql_str = sql_str[0:-2] + ') RETURNING id'
|
|
sql_cursor = self._psql_conn.cursor()
|
|
|
|
try:
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
except:
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
self._queue_error.put(device_id)
|
|
finally:
|
|
self._queue_ds[int(device_id)].put(['update_meta_id', _data['uuid'], int(sql_cursor.fetchone()[0])])
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
else:
|
|
sql_str = 'UPDATE "public"."recording_data_metas" SET '
|
|
sql_set = []
|
|
for item in _data.keys():
|
|
sql_str = sql_str + str(item) + ' = %s, '
|
|
if isinstance(_data[item], dict):
|
|
sql_set.append(json_stringify(_data[item]))
|
|
else:
|
|
sql_set.append(_data[item])
|
|
|
|
sql_str = sql_str[0:-2] + ' WHERE (id = %s)'
|
|
if isinstance(sql_id, tuple):
|
|
sql_id = sql_id[0]
|
|
sql_set.append(sql_id)
|
|
sql_cursor = self._psql_conn.cursor()
|
|
|
|
try:
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
except:
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
self._queue_error.put(device_id)
|
|
finally:
|
|
self._queue_ds[int(device_id)].put(['update_meta_id', _data['uuid'], int(sql_id)])
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
# _conn.close()
|
|
return True
|
|
|
|
# _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'
|
|
sql_set = [str(_id), ]
|
|
sql_cursor = self._psql_conn.cursor()
|
|
|
|
ret = None
|
|
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
ret = sql_cursor.fetchone()
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
|
|
return ret
|
|
|
|
elif _path is not None:
|
|
sql_str = 'SELECT * FROM "public"."recording_data_metas" WHERE path = %s'
|
|
sql_set = [str(_path), ]
|
|
sql_cursor = self._psql_conn.cursor()
|
|
|
|
ret = None
|
|
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
ret = sql_cursor.fetchone()
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
|
|
return ret
|
|
|
|
# _conn.close()
|
|
return None
|
|
|
|
# @calculate_time()
|
|
def data_meta_id_get_by_path(self, _path = None):
|
|
ret = None
|
|
if _path is not None:
|
|
sql_str = 'SELECT * FROM "public"."recording_data_metas" WHERE path = %s'
|
|
sql_set = [str(_path), ]
|
|
sql_cursor = self._psql_conn.cursor()
|
|
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
ret = 0
|
|
try:
|
|
ret = int(sql_cursor.fetchone()[0])
|
|
except:
|
|
return None
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
return ret
|
|
|
|
# _conn.close()
|
|
return None
|
|
|
|
# @calculate_time()
|
|
def data_raw_create(self, _data_dict, _channel_list, device_id):
|
|
if self._data_raw_create_sql_str == None:
|
|
sql_str_list = []
|
|
key_list = _data_dict[_channel_list[0]].keys()
|
|
key_str = ','.join(key_list)
|
|
values = ','.join(['%s'] * len(key_list))
|
|
sql_str_list = ['INSERT INTO "public"."{channel}_recording_data_raws" (', key_str, ') VALUES (', values, ') RETURNING id']
|
|
self._data_raw_create_sql_str = ''.join(sql_str_list)
|
|
try:
|
|
with self._psql_conn as conn:
|
|
with conn.cursor() as sql_cursor:
|
|
for _channel in _channel_list:
|
|
sql_set = list(map(lambda x: str(x), _data_dict[_channel].values()))
|
|
sql_cursor.execute(self._data_raw_create_sql_str.format(channel = _channel), sql_set)
|
|
self._queue_ds[int(device_id)].put(['update_raw_dict_id', _data_dict[_channel]['uuid'], _channel, int(sql_cursor.fetchone()[0])])
|
|
except (psycopg2.Error, RuntimeError) as e:
|
|
print('data_raw_create error', e)
|
|
self._queue_error.put(device_id)
|
|
|
|
return True
|
|
|
|
# @calculate_time()
|
|
def data_raw_update(self, _id, _channel, _data: JSON_OBJECT):
|
|
sql_str = 'UPDATE "public"."' + str(_channel) + '_recording_data_raws" SET '
|
|
sql_set = []
|
|
for item in _data.keys():
|
|
sql_str = sql_str + str(item) + ' = %s, '
|
|
if isinstance(_data[item], dict):
|
|
sql_set.append(json_stringify(_data[item]))
|
|
else:
|
|
sql_set.append(_data[item])
|
|
|
|
sql_str = sql_str[0:-2] + ' WHERE (id = %s)'
|
|
sql_set.append(_id)
|
|
|
|
sql_cursor = self._psql_conn.cursor()
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
return None
|
|
|
|
# @calculate_time()
|
|
def data_raw_recording(self, _id_dict, _channel_list, _data_dict, _id_db_save):
|
|
try:
|
|
with self._psql_conn as conn:
|
|
with conn.cursor() as sql_cursor:
|
|
para_list = []
|
|
for _channel in _channel_list:
|
|
temp_id = _id_dict[_channel] if _id_dict[_channel] != 0 else _id_db_save[_channel]
|
|
para_list.append([_channel, _data_dict[_channel] + '"***"', temp_id])
|
|
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(1)
|
|
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:
|
|
execute_batch(sql_cursor, self._new_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" ('
|
|
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, '
|
|
|
|
sql_str = sql_str[0:-2] + ') RETURNING id'
|
|
sql_cursor = self._psql_conn.cursor()
|
|
|
|
try:
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
except RuntimeError as e:
|
|
print('data_mini_create error', e)
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
self._queue_error.put(device_id)
|
|
finally:
|
|
self._queue_ds[int(device_id)].put(['update_mini_dict_id', _data_dict[_channel]['uuid'], _scale, _channel, int(sql_cursor.fetchone()[0])])
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
# _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:
|
|
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)
|
|
|
|
sql_str = 'UPDATE "public"."' + str(_channel) + '_recording_data_minis" SET data_mean = concat(data_mean, %s) where id = %s'
|
|
sql_set = [data_mean + '"***"', _id_dict[_channel]]
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
|
|
del str_mean, data_mean
|
|
|
|
for _channel in _channel_list:
|
|
del _start_time_dict[_channel], _data_mean_dict[_channel]
|
|
gc.collect()
|
|
|
|
return True
|
|
|
|
# @calculate_time(1)
|
|
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 = []
|
|
for item in _data.keys():
|
|
sql_str = sql_str + str(item) + ' = %s, '
|
|
if isinstance(_data[item], dict):
|
|
sql_set.append(json_stringify(_data[item]))
|
|
else:
|
|
sql_set.append(_data[item])
|
|
|
|
sql_str = sql_str[0:-2] + ' WHERE (id = %s)'
|
|
sql_set.append(_id)
|
|
|
|
sql_cursor = self._psql_conn.cursor()
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
|
|
return True
|
|
|
|
def data_meta_update_raw_empty_channel(self, meta_id, channel_list):
|
|
try:
|
|
sql_str = "UPDATE recording_data_metas SET raw_data = jsonb_set(raw_data, '{%s}', '[]'), mini_data = jsonb_set(mini_data, '{%s}', '{\"10\": [], \"100\": [], \"1000\": []}') WHERE id = %s;"
|
|
sql_set = []
|
|
for channel in channel_list:
|
|
sql_set.append([channel, channel, meta_id])
|
|
|
|
with self._psql_conn as conn:
|
|
with conn.cursor() as sql_cursor:
|
|
execute_batch(sql_cursor, sql_str, sql_set)
|
|
except psycopg2.Error as e:
|
|
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 = jsonb_insert(raw_data, '{%s, -1}', '%s', TRUE) WHERE id = %s;"
|
|
sql_set = []
|
|
for ch in raw_id_dict.keys():
|
|
sql_set.append([ch, raw_id_dict[ch], meta_id])
|
|
with self._psql_conn as conn:
|
|
with conn.cursor() as sql_cursor:
|
|
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
|
|
|
|
# @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], meta_id])
|
|
with self._psql_conn as conn:
|
|
with conn.cursor() as sql_cursor:
|
|
execute_batch(sql_cursor, sql_str, sql_set)
|
|
except psycopg2.Error as e:
|
|
print('e', e)
|
|
|
|
return True
|
|
|
|
def project_insert(self, device_id, _data):
|
|
sql_str = 'INSERT INTO "public"."meta_project_info" ('
|
|
sql_set = []
|
|
for item in _data.keys():
|
|
sql_str = sql_str + str(item) + ', '
|
|
if isinstance(_data[item], dict) or isinstance(_data[item], list):
|
|
sql_set.append(json_stringify(_data[item]))
|
|
else:
|
|
sql_set.append(_data[item])
|
|
|
|
sql_str = sql_str[0:-2] + ') VALUES ('
|
|
for item in _data.keys():
|
|
sql_str = sql_str + '%s, '
|
|
|
|
sql_str = sql_str[0:-2] + ') RETURNING id'
|
|
sql_cursor = self._psql_conn.cursor()
|
|
|
|
try:
|
|
sql_cursor.execute(sql_str, sql_set)
|
|
except BaseException as e:
|
|
print('insert error', e)
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|
|
self._queue_error.put()
|
|
finally:
|
|
self._queue_ds[int(device_id)].put(['project_id', int(sql_cursor.fetchone()[0])])
|
|
self._psql_conn.commit()
|
|
sql_cursor.close()
|