peterlu14
1036 lines
33 KiB
Python
1036 lines
33 KiB
Python
"""File Export/Import
|
|
|
|
**export/import file from command-line**
|
|
|
|
::
|
|
|
|
env PYTHONPATH=RaspBerryPi3/python python -m biopro.file.manager (export|import) -T TYPE -o OUTPUT SOURCE
|
|
|
|
**register export/import function**
|
|
|
|
|
|
::
|
|
|
|
from biopro.file.export import *
|
|
|
|
@register_export('txt', 'raw text', '.txt',
|
|
parameter('resample_rate', 'Re-Sampling Rate, None for using origin.', 'int?'),
|
|
parameter('resample_method', 'Re-Sampling Method', 'int'),
|
|
parameter('include_header', 'include file information', 'bool'))
|
|
def export_txt(meta: RecordingMetaFile,
|
|
export_path: Path,
|
|
resample_rate: int = None,
|
|
resample_method: int = 0,
|
|
include_comment=False):
|
|
pass
|
|
|
|
@register_import('txt', 'raw text', '.txt')
|
|
def import_txt(import_file: Path,
|
|
filepath: Path) -> RecordingMetaFile:
|
|
pass
|
|
"""
|
|
import abc
|
|
import os
|
|
from pathlib import Path
|
|
from subprocess import Popen
|
|
from time import time
|
|
from typing import List, Callable, Any, Dict, Type, Tuple, Union, Optional
|
|
|
|
# noinspection PyUnresolvedReferences
|
|
import biopro.ext.export as export
|
|
|
|
from biopro.recording.file import RecordingMetaFile
|
|
from biopro.util.cli import *
|
|
from biopro.util.json import JsonSerialize
|
|
from biopro.util.sys import current_command_header
|
|
from biopro.util.text import part_prefix, part_suffix, Table
|
|
from .api import FileOptions
|
|
|
|
EXPORT_TYPE: List['ExternalExportFunction'] = []
|
|
IMPORT_TYPE: List['ExternalImportFunction'] = []
|
|
|
|
|
|
def _parameter_type_expression(expr: str) -> Callable[[str], Any]:
|
|
def _type(_c: str) -> Callable[[str], Any]:
|
|
_, _t = part_prefix(_c, ':')
|
|
return _parameter_type_expression(_t)
|
|
|
|
if expr.endswith('?'):
|
|
return _type(expr[:-1])
|
|
elif expr == 'bool':
|
|
return bool
|
|
elif expr == 'int':
|
|
return int
|
|
elif expr == 'float':
|
|
return float
|
|
elif expr == 'str':
|
|
return str
|
|
elif expr.endswith(',...'):
|
|
func = _type(expr[:-4])
|
|
return lambda para: list(map(func, para.split(',')))
|
|
elif (expr.startswith('list[') or expr.startswith('List[')) and expr.endswith(']'):
|
|
func = _type(expr[5:-1])
|
|
return lambda para: list(map(func, para.split(',')))
|
|
else:
|
|
raise RuntimeError('unsupported type : ' + expr)
|
|
|
|
|
|
class ExternalParameter(JsonSerialize):
|
|
"""the parameter information of the export/import function.
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ExternalParameter = {
|
|
"name" :str # parameter name
|
|
"description" :str # parameter description
|
|
"type" :parameter_type
|
|
}
|
|
|
|
parameter_type :str
|
|
= primitive
|
|
| primitive "?" # optional type
|
|
| name:str ":" primitive # named type
|
|
| primitive ",..." # list type
|
|
| "list[" primitive "]" # list type
|
|
| "List[" primitive "]" # list type
|
|
|
|
primitive
|
|
= "bool" | "int" | "float" | "str"
|
|
"""
|
|
|
|
__slots__ = ('name', 'text', 'ptype')
|
|
|
|
def __init__(self, name: str, desp: str, ptype: str):
|
|
"""
|
|
|
|
:param name: parameter name
|
|
:param desp: parameter description
|
|
:param ptype: parameter type
|
|
"""
|
|
self.name = name
|
|
self.text = desp
|
|
self.ptype = ptype
|
|
|
|
@property
|
|
def option_name(self) -> str:
|
|
"""the url-style parameter name. replace the '_' with '-'."""
|
|
return self.name.replace('_', '-')
|
|
|
|
def as_json(self):
|
|
return {
|
|
'name': self.option_name,
|
|
'description': self.text,
|
|
'type': self.ptype,
|
|
}
|
|
|
|
|
|
class ExternalFunction(JsonSerialize):
|
|
"""the information of the export/import function.
|
|
|
|
**json format**
|
|
|
|
::
|
|
|
|
ExternalFunction = {
|
|
"name" :str # export/import function name
|
|
"description" :str # file description
|
|
"ext" :str # file extension name
|
|
"parameter" : [ExternalParameter]
|
|
}
|
|
|
|
"""
|
|
|
|
__slots__ = ('func', 'name', 'desp', 'ext', 'parameter')
|
|
|
|
def __init__(self, func: callable, name: str, desp: str, ext: str, *p: ExternalParameter):
|
|
"""
|
|
|
|
:param func: function instance
|
|
:param name: function name
|
|
:param desp: export/import file description
|
|
:param ext: export/import file extension name
|
|
:param p: parameter information
|
|
"""
|
|
|
|
self.func = func
|
|
self.name = name
|
|
self.desp = desp
|
|
self.ext = ext
|
|
self.parameter = p
|
|
|
|
for _p in p:
|
|
# check expression
|
|
try:
|
|
_parameter_type_expression(_p.ptype)
|
|
except RuntimeError as e:
|
|
raise RuntimeError('for parameter', _p.name) from e
|
|
|
|
@staticmethod
|
|
def _parameters(func) -> List[ExternalParameter]:
|
|
"""get the parameter information from function *func*.
|
|
|
|
:param func:
|
|
:return: list of parameter information.
|
|
"""
|
|
code = func.__code__
|
|
|
|
cnt: int = code.co_argcount
|
|
typ: Dict[str, str] = {}
|
|
ann: Dict[str, Type[Any]] = func.__annotations__
|
|
dfv: List[Any] = func.__defaults__ if func.__defaults__ is not None else []
|
|
|
|
desp: Dict[str, str] = {}
|
|
name: List[str] = code.co_varnames[:cnt]
|
|
ret = []
|
|
|
|
# __doc__ parsing, for desp
|
|
if func.__doc__ is not None:
|
|
for line in func.__doc__.split('\n'):
|
|
line: str = line.strip()
|
|
|
|
if line.startswith(':param '):
|
|
p_name, p_desp = part_suffix(line[7:], ':')
|
|
|
|
if p_desp is not None:
|
|
# type. we need to the type information from __doc__
|
|
# to get more information than type annotation such as list transform format.
|
|
# Furthermore, some python version doesn't introduce PEP 484 (type hints), we
|
|
# cannot get any type information from type annotation.
|
|
# For uniform code style, we encode type information in __doc__.
|
|
if p_desp.startswith('type('):
|
|
p_type, p_desp = part_prefix(p_desp[5:], ')')
|
|
typ[p_name] = p_type
|
|
p_desp = p_desp.strip()
|
|
|
|
# description
|
|
p_desp, _ = part_suffix(p_desp, '.')
|
|
|
|
desp[p_name] = p_desp
|
|
|
|
#
|
|
for i, p_name in enumerate(name):
|
|
if p_name in typ:
|
|
p_type = typ[p_name]
|
|
elif p_name in ann:
|
|
p_type = ann[p_name].__name__
|
|
else:
|
|
j = len(dfv) - cnt + i
|
|
|
|
if j < 0:
|
|
p_type = 'str'
|
|
else:
|
|
p_type = type(dfv[j]).__name__
|
|
|
|
p_desp = desp.get(p_name, '')
|
|
|
|
p = ExternalParameter(p_name, p_desp, p_type)
|
|
|
|
ret.append(p)
|
|
|
|
return ret
|
|
|
|
def _map_options(self, options: Dict[str, str]) -> Dict[str, Any]:
|
|
ret = options
|
|
|
|
for p in self.parameter:
|
|
if p.name in ret:
|
|
ret[p.name] = _parameter_type_expression(p.ptype)(ret[p.name])
|
|
|
|
return ret
|
|
|
|
def as_json(self):
|
|
return {
|
|
'name': self.name,
|
|
'description': self.desp,
|
|
'ext': self.ext,
|
|
'parameter': [p.as_json() for p in self.parameter],
|
|
}
|
|
|
|
|
|
class ExternalExportFunction(ExternalFunction):
|
|
"""the information of the export function """
|
|
|
|
__slots__ = ()
|
|
|
|
def __init__(self, func: callable, name: str, desp: str, ext: str, *p: ExternalParameter):
|
|
"""
|
|
|
|
:param func: function instance
|
|
:param name: function name
|
|
:param desp: export/import file description
|
|
:param ext: export/import file extension name
|
|
:param p: parameter information. if omit, program will auto scan the parameter of the *func*
|
|
"""
|
|
if len(p) == 0:
|
|
up = self._export_parameters(func)
|
|
else:
|
|
up = p
|
|
|
|
super().__init__(func, name, desp, ext, *up)
|
|
|
|
def export_file(self, meta: RecordingMetaFile, output_path: Path, **options: str):
|
|
"""export a meta file *meta* to *output_path*.
|
|
|
|
:param meta: recording meta file
|
|
:param output_path: output file path
|
|
:param options: export options
|
|
"""
|
|
self.func(meta, output_path, **self._map_options(options))
|
|
|
|
@staticmethod
|
|
def _export_parameters(func: callable) -> Tuple[ExternalParameter, ...]:
|
|
ret = ExternalFunction._parameters(func)
|
|
|
|
# remove leading required parameter: meta and output_path
|
|
return tuple(ret[2:])
|
|
|
|
|
|
class ExternalImportFunction(ExternalFunction):
|
|
"""the information of the import function"""
|
|
|
|
__slots__ = ()
|
|
|
|
def __init__(self, func: callable, name: str, desp: str, ext: str, *p: ExternalParameter):
|
|
"""
|
|
|
|
:param func: function instance
|
|
:param name: function name
|
|
:param desp: export/import file description
|
|
:param ext: export/import file extension name
|
|
:param p: parameter information. if omit, program will auto scan the parameter of the *func*
|
|
"""
|
|
if len(p) == 0:
|
|
up = self._import_parameters(func)
|
|
else:
|
|
up = p
|
|
|
|
super().__init__(func, name, desp, ext, *up)
|
|
|
|
def import_file(self, import_path: Path, target_path: Path, **options: str) -> RecordingMetaFile:
|
|
"""import a file from *import_path* to form a meta file at *target_path*.
|
|
|
|
:param import_path: source file path
|
|
:param target_path: meta file save path
|
|
:param options: import options
|
|
:return: recording meta file
|
|
"""
|
|
return self.func(import_path, target_path, **self._map_options(options))
|
|
|
|
@staticmethod
|
|
def _import_parameters(func: callable) -> Tuple[ExternalParameter, ...]:
|
|
ret = ExternalFunction._parameters(func)
|
|
|
|
# remove leading required parameter: import_path and target_path
|
|
return tuple(ret[2:])
|
|
|
|
|
|
def register_export(name: str, desp: str, ext: str, *p: ExternalParameter):
|
|
"""register a file export function. the prototype of the function should be compatible with
|
|
:func:`impl.file.ExternalExportFunction.export_file`
|
|
|
|
The export function options are as same as the function options in general. You can use :func:`parameter`
|
|
the export the options manually. Or left it empty, program will auto generate the export options
|
|
be reflection.
|
|
|
|
If using options auto generation feature. The *desp* will try to parsing the content of the function __doc__.
|
|
So as the lost information of the type of the options. It could get the information from
|
|
the type annotation (which introduce by PEP484) or in the function __doc__ with following format
|
|
``:param OPTION: type(TYPE) DESP``.
|
|
|
|
:param name: export type
|
|
:param desp: export file description
|
|
:param ext: export file extension
|
|
:param p: custom parameter, empty will auto generated by reflection
|
|
"""
|
|
|
|
def _f(f):
|
|
try:
|
|
EXPORT_TYPE.append(ExternalExportFunction(f, name, desp, ext, *p))
|
|
except RuntimeError as e:
|
|
raise RuntimeError('for function : ' + f.__name__) from e
|
|
return f
|
|
|
|
return _f
|
|
|
|
|
|
def register_import(name: str, desp: str, ext: str, *p: ExternalParameter):
|
|
"""register a file import function. the prototype of the function should be compatible with
|
|
:func:`impl.file.ExternalImportFunction.import_file`
|
|
|
|
see :func:`register_export` for more detail.
|
|
|
|
:param name: import type
|
|
:param desp: import file description
|
|
:param ext: import file extension
|
|
:param p: custom parameter, empty will auto generated by reflection
|
|
"""
|
|
|
|
def _f(f):
|
|
try:
|
|
IMPORT_TYPE.append(ExternalImportFunction(f, name, desp, ext, *p))
|
|
except RuntimeError as e:
|
|
raise RuntimeError('for function : ' + f.__name__) from e
|
|
return f
|
|
|
|
return _f
|
|
|
|
|
|
def parameter(name: str, desp: str, ptype: str) -> ExternalParameter:
|
|
"""alias function to create a :class:`ExternalParameter`.
|
|
|
|
:param name: parameter name
|
|
:param desp: parameter description
|
|
:param ptype: parameter type
|
|
"""
|
|
return ExternalParameter(name, desp, ptype)
|
|
|
|
|
|
class FileExporter:
|
|
"""file export/importer"""
|
|
|
|
def __init__(self):
|
|
# file export/import process
|
|
self._process: List[Popen] = []
|
|
|
|
def shutdown(self):
|
|
"""shutdown file manager. kill all sub process task."""
|
|
for p in self._process:
|
|
if p.poll() is None:
|
|
p.kill()
|
|
|
|
@staticmethod
|
|
def get_file_command(*argv: Any) -> List[str]:
|
|
env = current_command_header()
|
|
|
|
# program arguments
|
|
env.extend(map(str, argv))
|
|
|
|
return env
|
|
|
|
@staticmethod
|
|
def get_export_functions() -> Tuple[ExternalExportFunction, ...]:
|
|
"""registered export functions."""
|
|
return tuple(EXPORT_TYPE)
|
|
|
|
@staticmethod
|
|
def get_import_functions() -> Tuple[ExternalImportFunction, ...]:
|
|
"""registered import functions."""
|
|
return tuple(IMPORT_TYPE)
|
|
|
|
@classmethod
|
|
def get_export_func(cls, export_type: str) -> ExternalExportFunction:
|
|
|
|
for ele in EXPORT_TYPE:
|
|
if ele.name == export_type:
|
|
return ele
|
|
|
|
raise ValueError('unsupported export type : ' + export_type)
|
|
|
|
@classmethod
|
|
def get_import_func(cls, import_type: str) -> ExternalImportFunction:
|
|
for ele in IMPORT_TYPE:
|
|
if ele.name == import_type:
|
|
return ele
|
|
|
|
raise ValueError('unsupported import type : ' + import_type)
|
|
|
|
@classmethod
|
|
def get_export_parameters(cls, export_type: str) -> Tuple[ExternalParameter, ...]:
|
|
"""the options information the export function correspond to the *export_type*.
|
|
|
|
:param export_type:
|
|
:return: parameter information
|
|
:raises ValueError: unsupported *export_type*
|
|
"""
|
|
return cls.get_export_func(export_type).parameter
|
|
|
|
@classmethod
|
|
def get_import_parameters(cls, import_type: str) -> Tuple[ExternalParameter, ...]:
|
|
"""the options information the import function correspond to the *import_type*.
|
|
|
|
:param import_type:
|
|
:return: parameter information
|
|
:raises ValueError: unsupported *import_type*
|
|
"""
|
|
return cls.get_import_func(import_type).parameter
|
|
|
|
@classmethod
|
|
def export_file_strict(cls,
|
|
source_path: Path,
|
|
export_path: Path,
|
|
export_type: str,
|
|
**options):
|
|
"""export the meta in strict mode, which means that this function do not change, checking
|
|
the path of the arguments.
|
|
|
|
:param source_path: source recording meta file path
|
|
:param export_path: export file path
|
|
:param export_type: export file type
|
|
:param options: export function options
|
|
"""
|
|
export_func = cls.get_export_func(export_type)
|
|
meta = RecordingMetaFile.load(source_path)
|
|
|
|
export_func.export_file(meta, export_path, **options)
|
|
|
|
@classmethod
|
|
def import_file_strict(cls,
|
|
import_path: Path,
|
|
target_path: Path,
|
|
import_type: str,
|
|
**options) -> RecordingMetaFile:
|
|
"""import as meta file in strict mode, which means that this function do not change, checking
|
|
the path of the arguments.
|
|
|
|
:param import_path: import source file path
|
|
:param target_path: output result meta file path
|
|
:param import_type: import file type
|
|
:param options: import function options.
|
|
:return: output result meta file
|
|
"""
|
|
import_func = cls.get_import_func(import_type)
|
|
|
|
return import_func.import_file(import_path, target_path, **options)
|
|
|
|
def export_file(self,
|
|
source_path: Union[str, Path],
|
|
export_path: Union[str, Path],
|
|
export_type: str,
|
|
async_call=False,
|
|
overwrite=False,
|
|
**options) -> Path:
|
|
"""export mata file.
|
|
|
|
:param source_path: source recording meta file path
|
|
:param export_path: export file path
|
|
:param export_type: export file type
|
|
:param async_call: do export transform in child process
|
|
:param overwrite: overwrite the target file if *export_path* exists
|
|
:param options: export function options
|
|
:return: final export target file path with suffix
|
|
:raises ValueError: unsupported *export_type*
|
|
:raises FileNotFoundError: if *source_path* not a file
|
|
:raises FileExistsError: if *source_path* exists and *overwrite* is ``False``
|
|
:raises RuntimeError: has already request, or export fail.
|
|
"""
|
|
|
|
self.get_export_func(export_type) # check exist this export function
|
|
|
|
if isinstance(source_path, str):
|
|
source_path = Path(source_path)
|
|
|
|
if isinstance(export_path, str):
|
|
export_path = Path(export_path)
|
|
|
|
source_path = source_path.with_suffix(RecordingMetaFile.FILE_EXT)
|
|
|
|
if not source_path.is_file():
|
|
raise FileNotFoundError(str(source_path))
|
|
|
|
# export lock, Is another process use it?
|
|
export_lock = export_path.with_suffix('.export.' + export_type + '.lock')
|
|
|
|
try:
|
|
export_lock.touch(exist_ok=False)
|
|
except FileExistsError as e:
|
|
raise RuntimeError('already in request') from e
|
|
|
|
try:
|
|
# check file overwrite
|
|
if export_path.exists():
|
|
if overwrite:
|
|
export_path.unlink()
|
|
else:
|
|
raise FileExistsError(str(export_path))
|
|
|
|
if async_call:
|
|
if not self._export_file_async(source_path, export_path, export_type, **options):
|
|
raise RuntimeError('export fail')
|
|
else:
|
|
self.export_file_strict(source_path, export_path, export_type, **options)
|
|
|
|
finally:
|
|
# remove export lock
|
|
export_lock.unlink()
|
|
|
|
return export_path
|
|
|
|
def import_file(self,
|
|
import_path: Union[str, Path],
|
|
target_path: Union[str, Path],
|
|
import_type: str,
|
|
async_call=False,
|
|
overwrite=False,
|
|
**options) -> Path:
|
|
"""import as meta file.
|
|
|
|
:param import_path: import source file path
|
|
:param target_path: output result meta file path
|
|
:param import_type: import file type
|
|
:param async_call: do import transform in child process
|
|
:param overwrite: overwrite the result file if *target_path* exists
|
|
:param options: import function options.
|
|
:return: final import result meta file path
|
|
:raises ValueError: unsupported *import_type*
|
|
:raises FileExistsError: if *target_path* exists and *overwrite* is ``False``
|
|
:raises RuntimeError: has already request, or import fail.
|
|
"""
|
|
|
|
self.get_import_func(import_type) # check exist this import function
|
|
|
|
if isinstance(import_path, str):
|
|
import_path = Path(import_path)
|
|
|
|
if isinstance(target_path, str):
|
|
target_path = Path(target_path)
|
|
|
|
if not import_path.exists():
|
|
raise FileNotFoundError(str(import_path))
|
|
|
|
target_path = target_path.with_suffix(RecordingMetaFile.FILE_EXT)
|
|
|
|
# import lock. Is another process use it?
|
|
import_lock = target_path.with_suffix('.import.' + import_type + '.lock')
|
|
|
|
try:
|
|
import_lock.touch(exist_ok=False)
|
|
except FileExistsError as e:
|
|
raise RuntimeError('already in request') from e
|
|
|
|
try:
|
|
# check file overwrite
|
|
if target_path.exists():
|
|
if not overwrite:
|
|
raise FileExistsError(str(target_path))
|
|
else:
|
|
try:
|
|
RecordingMetaFile.remove_meta_file(target_path)
|
|
except RuntimeError:
|
|
pass
|
|
|
|
if async_call:
|
|
if not self._import_file_async(import_path, target_path, import_type, **options):
|
|
raise RuntimeError('import file fail')
|
|
|
|
meta = RecordingMetaFile.load(target_path)
|
|
else:
|
|
meta = self.import_file_strict(import_path, target_path, import_type, **options)
|
|
|
|
finally:
|
|
import_lock.unlink()
|
|
|
|
return Path(meta.filepath)
|
|
|
|
def _export_file_async(self, source_path: Path, export_path: Path, export_type: str, **options) -> bool:
|
|
cmd = self.get_file_command('export', '-q', '--strict', '-T', export_type, '-o', export_path, source_path)
|
|
|
|
for k, v in options.items():
|
|
cmd.append('--' + k + '=' + str(v))
|
|
|
|
env = dict(os.environ)
|
|
|
|
return self._wait_process(Popen(cmd, env=env))
|
|
|
|
def _import_file_async(self, import_path: Path, target_path: Path, import_type: str, **options) -> bool:
|
|
cmd = self.get_file_command('import', '-q', '--strict', '-T', import_type, '-o', target_path, import_path)
|
|
|
|
for k, v in options.items():
|
|
cmd.append('--' + k + '=' + str(v))
|
|
|
|
env = dict(os.environ)
|
|
|
|
return self._wait_process(Popen(cmd, env=env))
|
|
|
|
def _wait_process(self, p: Popen) -> bool:
|
|
print(*p.args)
|
|
|
|
self._process.append(p)
|
|
|
|
start = time()
|
|
try:
|
|
with p:
|
|
return p.wait() == 0
|
|
finally:
|
|
cost = time() - start
|
|
print('pid', p.pid, 'terminated, exit', p.returncode, ', use', '%.2f' % cost, 's')
|
|
|
|
try:
|
|
self._process.remove(p)
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
@register_export('txt', 'raw text', '.txt',
|
|
parameter('resample_rate', 'Re-Sampling Rate, 0 for using origin.', 'int'),
|
|
parameter('resample_method', 'Re-Sampling Method', 'int'),
|
|
parameter('shift_channel', 'start channel number from 1 instead 0', 'bool'))
|
|
def export_txt(meta: RecordingMetaFile,
|
|
export_path: Path,
|
|
resample_rate: int = 0,
|
|
resample_method: int = 0,
|
|
shift_channel=False):
|
|
export.export_txt(str(meta.filepath).encode(),
|
|
str(export_path).encode(),
|
|
resample_rate,
|
|
resample_method,
|
|
shift_channel)
|
|
|
|
|
|
@register_export('csv', 'Comma-Separated Values', '.csv',
|
|
parameter('channel', 'select channel, None for all.', 'CH:int,...'),
|
|
parameter('duration', 'the duration (seconds) for each column', 'int'),
|
|
parameter('resample_rate', 'Re-Sampling Rate, None for using origin.', 'int?'),
|
|
parameter('resample_method', 'Re-Sampling Method', 'int'),
|
|
parameter('include_header', 'include file information', 'bool'),
|
|
parameter('shift_channel', 'start channel number from 1 instead 0', 'bool'))
|
|
def export_csv(meta: RecordingMetaFile,
|
|
export_path: Path,
|
|
channel: List[int] = None,
|
|
duration: int = 60,
|
|
resample_rate: int = 0,
|
|
resample_method: int = 0,
|
|
include_header=False,
|
|
shift_channel=False):
|
|
if channel is None:
|
|
channel = meta.channels
|
|
|
|
export.export_csv(str(meta.filepath).encode(),
|
|
str(export_path).encode(),
|
|
channel,
|
|
duration,
|
|
resample_rate,
|
|
resample_method,
|
|
include_header,
|
|
shift_channel)
|
|
|
|
|
|
@register_export('nex5', 'Neuron Explorer', '.nex5',
|
|
parameter('channel', 'select channel, None for all.', 'CH:int,...'),
|
|
parameter('time_range', 'certain time range in unit ms', 'str'),
|
|
parameter('resample_rate', 'Re-Sampling Rate, None for using origin.', 'int?'),
|
|
parameter('resample_method', 'Re-Sampling Method', 'int'))
|
|
def export_nex(meta: RecordingMetaFile,
|
|
export_path: Path,
|
|
channel: List[int] = None,
|
|
time_range: str = None,
|
|
resample_rate: int = 0,
|
|
resample_method: int = 0):
|
|
from .nex5.main import MainExportNex5
|
|
|
|
main = MainExportNex5()
|
|
|
|
main.meta_file = meta.filepath
|
|
main.output_file = str(export_path)
|
|
main.channel = channel
|
|
|
|
if time_range is not None:
|
|
main.set_time_range(time_range)
|
|
|
|
main.resample_rate = resample_rate
|
|
main.resample_method = resample_method
|
|
|
|
main.run()
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class FileSubCommandMain(CliSubCommandMain, FileOptions, metaclass=abc.ABCMeta):
|
|
def __init__(self, command: str, options: Optional[FileOptions] = None):
|
|
super().__init__(command)
|
|
FileOptions.__init__(self)
|
|
|
|
if options is not None:
|
|
self.storage_root = options.storage_root
|
|
self.export_root = options.export_root
|
|
|
|
self.strict = False
|
|
self.quiet = False
|
|
self.overwrite = False
|
|
|
|
# TODO external source file paths
|
|
|
|
@cli_flags('-q', '--quiet')
|
|
def _quiet(self, opt: str):
|
|
"""quiet output"""
|
|
self.quiet = True
|
|
|
|
@cli_flags('--strict')
|
|
def _strict(self, opt: str):
|
|
"""strict mode"""
|
|
self.strict = True
|
|
|
|
@cli_flags('-f', '--overwrite')
|
|
def _overwrite(self, opt: str):
|
|
"""overwrite output file if it existed"""
|
|
self.overwrite = True
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class FileExportMain(FileSubCommandMain):
|
|
|
|
def __init__(self, command: str, options: Optional[FileOptions] = None):
|
|
super().__init__(command, options)
|
|
|
|
self.export_root = '.'
|
|
|
|
self.export_type: Optional[str] = None
|
|
self.source_path: Optional[str] = None
|
|
self.export_path: Optional[str] = None
|
|
self.export_para = {}
|
|
self.export_async = False
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
"""print help document or export function"""
|
|
if self.export_type is None:
|
|
self.print_help()
|
|
|
|
print('Export Function')
|
|
table = Table(column=2)
|
|
|
|
for ele in EXPORT_TYPE:
|
|
table.append(ele.name, ele.desp)
|
|
|
|
table.set_format(0, align_right=True)
|
|
table.set_format(1, align_right=False)
|
|
|
|
table.print(indent=' ')
|
|
|
|
else:
|
|
ret = FileExporter.get_export_parameters(self.export_type)
|
|
|
|
table = Table(column=2)
|
|
|
|
for para in ret:
|
|
table.append('--' + para.option_name, para.ptype)
|
|
|
|
table.set_format(0, align_right=False)
|
|
table.set_format(1, align_right=False)
|
|
|
|
for i, line in enumerate(table.lines()):
|
|
print(line)
|
|
print(' ', ret[i].text)
|
|
print()
|
|
|
|
@cli_options('-T', '--type', value='FORMAT')
|
|
def _export_type(self, opt: str, value: str):
|
|
"""export type"""
|
|
self.export_type = value
|
|
|
|
@cli_flags('-A', '--async')
|
|
def _export_async(self, opt: str):
|
|
"""async export"""
|
|
self.export_async = True
|
|
|
|
@cli_options('-o', '--output', value='PATH')
|
|
def _export_path(self, opt: str, value: str):
|
|
"""export to PATH."""
|
|
self.export_path = value
|
|
|
|
@cli_options('--', value='KEY=VALUE')
|
|
def _export_parameter(self, opt: str, value: str):
|
|
"""export parameter"""
|
|
key = opt[2:].replace('-', '_')
|
|
|
|
if value is None:
|
|
self.export_para[key] = True
|
|
else:
|
|
self.export_para[key] = value
|
|
|
|
@cli_arguments(0, value='PATH')
|
|
def _source_path(self, pos: int, value: str):
|
|
"""source PATH"""
|
|
self.source_path = value
|
|
|
|
def run(self):
|
|
if self.strict:
|
|
self._run_strict()
|
|
else:
|
|
ret = self._run()
|
|
|
|
if not self.quiet:
|
|
print(ret)
|
|
|
|
def _run(self) -> Path:
|
|
if self.source_path is None:
|
|
raise RuntimeError('lost PATH')
|
|
|
|
if self.export_path is None and self.export_type is None:
|
|
raise RuntimeError('lost -T FORMAT')
|
|
|
|
elif self.export_type is None:
|
|
try:
|
|
i = self.export_path.rindex('.')
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
self.export_type = self.export_path[i + 1:]
|
|
|
|
elif self.export_path is None:
|
|
source_path = Path(self.source_path)
|
|
export_path = Path(self.export_root) / source_path.name
|
|
|
|
if '.' in export_path.name:
|
|
export_path.with_suffix(self.export_type)
|
|
else:
|
|
export_path = export_path.with_name(export_path.name + '.' + self.export_type)
|
|
|
|
self.export_path = export_path
|
|
|
|
manager = FileExporter()
|
|
|
|
try:
|
|
ret = manager.export_file(self.source_path, self.export_path, self.export_type,
|
|
async_call=self.export_async,
|
|
overwrite=self.overwrite,
|
|
**self.export_para)
|
|
|
|
return ret
|
|
finally:
|
|
manager.shutdown()
|
|
|
|
def _run_strict(self) -> Path:
|
|
import sys
|
|
|
|
if self.export_type is None:
|
|
raise RuntimeError('lost -T FORMAT')
|
|
|
|
if self.source_path is None:
|
|
raise RuntimeError('lost SOURCE')
|
|
|
|
if self.export_path is None:
|
|
raise RuntimeError('lost PATH')
|
|
|
|
if self.export_async:
|
|
print('strict mode ignore async flag', file=sys.stderr)
|
|
|
|
export_path = Path(self.export_path)
|
|
|
|
FileExporter.export_file_strict(Path(self.source_path), export_path, self.export_type, **self.export_para)
|
|
|
|
return export_path
|
|
|
|
|
|
# noinspection PyUnusedLocal
|
|
class FileImportMain(FileSubCommandMain):
|
|
|
|
def __init__(self, command: str, options: Optional[FileOptions] = None):
|
|
super().__init__(command, options)
|
|
|
|
self.import_type: Optional[str] = None
|
|
self.source_path: Optional[str] = None
|
|
self.output_path: Optional[str] = None
|
|
self.import_para = {}
|
|
self.import_async = False
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
"""print help document or import function"""
|
|
if self.import_type is None:
|
|
self.print_help()
|
|
|
|
print('Import Function')
|
|
table = Table(column=2)
|
|
|
|
for ele in IMPORT_TYPE:
|
|
table.append(ele.name, ele.desp)
|
|
|
|
table.set_format(0, align_right=True)
|
|
table.set_format(1, align_right=False)
|
|
|
|
table.print(indent=' ')
|
|
|
|
else:
|
|
ret = FileExporter.get_import_parameters(self.import_type)
|
|
|
|
table = Table(column=2)
|
|
|
|
for para in ret:
|
|
table.append('--' + para.option_name, para.ptype)
|
|
|
|
for i, line in enumerate(table.lines()):
|
|
print(line)
|
|
print(' ', ret[i].text)
|
|
print()
|
|
|
|
@cli_options('-T', '--type', value='FORMAT')
|
|
def _import_type(self, opt: str, value: str):
|
|
"""import type"""
|
|
self.import_type = value
|
|
|
|
@cli_options('-o', '--output', value='PATH')
|
|
def _output_path(self, opt: str, value: str):
|
|
"""import to PATH."""
|
|
self.output_path = value
|
|
|
|
@cli_flags('-A', '--async')
|
|
def _export_async(self, opt: str):
|
|
"""async export"""
|
|
self.import_async = True
|
|
|
|
@cli_options('--', value='KEY=VALUE')
|
|
def _import_parameter(self, opt: str, value: str):
|
|
"""export parameter"""
|
|
key = opt[2:].replace('-', '_')
|
|
|
|
if value is None:
|
|
self.import_para[key] = True
|
|
else:
|
|
self.import_para[key] = value
|
|
|
|
@cli_arguments(0, value='PATH')
|
|
def _source_path(self, pos: int, value: str):
|
|
"""source PATH"""
|
|
self.source_path = value
|
|
|
|
def run(self):
|
|
if self.strict:
|
|
ret = self._run_strict()
|
|
else:
|
|
ret = self._run()
|
|
|
|
if not self.quiet:
|
|
print(ret)
|
|
|
|
def _run(self) -> Path:
|
|
if self.source_path is None:
|
|
raise RuntimeError('lost PATH')
|
|
|
|
if self.import_type is None:
|
|
try:
|
|
i = self.source_path.rindex('.')
|
|
except ValueError:
|
|
pass
|
|
else:
|
|
self.import_type = self.source_path[i + 1:]
|
|
|
|
if self.import_type is None:
|
|
raise RuntimeError('lost -T FORMAT')
|
|
|
|
manager = FileExporter()
|
|
|
|
try:
|
|
ret = manager.import_file(self.source_path, self.output_path, self.import_type,
|
|
overwrite=self.overwrite,
|
|
async_call=self.import_async,
|
|
**self.import_para)
|
|
|
|
# TODO path convert
|
|
return ret
|
|
finally:
|
|
manager.shutdown()
|
|
|
|
def _run_strict(self):
|
|
if self.import_type is None:
|
|
raise RuntimeError('lost -T FORMAT')
|
|
|
|
if self.source_path is None:
|
|
raise RuntimeError('lost SOURCE')
|
|
|
|
if self.output_path is None:
|
|
raise RuntimeError('lost PATH')
|
|
|
|
FileExporter.import_file_strict(Path(self.source_path), Path(self.output_path), self.import_type,
|
|
**self.import_para)
|