peterlu14
211 lines
6.6 KiB
Python
211 lines
6.6 KiB
Python
"""It is typing used of the abstract interface of RecordingData defined in c extension.
|
|
|
|
**RecordingData 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
|
|
-----------------------------------------------------------------
|
|
| format | device | channel | data width |
|
|
| time stamp |
|
|
| time stamp |
|
|
| ------------------------------------------------------------- |
|
|
| data | format = 0
|
|
| ------------------------------------------------------------- | -------------------
|
|
| data length | sampling rate | format = 1
|
|
| data[*] | * data length
|
|
| ------------------------------------------------------------- | -------------------
|
|
| data length | sampling rate | format = 2
|
|
| channel | data[*] | * data length
|
|
| ------------------------------------------------------------- | -------------------
|
|
| data length | sampling rate | format = 3
|
|
| invalid map | channel | data[8] | * data length
|
|
| ------------------------------------------------------------- | -------------------
|
|
|
|
|
|
data length
|
|
data length.
|
|
|
|
sampling rate
|
|
reciprocal of time delta (s) between each data.
|
|
0 meaning all data with same time stamp
|
|
|
|
format
|
|
data format.
|
|
|
|
time stamp
|
|
64 bits double, unit 1 msec
|
|
|
|
data length
|
|
type uint16 little endian, count of data
|
|
|
|
data width
|
|
data width in bytes
|
|
|
|
data
|
|
signed little endian integer.
|
|
|
|
invalid map
|
|
bit map, indicate next 8 data valid or not.
|
|
|
|
"""
|
|
|
|
from typing import Iterable, Tuple, Optional, IO, List
|
|
|
|
# noinspection PyUnresolvedReferences
|
|
import biopro.ext.recording as util_recording
|
|
|
|
__all__ = ['RecordingData']
|
|
|
|
inf = float("inf")
|
|
|
|
_RUNTIME_COMPILE = True
|
|
|
|
if _RUNTIME_COMPILE:
|
|
RecordingData = util_recording.RecordingData
|
|
|
|
else:
|
|
class RecordingData:
|
|
|
|
# XXX 1 byte channel may not fit the CV mode EliteZM,
|
|
# which channel also used to save the cycle number.
|
|
|
|
__slots__ = ('__impl', '__mod_count', '__mod_encode', '__encode_cache')
|
|
|
|
def __init__(self,
|
|
device: int,
|
|
time_stamp: float = 0,
|
|
sample_rate: int = 1):
|
|
"""new a RecordingData
|
|
|
|
:param device:
|
|
:param time_stamp: unit ms
|
|
:param sample_rate: unit 1/s
|
|
:return:
|
|
"""
|
|
|
|
if isinstance(device, int):
|
|
self.__impl: 'RecordingData' = util_recording.RecordingData(device, time_stamp, sample_rate)
|
|
else:
|
|
# internal call to wrap util_recording.RecordingData
|
|
|
|
# noinspection PyTypeChecker
|
|
self.__impl: 'RecordingData' = device
|
|
|
|
# encode() cache
|
|
# in commit eaa98054, I had implemented it in C, but it cause SIGSEGV,
|
|
# and I have no idea why.
|
|
self.__mod_count = 0
|
|
self.__mod_encode = 0
|
|
self.__encode_cache: Optional[bytes] = None
|
|
|
|
def __len__(self) -> int:
|
|
"""data count in this record.
|
|
|
|
if *format* equal 0, len return 1 iff data exist and valid.
|
|
otherwise, len return the number of the data no matter valid or invalid.
|
|
"""
|
|
return len(self.__impl)
|
|
|
|
def __getitem__(self, item: int) -> Tuple[int, Optional[int]]:
|
|
return self.__impl[item]
|
|
|
|
@property
|
|
def device(self) -> int:
|
|
return self.__impl.device
|
|
|
|
@property
|
|
def time_stamp(self) -> int:
|
|
"""
|
|
|
|
:return: time stamp relative to beginning. unit msec
|
|
"""
|
|
return self.__impl.time_stamp
|
|
|
|
@time_stamp.setter
|
|
def time_stamp(self, value: int):
|
|
self.__impl.time_stamp = value
|
|
|
|
@property
|
|
def data_width(self) -> int:
|
|
return self.__impl.data_width
|
|
|
|
@property
|
|
def data_capacity(self) -> int:
|
|
return self.__impl.data_capacity
|
|
|
|
@property
|
|
def data_size(self) -> int:
|
|
return self.__impl.data_size
|
|
|
|
@property
|
|
def to_time(self) -> int:
|
|
return self.__impl.to_time
|
|
|
|
@property
|
|
def sample_rate(self) -> int:
|
|
return self.__impl.sample_rate
|
|
|
|
@sample_rate.setter
|
|
def sample_rate(self, value: int):
|
|
self.__impl.sample_rate = value
|
|
|
|
def clear(self):
|
|
self.__impl.clear()
|
|
|
|
def channels(self) -> List[int]:
|
|
return self.__impl.channels()
|
|
|
|
def in_time_range(self, time_start: float, time_stop: float) -> bool:
|
|
return self.__impl.in_time_range(time_start, time_stop)
|
|
|
|
def append_data(self, channel: int, value: Optional[int]):
|
|
"""append data.
|
|
|
|
:param channel:
|
|
:param value: data value, None for invalid data
|
|
"""
|
|
self.__impl.append_data(channel, value)
|
|
self.__mod_count += 1
|
|
|
|
def encode(self) -> bytes:
|
|
if self.__encode_cache is not None:
|
|
if self.__mod_encode == self.__mod_count:
|
|
return self.__encode_cache
|
|
ret = self.__impl.encode()
|
|
|
|
self.__encode_cache = ret
|
|
self.__mod_encode = self.__mod_count
|
|
|
|
return ret
|
|
|
|
@staticmethod
|
|
def decode(data: bytes) -> 'RecordingData':
|
|
ret = RecordingData(util_recording.RecordingData.decode(data))
|
|
|
|
# set encode cache
|
|
ret.__encode_cache = data
|
|
ret.__mod_encode = ret.__mod_count
|
|
|
|
return ret
|
|
|
|
@staticmethod
|
|
def read_block(file: IO) -> Optional['RecordingData']:
|
|
ret = util_recording.RecordingData.read_block(file)
|
|
|
|
if ret is None:
|
|
return None
|
|
else:
|
|
return RecordingData(ret)
|
|
|
|
def __iter__(self) -> Iterable[Tuple[float, int, Optional[int]]]:
|
|
"""same as entry_iter(0)"""
|
|
return iter(self.__impl)
|
|
|
|
def entry_iter(self,
|
|
time_start: float = 0.0,
|
|
time_stop: float = inf) -> Iterable[Tuple[float, int, Optional[int]]]:
|
|
"""iterator data in this recording data"""
|
|
return self.__impl.entry_iter(time_start, time_stop)
|