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 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) ) self._mqtt_send_data_ch_level[50+ch] = MqttDataMessageHandler(self._mqtt_thread, 'data_server/device_data_stream/' + str(result.device) + '/' + str(ch+50) ) 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