import abc import struct import math import numpy 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', '_show_data') 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 = [] self._show_data = False @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 voltage = 0 mem_cnt = data[1] time_stamp: float = struct.unpack(' 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', '_show_data') 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 self._show_data = False 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 = [] hsrtia_a = [] hsrtia_b = [] hsrtia_c = [] hsrtia_d = [] phase_coeff = [] phase_offset = [] # phase_coeff = [[0]*4 for i in range(4)] # phase_offset = [[0]*4 for i in range(4)] phase_coeff = numpy.zeros([4, 4], dtype = int) phase_offset = numpy.zeros([4, 4], dtype = int) ######################################## # phase_coeff # [[gain0, g1, g2, g3] ----->最高頻 # [gain0, g1, g2, g3] ----->中頻 # [gain0, g1, g2, g3] ----->低頻 # [gain0, g1, g2, g3] ----->最低頻 # ] ####################################### # 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 g = 0 phase_coeff[0][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[0][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[1][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[1][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[0] index = 40 g = 0 phase_coeff[2][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[2][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[3][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[3][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[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]) #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]) #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]) #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]) index = 140 g = 1 phase_coeff[0][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[0][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[1][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[1][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[0] index = 160 g = 1 phase_coeff[2][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[2][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[3][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[3][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[0] index = 180 g = 2 phase_coeff[0][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[0][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[1][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[1][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[0] index = 200 g = 2 phase_coeff[2][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[2][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[3][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[3][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[0] index = 220 g = 3 phase_coeff[0][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[0][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[1][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[1][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[0] index = 240 g = 3 phase_coeff[2][g] = struct.unpack('>i', cali_coeff[index+1:index+5])[0] phase_offset[2][g] = struct.unpack('>i', cali_coeff[index+5:index+9])[0] phase_coeff[3][g] = struct.unpack('>i', cali_coeff[index+9:index+13])[0] phase_offset[3][g] = struct.unpack('>i', cali_coeff[index+13:index+17])[0] # print('cutoff_freq', cutoff_freq) # print('hsrtia_a', hsrtia_a) # print('hsrtia_b', hsrtia_b) # print('hsrtia_c', hsrtia_c) # print('phase_coeff') # print(phase_coeff) # print('phase_offset') # print(phase_offset) cali_table.append((cutoff_freq, phase_coeff, phase_offset, 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+3:5+3])[0] # unit: 1/1000 nA ch2 = struct.unpack('>i', data[5+3:9+3])[0] # unit: mV ch3 = struct.unpack('>i', data[9+3:13+3])[0] # unit: kOm time_stamp: float = struct.unpack('H', data[17+3:19+3])[0] d19 = data[19+3] 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: hsrtia_a = [] hsrtia_b = [] hsrtia_c = [] hsrtia_d = [] cutoff_freq, phase_coeff, phase_offset, 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 fre_idx = 0 rolloff_cali = cutoff_freq/1e5 voltage_mag = math.sqrt(img ** 2 + real ** 2) * (1 + freq ** 2 / rolloff_cali ** 2 / 1e4) # 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] # 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 raw_phase = math.atan2(img , real) * 180 / math.pi if (freq >= 1000000): # 10000 Hz fre_idx = 0 elif (freq >= 10000): # 100 Hz fre_idx = 1 elif (freq >= 1000): # 10 Hz fre_idx = 2 elif (freq >= 1): # 0.01 Hz fre_idx = 3 ideal_raw_phase = phase_coeff[gain][fre_idx] /1e10 * freq + phase_offset[gain][fre_idx] / 1e6 phase = raw_phase - ideal_raw_phase phase = phase % 180 if phase % 180<=90 else phase % 180-180 # last_phase_to90 = self._last_phase % 180 if self._last_phase % 180<=90 else self._last_phase % 180-180 # diff = phase - last_phase_to90 # if (self._first_phase_flag): # # self._last_phase = phase # self._first_phase_flag = 0 # elif (abs(diff) >= 90): # phase = self._last_phase + diff + (180 if diff<0 else-180) # else: # phase = self._last_phase + diff # 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._show_data: 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(0, ch1) #Raw Imag ret.append_data(1, ch2) #Raw Real ret.append_data(2, ch3 * 10) #Frequency [mHz] ret.append_data(3, cycle_number) 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(0, ch1) #Iin [nA] ret.append_data(1, ch2) #Vset [nV] ret.append_data(2, ch3) #Vout [nV] ret.append_data(3, cycle_number) 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