peterlu14
391 lines
12 KiB
Python
391 lines
12 KiB
Python
from pathlib import Path
|
|
from typing import List, Optional
|
|
|
|
from biopro.recording.file import RecordingMetaFile
|
|
from biopro.recording.main import MainMeta as RecordingMetaMain
|
|
from biopro.util.address import address_str, bytes_address
|
|
from biopro.util.cli import *
|
|
from biopro.util.console import hex_line, hex_value, hex_table
|
|
from biopro.util.text import print_block, Table
|
|
from .data import *
|
|
from .encoder import DataMessageDecoder
|
|
from .request import DataMessageRequest, TimeDurationDataMessageResponse, TimeRangeDataMessageResponse
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class Main(CliMain):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
"""print help document"""
|
|
self.print_help()
|
|
|
|
@cli_command('reader')
|
|
def _reader_command(self, act: str, argv: List[str]):
|
|
"""Recording File Reader"""
|
|
return _MainReader
|
|
|
|
@cli_command('encode')
|
|
def _encode_command(self, act: str, argv: List[str]):
|
|
"""encode message"""
|
|
return _MainEncode
|
|
|
|
@cli_command('decode')
|
|
def _decode_command(self, act: str, argv: List[str]):
|
|
"""decode message"""
|
|
return MainDecode
|
|
|
|
def run(self):
|
|
self.print_help()
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class _MainReader(CliSubCommandMain):
|
|
def __init__(self, command: str):
|
|
super().__init__(command)
|
|
|
|
self.meta_file_path: Optional[str] = None
|
|
self.print_info = False
|
|
|
|
@cli_flags('-i')
|
|
def _print_info(self, flag: str):
|
|
""""""
|
|
self.print_info = True
|
|
|
|
@cli_arguments(0, 'FILE')
|
|
def _meta_file(self, pos: int, value: str):
|
|
"""meta file path"""
|
|
self.meta_file_path = value
|
|
|
|
def run(self):
|
|
if self.meta_file_path is None:
|
|
raise ValueError('lost FILE')
|
|
|
|
reader = RecordingFileReader(self.meta_file_path)
|
|
|
|
if self.print_info:
|
|
t = Table(column=2)
|
|
|
|
t.append('file_path', reader.file_path.decode())
|
|
t.append('version_number', reader.version_number)
|
|
t.append('create_time', reader.create_time)
|
|
t.append('file_uuid', hex_line(reader.file_uuid))
|
|
t.append('data_format', hex_line(reader.data_format))
|
|
t.append('device_name', reader.device_name.decode())
|
|
t.append('device_address', address_str(bytes_address(reader.device_address)))
|
|
t.append('device_serial', hex_line(reader.device_serial))
|
|
t.append('parameter_count', reader.parameter_count)
|
|
for i in range(reader.parameter_count):
|
|
name, value = reader.parameter(i)
|
|
t.append(name, value)
|
|
|
|
t.append('recording_file_size', reader.recording_file_size)
|
|
for i in range(reader.recording_file_size):
|
|
t.append(i, reader.recording_file_name(i).decode())
|
|
|
|
t.set_format(1, align_right=False)
|
|
t.print()
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class MainDecode(CliSubCommandMain):
|
|
def __init__(self, command: str):
|
|
super().__init__(command)
|
|
|
|
self.flag_raw_file = False
|
|
self.file_path = None
|
|
|
|
self.flag_skip_null = False
|
|
|
|
self.raw_data = []
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
"""print help document"""
|
|
self.print_help()
|
|
|
|
@cli_flags('--file')
|
|
def _raw_file(self, opt: str):
|
|
"""argument as raw data file"""
|
|
self.flag_raw_file = True
|
|
|
|
@cli_flags('--skip-null')
|
|
def _skip_null(self, opt: str):
|
|
"""do not print NULL data"""
|
|
self.flag_skip_null = True
|
|
|
|
@cli_arguments(0, value='ARGS')
|
|
def _first_arg(self, pos: int, value: str):
|
|
"""raw data or meta file"""
|
|
if self.flag_raw_file:
|
|
self.file_path = value
|
|
|
|
else:
|
|
if ' ' in value:
|
|
self.raw_data.extend(value.split(' '))
|
|
else:
|
|
self.raw_data.append(value)
|
|
|
|
@cli_arguments('*', value='')
|
|
def _left_arg(self, pos: int, value: str):
|
|
if self.flag_raw_file:
|
|
raise RuntimeError()
|
|
else:
|
|
if ' ' in value:
|
|
self.raw_data.extend(value.split(' '))
|
|
else:
|
|
self.raw_data.append(value)
|
|
|
|
def run(self):
|
|
if self.flag_raw_file:
|
|
message = self._run_raw_data()
|
|
else:
|
|
message = self._decode_raw_data()
|
|
|
|
decoder = DataMessageDecoder.decode(message)
|
|
self.print_decoder_result(decoder, skip_null=self.flag_skip_null)
|
|
|
|
def _run_raw_data(self) -> bytes:
|
|
if self.file_path is None:
|
|
raise RuntimeError('lost PATH')
|
|
|
|
ret = []
|
|
|
|
with open(self.file_path) as f:
|
|
for line in f:
|
|
|
|
if line.startswith('#'):
|
|
continue
|
|
|
|
# example line
|
|
# 0x0030: 01 F1 01 F3 01 F2 01 E6 01 E2 01 C9 01 C1 01 9D ................
|
|
i = line.index(' ')
|
|
j = line.find(' ', i + 2)
|
|
|
|
if j > 0:
|
|
data = line[i + 2:j].strip()
|
|
else:
|
|
data = line[i + 2:].strip()
|
|
|
|
ret.append(bytes(map(hex_value, data.split(' '))))
|
|
|
|
return b''.join(ret)
|
|
|
|
def _decode_raw_data(self) -> bytes:
|
|
return bytes(map(hex_value, self.raw_data))
|
|
|
|
@staticmethod
|
|
def print_decoder_result(decoder: DataMessageDecoder, skip_null=False):
|
|
print('message_size', decoder.message_size)
|
|
|
|
for entry in decoder.data_entry:
|
|
t = Table(column=2)
|
|
|
|
t.append('device', entry.device)
|
|
t.append('channel', entry.channel)
|
|
t.append('time stamp [ms]', entry.time_stamp)
|
|
t.append('present mode', entry.present_mode.name)
|
|
t.append('sample rate [1/s]', entry.sample_rate)
|
|
t.append('time delta [ms]', '%.2f' % (1000 / entry.sample_rate))
|
|
t.append('entry size', len(entry.data_entry))
|
|
t.append('%5s %8s' % ('index', 'time'), 'value')
|
|
|
|
for i, value in enumerate(entry.data_entry):
|
|
if skip_null and value is None:
|
|
continue
|
|
|
|
t.append('%5d %8.2f' % (i, entry.time_stamp + i * 1000 / entry.sample_rate),
|
|
str(value) if value is not None else 'NULL')
|
|
|
|
t.set_format(0, align_right=False)
|
|
|
|
for line in t.lines():
|
|
print(line)
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class _MainEncode(CliSubCommandMain):
|
|
def __init__(self, command: str):
|
|
super().__init__(command)
|
|
|
|
self.storage_root = '.'
|
|
|
|
self.output_type = 'binary'
|
|
self.output_path = None
|
|
|
|
self.path: str = None
|
|
self.request_command = []
|
|
|
|
# options for -t table
|
|
self.output_table_column = 16
|
|
|
|
# options for -t decoder
|
|
self.output_decoder_skip_null = False
|
|
|
|
self.use_c_extension = False
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
"""print help document"""
|
|
self.print_help()
|
|
|
|
@cli_flags('--help-request', force_return=True)
|
|
def _help_request(self, opt: str):
|
|
"""print request command format"""
|
|
print_block(DataMessageRequest.__doc__)
|
|
|
|
@cli_options('-C', '--root', value='PATH')
|
|
def _storage_root(self, opt: str, value: str):
|
|
"""change the storage root"""
|
|
self.storage_root = value
|
|
|
|
@cli_options('-t', '--output-type', value='VALUE')
|
|
def _output_type(self, opt: str, value: str):
|
|
"""output type. could be
|
|
binary : binary format (default). this require '-o' options.
|
|
line : hex value
|
|
table : hex table
|
|
decoder : pass message to decoder.
|
|
"""
|
|
|
|
if value not in ('binary', 'line', 'table', 'decoder'):
|
|
raise RuntimeError('illegal output type : ' + value)
|
|
|
|
self.output_type = value
|
|
|
|
@cli_options('-o', '--output', value='FILE')
|
|
def _output_path(self, opt: str, value: str):
|
|
"""output file path"""
|
|
self.output_path = value
|
|
|
|
@cli_options('-c', '--column', value='VALUE')
|
|
@cli_help_section('OPTIONS(--output=table)')
|
|
def _output_table_column(self, opt: str, value: str):
|
|
"""table column (default = 16)"""
|
|
self.output_table_column = int(value)
|
|
|
|
@cli_flags('--skip-null')
|
|
@cli_help_section('OPTIONS(--output=decoder)')
|
|
def _output_decoder_skip_null(self, opt: str):
|
|
"""skip null in the output of the decoder"""
|
|
self.output_decoder_skip_null = True
|
|
|
|
@cli_flags('--use-extension')
|
|
def _use_c_extension(self, opt: str):
|
|
"""use c extension to improve performance."""
|
|
self.use_c_extension = True
|
|
|
|
@cli_arguments(0, value='FILE')
|
|
def _path(self, pos: int, value: str):
|
|
"""source meta file"""
|
|
self.path = value
|
|
|
|
@cli_arguments('*', value='REQUEST')
|
|
def _request(self, pos: int, value: str):
|
|
"""request command"""
|
|
self.request_command.append(value)
|
|
|
|
def run(self):
|
|
if self.path is None:
|
|
raise RuntimeError('lost FILE')
|
|
|
|
if self.output_type == 'binary' and self.output_path is None:
|
|
raise RuntimeError('lost output path which is required for binary output format')
|
|
|
|
meta = RecordingMetaMain.find_meta_by_name(Path(self.storage_root), self.path)
|
|
|
|
if meta is None:
|
|
raise RuntimeError('not a meta file : ' + self.path)
|
|
|
|
request = DataMessageRequest()
|
|
|
|
for cmd in self.request_command:
|
|
response = request.update_command(cmd)
|
|
|
|
if response is not None:
|
|
|
|
if isinstance(response, TimeDurationDataMessageResponse):
|
|
raise RuntimeError('file cannot accept time duration response')
|
|
|
|
elif isinstance(response, TimeRangeDataMessageResponse):
|
|
self._run_response(meta, request, response)
|
|
|
|
else:
|
|
raise RuntimeError()
|
|
|
|
def _run_response(self,
|
|
meta: RecordingMetaFile,
|
|
request: DataMessageRequest,
|
|
response: TimeRangeDataMessageResponse):
|
|
encoder = request.new_encoder(response)
|
|
|
|
if encoder is None:
|
|
print('None (no listen device/channel)')
|
|
|
|
elif isinstance(encoder, DataMessageEncoder):
|
|
if self.use_c_extension:
|
|
message = self._run_c_extension(meta, encoder, response)
|
|
else:
|
|
print('No C extension')
|
|
|
|
if message is None:
|
|
print('None')
|
|
|
|
elif len(message) == 0:
|
|
print('empty')
|
|
else:
|
|
self._run_output_message(message)
|
|
|
|
else:
|
|
print('None (too many device)')
|
|
|
|
def _run_c_extension(self,
|
|
meta: RecordingMetaFile,
|
|
encoder: DataMessageEncoder,
|
|
response: TimeRangeDataMessageResponse) -> bytes:
|
|
reader = RecordingFileReader(meta)
|
|
|
|
encoder.append_file(reader, response.time_start, response.time_stop)
|
|
|
|
return encoder.message()
|
|
|
|
def _run_output_message(self, message: bytes):
|
|
if self.output_type == 'binary':
|
|
with open(self.output_path, 'wb') as f:
|
|
f.write(message)
|
|
|
|
elif self.output_type == 'line':
|
|
content = hex_line(message)
|
|
|
|
if self.output_path is None:
|
|
print(content)
|
|
|
|
else:
|
|
with open(self.output_path, 'w') as f:
|
|
print(content, file=f)
|
|
|
|
elif self.output_type == 'table':
|
|
if self.output_path is not None:
|
|
f = open(self.output_path, 'w')
|
|
|
|
else:
|
|
f = None
|
|
|
|
try:
|
|
hex_table(message, column=self.output_table_column, file=f)
|
|
|
|
finally:
|
|
if f is not None:
|
|
f.close()
|
|
|
|
elif self.output_type == 'decoder':
|
|
decoder = DataMessageDecoder.decode(message)
|
|
|
|
MainDecode.print_decoder_result(decoder, skip_null=self.output_decoder_skip_null)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
Main().main()
|