Files
controller-wisetopdataserver/python/biopro/server/recording.py
T

683 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
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.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._ac_amp = self._meta_file.configuration.get_parameter('AC_AMP')
decoder._mode = self._meta_file.configuration.get_parameter('MODE')
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
decoder = self.data_format()
# 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 = 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 = decoder.decode(section)
# if self._isTimeOut:
# try:
# print('result: ', result.data_size)
# except:
# pass
try:
if isinstance(decoder, I4V4Z4T4DataDecoder) or isinstance(decoder, EISZeroOneDataDecoder):
if decoder.isFinishMode is not None and decoder.isFinishMode() == 1:
self._queue_msg.put(['ds', self._device, 'interrupt'])
# 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)
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)
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):
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]', raw_data[0])
return save
elif colum_total == 2:
print('this ram data < 3 records, colum_total = ', colum_total)
print('raw_data[0]', raw_data[0])
print('raw_data[1]', 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()
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