import abc import struct import math from typing import Optional, TypeVar, Generic, Tuple, Dict, List, AnyStr from datetime import datetime from biopro.recording import RecordingData, RecordingFileDataFormat # from biopro.util.console import hex_line T = TypeVar('T') class EODInterrupt(RuntimeError): """End of Data interrupt. """ __slots__ = () class DataDecodeFormat(Generic[T], metaclass=abc.ABCMeta): """data decoder. do not share this instance between data decoding handle, except you make sure it is safe. """ __slots__ = () @property @abc.abstractmethod def name(self) -> AnyStr: """decoder name make sure that decoder and ``DataDecodeFormat.parse(decoder.name)`` are functional equally. :return: decoder name. """ pass @property @abc.abstractmethod def format(self) -> bytes: """recording meta file format :return: """ pass def message(self) -> Optional[str]: """send message to front-end to indicate that data has changed something. :return: message content """ return None @abc.abstractmethod def decode(self, data: bytes) -> Optional[T]: """ :param data: raw data :return: decoded data """ pass @classmethod def parse(cls, expr: AnyStr) -> 'DataDecodeFormat': if expr == NulDataDecoder.NAME: return NulDataDecoder.INSTANCE elif expr == RawDataDecoder.NAME: return RawDataDecoder.INSTANCE elif expr == TC4VAF2DataDecoder.NAME: return TC4VAF2DataDecoder() elif expr == TDC4VAF2DataDecoder.NAME: return TDC4VAF2DataDecoder() elif expr == TDC4VCDataDecoder.NAME: return TDC4VCDataDecoder() elif expr == I4V4Z4T4DataDecoder.NAME: return I4V4Z4T4DataDecoder() elif isinstance(expr, str): raise RuntimeError('current not support custom data decode format : ' + expr) elif isinstance(expr, bytes): if expr.startswith(b'TDC4VAF2:'): return TDC4VAF2DataDecoder(expr[9:]) elif expr.startswith(b'TDC4VC:'): return TDC4VCDataDecoder(expr[7:]) elif expr.startswith(b'NeuliveThreeOne:'): return NeuliveThreeOneDataDecoder(expr[16:]) elif expr.startswith(b'EISZeroOne:'): return EISZeroOneDataDecoder(expr[11:]) raise RuntimeError('current not support custom data decode format : ' + str(expr)) def __str__(self): return self.__class__.__name__ __repr__ = __str__ class NulDataDecoder(DataDecodeFormat[None]): """drop all data, decode always return None.""" NAME = 'NUL' __slots__ = () @property def name(self) -> str: return self.NAME @property def format(self) -> bytes: return RecordingFileDataFormat.RAW_DATA def decode(self, data: bytes) -> None: """ :param data: raw data :return: always None """ return None NulDataDecoder.INSTANCE = NulDataDecoder() class RawDataDecoder(DataDecodeFormat[bytes]): NAME = 'RAW' __slots__ = () @property def name(self) -> str: return self.NAME @property def format(self) -> bytes: return RecordingFileDataFormat.RAW_DATA def decode(self, data: bytes) -> bytes: """ :param data: raw data :return: *data* """ return data RawDataDecoder.INSTANCE = RawDataDecoder() class Timer: OVERFLOW = (int)(0xFFFF_FFFF / 31) # CC2650 device overflow value __slots__ = '_start_time', '_prev_time', '_overflow', '_sum_timestamp', '_prev_prev_time' def __init__(self): self._start_time: Dict[int, float] = {} self._prev_time: Dict[int, float] = {} self._overflow: Dict[int, float] = {} self._prev_prev_time: Dict[int, float] = {} for dev_num in range(8): self._start_time[dev_num] = None def eis_get_time_stamp(self, device: int, time_stamp: float) -> Tuple[Optional[float], Optional[float]]: if self._start_time[device] is None: self._start_time[device] = time_stamp self._prev_time[device] = time_stamp self._prev_prev_time[device] = time_stamp self._overflow[device] = 0 return 0, 0, False prev_time = self._prev_time.get(device, None) prev_prev_time = self._prev_prev_time.get(device, None) overflow = self._overflow.get(device, 0) start_time = self._start_time[device] if prev_time is None: self._prev_time[device] = time_stamp self._overflow[device] = 0 ret = time_stamp - start_time print("ignore prev_time is None; ", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) return ret, 0, False # data duplicate elif time_stamp == prev_time: print("ignore time_stamp == prev_time; ", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) return None, None, True # if (time_stamp > self.OVERFLOW - 1200) or (time_stamp < 0x0000_0050): # print("right time [time:", time_stamp, ", prev_time:", prev_time, ']', datetime.now()) # time stamp overflow if time_stamp < 0x0100_0000 and 0x0800_0000 < prev_time: overflow += 1 self._overflow[device] = overflow self._prev_time[device] = time_stamp delta = time_stamp + (self.OVERFLOW + 1) - prev_time ret = (self.OVERFLOW + 1) * overflow + time_stamp - start_time print("time stamp overflow;", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, "; delta:", delta, ",", datetime.now()) print("overflow:", overflow, "; device:", device) return ret, delta, False # ignore negative delta elif time_stamp < prev_time and time_stamp != 0: print("ignore negative delta; ", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) return None, None, True # # ignore big jump, if delta t > 30s, ignore # elif time_stamp > prev_time and (time_stamp - prev_time > 60000) and prev_time != 0: # print("ignore big jump;", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) # return None, None, True else: delta = time_stamp - prev_time self._prev_time[device] = time_stamp overflow = self._overflow[device] ret = time_stamp + (self.OVERFLOW + 1) * overflow - start_time self._prev_prev_time[device] = prev_time return ret, delta, False def get_time_stamp(self, device: int, time_stamp: float) -> Tuple[Optional[float], Optional[float]]: if self._start_time[device] is None: self._start_time[device] = time_stamp self._prev_time[device] = time_stamp self._prev_prev_time[device] = time_stamp self._overflow[device] = 0 return 0, 0, False prev_time = self._prev_time.get(device, None) prev_prev_time = self._prev_prev_time.get(device, None) overflow = self._overflow.get(device, 0) start_time = self._start_time[device] if prev_time is None: self._prev_time[device] = time_stamp self._overflow[device] = 0 ret = time_stamp - start_time print("ignore prev_time is None; ", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) return ret, 0, False # data duplicate elif time_stamp == prev_time: print("ignore time_stamp == prev_time; ", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) return None, None, True # if (time_stamp > self.OVERFLOW - 1200) or (time_stamp < 0x0000_0050): # print("right time [time:", time_stamp, ", prev_time:", prev_time, ']', datetime.now()) # time stamp overflow if time_stamp < 0x0100_0000 and 0x0800_0000 < prev_time: overflow += 1 self._overflow[device] = overflow self._prev_time[device] = time_stamp delta = time_stamp + (self.OVERFLOW + 1) - prev_time ret = (self.OVERFLOW + 1) * overflow + time_stamp - start_time print("time stamp overflow;", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, "; delta:", delta, ",", datetime.now()) print("overflow:", overflow, "; device:", device) return ret, delta, False # ignore negative delta elif time_stamp < prev_time and time_stamp != 0: print("ignore negative delta; ", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) return None, None, True # ignore big jump, if delta t > 30s, ignore elif time_stamp > prev_time and (time_stamp - prev_time > 60000) and prev_time != 0: print("ignore big jump;", "time:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, ",", datetime.now()) return None, None, True else: delta = time_stamp - prev_time self._prev_time[device] = time_stamp overflow = self._overflow[device] ret = time_stamp + (self.OVERFLOW + 1) * overflow - start_time self._prev_prev_time[device] = prev_time return ret, delta, False class RecDataDecoder(DataDecodeFormat[RecordingData], metaclass=abc.ABCMeta): OVERFLOW = 0xFFFF_FFFF __slots__ = '_device', 'timer' def __init__(self): self._device: int = 0 self.timer = Timer() @property def device(self) -> int: return self._device @device.setter def device(self, value: int): if not (0 <= value): raise ValueError('illegal device value : ' + str(value)) self._device = value @property def format(self) -> bytes: return RecordingFileDataFormat.REC_DATA def get_time_stamp(self, time_stamp: float) -> Tuple[Optional[float], Optional[float]]: return self.timer.get_time_stamp(self.device, time_stamp) def eis_get_time_stamp(self, time_stamp: float) -> Tuple[Optional[float], Optional[float]]: return self.timer.eis_get_time_stamp(self.device, time_stamp) @abc.abstractmethod def decode(self, data: bytes) -> Optional[RecordingData]: pass class TC4VAF2DataDecoder(RecDataDecoder): """ :: | | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 ----------------------------------------------------------------- | header | length | | timestamp | | data [0] | data[1] | | ............................ | data[N] | *data* :: | | 1 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 --------------------------------- |channel| value | F | """ NAME = 'TC4VAF2' __slots__ = ('_prev_data',) def __init__(self): super().__init__() self._prev_data: Optional[bytes] = None @property def name(self) -> str: return self.NAME def decode(self, data: bytes) -> Optional[RecordingData]: if len(data) < 6: return None # hex_line(data) data_length = int(data[1]) if data_length == 0: return None time_stamp: float = struct.unpack('> 4 value = ((d1 & 0x0F) << 6) | ((d2 & 0xFC) >> 2) flag = d2 & 0b0011 if flag == 0: ret.append_data(channel, value - 512) else: ret.append_data(channel, None) return ret class TDC4VAF2DataDecoder(RecDataDecoder): """ data format ----------- :: | | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 ----------------------------------------------------------------- | header | length = N | | timestamp | | time delta | | data [0] | data[1] | | ............................ | data[N] | *data* :: | | 1 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 --------------------------------- |channel| value | F | cali coeff table format ----------------------- TODO cali coeff table format :: | | 1 | 2 | 3 | 4 | 5 | 6 | 01234567890123456789012345678901234567890123456789012345678901234 ----------------------------------------------------------------- | | c0 g0 amp | c0 g1 amp | c0 g2.. .. amp | | c0 g0 off | c0 g1.. .. off | c0 g2 off | | c1 g0.. | ............................................................. | | cN g0 amp | cN g1 amp | cN g2 amp |.. (4 bytes) ..| | cN g0 off | cN g1 off | cN g2 off |.. (4 bytes) ..| """ NAME = 'TDC4VAF2' AMP_GAIN = (35, 75, 325) __slots__ = 'amp_gain', 'data_counter', 'cali_coeff', '_cali_coeff' def __init__(self, cali_coeff: bytes = None): super().__init__() self.amp_gain: int = 0 self.data_counter: int = 0 self._cali_coeff: Optional[bytes] = None self.cali_coeff: Optional[List[Tuple[int, int]]] = None if cali_coeff is not None: self._cali_coeff = cali_coeff self.cali_coeff = self._decode_cali_coeff(cali_coeff) @staticmethod def _decode_cali_coeff(self, cali_coeff: bytes) -> Optional[List[Tuple[int, int]]]: if cali_coeff != b'': cali_table = [] for i in range(0, 12): # foreach channel for j in range(20 * i + 3, 20 * i + 9, 2): amp = struct.unpack('>h', cali_coeff[j:j + 2])[0] offset = struct.unpack('>h', cali_coeff[j + 10:j + 12])[0] cali_table.append((amp, offset)) return cali_table else: return None @property def name(self) -> AnyStr: if self._cali_coeff is None: return self.NAME else: return self.NAME.encode() + b':' + self._cali_coeff def decode(self, data: bytes) -> Optional[RecordingData]: """ data_counter is used for excluding data which were send from the beginning ( < 15 )with the wrong timestamp, and these data will cause error while recording. """ if self.data_counter < 15: self.data_counter += 1 else: if len(data) < 8: return None # print(hex_line(data)) data_length = int(data[1]) if data_length == 0: return None time_stamp: float = struct.unpack('> 4 value = ((d1 & 0x0F) << 6) | ((d2 & 0xFC) >> 2) flag = d2 & 0b0011 if flag != 0: ret.append_data(channel, None) else: if channel < 12 and self.cali_coeff is not None: gain, offset = self.cali_coeff[channel * 3 + amp_gain] gain /= 2 if gain == 0: cali_value = value - 512 else: cali_value = int(1000 * ((value - 512 - offset) / gain)) else: cali_value = value - 512 ret.append_data(channel, cali_value) return ret class TDC4VCDataDecoder(RecDataDecoder): """ :: | | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 ----------------------------------------------------------------- | header | length = N | | timestamp | | data [0] | data[1] | | ............................ | data[15] | *data* :: | | 1 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 --------------------------------- |channel| value | F | """ NAME = 'TDC4VC' AMP_GAIN = (100, 400, 800, 25) CH_NUMBER = 8 GAIN_LEVEL_NUMBER = 4 __slots__ = "_start_return_data", "_time_stamp", "_time_stamp_list", "_discard_data_counter", "_channel", 'amp_gain', 'data_counter', 'cali_coeff', '_cali_coeff', "_cycle_start_time", "_adc_clock", "_axis_ch", "_prev_delta_time","_prev_data", "_prev_time_stamp" def __init__(self, cali_coeff: bytes = None): super().__init__() self._start_return_data = False self._time_stamp = 0 self._time_stamp_list = [] self._discard_data_counter = 0 self._channel = 0 self._adc_clock = 0 self._axis_ch = 0 self.amp_gain: int = 0 self.data_counter: int = 0 self._cali_coeff: Optional[bytes] = None self.cali_coeff: Optional[List[Tuple[int, int]]] = None self._cycle_start_time = [] if cali_coeff is not None: self._cali_coeff = cali_coeff self.cali_coeff = self._decode_cali_coeff(cali_coeff) self._prev_data = None self._prev_delta_time = None self._prev_time_stamp = None @staticmethod def _decode_cali_coeff(cali_coeff: bytes) -> Optional[List[Tuple[int, int]]]: if cali_coeff != b'': cali_table = [] max_length = 4 size_of_cali_data_per_amp_gain_level = 33 for i in range(0, max_length): for j in range(1, size_of_cali_data_per_amp_gain_level, 4): offset = i * size_of_cali_data_per_amp_gain_level + j amp = struct.unpack('>h', cali_coeff[offset: offset + 2])[0] offset = struct.unpack('>h', cali_coeff[offset + 2: offset + 4])[0] cali_table.append((amp, offset)) # print('decode cali', i , j , amp , offset) # print('cali_table', cali_table) return cali_table else: print() return None @property def name(self) -> AnyStr: if self._cali_coeff is None: return self.NAME else: return self.NAME.encode() + b':' + self._cali_coeff @property def real_sample_rate(self) -> int: result = 0 if self._adc_clock != 0: result = (200 / self._adc_clock) * 1000 return result def decode(self, data: bytes) -> Optional[RecordingData]: if len(data) < 20: return None # data_packet contains (data_length) pair of datas data_length = int(data[1]) # axis data start in index (axis_start_index) axis_start_index = 8 + data_length * 2 # data_packet contains (axis_data_length) pair of datas axis_data_length = int((len(data) - axis_start_index) / 6) if data_length == 0: return None time_stamp: float = struct.unpack('>I', data[2:6])[0] time_delta: float = struct.unpack('>H', data[6:8])[0] # print('absolute time_stamp', time_stamp * 1000 / 32) """ why multiple first then divide at last """ # time_stamp = time_stamp * 32 # unit: 1/32K sec, 1/32 ms # time_delta = time_delta * 32 # unit: as above ''' garbage -> 0 -> real data''' ''' False -> True -> True ''' ''' Some device will start early before enter decoder, so start flag will never enable''' if time_stamp == 0 or (self._time_stamp != 0 and self._time_stamp < time_stamp): self._start_return_data = True if time_stamp is None or time_stamp == 0 or not self._start_return_data: self._time_stamp = time_stamp return None if self._discard_data_counter < 10: self._discard_data_counter += 1 return None stop = len(data) if time_stamp != 0 and self._start_return_data: time_stamp, _ , ret_get_time_stamp = self.get_time_stamp(time_stamp) if data_length == 1 or int(time_delta) == 0: sample_rate = 0 if len(self._prev_time_stamp) > 0 and time_stamp is not None: # print('prev_time_stamp, time_stamp', self._prev_time_stamp[0], time_stamp) data_length_for_sample_rate = data_length data_packets_time_delta = time_stamp - self._prev_time_stamp[0] if self._prev_time_stamp[0] > time_stamp: return None if int(data_packets_time_delta / self._prev_delta_time[0]) >= 2 or self._prev_time_stamp[0] is None: data_packets_time_delta = self._prev_delta_time[0] # unit 1/s sample_rate = 1000 * (data_length_for_sample_rate) / data_packets_time_delta sample_rate_channel_upper_256 = sample_rate * axis_data_length / data_length # print('sample_rate', sample_rate) ret = RecordingData(self.device, (self._prev_time_stamp[0] * 1000), int(sample_rate), int(sample_rate_channel_upper_256), 4) ret.cycle = self._cycle_start_time # get user setting gain level # and set amp_gain=0~3 => gain=25,100,400,800 amp_gain = self.amp_gain amp_gain = amp_gain + 1 if amp_gain == 4: amp_gain = 0 # decode recording data for i in range(data_length): # avoid index out of range offset = 8 + 2 * i if offset + 1 < stop: d1 = (data[offset] & 0xF0) >> 4 d2 = (data[offset] & 0x0F) d3 = data[offset + 1] channel = d1 value = d2 << 8 | d3 # print("\nNeulive 3.x") # print("channel", channel + 1) # print("value", value - 2048) # print("amp_gain = ", amp_gain) # print("cali:", self.cali_coeff) if channel < 8 and self._prev_data[0] is not None: gain, offset = self._prev_data[0][8 * amp_gain + channel] # print('cali_gain', gain) # print('cali_offset', offset) if gain == 0: cali_value = value - 2048 elif gain == 1 and offset == 0: cali_value = value - 2048 else: cali_value = int(1000 * ((value - 2048 - offset) / gain)) else: cali_value = value - 2048 # print('c,v', channel, cali_value) ret.append_data(channel, cali_value) # print('\n') # decode 3-axis acc data x_axis_acc_ch = 256 y_axis_acc_ch = 257 z_axis_acc_ch = 258 axis_acc_ch = 259 if self._axis_ch > 0: axis_channel_allow = bin(self._axis_ch)[2:] axis_channel_allow_list = [ch for ch in axis_channel_allow] axis_channel_allow_list.reverse() for i in range(axis_data_length): x_axis_index: int = i*6 + axis_start_index y_axis_index: int = i*6 + axis_start_index + 2 z_axis_index: int = i*6 + axis_start_index + 4 x_axis_data = data[x_axis_index] << 8 | data[x_axis_index + 1] y_axis_data = data[y_axis_index] << 8 | data[y_axis_index + 1] z_axis_data = data[z_axis_index] << 8 | data[z_axis_index + 1] # using 2's complement if x_axis_data >= 0x8000: x_axis_data = x_axis_data - 0xFFFF if y_axis_data >= 0x8000: y_axis_data = y_axis_data - 0xFFFF if z_axis_data >= 0x8000: z_axis_data = z_axis_data - 0xFFFF axis_data = int(math.sqrt(x_axis_data ** 2 + y_axis_data ** 2 + z_axis_data ** 2)) if len(axis_channel_allow_list) >= 1: if int(axis_channel_allow_list[0]) == 1: ret.append_data(x_axis_acc_ch, int(x_axis_data / 100)) if len(axis_channel_allow_list) >= 2: if int(axis_channel_allow_list[1]) == 1: ret.append_data(y_axis_acc_ch, int(y_axis_data / 100)) if len(axis_channel_allow_list) >= 3: if int(axis_channel_allow_list[2]) == 1: ret.append_data(z_axis_acc_ch, int(z_axis_data / 100)) if len(axis_channel_allow_list) >= 4: if int(axis_channel_allow_list[3]) == 1: ret.append_data(axis_acc_ch, int(axis_data / 100)) self._prev_data.pop() self._prev_time_stamp.pop() self._prev_delta_time.pop() self._prev_data.append(self.cali_coeff) self._prev_time_stamp.append(time_stamp) self._prev_delta_time.append(time_delta) return ret else: if time_stamp is not None: self._prev_data.append(self.cali_coeff) self._prev_time_stamp.append(time_stamp) self._prev_delta_time.append(time_delta) return None class I4V4Z4T4DataDecoder(RecDataDecoder): """ :: | | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 ----------------------------------------------------------------- | header | | current | | voltage | | impedance | | time stamp | | cycle number | cycle number for cyclic voltammetry use, we save it as channel number. 0xFF """ NAME = 'I4V4Z4T4' MESSAGE_CYCLE_COMPLETE = 'CycleComplete' __slots__ = ('_message', '_cycle_number', '_start_return_data', '_time_stamp', '_total_time_stamp', '_mode', '_cycle_start_time', '_mode_stop') def __init__(self): super().__init__() self._cycle_number = 0 self._message: Optional[str] = None self._start_return_data = False self._time_stamp = 0 self._total_time_stamp = 0 self._mode_stop = 0 self._mode = 0 self._cycle_start_time = [] @property def name(self) -> str: return self.NAME def message(self) -> Optional[str]: ret = self._message self._message = None return ret def decode(self, data: bytes) -> Optional[RecordingData]: if len(data) < 18: return None current = struct.unpack('>i', data[1:5])[0] # unit: 1/1000 nA voltage = struct.unpack('>i', data[5:9])[0] # unit: mV impedance = struct.unpack('>i', data[9:13])[0] # unit: kOm time_stamp: float = struct.unpack('H', data[17:19])[0] finish_mode_falg = data[19] # print('decode', list(data[20:])) mem_wrong = data[20] mem_retry_cnt = data[21] mem_green_wrong = data[22] mem_green_retry_cnt = data[23] ram_num = data[24] broken_flag = data[-1] if (finish_mode_falg & 0b11110000 == 0b10100000): finishMode = True else: finishMode = False if time_stamp == 0: self._start_return_data = True if time_stamp is None or time_stamp == 0 or not self._start_return_data: return None if time_stamp != 0 and self._start_return_data == True: time_stamp, delta, ret_get_time_stamp = self.get_time_stamp(time_stamp) if ret_get_time_stamp == True: print("error timeStamp full data:", list(data), datetime.now(), '\n') return None else: # print('|', time_stamp, '|', delta, '|', current, '|', voltage, '|', impedance, # '|', cycle_number, '|', finishMode, '@', str(self.device)) # print('|', '{:10}'.format(time_stamp), # '|', '{:10}'.format(delta), # '|', '{:10}'.format(current), # '|', '{:10}'.format(voltage), # '|', '{:10}'.format(impedance), # '|', '{:5}'.format(cycle_number), # '|', '{:1}'.format(finishMode), # '@', str(self.device), '|') # print('|', '{:5}'.format(mem_wrong), # '|', '{:5}'.format(mem_retry_cnt), # '|', '{:5}'.format(mem_green_wrong), # '|', '{:5}'.format(mem_green_retry_cnt), # '|', '{:5}'.format(ram_num), # '|', '{:5}'.format(broken_flag), # '@', str(self.device), '|') pass if finishMode == True: print("finishMode full data:", list(data), datetime.now()) self._mode_stop = 1 else: self._mode_stop = 0 ret = RecordingData(self.device, int(time_stamp * 1000 / 2), 0) ret.append_data(0, current) ret.append_data(1, voltage) ret.append_data(2, impedance) ret.append_data(3, cycle_number) # memoryboard information # ret.append_data(4, mem_wrong) # ret.append_data(5, mem_retry_cnt) # ret.append_data(6, mem_green_wrong) # ret.append_data(7, mem_green_retry_cnt) # ret.append_data(8, ram_num) # ret.append_data(9, broken_flag) # ret.append_data(4, ram_num) if cycle_number != self._cycle_number: # notify cycle_number change self._message = self.MESSAGE_CYCLE_COMPLETE + '=%d' % cycle_number self._cycle_number = cycle_number return ret def isFinishMode(self) -> int: return self._mode_stop class TDBC4VCTHDataDecoder(RecDataDecoder): """ :: | | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 ----------------------------------------------------------------- | header | length | | timestamp | | | battery voltage | |channel| channel voltage | | stimulation remind times | |channel| channel voltage | | stimulation remind times | """ NAME = 'TDBC4VCTH' __slots__ = ('_remind',) def __init__(self): super().__init__() self._remind: int = 1 @property def name(self) -> str: return self.NAME def decode(self, data: bytes) -> Optional[RecordingData]: if self._remind == 0: raise EODInterrupt() if len(data) < 6: return None data_length = int(data[1]) if data_length == 0: return None time_stamp: float = struct.unpack('= 8: d1 = data[6] d2 = data[7] voltage = ((d1 & 0xF0) << 8) | (d2 & 0xFF) voltage = 4300 * voltage / 4096 ret.append_data(0, int(voltage)) remind = 0 for i in range(8, data_length, 4): d1 = data[i] d2 = data[i + 1] d3 = data[i + 2] d4 = data[i + 3] channel = (d1 >> 4) & 0x0F value = ((d1 & 0x0F) << 8) | (d2 & 0xFF) remind += ((d3 & 0xFF) << 8) | (d4 & 0xFF) value = 26 * value - 25 * voltage ret.append_data(channel, int(value)) self._remind = remind return ret class NeuliveThreeOneDataDecoder(RecDataDecoder): """ :: | | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 ----------------------------------------------------------------- | timestamp | |channel data # | | delta time | | channel data | | x-axis acc data | | y-axis acc data | | z-axis acc data | """ NAME = 'NeuliveThreeOne' AMP_GAIN = (100, 400, 800, 25) CH_NUMBER = 8 GAIN_LEVEL_NUMBER = 4 __slots__ = ("_start_return_data", "_time_stamp", "_time_stamp_list", "_discard_data_counter", "_channel", 'amp_gain', 'data_counter', 'cali_coeff', '_cali_coeff', "_cycle_start_time", "_adc_clock", "_axis_ch", "_prev_delta_time","_prev_data", "_prev_time_stamp", "_accelerator_sensitivity", '_prev_packet_delta_time', '_prev_packet_delta_mean') def __init__(self, cali_coeff: bytes = None): super().__init__() self._start_return_data = False self._time_stamp = 0 self._time_stamp_list = [] self._discard_data_counter = 0 self._channel = 0 self._adc_clock = 0 self._axis_ch = 0 self._accelerator_sensitivity = 0 self.amp_gain: int = 0 self.data_counter: int = 0 self._cali_coeff: Optional[bytes] = None self.cali_coeff: Optional[List[Tuple[int, int]]] = None self._cycle_start_time = [] if cali_coeff is not None: self._cali_coeff = cali_coeff self.cali_coeff = self._decode_cali_coeff(cali_coeff) self._prev_data = None self._prev_delta_time = None self._prev_time_stamp = None self._prev_packet_delta_time = [] self._prev_packet_delta_mean = None @staticmethod def _decode_cali_coeff(cali_coeff: bytes) -> Optional[List[Tuple[int, int]]]: if cali_coeff != b'': cali_table = [] max_length = 4 size_of_cali_data_per_amp_gain_level = 64 for i in range(0, max_length): for j in range(0, size_of_cali_data_per_amp_gain_level, 4): index = i * size_of_cali_data_per_amp_gain_level + j amp = struct.unpack('>h', cali_coeff[index: index + 2])[0] offset = struct.unpack('>h', cali_coeff[index + 2: index + 4])[0] cali_table.append((amp, offset)) # print('decode cali', i , j , amp , offset) # print('cali_table', cali_table) return cali_table else: print() return None @property def name(self) -> AnyStr: if self._cali_coeff is None: return self.NAME else: return self.NAME.encode() + b':' + self._cali_coeff @property def real_sample_rate(self) -> int: result = 0 if self._adc_clock != 0: result = (200 / self._adc_clock) * 1000 return result def decode(self, data: bytes) -> Optional[RecordingData]: if data is None: return None data_length = data[5] # channel data & acc data number should be data_length+2 if (len(data)-8)/3 != data_length+2: return None axis_start_index = 8 + data_length * 3 # 239 axis_data_length = int((len(data) - axis_start_index) / 6) time_stamp: int = struct.unpack(' 0: if self._accelerator_sensitivity == 0: divide_ratio = 16384 elif self._accelerator_sensitivity == 1: divide_ratio = 8192 elif self._accelerator_sensitivity == 2: divide_ratio = 4096 elif self._accelerator_sensitivity == 3: divide_ratio = 2048 axis_channel_allow = bin(self._axis_ch)[2:] axis_channel_allow_list = [ch for ch in axis_channel_allow] axis_channel_allow_list.reverse() for i in range(axis_data_length): x_axis_index: int = i*6 + axis_start_index y_axis_index: int = i*6 + axis_start_index + 2 z_axis_index: int = i*6 + axis_start_index + 4 x_axis_data = data[x_axis_index + 1] << 8 | data[x_axis_index] y_axis_data = data[y_axis_index + 1] << 8 | data[y_axis_index] z_axis_data = data[z_axis_index + 1] << 8 | data[z_axis_index] # print("x_axis_data = ", x_axis_data, "y_axis_data = ", y_axis_data, "z_axis_data", z_axis_data) # using 2's complement if x_axis_data >= 0x8000: x_axis_data = x_axis_data - 0xFFFF if y_axis_data >= 0x8000: y_axis_data = y_axis_data - 0xFFFF if z_axis_data >= 0x8000: z_axis_data = z_axis_data - 0xFFFF axis_data = int(math.sqrt(x_axis_data ** 2 + y_axis_data ** 2 + z_axis_data ** 2)) if self._axis_ch >= 1: ret.append_data(x_axis_acc_ch, int((x_axis_data) / divide_ratio)) if self._axis_ch >= 2: ret.append_data(y_axis_acc_ch, int((y_axis_data) / divide_ratio)) if self._axis_ch >= 4: ret.append_data(z_axis_acc_ch, int((z_axis_data) / divide_ratio)) if self._axis_ch >= 8: ret.append_data(axis_acc_ch, int((axis_data) / divide_ratio)) # update previous time stamp self._prev_time_stamp.pop() self._prev_delta_time.pop() self._prev_time_stamp.append(time_stamp) self._prev_delta_time.append(time_delta) return ret class EISZeroOneDataDecoder(RecDataDecoder): """ ::S | | 1 | 2 | 3 | 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 ----------------------------------------------------------------- | id | | ch1 | | ch2 | | ch3 | | timestamp | | None | | finishMode | """ NAME = 'EISZeroOne' MESSAGE_CYCLE_COMPLETE = 'CycleComplete' __slots__ = ('_message', '_cycle_number', '_start_return_data', '_time_stamp', '_total_time_stamp', '_mode', '_cycle_start_time', '_mode_stop', '_last_time_stamp', '_last_delta', '_cali_coeff', 'cali_coeff', '_ac_amp', '_mode', '_freq_start', '_freq_stop', '_freq_direction', '_last_phase', '_first_phase_flag') def __init__(self, cali_coeff: bytes = None): super().__init__() self._cycle_number = 0 self._message: Optional[str] = None self._start_return_data = False self._time_stamp = 0 self._total_time_stamp = 0 self._mode_stop = 0 self._cycle_start_time = [] self._ac_amp: int = 0 self._mode: int = 0 self._freq_start: int = 0 self._freq_stop: int = 0 self._freq_direction = 0 self._last_phase = 0 self._first_phase_flag = 1 self._cali_coeff: Optional[bytes] = None self.cali_coeff: Optional[List[Tuple[int, int]]] = None if cali_coeff is not None: self._cali_coeff = cali_coeff self.cali_coeff = self._decode_cali_coeff(cali_coeff) @staticmethod def _decode_cali_coeff(cali_coeff: bytes) -> Optional[List[Tuple[int, int]]]: if cali_coeff != b'': cali_table = [] phase_para_a = [] phase_para_b = [] hsrtia_a = [] hsrtia_b = [] hsrtia_c = [] hsrtia_d = [] # print('cali_coeff', cali_coeff) cutoff_freq = struct.unpack('>I', cali_coeff[1:5])[0] * 100 #4 # temp = struct.unpack('>B', cali_coeff[5:6])[0] #1 # hsrtia_200r = struct.unpack('>B', cali_coeff[6:7])[0] #1 # hsrtia_5k = struct.unpack('>H', cali_coeff[7:9])[0] #2 # hsrtia_20k = struct.unpack('>H', cali_coeff[6:8])[0] #2 # hsrtia_160k = struct.unpack('>I', cali_coeff[8:12])[0] #4 index = 20 for i in range(index, index+16, 8): phase_para_a.append(struct.unpack('>i', cali_coeff[i+1:i+5])[0]) phase_para_b.append(struct.unpack('>i', cali_coeff[i+5:i+9])[0]) index = 40 for i in range(index, index+16, 8): phase_para_a.append(struct.unpack('>i', cali_coeff[i+1:i+5])[0]) phase_para_b.append(struct.unpack('>i', cali_coeff[i+5:i+9])[0]) #Lv[0] 160k index = 60 hsrtia_a.append(struct.unpack('>i', cali_coeff[index+1:index+5])[0]/1e8) hsrtia_b.append(struct.unpack('>i', cali_coeff[index+5:index+9])[0]/1e8) hsrtia_c.append(struct.unpack('>i', cali_coeff[index+9:index+13])[0]/1e4) hsrtia_d.append(struct.unpack('>B', cali_coeff[index+13:index+14])[0]) #Lv[1] 20k index = 80 hsrtia_a.append(struct.unpack('>i', cali_coeff[index+1:index+5])[0]/1e8) hsrtia_b.append(struct.unpack('>i', cali_coeff[index+5:index+9])[0]/1e8) hsrtia_c.append(struct.unpack('>i', cali_coeff[index+9:index+13])[0]/1e4) hsrtia_d.append(struct.unpack('>B', cali_coeff[index+13:index+14])[0]) #Lv[2] 5k index = 100 hsrtia_a.append(struct.unpack('>i', cali_coeff[index+1:index+5])[0]/1e8) hsrtia_b.append(struct.unpack('>i', cali_coeff[index+5:index+9])[0]/1e8) hsrtia_c.append(struct.unpack('>i', cali_coeff[index+9:index+13])[0]/1e4) hsrtia_d.append(struct.unpack('>B', cali_coeff[index+13:index+14])[0]) #Lv[3] 200R index = 120 hsrtia_a.append(struct.unpack('>i', cali_coeff[index+1:index+5])[0]/1e8) hsrtia_b.append(struct.unpack('>i', cali_coeff[index+5:index+9])[0]/1e8) hsrtia_c.append(struct.unpack('>i', cali_coeff[index+9:index+13])[0]/1e4) hsrtia_d.append(struct.unpack('>B', cali_coeff[index+13:index+14])[0]) # hsrtia_a.append(struct.unpack('>I', cali_coeff[index+1:index+5])[0]) # hsrtia_b.append(struct.unpack('>I', cali_coeff[index+5:index+9])[0]/1e6) # hsrtia_c.append(struct.unpack('>I', cali_coeff[index+9:index+13])[0]/1e5) # hsrtia_d.append(struct.unpack('>I', cali_coeff[index+13:index+17])[0]/1e6) # print('cutoff_freq', cutoff_freq) # print('hsrtia_a', hsrtia_a) # print('hsrtia_b', hsrtia_b) # print('hsrtia_c', hsrtia_c) # print('hsrtia_d', hsrtia_d) # print('phase_para_a', phase_para_a) # print('phase_para_b', phase_para_b) cali_table.append((cutoff_freq, phase_para_a, phase_para_b, hsrtia_a, hsrtia_b, hsrtia_c, hsrtia_d)) return cali_table else: print() return None @property def name(self) -> AnyStr: if self._cali_coeff is None: return self.NAME else: return self.NAME.encode() + b':' + self._cali_coeff def message(self) -> Optional[str]: ret = self._message self._message = None return ret def decode(self, data: bytes) -> Optional[RecordingData]: if len(data) < 18: return None ch1 = struct.unpack('>i', data[1:5])[0] # unit: 1/1000 nA ch2 = struct.unpack('>i', data[5:9])[0] # unit: mV ch3 = struct.unpack('>i', data[9:13])[0] # unit: kOm time_stamp: float = struct.unpack('H', data[17:19])[0] d19 = data[19] gain = (d19 & 0x0F) finishMode = (d19 & 0x80) >> 7 if time_stamp == 0: self._start_return_data = True if time_stamp is None or time_stamp == 0 or not self._start_return_data: return None if time_stamp != 0 and self._start_return_data == True: if (self._mode == 0): time_stamp, delta, ret_get_time_stamp = self.eis_get_time_stamp(time_stamp) else: time_stamp, delta, ret_get_time_stamp = self.get_time_stamp(time_stamp) if time_stamp is None or time_stamp < 0 or delta < 0: print("error timeStamp full data:", list(data), datetime.now(), '\n') return None else: if self.cali_coeff is not None and self._mode == 0: phase_para_a = [] phase_para_b = [] hsrtia_a = [] hsrtia_b = [] hsrtia_c = [] hsrtia_d = [] cutoff_freq, phase_para_a, phase_para_b, hsrtia_a, hsrtia_b, hsrtia_c, hsrtia_d = self.cali_coeff[0] voltage_amp = round(self._ac_amp * 800 / 2047) # use UI value if (self._freq_start > self._freq_stop): self._freq_direction = 0 else: self._freq_direction = 1 if (self._mode == 0): img = ch1 real = ch2 freq = ch3 voltage_mag = math.sqrt(img ** 2 + real ** 2) * (1 + freq ** 2 / cutoff_freq ** 2) # if (gain == 3): # current = hsrtia_a[gain] * math.exp(hsrtia_b[gain] * voltage_mag) + hsrtia_c[gain] * math.exp(hsrtia_d[gain] * voltage_mag) # else: current = voltage_mag ** 2 * hsrtia_a[gain] + voltage_mag * hsrtia_b[gain] + hsrtia_c[gain] # print(current) # print(voltage_mag) # print(hsrtia_a[gain]) # print(hsrtia_b[gain]) # print(hsrtia_c[gain]) if (current != 0): # impedance = voltage_amp * 1000_000 / 1.414213 / current impedance = voltage_amp * 707106.78 / current else: impedance = 0 if (real > 0): raw_phase = math.atan(img / real) * 180 / math.pi elif (real == 0): raw_phase = 90 else: raw_phase = math.atan(img / real) * 180 / math.pi + 180 if (freq >= 1000000): # 10000 Hz i = 0 elif (freq >= 10000): # 100 Hz i = 1 elif (freq >= 1000): # 10 Hz i = 2 elif (freq >= 1): # 0.01 Hz i = 3 ideal_raw_phase = phase_para_a[i] /1e10 * freq + phase_para_b[i] / 1e6 phase = raw_phase - ideal_raw_phase if (self._first_phase_flag): self._last_phase = phase self._first_phase_flag = 0 elif (abs(phase - self._last_phase) >= 200): phase -= 360 self._last_phase = phase imag_after_cal = impedance * math.sin(round(phase) * math.pi / 180) real_after_cal = impedance * math.cos(round(phase) * math.pi / 180) if (self._mode == 0): # print('|', '{:10}'.format(time_stamp), # '|', '{:5}'.format(delta), # '|', '{:6}'.format(ch1), # '|', '{:6}'.format(ch2), # '|', '{:8}'.format(ch3 / 100), # '|', '{:6}'.format(round(voltage_mag)), # '|', '{:5}'.format(int(imag_after_cal)), # '|', '{:5}'.format(int(real_after_cal)), # '|', '{:5}'.format(round(impedance)), # '|', '{:5}'.format(round(phase, 1)), # '|', '{:5}'.format(round(current, 3)), # '|', '{:1}'.format(gain), # '|', '{:1}'.format(finishMode), # '@', str(self.device), '|') pass else: # print('|', '{:10}'.format(time_stamp), # '|', '{:5}'.format(delta), # '|', '{:5}'.format(ch1), # '|', '{:5}'.format(ch2), # '|', '{:5}'.format(ch3), # '|', '{:5}'.format(cycle_number), # '|', '{:1}'.format(gain), # '|', '{:1}'.format(finishMode), # '@', str(self.device), '|') pass if finishMode == True: print("finishMode full data:", list(data), datetime.now()) self._mode_stop = 1 else: self._mode_stop = 0 ret = RecordingData(self.device, int(time_stamp * 1000 / 2), 0) if (self._mode == 0): #EIS Mode ret.append_data(3, cycle_number) ret.append_data(0, ch1) #Raw Imag ret.append_data(1, ch2) #Raw Real ret.append_data(2, ch3 * 10) #Frequency [mHz] ret.append_data(4, round(imag_after_cal)) #Z_imag [Ohm] ret.append_data(5, round(real_after_cal)) #Z_real [Ohm] ret.append_data(6, round(impedance)) #Impedance [Ohm] ret.append_data(7, round(phase)) #Phase [degree] ret.append_data(8, round(current)) #Current [nA] ret.append_data(9, gain) #Gain Level else: #CV Mode ret.append_data(3, cycle_number) ret.append_data(0, ch1) #Iin [nA] ret.append_data(1, ch2) #Vset [nV] ret.append_data(2, ch3) #Vout [nV] if cycle_number != self._cycle_number: # notify cycle_number change self._message = self.MESSAGE_CYCLE_COMPLETE + '=%d' % cycle_number self._cycle_number = cycle_number # print("$$$ real_time_stamp $$$") # print(time_stamp) return ret def isFinishMode(self) -> int: return self._mode_stop