Roy
1271 lines
51 KiB
Python
1271 lines
51 KiB
Python
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 == RawDataDecoder.NAME:
|
|
return RawDataDecoder.INSTANCE
|
|
|
|
elif expr == TDC4VCDataDecoder.NAME:
|
|
return TDC4VCDataDecoder()
|
|
|
|
elif expr == I4V4Z4T4DataDecoder.NAME:
|
|
return I4V4Z4T4DataDecoder()
|
|
|
|
elif expr == PEL_FMT1_DataDecoder.NAME:
|
|
return PEL_FMT1_DataDecoder()
|
|
|
|
elif isinstance(expr, str):
|
|
raise RuntimeError('current not support custom data decode format : ' + expr)
|
|
|
|
elif isinstance(expr, bytes):
|
|
if expr.startswith(b'TDC4VC:'):
|
|
return TDC4VCDataDecoder(expr[7:])
|
|
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 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:
|
|
CC2650_TIMESTAMP_OVERFLOW = (int)(0xFFFF_FFFF / 31)
|
|
BMD380_TIMESTAMP_OVERFLOW = (0xFFFF_FFFF)
|
|
|
|
__slots__ = '_start_time', '_prev_time', '_prev_prev_time', '_overflow'
|
|
|
|
def __init__(self):
|
|
self._start_time: Dict[int, float] = {}
|
|
self._prev_time: Dict[int, float] = {}
|
|
self._prev_prev_time: Dict[int, float] = {}
|
|
self._overflow: Dict[int, float] = {}
|
|
|
|
for dev_num in range(8):
|
|
for d in [self._start_time, self._prev_time, self._prev_prev_time, self._overflow]:
|
|
d[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
|
|
|
|
# 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.CC2650_TIMESTAMP_OVERFLOW + 1) - prev_time
|
|
ret = (self.CC2650_TIMESTAMP_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
|
|
|
|
else:
|
|
delta = time_stamp - prev_time
|
|
self._prev_time[device] = time_stamp
|
|
overflow = self._overflow[device]
|
|
ret = time_stamp + (self.CC2650_TIMESTAMP_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
|
|
|
|
# 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.CC2650_TIMESTAMP_OVERFLOW + 1) - prev_time
|
|
ret = (self.CC2650_TIMESTAMP_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.CC2650_TIMESTAMP_OVERFLOW + 1) * overflow - start_time
|
|
self._prev_prev_time[device] = prev_time
|
|
return ret, delta, False
|
|
|
|
def get_bmd380_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
|
|
|
|
start_time = self._start_time[device]
|
|
prev_time = self._prev_time[device]
|
|
prev_prev_time = self._prev_prev_time[device]
|
|
overflow = self._overflow[device]
|
|
|
|
# timestamp overflow
|
|
if time_stamp < 0x0100_0000 and 0x0F00_0000 < prev_time:
|
|
overflow += 1
|
|
delta = time_stamp + (self.BMD380_TIMESTAMP_OVERFLOW + 1) - prev_time
|
|
ret = (self.BMD380_TIMESTAMP_OVERFLOW + 1) * overflow + time_stamp - start_time
|
|
print("time_stamp overflow:", "time_stamp:", time_stamp, "; prev_time:", prev_time, "; prev_prev_time:", prev_prev_time, "; delta:", delta, ",", datetime.now())
|
|
print("time_stamp overflow:", "overflow_times:", overflow, "; device:", device)
|
|
|
|
else:
|
|
delta = time_stamp - prev_time
|
|
ret = (self.BMD380_TIMESTAMP_OVERFLOW + 1) * overflow + time_stamp - start_time
|
|
|
|
self._overflow[device] = overflow
|
|
self._prev_prev_time[device] = prev_time
|
|
self._prev_time[device] = time_stamp
|
|
|
|
return ret, delta, False
|
|
|
|
class RecDataDecoder(DataDecodeFormat[RecordingData], metaclass=abc.ABCMeta):
|
|
|
|
__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)
|
|
|
|
def get_bmd380_time_stamp(self, time_stamp: float) -> Tuple[Optional[float], Optional[float]]:
|
|
return self.timer.get_bmd380_time_stamp(self.device, time_stamp)
|
|
|
|
@abc.abstractmethod
|
|
def decode(self, data: bytes) -> Optional[RecordingData]:
|
|
pass
|
|
|
|
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',
|
|
'_last_mem_wrong_information', '_last_mem_cnt', '_last_elite_notify_times')
|
|
|
|
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
|
|
|
|
self._last_mem_wrong_information = -1
|
|
self._last_mem_cnt = -1
|
|
self._last_elite_notify_times = -1
|
|
|
|
@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
|
|
|
|
#/* Elite Notify data:
|
|
# * +--------+----------+---------+---------+---------+-----------+-----------------+
|
|
# * | id(1B) | time(4B) | ch1(4B) | ch2(4B) | ch3(4B) | cycle(2B) | finish_flag(1B) |
|
|
# * | bat(4B) | notify#(1B) | ch4(4B) | ch5(4B) | ch6(4B) | __(3B) |
|
|
# * +---------+-------------+---------+---------+---------+--------+
|
|
# */
|
|
#/*
|
|
# * EliteADCControl(): use ADC plot, and send what data to controller
|
|
# * +---------------------------+-----------+-----------+-----------+-----------+-----------+
|
|
# * | MODE | ch1 | ch2 | ch3 | cycle | ch4 |
|
|
# * +---------------------------+-----------+-----------+-----------+-----------+-----------+
|
|
# * | CURVE_IV | Iin | Vout | Vin | | Vmon |
|
|
# * | CURVE_IV_CY | Iin | Vout | Vin | v | Vmon |
|
|
# * | CURVE_VO | Iin | Vout | Vin | | Vmon |
|
|
# * | CURVE_RT | Iin | Vout | R | | Vmon |
|
|
# * | CURVE_VT | Iin | Vin | | | |
|
|
# * | CURVE_IT | Iin | Vin | Vout | | Vmon |
|
|
# * | CURVE_CC | Iin | Vin | Vout | | Vmon |
|
|
# * | CURVE_CP | Iin | Vout-Vin | Vout | | Vmon |
|
|
# * | CURVE_CV | Iin | Vout-Vin | Vout | v | Vmon |
|
|
# * | CURVE_LSV | Iin | Vout-Vin | Vout | | Vmon |
|
|
# * | CURVE_CA | Iin | Vout-Vin | Vout | | Vmon |
|
|
# * | CURVE_OCP | Iin | Vmon-Vin | Vin | | Vmon |
|
|
# * | CURVE_UNI_PULSE | pul1_Iin | pul2_Iin | | | |
|
|
# * | CURVE_DPV | c1&c2_avg | Vout-Vin | Vout | | Vmon |
|
|
# * | CURVE_DPV_SMPRATE | Iin | Vout-Vin | Vout | | Vmon |
|
|
# * | CURVE_DPV_ADVANCE | c1&c2_avg | Vout-Vin | Vout | | Vmon |
|
|
# * | CURVE_DPV_ADVANCE_SMPRATE | Iin | Vout-Vin | Vout | | Vmon |
|
|
# * +---------------------------+-----------+-----------+-----------+-----------+-----------+
|
|
# *
|
|
# * ps. c1_avg = pul1_Iin
|
|
# * ps. c2_avg = pul2_Iin
|
|
# */
|
|
|
|
mem_cnt = data[1]
|
|
time_stamp: float = struct.unpack('<I', data[1+3:5+3])[0]
|
|
ch1 = struct.unpack('<i', data[5+3:9+3])[0] # unit: nA
|
|
ch2 = struct.unpack('<i', data[9+3:13+3])[0] # unit: uV
|
|
ch3 = struct.unpack('<i', data[13+3:17+3])[0] # unit: mOm
|
|
|
|
cycle_number = struct.unpack('<H', data[17+3:19+3])[0]
|
|
finish_mode_falg = data[19+3]
|
|
battery = struct.unpack('<i', data[20+3:24+3])[0]
|
|
elite_notify_times = data[24+3]
|
|
ch4 = struct.unpack('<i', data[25+3:29+3])[0]
|
|
ch5 = struct.unpack('<i', data[29+3:33+3])[0]
|
|
ch6 = struct.unpack('<i', data[33+3:37+3])[0]
|
|
|
|
mem_wrong_information = struct.unpack('<i', data[43:47])[0] # mem_wrong_information = green retry, green wrong, red retry, red wrong
|
|
|
|
if mem_wrong_information != self._last_mem_wrong_information:
|
|
print(datetime.now(), 'device', str(self.device), 'mem_wrong_information[43:47]:', data[43:47], mem_wrong_information, self._last_mem_wrong_information, flush = True)
|
|
if mem_cnt != self._last_mem_cnt+1:
|
|
if not (mem_cnt == 0 and self._last_mem_cnt == 255):
|
|
print(datetime.now(), 'device', str(self.device), 'mem_cnt:', mem_cnt, 'self._last_mem_cnt:', self._last_mem_cnt, flush = True)
|
|
if (elite_notify_times != self._last_elite_notify_times+1) and not (elite_notify_times == 0 and self._last_elite_notify_times == 0):
|
|
if not (elite_notify_times == 0 and self._last_elite_notify_times == 255):
|
|
print(datetime.now(), 'device', str(self.device), 'elite_notify_times:', elite_notify_times, 'self._elite_notify_times:', self._last_elite_notify_times, flush = True)
|
|
self._last_mem_wrong_information = mem_wrong_information
|
|
self._last_mem_cnt = mem_cnt
|
|
self._last_elite_notify_times = elite_notify_times
|
|
|
|
ram_num = data[47]
|
|
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:
|
|
if self._show_data:
|
|
print('|', time_stamp, '|', delta, '|', int(time_stamp * 1000 / 2),
|
|
'|', ch1, '|', ch2, '|', ch3, '|', cycle_number,
|
|
'|', ch4, '|', ch5, '|', ch6,
|
|
'|', finishMode, '@', str(self.device), flush = True)
|
|
|
|
if finishMode == True:
|
|
print("finishMode full data:", list(data), datetime.now())
|
|
self._mode_stop = 1
|
|
else:
|
|
self._mode_stop = 0
|
|
|
|
channels = [0, 1, 2, 3, 4, 5, 6]
|
|
values = [
|
|
ch1,
|
|
ch2,
|
|
ch3,
|
|
cycle_number,
|
|
ch4,
|
|
ch5,
|
|
ch6
|
|
]
|
|
|
|
result = {
|
|
"time": int(time_stamp * 1000 / 2),
|
|
"channels": channels,
|
|
"data": values,
|
|
"device": self.device
|
|
}
|
|
|
|
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 result
|
|
|
|
def isFinishMode(self) -> int:
|
|
return self._mode_stop
|
|
|
|
class PEL_FMT1_DataDecoder(RecDataDecoder):
|
|
NAME = 'PEL_FMT1_'
|
|
|
|
MESSAGE_CYCLE_COMPLETE = 'CycleComplete'
|
|
|
|
__slots__ = ('_message', '_mode', '_mode_stop', '_show_data')
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
self._message: Optional[str] = None
|
|
self._mode_stop = 0
|
|
self._mode = 0
|
|
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]:
|
|
mem_board_id: int = struct.unpack('<B', data[0+3:1+3])[0]
|
|
packet_sequence: int = struct.unpack('<B', data[1+3:2+3])[0]
|
|
raw_time_stamp: float = struct.unpack('<I', data[2+3:6+3])[0]
|
|
|
|
if all(b == 0 for b in data[0+3:42+3]):
|
|
hex_string = ' '.join(format(i, '02X') for i in data)
|
|
print(hex_string, flush=True)
|
|
return None
|
|
|
|
timestamp, delta, ret_get_time_stamp = self.get_bmd380_time_stamp(raw_time_stamp)
|
|
output_vo = struct.unpack('<i', data[6+3:10+3])[0]
|
|
output_vc = struct.unpack('<i', data[10+3:14+3])[0]
|
|
output_ve = struct.unpack('<i', data[14+3:18+3])[0]
|
|
hold_OUT = struct.unpack('<f', data[18+3:22+3])[0]
|
|
hold_VCC = struct.unpack('<f', data[22+3:26+3])[0]
|
|
hold_VEE = struct.unpack('<f', data[26+3:30+3])[0]
|
|
resis_pattern_id = struct.unpack('<H', data[30+3:32+3])[0]
|
|
resis_bitsmask = struct.unpack('<H', data[32+3:34+3])[0]
|
|
resis_g_val = struct.unpack('<f', data[34+3:38+3])[0]
|
|
mode_flag = struct.unpack('<I', data[38+3:42+3])[0]
|
|
flag_mode_complete = (mode_flag & 0x0000000F)
|
|
val1 = 0
|
|
val2 = 0
|
|
val3 = 0
|
|
|
|
values = [val1, val2, output_vo, output_vc, output_ve, val3, resis_g_val, hold_OUT, hold_VCC, hold_VEE, resis_pattern_id, resis_bitsmask]
|
|
channels = [0,1,2,3,4,5,6,7,8,9,10,11]
|
|
|
|
if self._mode == 3:
|
|
VO_VEE = hold_OUT - hold_VEE
|
|
current = VO_VEE * resis_g_val
|
|
VO_VCC = hold_OUT - hold_VCC
|
|
values.extend([VO_VEE, current, VO_VCC])
|
|
channels.extend([12,13,14])
|
|
if self._mode == 4:
|
|
VCC_VO = hold_VCC - hold_OUT
|
|
current = VCC_VO * resis_g_val
|
|
VO_VEE = hold_OUT - hold_VEE
|
|
values.extend([VCC_VO, current, VO_VEE])
|
|
channels.extend([12,13,14])
|
|
if self._mode == 5:
|
|
VO_VEE = hold_OUT - hold_VEE
|
|
current = VO_VEE * resis_g_val
|
|
VO_VCC = hold_OUT - hold_VCC
|
|
values.extend([VO_VEE, current, VO_VCC])
|
|
channels.extend([12,13,14])
|
|
|
|
print(mem_board_id, packet_sequence, raw_time_stamp, timestamp, delta, output_vo, output_vc, output_ve, hold_OUT, hold_VCC, hold_VEE, resis_pattern_id, resis_bitsmask, resis_g_val, mode_flag, flush=True)
|
|
|
|
result = {
|
|
"time": timestamp,
|
|
"channels": channels,
|
|
"data": values,
|
|
"device": self.device
|
|
}
|
|
|
|
if (flag_mode_complete == 0x0A):
|
|
print("!!!!!!finish Mode", datetime.now(), flush = True)
|
|
self._mode_stop = 1
|
|
else:
|
|
self._mode_stop = 0
|
|
|
|
# Use a list comprehension to unpack all values
|
|
# values = [struct.unpack(fmt, data[start:end])[0] for start, end, fmt in data_specs]
|
|
return result
|
|
|
|
def isFinishMode(self) -> int:
|
|
return self._mode_stop
|
|
|
|
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_package', 'cali_coeff', '_ac_amp', '_mode',
|
|
'_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._mode: int = 0
|
|
self._last_phase = 0
|
|
self._first_phase_flag = 1
|
|
|
|
self._cali_package: Optional[bytes] = None
|
|
self.cali_coeff: Optional[List[Tuple[int, int]]] = None
|
|
|
|
self._show_data = False
|
|
|
|
if self._cali_package is None:
|
|
self._cali_package = cali_coeff
|
|
self.cali_coeff = self._decode_cali_coeff(self._cali_package)
|
|
|
|
@staticmethod
|
|
def _decode_cali_coeff(cali_coeff: bytes) -> Optional[List[Tuple[int, int]]]:
|
|
if cali_coeff != b'':
|
|
cis_data_len = 20
|
|
cali_table = []
|
|
hsrtia_a = []
|
|
hsrtia_b = []
|
|
rolloff = []
|
|
phase_coeff = []
|
|
phase_offset = []
|
|
|
|
phase_coeff = numpy.zeros([8, 4], dtype = int)
|
|
phase_offset = numpy.zeros([8, 4], dtype = int)
|
|
|
|
########################################
|
|
# phase_coeff
|
|
# [[freq0, freq1, freq2, freq3] ----->gain0
|
|
# [freq0, freq1, freq2, freq3] ----->gain1
|
|
# [freq0, freq1, freq2, freq3] ----->gain2
|
|
# [freq0, freq1, freq2, freq3] ----->gain3
|
|
# ]
|
|
#######################################
|
|
|
|
#hstia=0
|
|
cis_cali_packet = 1
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 2
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 0
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 3
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 0
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
#hstia=1
|
|
cis_cali_packet = 4
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 5
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 1
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 6
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 1
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
#hstia=2
|
|
cis_cali_packet = 7
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 8
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 2
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 9
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 2
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
#hstia=3
|
|
cis_cali_packet = 10
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 11
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 3
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 12
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 3
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
#hstia=4
|
|
cis_cali_packet = 13
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 14
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 4
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 15
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 4
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
#hstia=5
|
|
cis_cali_packet = 16
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 17
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 5
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 18
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 5
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
#hstia=6
|
|
cis_cali_packet = 19
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 20
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 6
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 21
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 6
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
#hstia=7
|
|
cis_cali_packet = 22
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
hsrtia_a.append(struct.unpack('>i', cali_coeff[index+2:index+6])[0])
|
|
hsrtia_b.append(struct.unpack('>q', cali_coeff[index+6:index+14])[0])
|
|
rolloff.append(struct.unpack('>i', cali_coeff[index+14:index+18])[0])
|
|
|
|
cis_cali_packet = 23
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 7
|
|
phase_coeff[g][0] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][0] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][1] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][1] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
|
|
cis_cali_packet = 24
|
|
index = (cis_cali_packet - 1) * cis_data_len
|
|
g = 7
|
|
phase_coeff[g][2] = struct.unpack('>i', cali_coeff[index+2:index+6])[0]
|
|
phase_offset[g][2] = struct.unpack('>i', cali_coeff[index+6:index+10])[0]
|
|
phase_coeff[g][3] = struct.unpack('>i', cali_coeff[index+10:index+14])[0]
|
|
phase_offset[g][3] = struct.unpack('>i', cali_coeff[index+14:index+18])[0]
|
|
print('hsrtia_a', hsrtia_a)
|
|
print('hsrtia_b', hsrtia_b)
|
|
print('rolloff', rolloff)
|
|
print('phase_coeff')
|
|
print(phase_coeff)
|
|
print('phase_offset')
|
|
print(phase_offset)
|
|
|
|
cali_table.append((phase_coeff, phase_offset, hsrtia_a, hsrtia_b, rolloff))
|
|
|
|
return cali_table
|
|
else:
|
|
print()
|
|
return None
|
|
|
|
@property
|
|
def name(self) -> AnyStr:
|
|
if self._cali_package is None:
|
|
return self.NAME
|
|
else:
|
|
return self.NAME.encode() + b':' + self._cali_package
|
|
|
|
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('<I', data[13+3:17+3])[0] # unit: ms
|
|
cycle_number = struct.unpack('>H', data[17+3:19+3])[0]
|
|
d19 = data[19+3]
|
|
gain = data[20+3]
|
|
finishMode = (d19 & 0x80) >> 7
|
|
ch4 = struct.unpack('<i', data[21+3:25+3])[0] # Amp[uV]
|
|
|
|
notify_one = struct.unpack('<i', data[25+3:29+3])[0]
|
|
notify_two = struct.unpack('<i', data[29+3:33+3])[0]
|
|
notify_three = struct.unpack('<i', data[33+3:37+3])[0]
|
|
|
|
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 or self._mode == 5):
|
|
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 or self._mode == 5):
|
|
hsrtia_a = []
|
|
hsrtia_b = []
|
|
rolloff = []
|
|
phase_coeff, phase_offset, hsrtia_a, hsrtia_b, rolloff = self.cali_coeff[0]
|
|
|
|
if (self._mode == 0 or self._mode == 5):
|
|
img = ch1
|
|
real = ch2
|
|
freq = ch3
|
|
fre_idx = 0
|
|
voltage_amp = round(ch4 / 1000) # Amp[mV]
|
|
rolloff_cali = rolloff[gain]
|
|
|
|
voltage_mag = math.sqrt(img ** 2 + real ** 2) * (1 + freq ** 2 / rolloff_cali ** 2 / 1e4)
|
|
current = (voltage_mag ** 2 * hsrtia_a[gain] + voltage_mag * hsrtia_b[gain]) / 1e8 #[nA]
|
|
|
|
if (current != 0):
|
|
# impedance[mOhm] = voltage_amp[mv] * 1000000 / 1.414213 / current[nA] #RMS=amp*SQRT(2), SQRT(2)=1.414213
|
|
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
|
|
|
|
imag_after_cal = round(impedance * math.sin(phase * math.pi / 180))
|
|
real_after_cal = round(impedance * math.cos(phase * math.pi / 180))
|
|
|
|
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 in (0, 5):
|
|
channels = [0, 1, 2, 3, 4, 5, 6, 7, 8, 9,
|
|
10, 11, 12, 13]
|
|
values = [
|
|
ch1, # 0 raw_img
|
|
ch2, # 1 raw_real
|
|
ch3 * 10, # 2 Frequency [mHz]
|
|
cycle_number, # 3 cycle
|
|
imag_after_cal, # 4 Z_imag [Ω]
|
|
real_after_cal, # 5 Z_real [Ω]
|
|
round(impedance), # 6 Impedance [Ω]
|
|
round(phase * 1000), # 7 Phase [millidegree]
|
|
round(current), # 8 Current [nA]
|
|
gain, # 9 gain
|
|
notify_one, # 10 debug1
|
|
notify_two, # 11 debug2
|
|
notify_three, # 12 debug3
|
|
voltage_amp # 13 Amp [mV]
|
|
]
|
|
|
|
else:
|
|
#CV Mode
|
|
channels = [0, 1, 2, 3]
|
|
values = [
|
|
ch1, # 0 Iin [nA]
|
|
ch2, # 1 Vset [nV]
|
|
ch3, # 2 Vout [nV]
|
|
cycle_number # 3 cycle
|
|
]
|
|
|
|
result = {
|
|
"time": int(time_stamp * 1000 / 2),
|
|
"channels": channels,
|
|
"data": values,
|
|
"device": self.device
|
|
}
|
|
|
|
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 result
|
|
|
|
def isFinishMode(self) -> int:
|
|
return self._mode_stop
|
|
|