peterlu14
991 lines
31 KiB
Python
991 lines
31 KiB
Python
"""Artifact data generator.
|
|
|
|
cli options
|
|
-----------
|
|
|
|
::
|
|
|
|
./BioProController/RaspBerryPi3/python/biopro/generator.py [OPTIONS] [FILE]
|
|
|
|
|
|
OPTIONS
|
|
-D, --device DEVICE device ID
|
|
|
|
-L, --library PATH library search path
|
|
|
|
-T, --type TYPE output data format.
|
|
raw : raw data (default)
|
|
bprf : BioPro Recording File (output to file only)
|
|
|
|
-a, --amplitude VALUE signal amplitude, default 500
|
|
|
|
-c, --channel CH[,CH] channel list, default [0]
|
|
|
|
-g, --generator GENERATOR generator, could be:
|
|
zero : all zero
|
|
sin : sine save (default)
|
|
ramp : ramp
|
|
(Elite, Neulive serial)
|
|
TC4VAF2 Elite_Legacy
|
|
TDC4VAF2 Elite Neulive
|
|
(NeuliveSTI serial)
|
|
NeuliveSTI NeuliveSTI1.0 TDBC4VCTH
|
|
(EliteZM serial)
|
|
EliteZM I4V4Z4T4
|
|
(mode :IV, :FG, :CV, :ZT, :VT, :IT)
|
|
|
|
-h, --help print help document
|
|
|
|
-n, --noise VALUE noise factor, default 0
|
|
|
|
-p just list parameter, do not run anything
|
|
|
|
-s, --sample-rate RATE sampling rate in unit 1/s.
|
|
could be [N]K, default 100.
|
|
|
|
-t, --duration VALUE total time duration in unit second.
|
|
could be [N]h[N]m[N][.NNN]. also can use 'inf' to run forever. default 1.
|
|
|
|
|
|
OPTIONS (--type=bprf)
|
|
--bprf-verbose [NEWLINE] print what it write to stdout, default NEWLINE = \n
|
|
|
|
--split-size SIZE file split for every size, could be [N]K, [N]M, [N]G.
|
|
|
|
--split-time TIME file split for every second
|
|
|
|
|
|
ARGUMENTS
|
|
FILE output file, '-' output to stdout (default)
|
|
|
|
Class
|
|
-----
|
|
|
|
"""
|
|
|
|
import sys
|
|
from math import sin, pi
|
|
from random import random as _random
|
|
from typing import IO, Type
|
|
|
|
from biopro.devlib.data import I4V4Z4T4DataDecoder, EODInterrupt
|
|
from biopro.devlib.library import *
|
|
from biopro.util.cli import *
|
|
from biopro.util.text import part_prefix, part_suffix
|
|
from .recording import RecordingData, RecordingMetaFile, RecordingFileWriter
|
|
|
|
T = TypeVar('T')
|
|
|
|
|
|
def _sign(v: float) -> int:
|
|
return 1 if v > 0 else -1
|
|
|
|
|
|
RT_1 = Tuple[float, float, int, Optional[RecordingData]]
|
|
|
|
|
|
class Generator(metaclass=abc.ABCMeta):
|
|
"""abstract artifact data generator"""
|
|
|
|
__slots__ = ('device', 'configuration', 'channel', 'sample_rate', 'amplitude', 'noise',
|
|
'_start_time_stamp', '_prev_time_stamp')
|
|
|
|
def __init__(self, options: 'GeneratorOptions', configuration: Optional[DeviceConfiguration] = None):
|
|
# basic options
|
|
self.device = options.device
|
|
|
|
# addiction device configuration
|
|
self.configuration = configuration
|
|
|
|
self.channel = tuple(options.channel)
|
|
self.sample_rate = options.sample_rate
|
|
self.amplitude = options.amplitude
|
|
self.noise = options.noise
|
|
|
|
# internal data
|
|
self._start_time_stamp: Optional[float] = None
|
|
self._prev_time_stamp: Optional[float] = None
|
|
|
|
def data_should_generated(self, time_stamp: float, max_time_delta: float = None) -> RT_1:
|
|
"""calculate how many data should be generated according to the *sample_rate* and how
|
|
long did previous function be called.
|
|
|
|
:param time_stamp: current time stamp
|
|
:param max_time_delta: max time delta
|
|
:return: time stamp, time pass and the number of data should be generated.
|
|
"""
|
|
if self._start_time_stamp is None:
|
|
self._start_time_stamp = time_stamp
|
|
self._prev_time_stamp = time_stamp
|
|
return time_stamp, 0.0, 0, None
|
|
|
|
else:
|
|
delta = time_stamp - self._prev_time_stamp
|
|
|
|
if delta < 0:
|
|
return self._prev_time_stamp, 0.0, 0, None
|
|
|
|
if max_time_delta is not None and max_time_delta < delta:
|
|
delta = max_time_delta
|
|
|
|
count = int(delta * self.sample_rate)
|
|
|
|
if count == 0:
|
|
return time_stamp, 0.0, 0, None
|
|
|
|
elif count > 0xFFFF:
|
|
count = 0xFFF0
|
|
delta = count / self.sample_rate
|
|
|
|
data = self.new_data(int(1000 * (self._prev_time_stamp - self._start_time_stamp)))
|
|
|
|
else:
|
|
data = self.new_data(int(1000 * (self._prev_time_stamp - self._start_time_stamp)))
|
|
|
|
self._prev_time_stamp += delta
|
|
return self._prev_time_stamp, delta, count, data
|
|
|
|
def new_data(self, time_stamp: int) -> RecordingData:
|
|
return RecordingData(self.device, time_stamp, self.sample_rate)
|
|
|
|
@abc.abstractmethod
|
|
def get_data(self, time_stamp: float) -> List[RecordingData]:
|
|
""" generating data.
|
|
|
|
:param time_stamp: current time stamp in unit second
|
|
:return: list of (channel data)
|
|
"""
|
|
pass
|
|
|
|
def message(self) -> Optional[str]:
|
|
"""this method behavior as same as :func:`biopro.devlib.data.DataDecodeFormat.message`.
|
|
|
|
:return:
|
|
"""
|
|
return None
|
|
|
|
def factor_noise(self) -> float:
|
|
"""
|
|
|
|
:return: noise factor
|
|
"""
|
|
return 1 + self.noise * _random() - self.noise / 2
|
|
|
|
def new_meta_configuration(self) -> SimpleDeviceConfiguration:
|
|
return SimpleDeviceConfiguration({
|
|
DeviceConfiguration.LIBRARY: 'GENERATOR',
|
|
DeviceConfiguration.VERSION: '0.4.0',
|
|
DeviceConfiguration.SAMPLE_RATE: self.sample_rate,
|
|
DeviceConfiguration.CHANNEL: self.channel,
|
|
DeviceConfiguration.AMP_GAIN: 1,
|
|
})
|
|
|
|
|
|
class ZeroGenerator(Generator):
|
|
__slots__ = ()
|
|
|
|
def __init__(self, options: 'GeneratorOptions', configuration: Optional[DeviceConfiguration] = None):
|
|
super().__init__(options, configuration)
|
|
|
|
def get_data(self, time_stamp: float) -> List[RecordingData]:
|
|
if time_stamp < 0:
|
|
raise ValueError('negative time stamp value : ' + str(time_stamp))
|
|
|
|
ret = []
|
|
|
|
next_time_stamp = -1
|
|
|
|
while next_time_stamp < time_stamp:
|
|
next_time_stamp, time_pass, count, data = self.data_should_generated(time_stamp)
|
|
|
|
if data is not None:
|
|
for i in range(count):
|
|
channel_id = self.channel[i % len(self.channel)]
|
|
value = int(self.amplitude * self.factor_noise())
|
|
data.append_data(channel_id, value)
|
|
|
|
ret.append(data)
|
|
|
|
return ret
|
|
|
|
def new_meta_configuration(self) -> SimpleDeviceConfiguration:
|
|
ret = super().new_meta_configuration()
|
|
|
|
ret.set_parameter('MODE', 'zero')
|
|
|
|
return ret
|
|
|
|
|
|
class SinWaveGenerator(Generator):
|
|
"""Artifact sin ware data generator.
|
|
|
|
"""
|
|
|
|
TIME_CONSTANT = 'TIME_CONSTANT'
|
|
|
|
__slots__ = ('time_constant', '_phase')
|
|
|
|
def __init__(self, options: 'GeneratorOptions', configuration: Optional[DeviceConfiguration] = None):
|
|
super().__init__(options, configuration)
|
|
|
|
self.time_constant = 1
|
|
self._phase = 0.0
|
|
|
|
def get_data(self, time_stamp: float) -> List[RecordingData]:
|
|
if time_stamp < 0:
|
|
raise ValueError('negative time stamp value : ' + str(time_stamp))
|
|
|
|
ret = []
|
|
|
|
t = self.time_constant
|
|
channel_size = len(self.channel)
|
|
|
|
next_time_stamp = -1
|
|
|
|
while next_time_stamp < time_stamp:
|
|
next_time_stamp, time_pass, count, data = self.data_should_generated(time_stamp)
|
|
|
|
if data is not None:
|
|
|
|
for i in range(count):
|
|
channel_id = self.channel[i % channel_size]
|
|
value = int(self.amplitude * sin(self._phase / t * pi / 180))
|
|
data.append_data(channel_id, value)
|
|
|
|
self._phase += 1
|
|
|
|
ret.append(data)
|
|
|
|
return ret
|
|
|
|
def new_meta_configuration(self) -> SimpleDeviceConfiguration:
|
|
ret = super().new_meta_configuration()
|
|
|
|
ret.set_parameter('MODE', 'sin')
|
|
ret.set_parameter(self.TIME_CONSTANT, self.time_constant)
|
|
|
|
return ret
|
|
|
|
|
|
class RampGenerator(Generator):
|
|
"""Artifact ramp data generator"""
|
|
|
|
__slots__ = ('_count',)
|
|
|
|
def __init__(self, options: 'GeneratorOptions', configuration: Optional[DeviceConfiguration] = None):
|
|
super().__init__(options, configuration)
|
|
|
|
self._count = -self.amplitude
|
|
|
|
def get_data(self, time_stamp: float) -> List[RecordingData]:
|
|
if time_stamp < 0:
|
|
raise ValueError('negative time stamp value : ' + str(time_stamp))
|
|
|
|
ret = []
|
|
|
|
next_time_stamp = -1
|
|
|
|
while next_time_stamp < time_stamp:
|
|
next_time_stamp, time_pass, count, data = self.data_should_generated(time_stamp)
|
|
|
|
if data is not None:
|
|
amp = self.amplitude
|
|
cnt = self._count
|
|
|
|
for i in range(count):
|
|
channel_id = self.channel[i % len(self.channel)]
|
|
value = int(cnt * self.factor_noise())
|
|
data.append_data(channel_id, value)
|
|
|
|
cnt += 1
|
|
|
|
if cnt > amp:
|
|
cnt = -amp
|
|
|
|
self._count = cnt
|
|
ret.append(data)
|
|
|
|
return ret
|
|
|
|
def new_meta_configuration(self) -> SimpleDeviceConfiguration:
|
|
ret = super().new_meta_configuration()
|
|
|
|
ret.set_parameter('MODE', 'ramp')
|
|
|
|
return ret
|
|
|
|
|
|
class DeviceGenerator(Generator):
|
|
SUPPORT = {
|
|
'TC4VAF2': ('Elite_Legacy', '0.0.0'),
|
|
'Elite_Legacy': ('Elite_Legacy', '0.0.0'),
|
|
'TDC4VAF2': ('Elite', '0.1.0'),
|
|
'Elite': ('Elite', '0.1.0'),
|
|
'Neulive': ('Neulive', '1.2.0'),
|
|
}
|
|
|
|
__slots__ = ('_phase',)
|
|
|
|
def __init__(self, options: 'GeneratorOptions', configuration: DeviceConfiguration):
|
|
if configuration is None:
|
|
raise TypeError('NoneType configuration')
|
|
|
|
super().__init__(options, configuration)
|
|
|
|
self.amplitude = min(512.0, abs(self.amplitude))
|
|
self._phase = 0
|
|
|
|
def get_data(self, time_stamp: float) -> List[RecordingData]:
|
|
if time_stamp < 0:
|
|
raise ValueError('negative time stamp value : ' + str(time_stamp))
|
|
|
|
ret = []
|
|
|
|
time_step = 10 / self.sample_rate
|
|
nxt = -1 # next_time_stamp
|
|
|
|
while nxt < time_stamp:
|
|
nxt, time_pass, count, data = self.data_should_generated(time_stamp, time_step)
|
|
|
|
if data is not None and len(self.channel) != 0:
|
|
|
|
for i in range(count):
|
|
channel_id = self.channel[i % len(self.channel)]
|
|
value = int(self.amplitude * sin(self._phase * pi / 180) * self.factor_noise())
|
|
data.append_data(channel_id, value)
|
|
|
|
self._phase += 1
|
|
|
|
ret.append(data)
|
|
|
|
return ret
|
|
|
|
def new_meta_configuration(self) -> SimpleDeviceConfiguration:
|
|
conf = SimpleDeviceConfiguration(self.configuration)
|
|
|
|
conf.set_parameter(DeviceConfiguration.SAMPLE_RATE, self.sample_rate)
|
|
conf.set_parameter(DeviceConfiguration.CHANNEL, self.channel)
|
|
conf.set_parameter(DeviceConfiguration.AMP_GAIN, 1)
|
|
|
|
return conf
|
|
|
|
|
|
class NeuliveSTIGenerator(Generator):
|
|
SUPPORT = {
|
|
'NeuliveSTI': ('NeuliveSTI1.0', '1.0.0'),
|
|
'NeuliveSTI1.0': ('NeuliveSTI1.0', '1.0.0'),
|
|
'TDBC4VCTH': ('NeuliveSTI1.0', '1.0.0'),
|
|
}
|
|
|
|
__slots__ = ('voltage_drop_rate', '_voltage')
|
|
|
|
def __init__(self, options: 'GeneratorOptions', configuration: DeviceConfiguration):
|
|
if configuration is None:
|
|
raise TypeError('NoneType configuration')
|
|
|
|
super().__init__(options, configuration)
|
|
|
|
s = set(self.channel)
|
|
|
|
# NeuliveSTI only has 3 channels
|
|
s.intersection({0, 1, 2})
|
|
|
|
# force add zero channel (battery channel)
|
|
s.add(0)
|
|
|
|
self.channel = tuple(sorted(s))
|
|
|
|
# sample rate force to 1/s
|
|
self.sample_rate = 1
|
|
|
|
self.voltage_drop_rate = 10 # mv / sec
|
|
|
|
self._voltage = 4300 # mv
|
|
|
|
def get_data(self, time_stamp: float) -> List[RecordingData]:
|
|
if time_stamp < 0:
|
|
raise ValueError('negative time stamp value : ' + str(time_stamp))
|
|
|
|
ret = []
|
|
|
|
time_step = 1 # 1 second
|
|
nxt = -1 # next_time_stamp
|
|
|
|
while nxt < time_stamp:
|
|
nxt, _, count, data = self.data_should_generated(time_stamp, time_step)
|
|
|
|
if data is not None and len(self.channel) != 0:
|
|
|
|
for _ in range(count):
|
|
data.append_data(0, self._voltage)
|
|
|
|
if 1 in self.channel:
|
|
data.append_data(1, int(100 * self.factor_noise()))
|
|
|
|
if 2 in self.channel:
|
|
data.append_data(2, int(100 * self.factor_noise()))
|
|
|
|
self._voltage = int(self._voltage - self.sample_rate)
|
|
|
|
ret.append(data)
|
|
|
|
return ret
|
|
|
|
def new_data(self, time_stamp: int) -> RecordingData:
|
|
return RecordingData(self.device, time_stamp, 0)
|
|
|
|
def new_meta_configuration(self) -> SimpleDeviceConfiguration:
|
|
conf = SimpleDeviceConfiguration(self.configuration)
|
|
|
|
conf.set_parameter(DeviceConfiguration.SAMPLE_RATE, self.sample_rate)
|
|
conf.set_parameter(DeviceConfiguration.CHANNEL, self.channel)
|
|
conf.set_parameter(DeviceConfiguration.AMP_GAIN, 1)
|
|
|
|
return conf
|
|
|
|
|
|
class EliteZMGenerator(Generator):
|
|
MODE = (
|
|
"I-V Curve",
|
|
"Cyclic Voltammetry",
|
|
"Function Generator",
|
|
"R-T Curve",
|
|
"V-T Curve",
|
|
"I-T Curve",
|
|
)
|
|
|
|
VOLTAGE = tuple(map(lambda v: (v + 1) * 0x0001, range(65536)))
|
|
|
|
VOLTAGE_STEP = (30, 60, 90, 120)
|
|
|
|
STEP_TIME = (0.5, 1.0, 1.5, 2.0)
|
|
|
|
SUPPORT = {
|
|
'EliteZM': ('EliteZM', '1.2.30'),
|
|
'I4V4Z4T4': ('EliteZM', '1.2.30'),
|
|
}
|
|
|
|
__slots__ = ('_data_format', 'working_mode', 'voltage_origin', 'voltage_final', 'voltage_step',
|
|
'_voltage', '_voltage_step', '_cycle_number', '_phase', '_message', '_is_eod')
|
|
|
|
def __init__(self, options: 'GeneratorOptions', configuration: DeviceConfiguration):
|
|
if configuration is None:
|
|
raise TypeError('NoneType configuration')
|
|
|
|
super().__init__(options, configuration)
|
|
|
|
self.working_mode = configuration.get('MODE', 0)
|
|
|
|
if isinstance(self.working_mode, str):
|
|
self.working_mode = self.MODE.index(self.working_mode)
|
|
|
|
if self.working_mode == 0: # IV
|
|
self.channel = (0, 1)
|
|
elif self.working_mode == 1: # CV
|
|
self.channel = (0, 1)
|
|
elif self.working_mode == 2: # FG
|
|
self.channel = (1,)
|
|
elif self.working_mode == 3: # ZT
|
|
self.channel = (2,)
|
|
elif self.working_mode == 4: # VT
|
|
self.channel = (1,)
|
|
elif self.working_mode == 5: # IT
|
|
self.channel = (0,)
|
|
else:
|
|
raise ValueError('unsupported mode : ' + str(self.working_mode))
|
|
|
|
# sample rate force to 1/s
|
|
self.sample_rate = 1
|
|
|
|
self.voltage_origin = configuration.get('VOLT_ORIGIN', self.VOLTAGE[0]) # unit: V
|
|
self.voltage_final = configuration.get('VOLT_FINAL', self.VOLTAGE[-1]) # unit: V
|
|
self.voltage_step = configuration.get('VOLT_STEP', self.VOLTAGE_STEP[0]) # unit: mV
|
|
|
|
# internal data
|
|
self._voltage = self.voltage_origin * 1000
|
|
|
|
if self.voltage_origin <= self.voltage_final:
|
|
self._voltage_step = self.voltage_step
|
|
else:
|
|
self._voltage_step = -self.voltage_step
|
|
|
|
self._cycle_number = 0
|
|
self._phase = 0
|
|
self._message: Optional[str] = None
|
|
self._is_eod = False
|
|
|
|
def get_data(self, time_stamp: float) -> List[RecordingData]:
|
|
if time_stamp < 0:
|
|
raise ValueError('negative time stamp value : ' + str(time_stamp))
|
|
|
|
if self._is_eod:
|
|
raise EODInterrupt()
|
|
|
|
ret = []
|
|
|
|
nxt = -1 # next_time_stamp
|
|
|
|
while nxt < time_stamp:
|
|
nxt, data = self._gen_data(time_stamp)
|
|
|
|
if data is not None:
|
|
ret.append(data)
|
|
|
|
if self._is_eod:
|
|
break
|
|
|
|
return ret
|
|
|
|
def _gen_data(self, time_stamp: float) -> Tuple[float, Optional[RecordingData]]:
|
|
nxt, _, count, data = self.data_should_generated(time_stamp, 1)
|
|
|
|
if data is not None:
|
|
for _ in range(count):
|
|
current = 500 * (sin(self._phase * pi / 180) + 1.1)
|
|
|
|
# current
|
|
if 0 in self.channel:
|
|
data.append_data(0 + 3 * self._cycle_number, int(current))
|
|
|
|
# voltage
|
|
if 1 in self.channel:
|
|
data.append_data(1 + 3 * self._cycle_number, int(self._voltage))
|
|
|
|
# impedance
|
|
if 2 in self.channel:
|
|
if current == 0:
|
|
data.append_data(2 + 3 * self._cycle_number, 0)
|
|
else:
|
|
data.append_data(2 + 3 * self._cycle_number, int(self._voltage / current))
|
|
|
|
self._next_voltage()
|
|
|
|
if self._is_eod:
|
|
break
|
|
|
|
self._phase += 1
|
|
|
|
return nxt, data
|
|
|
|
def _next_voltage(self):
|
|
"""change voltage value. """
|
|
|
|
nxt_voltage = self._voltage + self._voltage_step
|
|
|
|
if nxt_voltage >= self.voltage_final * 1000:
|
|
self._voltage = self.voltage_final * 1000
|
|
self._voltage_step = -abs(self._voltage_step)
|
|
|
|
if self.working_mode == 0:
|
|
self._is_eod = True
|
|
|
|
elif self.working_mode == 1 and self.voltage_final < self.voltage_origin:
|
|
self._cycle_number += 1
|
|
self._message = I4V4Z4T4DataDecoder.MESSAGE_CYCLE_COMPLETE
|
|
|
|
elif nxt_voltage <= self.voltage_origin * 1000:
|
|
self._voltage = self.voltage_origin * 1000
|
|
self._voltage_step = abs(self._voltage_step)
|
|
|
|
if self.working_mode == 0:
|
|
self._is_eod = True
|
|
|
|
elif self.working_mode == 1 and self.voltage_origin < self.voltage_final:
|
|
self._cycle_number += 1
|
|
self._message = I4V4Z4T4DataDecoder.MESSAGE_CYCLE_COMPLETE
|
|
|
|
else:
|
|
self._voltage = nxt_voltage
|
|
|
|
def message(self) -> Optional[str]:
|
|
ret = self._message
|
|
self._message = None
|
|
|
|
return ret
|
|
|
|
def new_data(self, time_stamp: int) -> RecordingData:
|
|
return RecordingData(self.device, time_stamp, 0)
|
|
|
|
def new_meta_configuration(self) -> SimpleDeviceConfiguration:
|
|
conf = SimpleDeviceConfiguration(self.configuration)
|
|
|
|
conf.set_parameter(DeviceConfiguration.SAMPLE_RATE, self.sample_rate)
|
|
conf.set_parameter(DeviceConfiguration.CHANNEL, self.channel)
|
|
conf.set_parameter(DeviceConfiguration.AMP_GAIN, 1)
|
|
###
|
|
conf.set_parameter('MODE', self.MODE[self.working_mode])
|
|
conf.set_parameter('VOLT_ORIGIN', self.voltage_origin)
|
|
conf.set_parameter('VOLT_FINAL', self.voltage_final)
|
|
conf.set_parameter('VOLT_STEP', self.voltage_step)
|
|
conf.set_parameter('STEP_TIME', 1.0)
|
|
|
|
return conf
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class GeneratorOptions(CliOptions):
|
|
def __init__(self):
|
|
# general options
|
|
self.generator_type: str = 'sin'
|
|
self.generator_cls: Type[Generator] = SinWaveGenerator
|
|
self.configuration: Optional[DeviceConfiguration] = None
|
|
|
|
self.device: int = 0
|
|
self.channel: List[int] = [0]
|
|
self.sample_rate = 100
|
|
self.amplitude = 500
|
|
self.noise = 0.0
|
|
|
|
# library path
|
|
self.repository = DeviceLibraryRepository()
|
|
|
|
@cli_options('-g', '--generator', value='GENERATOR')
|
|
def _generator_type(self, opt: str, value: str):
|
|
"""generator, could be:
|
|
zero : all zero
|
|
sin : sine save (default)
|
|
ramp : ramp
|
|
(Elite, Neulive serial)
|
|
TC4VAF2 Elite_Legacy
|
|
TDC4VAF2 Elite Neulive
|
|
(NeuliveSTI serial)
|
|
NeuliveSTI NeuliveSTI1.0 TDBC4VCTH
|
|
(EliteZM serial)
|
|
EliteZM I4V4Z4T4
|
|
(mode :IV, :FG, :CV, :ZT, :VT, :IT)
|
|
"""
|
|
|
|
self.set_generator(value)
|
|
|
|
def set_generator(self, generator_type: str):
|
|
self.generator_type = generator_type
|
|
|
|
if generator_type == 'zero':
|
|
self.generator_cls = ZeroGenerator
|
|
self.configuration = None
|
|
|
|
elif generator_type == 'sin':
|
|
self.generator_cls = SinWaveGenerator
|
|
self.configuration = None
|
|
|
|
elif generator_type == 'ramp':
|
|
self.generator_cls = RampGenerator
|
|
self.configuration = None
|
|
|
|
else:
|
|
try:
|
|
lib, ver = DeviceGenerator.SUPPORT[generator_type]
|
|
self.configuration = self._new_device_configuration(lib, ver)
|
|
self.generator_cls = DeviceGenerator
|
|
return
|
|
except KeyError:
|
|
pass
|
|
|
|
try:
|
|
lib, ver = NeuliveSTIGenerator.SUPPORT[generator_type]
|
|
self.configuration = self._new_device_configuration(lib, ver)
|
|
self.generator_cls = NeuliveSTIGenerator
|
|
return
|
|
except KeyError:
|
|
pass
|
|
|
|
try:
|
|
lib, ver = EliteZMGenerator.SUPPORT[generator_type]
|
|
self.configuration = self._new_device_configuration(lib, ver)
|
|
self.generator_cls = EliteZMGenerator
|
|
|
|
# init ZM options
|
|
self.configuration.set_parameter('VOLT_ORIGIN', EliteZMGenerator.VOLTAGE[0])
|
|
self.configuration.set_parameter('VOLT_FINAL', EliteZMGenerator.VOLTAGE[-1])
|
|
|
|
return
|
|
except KeyError:
|
|
pass
|
|
|
|
raise RuntimeError('unknown generator type : ' + generator_type)
|
|
|
|
@cli_options('-D', '--device', value='DEVICE')
|
|
def _device(self, opt: str, value: str):
|
|
"""device ID"""
|
|
self.device = int(value)
|
|
|
|
@cli_options('-c', '--channel', value='CH[,CH]')
|
|
def _channel(self, opt: str, value: str):
|
|
"""channel list, default [0]"""
|
|
self.channel = list(map(int, value.split(',')))
|
|
|
|
@cli_options('-s', '--sample-rate', value='RATE')
|
|
def _sample_rate(self, opt: str, value: str):
|
|
"""sampling rate in unit 1/s.
|
|
could be [N]K, default 100.
|
|
|
|
"""
|
|
if value.endswith('K'):
|
|
self.sample_rate = int(value[:-1]) * 1000
|
|
else:
|
|
self.sample_rate = int(value)
|
|
|
|
@cli_options('-a', '--amplitude', value='VALUE')
|
|
def _amplitude(self, opt: str, value: str):
|
|
"""signal amplitude, default 500"""
|
|
self.amplitude = int(value)
|
|
|
|
@cli_options('-n', '--noise', value='VALUE')
|
|
def _noise(self, opt: str, value: str):
|
|
"""noise factor, default 0"""
|
|
self.noise = float(value)
|
|
|
|
@cli_options('-L', '--library', value='PATH')
|
|
def _library_path(self, opt: str, value: str):
|
|
"""library search path"""
|
|
self.repository.add_library_path(value)
|
|
|
|
def get_generator(self) -> Generator:
|
|
return self.generator_cls(self, self.configuration)
|
|
|
|
def _new_device_configuration(self, library_name: str, library_version: str) -> DeviceConfiguration:
|
|
library = self.repository.get_library(library_name, library_version)
|
|
if library is None:
|
|
raise RuntimeError('library ' + library_name + ' ' + library_version + ' not found')
|
|
|
|
parameter = DeviceParameter(library)
|
|
|
|
return DeviceParameterConfiguration(library, parameter)
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class Main(CliMain, GeneratorOptions):
|
|
def __init__(self):
|
|
super().__init__()
|
|
GeneratorOptions.__init__(self)
|
|
|
|
# internal property
|
|
self._generator: Generator = None
|
|
|
|
# output
|
|
self.output_type = 'raw'
|
|
self.output_file = '-'
|
|
|
|
# time
|
|
self.duration: Optional[float] = 1.0
|
|
'''time duration in unit second'''
|
|
|
|
self.list_parameter = False
|
|
|
|
# recording file writer options
|
|
self.writer_split_time: Optional[int] = None
|
|
self.writer_split_size: Optional[int] = None
|
|
self.writer_print_data = None
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
"""print help document"""
|
|
self.print_help()
|
|
|
|
def set_generator(self, generator_type: str):
|
|
super().set_generator(generator_type)
|
|
|
|
c = self.configuration
|
|
|
|
if c is None:
|
|
return
|
|
|
|
m: Dict[str, str] = {}
|
|
h: List[CliHandleOption] = []
|
|
|
|
for k in c.keys():
|
|
if k.startswith('_'):
|
|
continue
|
|
|
|
if k in (DeviceConfiguration.CHANNEL, DeviceConfiguration.SAMPLE_RATE):
|
|
# do not include channel and sample_rate options
|
|
continue
|
|
|
|
o = '--' + k.lower().replace('_', '-')
|
|
m[o] = k
|
|
|
|
h.append(CliHandleOption(o, help_doc=k))
|
|
|
|
if len(h) > 0:
|
|
def _callback(flag: str, value: Optional[str]):
|
|
c.set_parameter(m[flag], value)
|
|
|
|
self.extend_options_callback(_callback, *h,
|
|
help_section='Configuration for ' + c.library)
|
|
|
|
@cli_options('-T', '--type', value='TYPE')
|
|
def _output_type(self, opt: str, value: str):
|
|
"""output data format.
|
|
raw : raw data (default)
|
|
bprf : BioPro Recording File (output to file only)
|
|
|
|
"""
|
|
self.output_type = value
|
|
|
|
@cli_options('-t', '--duration', value='VALUE')
|
|
def _duration(self, opt: str, value: str):
|
|
"""total time duration in unit second.
|
|
could be [N]h[N]m[N][.NNN]. also can use 'inf' to run forever. default 1.
|
|
"""
|
|
if value == 'inf':
|
|
self.duration = None
|
|
else:
|
|
self.duration = self._parse_time(value)
|
|
|
|
@cli_flags('-p')
|
|
def _list_parameter(self, flag: str):
|
|
"""just list parameter, do not run anything"""
|
|
self.list_parameter = True
|
|
|
|
@cli_arguments('?', value='FILE')
|
|
def _file(self, pos: int, value: str):
|
|
"""output file, '-' output to stdout (default)"""
|
|
self.output_file = value
|
|
|
|
@cli_options('--split-time', value='TIME')
|
|
@cli_help_section('OPTIONS (--type=bprf)')
|
|
def _split_time(self, opt: str, value: str):
|
|
"""file split for every second"""
|
|
if self.output_type != 'bprf':
|
|
print('WARN', opt + ' only work for --type=bprf')
|
|
|
|
self.writer_split_time = int(value)
|
|
|
|
@cli_options('--split-size', value='SIZE')
|
|
@cli_help_section('OPTIONS (--type=bprf)')
|
|
def _split_size(self, opt: str, value: str):
|
|
"""file split for every size, could be [N]K, [N]M, [N]G."""
|
|
if self.output_type != 'bprf':
|
|
print('WARN', opt + ' only work for --type=bprf')
|
|
|
|
self.writer_split_size = self._parse_size(value)
|
|
|
|
@cli_options('--bprf-verbose', value='NEWLINE', optional=True)
|
|
@cli_help_section('OPTIONS (--type=bprf)')
|
|
def _bprf_verbose(self, opt: str, value: str):
|
|
"""print what it write to stdout, default NEWLINE = \\n"""
|
|
if self.output_type != 'bprf':
|
|
print('WARN', opt + ' only work for --type=bprf')
|
|
|
|
if value is None:
|
|
self.writer_print_data = '\n'
|
|
else:
|
|
self.writer_print_data = value
|
|
|
|
@staticmethod
|
|
def _parse_size(expr: str) -> int:
|
|
factor = 1
|
|
|
|
if expr.endswith('K'):
|
|
factor = 1000
|
|
expr = expr[:-1]
|
|
|
|
elif expr.endswith('M'):
|
|
factor = 1000 * 1000
|
|
expr = expr[:-1]
|
|
|
|
elif expr.endswith('G'):
|
|
factor = 1000 * 1000 * 1000
|
|
expr = expr[:-1]
|
|
|
|
return int(expr) * factor
|
|
|
|
@staticmethod
|
|
def _parse_time(expr: str) -> float:
|
|
"""parse time expression
|
|
|
|
:param expr: expression in [N]h[N]m[N][.NNN]
|
|
:return: second
|
|
"""
|
|
t_h, expr = part_prefix(expr, 'h', missing=0)
|
|
t_h = int(t_h)
|
|
|
|
if len(expr) > 0:
|
|
t_m, expr = part_prefix(expr, 'm', missing=0)
|
|
t_m = int(t_m)
|
|
else:
|
|
t_m = 0
|
|
|
|
if len(expr) > 0:
|
|
t_s, t_u = part_suffix(expr, '.')
|
|
t_s = int(t_s)
|
|
|
|
if t_u is not None:
|
|
t_u = t_u[:3]
|
|
if len(t_u) < 3:
|
|
t_u += '0'
|
|
t_u = int(t_u)
|
|
else:
|
|
t_u = 0
|
|
|
|
else:
|
|
t_s = t_u = 0
|
|
|
|
return t_u * 1e-3 + t_s + t_m * 60 + t_h * 60 * 60
|
|
|
|
def run(self):
|
|
self._generator = self.get_generator()
|
|
|
|
if self.list_parameter:
|
|
configuration = self._generator.configuration
|
|
|
|
for para in configuration.keys():
|
|
print(para)
|
|
|
|
else:
|
|
|
|
if self.output_type == 'raw':
|
|
if self.output_file == '-':
|
|
self._run(sys.stdout)
|
|
else:
|
|
with open(self.output_file, 'w') as output:
|
|
self._run(output)
|
|
|
|
elif self.output_type == 'bprf':
|
|
if self.output_file == '-':
|
|
raise RuntimeError('bprf cannot output to stdout')
|
|
|
|
meta = RecordingMetaFile(self.output_file)
|
|
|
|
meta.configuration = self._generator.new_meta_configuration()
|
|
|
|
with RecordingFileWriter(meta) as output:
|
|
output.splitting_threshold_size = self.writer_split_size
|
|
output.splitting_threshold_time = self.writer_split_time
|
|
|
|
self._run(output)
|
|
|
|
else:
|
|
raise RuntimeError('unknown output type : ' + self.output_type)
|
|
|
|
def _run(self, output: Union[IO, RecordingFileWriter]):
|
|
if self.duration is None:
|
|
raise RuntimeError()
|
|
|
|
is_writer = isinstance(output, RecordingFileWriter)
|
|
|
|
# init first time
|
|
self._generator.get_data(0.0)
|
|
|
|
try:
|
|
|
|
if self.writer_print_data or not is_writer:
|
|
for data in self._generator.get_data(self.duration):
|
|
if len(data) > 0:
|
|
if is_writer:
|
|
output.write(data)
|
|
|
|
for time_stamp, channel_id, value in data.entry_iter():
|
|
print('%.4f' % time_stamp, channel_id, value, end=self.writer_print_data)
|
|
else:
|
|
for data in self._generator.get_data(self.duration):
|
|
if len(data) > 0:
|
|
output.write(data)
|
|
|
|
except EODInterrupt:
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
Main().main()
|