peterlu14
692 lines
24 KiB
Python
692 lines
24 KiB
Python
import base64
|
|
import re
|
|
import zlib
|
|
import sys
|
|
import gc
|
|
from collections import Counter
|
|
from typing import Iterable
|
|
from time import time, sleep
|
|
from biopro.devlib.data import *
|
|
|
|
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, RecordingFileWriter
|
|
from biopro.devlib.data import DataDecodeFormat
|
|
from biopro.util.logger import logging_info, Logger
|
|
|
|
from biopro.util.json import JSON_OBJECT
|
|
from .socket import ServerThread
|
|
|
|
from multiprocessing import Process, Queue
|
|
from .mqtt import MqttDataMessageHandler, MqttThread
|
|
|
|
import biopro.server._identify
|
|
|
|
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
_RUNTIME_COMPILE = False
|
|
|
|
|
|
def json_stringify(o: Any) -> str:
|
|
return _json_stringify(o, separators=(',', ':'))
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class RecordingProcess(Process):
|
|
"""
|
|
connect to DB and send sql command
|
|
"""
|
|
|
|
def __init__(self, client_id: str, log_verbose, device_id, queue_rec, queue_ds, queue_msg, meta_path,
|
|
device_info, parameter, parent, recording_file_name, data_format,
|
|
mqtt_ip, mqtt_port,
|
|
name = 'Recording-Process', database = None):
|
|
super(RecordingProcess, self).__init__(name = name)
|
|
|
|
self._database = database
|
|
self._device = device_id
|
|
|
|
self._meta_file = RecordingMetaFile.load(meta_path, self._database)
|
|
self._meta_file._device = device_info
|
|
self._meta_file._filepath = meta_path
|
|
self._meta_file._parameter = parameter
|
|
self._meta_file._parent = parent
|
|
self._meta_file._recording_file_name = recording_file_name
|
|
self._meta_file._device_id = device_id
|
|
print(self._meta_file._device, self._meta_file._parameter, self._meta_file._parent, self._meta_file._filepath)
|
|
self._meta_file.write(self._database)
|
|
|
|
self._data_format = data_format
|
|
self._timer = None
|
|
|
|
self.sync_file_request = False
|
|
self._sync_started = False
|
|
|
|
self._writer = None
|
|
|
|
|
|
self.log_verbose = log_verbose
|
|
self._mqtt_thread = MqttThread(None,
|
|
mqtt_ip,
|
|
mqtt_port,
|
|
'device-' + str(self._device) + '-' + str(self._meta_file._file_uuid),
|
|
self.log_verbose,
|
|
'MQTT-device-' + str(self._device),
|
|
is_sub = False)
|
|
# self._mqtt_thread.start()
|
|
self._mqtt_send_data_ch_level = {}
|
|
|
|
self._queue_rec = queue_rec
|
|
self._queue_ds = queue_ds
|
|
self._queue_msg = queue_msg
|
|
self.queue_flag = True
|
|
|
|
self._is_close = True
|
|
|
|
self._prev_data = []
|
|
self._prev_delta_time = []
|
|
self._prev_time_stamp = []
|
|
|
|
self._isTimeOut = False
|
|
|
|
# section use
|
|
self._skip_ram_cnt = 0
|
|
self._get_first_last_cnt = [True, True, True, True, True, True, True, True]
|
|
self._last_cnt = [-1, -1, -1, -1, -1, -1, -1, -1]
|
|
self._wrong_cnt = 0
|
|
|
|
# elite data length
|
|
self._elite_data_len = 40
|
|
self._mem_header_len = 3
|
|
self._mem_tailer_len = 6
|
|
self._single_data_len = self._elite_data_len + self._mem_header_len + self._mem_tailer_len
|
|
|
|
self._decoder = self.data_format()
|
|
self._start_time = time()
|
|
|
|
def ensure_data_format(self) -> DataDecodeFormat:
|
|
if isinstance(self._data_format, (str, bytes)):
|
|
self._data_format = DataDecodeFormat.parse(self._data_format)
|
|
|
|
# add device information
|
|
if isinstance(self._data_format, RecDataDecoder):
|
|
self._data_format.device = self._device
|
|
|
|
return self._data_format
|
|
|
|
def data_format(self) -> DataDecodeFormat:
|
|
decoder = self.ensure_data_format()
|
|
|
|
# transmit calibration gain level to decoder
|
|
if isinstance(decoder, TDC4VAF2DataDecoder):
|
|
# get amp_gain from meta file
|
|
decoder.amp_gain = self._meta_file.configuration.amp_gain
|
|
elif isinstance(decoder, TDC4VCDataDecoder):
|
|
# get amp_gain from meta file
|
|
decoder.amp_gain = self._meta_file.configuration.amp_gain
|
|
decoder._channel = self._meta_file.configuration.channel
|
|
decoder._adc_clock = self._meta_file.configuration.get_parameter('ADC_CLOCK')
|
|
# decoder._time_stamp_list = self._time_stamp_list
|
|
decoder._axis_ch = self._meta_file.configuration.get_parameter('AXIS_CH')
|
|
|
|
decoder._prev_data = self._prev_data
|
|
decoder._prev_delta_time = self._prev_delta_time
|
|
decoder._prev_time_stamp = self._prev_time_stamp
|
|
elif isinstance(decoder, I4V4Z4T4DataDecoder):
|
|
# get cycle_time from meta file
|
|
decoder._mode = self._meta_file.configuration.get_parameter('MODE')
|
|
# decoder._cycle_start_time = self._cycle_start_time
|
|
elif isinstance(decoder, NeuliveThreeOneDataDecoder):
|
|
# get amp_gain from meta file
|
|
decoder.amp_gain = self._meta_file.configuration.amp_gain
|
|
decoder._channel = self._meta_file.configuration.channel
|
|
decoder._adc_clock = self._meta_file.configuration.get_parameter('ADC_CLOCK')
|
|
# decoder._time_stamp_list = self._time_stamp_list
|
|
decoder._axis_ch = self._meta_file.configuration.get_parameter('AXIS_CH')
|
|
decoder._accelerator_sensitivity = self._meta_file.configuration.get_parameter('ACCELERATOR_S')
|
|
|
|
decoder._prev_data = self._prev_data
|
|
decoder._prev_delta_time = self._prev_delta_time
|
|
decoder._prev_time_stamp = self._prev_time_stamp
|
|
elif isinstance(decoder, EISZeroOneDataDecoder):
|
|
# get amp_gain from meta file
|
|
decoder._mode = self._meta_file.configuration.get_parameter('MODE')
|
|
decoder._ac_amp = self._meta_file.configuration.get_parameter('AC_AMP')
|
|
decoder._freq_start = self._meta_file.configuration.get_parameter('FREQ_START')
|
|
decoder._freq_stop = self._meta_file.configuration.get_parameter('FREQ_STOP')
|
|
|
|
return decoder
|
|
|
|
def recv_data_foreach_runtime(self) -> bool:
|
|
"""foreach data runtime to receive data. called by :class:`DataRuntimeThread`.
|
|
|
|
:return: has runtime received data
|
|
"""
|
|
q = None
|
|
try:
|
|
# now = time()
|
|
q = self._queue_rec.get()
|
|
# get_queue_time_cost = time() - now
|
|
# if get_queue_time_cost > 2:
|
|
# print('get queue cost time', get_queue_time_cost)
|
|
except:
|
|
return False
|
|
finally:
|
|
if self._queue_rec.qsize() > 10:
|
|
print('qsize: ', self._device, self._queue_rec.qsize())
|
|
|
|
if q == 'close':
|
|
self.log_verbose('RecordingProcess file close:', self._device)
|
|
self.rec_update()
|
|
self.final_write()
|
|
self.is_closed = True
|
|
return False
|
|
elif q == 'show_data':
|
|
self._decoder._show_data = not self._decoder._show_data
|
|
else:
|
|
self.rec_update()
|
|
self.sync_data(q)
|
|
|
|
del q
|
|
|
|
return True
|
|
|
|
def sync_data(self, data = None):
|
|
self._isTimeOut = False
|
|
current_time = time()
|
|
if self._timer is not None:
|
|
if current_time - self._timer > 1.5:
|
|
print('time, sync_data routine_time', 'device:', self._device, datetime.now(), current_time, current_time - self._timer)
|
|
self._isTimeOut = True
|
|
self._timer = current_time
|
|
|
|
# print('sync_data')
|
|
# print('data',data)
|
|
|
|
# print('server/data', self._prev_delta_time,self._prev_time_stamp, self._prev_data)
|
|
|
|
if data is None or len(data) == 0:
|
|
result = self._decoder.decode(b'')
|
|
if result is not None:
|
|
ret = result
|
|
|
|
else:
|
|
return None
|
|
|
|
else:
|
|
ret = []
|
|
# print('decoder data')
|
|
|
|
for offset, section in self._foreach_data_section(data):
|
|
# for section in self._neu_foreach_data_section(data):
|
|
|
|
result = self._decoder.decode(section)
|
|
# if self._isTimeOut:
|
|
# try:
|
|
# print('result: ', result.data_size)
|
|
# except:
|
|
# pass
|
|
|
|
if result is not None:
|
|
##
|
|
if len(self._mqtt_send_data_ch_level) == 0:
|
|
for ch in result.channels():
|
|
self._mqtt_send_data_ch_level[ch] = MqttDataMessageHandler(self._mqtt_thread, 'data_server/device_data_stream/' + str(result.device) + '/' + str(ch) )
|
|
ret.append(result)
|
|
else:
|
|
continue
|
|
|
|
# write back
|
|
# ctime1 = time()
|
|
if self._writer is not None and len(ret) > 0:
|
|
if len(self._writer.channel_list) == 0:
|
|
self._writer.channels_update(ret[0].channels())
|
|
self._writer.write(ret, self._mqtt_send_data_ch_level)
|
|
# print('write time: ', time() - ctime1)
|
|
# print(ret)
|
|
|
|
try:
|
|
time_duration = self._meta_file.configuration.get_parameter('TIME_DURATION')
|
|
if time_duration and time_duration is not 0 and time() - self._start_time >= time_duration:
|
|
self._queue_msg.put(['ds', self._device, 'interrupt'])
|
|
return None
|
|
|
|
if isinstance(self._decoder, I4V4Z4T4DataDecoder) or isinstance(self._decoder, EISZeroOneDataDecoder):
|
|
if self._decoder.isFinishMode is not None and self._decoder.isFinishMode() == 1:
|
|
self._queue_msg.put(['ds', self._device, 'interrupt'])
|
|
return None
|
|
# content = {}
|
|
# content['header'] = 'device_instruction/0'
|
|
# content['device'] = result.device
|
|
# content['instruction'] = 'interrupt'
|
|
# self._mqtt_thread.publish('device_instruction',json_stringify(content), True)
|
|
# self.server.stop_sync(self.device)
|
|
except RuntimeError as e:
|
|
print(e)
|
|
|
|
del ret
|
|
del data
|
|
return
|
|
|
|
def _check_cnt(self, raw_data: bytes, device: int):
|
|
# check memory board cnt, if jump cnt, don't save data
|
|
save = True
|
|
head = raw_data[0]
|
|
head_counter = raw_data[1]
|
|
device_id = raw_data[3]
|
|
|
|
try:
|
|
# assign last_data_cnt value and handle first last_data_cnt
|
|
if (device == 4):
|
|
if (self._get_first_last_cnt[4] == True):
|
|
self._get_first_last_cnt[4] = False
|
|
self._last_cnt[4] = head_counter - 1
|
|
last_data_cnt = self._last_cnt[4]
|
|
|
|
elif (device == 5):
|
|
if (self._get_first_last_cnt[5] == True):
|
|
self._get_first_last_cnt[5] = False
|
|
self._last_cnt[5] = head_counter - 1
|
|
last_data_cnt = self._last_cnt[5]
|
|
|
|
elif (device == 6):
|
|
if (self._get_first_last_cnt[6] == True):
|
|
self._get_first_last_cnt[6] = False
|
|
self._last_cnt[6] = head_counter - 1
|
|
last_data_cnt = self._last_cnt[6]
|
|
|
|
elif (device == 7):
|
|
if (self._get_first_last_cnt[7] == True):
|
|
self._get_first_last_cnt[7] = False
|
|
self._last_cnt[7] = head_counter - 1
|
|
last_data_cnt = self._last_cnt[7]
|
|
|
|
# dont save to section when head or id is wrong
|
|
if (head != 255 or device_id != device or head_counter > 255 or head_counter < 0):
|
|
save = False
|
|
last_data_cnt = last_data_cnt + 1
|
|
|
|
return save
|
|
|
|
if (last_data_cnt == 255):
|
|
if (head_counter - last_data_cnt != -255):
|
|
save = False
|
|
last_data_cnt = 0
|
|
|
|
else:
|
|
last_data_cnt = head_counter
|
|
|
|
else:
|
|
if (head_counter - last_data_cnt != 1):
|
|
save = False
|
|
last_data_cnt = last_data_cnt + 1
|
|
|
|
else:
|
|
last_data_cnt = head_counter
|
|
|
|
finally:
|
|
|
|
if (device == 4):
|
|
self._last_cnt[4] = last_data_cnt
|
|
|
|
elif (device == 5):
|
|
self._last_cnt[5] = last_data_cnt
|
|
|
|
elif (device == 6):
|
|
self._last_cnt[6] = last_data_cnt
|
|
|
|
elif (device == 7):
|
|
self._last_cnt[7] = last_data_cnt
|
|
|
|
if (save == False):
|
|
self._wrong_cnt += 1
|
|
print('jump cnt', self._wrong_cnt, 'times, device:', device, ',', datetime.now())
|
|
|
|
return save
|
|
|
|
def _check_jump_ram(self, raw_data: bytes, colum_now: int, colum_total: int, save: bool, device: int, data_len: int):
|
|
col = colum_now
|
|
save = False
|
|
|
|
if raw_data[col][0] != 255 or raw_data[col][3] != device:
|
|
return save
|
|
|
|
if colum_total == 1:
|
|
print('this ram data < 3 records, colum_total = ', colum_total)
|
|
print('raw_data[0]', list(raw_data[0]))
|
|
save = True
|
|
# return save
|
|
|
|
elif colum_total == 2:
|
|
print('this ram data < 3 records, colum_total = ', colum_total)
|
|
print('raw_data[0]', list(raw_data[0]))
|
|
print('raw_data[1]', list(raw_data[1]))
|
|
if (raw_data[col + 1][1] - raw_data[col][1] == 1) or (raw_data[col][1] == 255 and raw_data[col + 1][1] == 0):
|
|
save = True
|
|
|
|
elif colum_total >= 3:
|
|
if (raw_data[col + 1][1] - raw_data[col][1] == 1 and raw_data[col + 2][1] - raw_data[col + 1][1] == 1) or\
|
|
(raw_data[col][1] == 255 and raw_data[col + 1][1] == 0 and raw_data[col + 2][1] == 1) or\
|
|
(raw_data[col][1] == 254 and raw_data[col + 1][1] == 255 and raw_data[col + 2][1] == 0):
|
|
save = True
|
|
|
|
if save == True:
|
|
self._skip_ram_cnt += 1
|
|
last_data_cnt = raw_data[col][1]
|
|
if (device == 4):
|
|
print('self._last_cnt[4]:', self._last_cnt[4], ', last_data_cnt:', last_data_cnt)
|
|
if (device == 5):
|
|
print('self._last_cnt[5]:', self._last_cnt[5], ', last_data_cnt:', last_data_cnt)
|
|
if (device == 6):
|
|
print('self._last_cnt[6]:', self._last_cnt[6], ', last_data_cnt:', last_data_cnt)
|
|
if (device == 7):
|
|
print('self._last_cnt[7]:', self._last_cnt[7], ', last_data_cnt:', last_data_cnt)
|
|
|
|
print('skip RAM, but data is right, solve data', list(raw_data[col][0:3]),'|', list(raw_data[colum_total-1][0:3]), 'skip Ram', self._skip_ram_cnt, 'times \n')
|
|
|
|
if (device == 4):
|
|
self._last_cnt[4] = last_data_cnt
|
|
|
|
elif (device == 5):
|
|
self._last_cnt[5] = last_data_cnt
|
|
|
|
elif (device == 6):
|
|
self._last_cnt[6] = last_data_cnt
|
|
|
|
elif (device == 7):
|
|
self._last_cnt[7] = last_data_cnt
|
|
|
|
return save
|
|
|
|
def _neu_foreach_data_section(self, data: bytes):
|
|
"""
|
|
:param data:
|
|
:return:
|
|
"""
|
|
_device = 6
|
|
|
|
offset = 0
|
|
ret = [_device]
|
|
total_sample = 0
|
|
ble_packet_end = 0
|
|
header = None
|
|
|
|
# packet header format
|
|
# - header (0xFF)
|
|
# - data counter
|
|
# - data content len (ble_packet_end)
|
|
# - chip id
|
|
def _discard_packet_header(offset: int, data: bytes) -> List[int, int]:
|
|
while offset < len(data):
|
|
data_header = data[offset]
|
|
if data_header != 0xFF:
|
|
offset += 1
|
|
continue
|
|
|
|
head_counter = data[offset + 1]
|
|
ble_packet_len = data[offset + 2]
|
|
device_id = data[offset + 3]
|
|
|
|
if device_id != _device:
|
|
offset += 1
|
|
continue
|
|
else:
|
|
ble_packet_end = offset + ble_packet_len + 3
|
|
# print("head_counter", head_counter)
|
|
return offset+4, ble_packet_end
|
|
return len(data), 0
|
|
|
|
# decoder header format:
|
|
# - timestamp (LSB)
|
|
# - channnel data number
|
|
# - delta time
|
|
def _get_ret_header(offset: int, data: bytes) -> List[int, int, Optional[bytes]]:
|
|
total_sample = data[offset + 4]
|
|
decoder_header = data[offset : offset+7]
|
|
offset += 7
|
|
return [offset, total_sample, decoder_header]
|
|
|
|
def _is_ret_need_header(ret: bytes) -> bool:
|
|
if len(ret) < 7:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def _is_ret_complete(ret: bytes) -> bool:
|
|
if len(ret) < 8:
|
|
return False
|
|
|
|
sample_length = 3
|
|
sample_number = int((len(ret) - 8) / sample_length)
|
|
|
|
if sample_number >= total_sample:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def _extend_ret(ret: bytes, offset: int, data: bytes) -> List[int, bytes]:
|
|
expect_ret_len = 8 + total_sample*3 # header_len = 8
|
|
extend_len = expect_ret_len - len(ret)
|
|
|
|
if (offset + extend_len) < ble_packet_end:
|
|
ret.extend(data[offset : offset + extend_len])
|
|
offset += extend_len
|
|
|
|
else:
|
|
ret.extend(data[offset : ble_packet_end])
|
|
offset = ble_packet_end
|
|
|
|
return offset, ret
|
|
|
|
# print("pre decode data len = ", len(data))
|
|
# print("pre decode data = ", list(data))
|
|
while offset < len(data):
|
|
if offset >= ble_packet_end:
|
|
offset, ble_packet_end = _discard_packet_header(offset, data)
|
|
|
|
if _is_ret_need_header(ret):
|
|
# get ret header
|
|
offset, total_sample, header = _get_ret_header(offset, data)
|
|
total_sample += 2 # acc data can be view as two sample
|
|
ret.extend(header)
|
|
|
|
# extend ret
|
|
offset, ret = _extend_ret(ret, offset, data)
|
|
|
|
# is ret complete?
|
|
if _is_ret_complete(ret):
|
|
# print("ret = ", ret)
|
|
# print("\n")
|
|
yield bytes(ret)
|
|
ret = [_device]
|
|
|
|
|
|
def log_data_receive_statistics(self, logger: Logger):
|
|
c = self._data_counter
|
|
|
|
def _foreach_data_section(self, data: bytes):
|
|
"""
|
|
:param data:
|
|
:return:
|
|
"""
|
|
raw_data = []
|
|
records = 0
|
|
print_flag = False
|
|
device = self._device
|
|
single_data_len = self._single_data_len
|
|
mem_header_len = self._mem_header_len
|
|
|
|
# packet section data use two microarray
|
|
for i in range(0, len(data), single_data_len):
|
|
records = records + 1
|
|
raw_data.append(data[i : i + single_data_len])
|
|
if len(data[i : i + 10]) < 10:
|
|
print('data dont be package:', raw_data[records - 1], 'index:', records - 1)
|
|
del raw_data[records - 1]
|
|
records = records - 1
|
|
|
|
# check cnt
|
|
for i in range(records):
|
|
|
|
# if jump cnt, don't save data
|
|
save = self._check_cnt(raw_data[i], device)
|
|
|
|
# # if jump Ram, also can handle data
|
|
if save == False:
|
|
save = self._check_jump_ram(raw_data, i, records, save, device, single_data_len)
|
|
|
|
if (save == True):
|
|
section = raw_data[i]
|
|
|
|
check_sum = sum(section[0 : single_data_len - 1]) & 0b11111111
|
|
|
|
if (check_sum != section[single_data_len - 1]):
|
|
print('check number not match,',
|
|
list(section[0 : mem_header_len]), section[-1], '!=', check_sum, 'device:', device, ',', datetime.now())
|
|
if (i-1 >= 0):
|
|
print('check number not match', list(raw_data[i-1]), 'raw:', i-1)
|
|
|
|
print('check number not match', list(raw_data[i]), 'raw:', i)
|
|
|
|
if (i+1 <= records):
|
|
print('check number not match', list(raw_data[i+1]), 'raw:', i+1, '\n')
|
|
section = section + b'\x01'
|
|
|
|
print('delete data:', list(raw_data[i][0:3]), 'raw:', i, '\n')
|
|
continue
|
|
else:
|
|
section = section + b'\x00'
|
|
|
|
# section = section[3:]
|
|
yield None, section
|
|
|
|
else:
|
|
print_flag = True
|
|
print('delete data:', list(raw_data[i][0:3]), 'raw:', i, '\n')
|
|
|
|
# if print_flag:
|
|
# print('delete data context:')
|
|
# for i in range(0, len(data), single_data_len):
|
|
# print(list(data[i:i+single_data_len]))
|
|
# print()
|
|
|
|
return
|
|
|
|
def run(self) -> None:
|
|
self.setup()
|
|
# self.routine()
|
|
|
|
def setup(self) -> None:
|
|
self.log_verbose('RecordingProcess setup:', self._device)
|
|
self._is_close = False
|
|
if self._mqtt_thread is not None:
|
|
self._mqtt_thread.start()
|
|
self._writer = RecordingFileWriter(self._meta_file, self._device, self._database)
|
|
self.routine()
|
|
|
|
def routine(self) -> None:
|
|
self.log_verbose('RecordingProcess routine:', self._device)
|
|
self._sync_started = True
|
|
running = True
|
|
while running:
|
|
running = self.recv_data_foreach_runtime()
|
|
self.log_verbose('RecordingProcess routine done:', self._device)
|
|
if self._mqtt_thread is not None:
|
|
self._mqtt_thread.shutdown()
|
|
return None
|
|
|
|
def shutdown(self) -> None:
|
|
self.log_verbose('RecordingProcess shutdown:', self._device)
|
|
self._sync_started = False
|
|
self._queue_rec.put('close')
|
|
self.queue_flag = True
|
|
|
|
|
|
def final_write(self):
|
|
if self._writer is not None:
|
|
return self._writer.close(self._mqtt_send_data_ch_level)
|
|
else:
|
|
return False
|
|
|
|
# def put_queue(self, request: list) -> None:
|
|
# self._queue.put(request)
|
|
# del request
|
|
# return
|
|
|
|
@property
|
|
def filepath(self):
|
|
return self._meta_file.filepath
|
|
|
|
@property
|
|
def meta_file(self) -> RecordingMetaFile:
|
|
return self._meta_file
|
|
|
|
@property
|
|
def file_writer(self) -> Optional[RecordingFileWriter]:
|
|
return self._writer
|
|
|
|
@property
|
|
def device(self) -> int:
|
|
return self._device
|
|
|
|
@property
|
|
def sync_started(self) -> bool:
|
|
return self._sync_started
|
|
|
|
# @sync_started.setter
|
|
# def sync_started(self, value: bool):
|
|
# self._sync_started = value
|
|
|
|
# if value:
|
|
# self._open()
|
|
# else:
|
|
# self._close()
|
|
|
|
def put_rec_queue(self, data) -> None:
|
|
self._queue_rec.put(data)
|
|
return
|
|
|
|
def rec_update(self) -> bool:
|
|
while self._queue_ds.qsize() > 0:
|
|
try:
|
|
q = self._queue_ds.get(block=False)
|
|
except:
|
|
return False
|
|
if q is not None:
|
|
func_name = q[0]
|
|
args = q[1:]
|
|
try:
|
|
self.solve_for(func_name, args = args)
|
|
except RuntimeError as e:
|
|
print('solve func error', func_name, args)
|
|
return False
|
|
return True
|
|
|
|
def solve_for(self, name: str, args: list):
|
|
do = f"{name}"
|
|
if callable(getattr(self, do)):
|
|
try:
|
|
getattr(self, do)(*args)
|
|
except:
|
|
pass
|
|
return None
|
|
|
|
def update_meta_id(self, _uuid, _id):
|
|
if self._writer is not None:
|
|
self._writer.update_meta_id(_id)
|
|
return
|
|
|
|
def update_raw_dict_id(self, _uuid, _channel, _id):
|
|
if self._writer is not None:
|
|
self._writer.update_raw_dict_id(_channel, _id)
|
|
return
|
|
|
|
def update_mini_dict_id(self, _uuid, _scale, _channel, _id):
|
|
if self._writer is not None:
|
|
self._writer.update_mini_dict_id(_scale, _channel, _id)
|
|
return
|
|
|