import collections.abc
from pathlib import Path
from typing import Union
import numpy as np
from typeguard import typechecked
from espnet2.fileio.read_text import read_2columns_text
[docs]class NpyScpWriter:
"""Writer class for a scp file of numpy file.
Examples:
key1 /some/path/a.npy
key2 /some/path/b.npy
key3 /some/path/c.npy
key4 /some/path/d.npy
...
>>> writer = NpyScpWriter('./data/', './data/feat.scp')
>>> writer['aa'] = numpy_array
>>> writer['bb'] = numpy_array
"""
@typechecked
def __init__(self, outdir: Union[Path, str], scpfile: Union[Path, str]):
self.dir = Path(outdir)
self.dir.mkdir(parents=True, exist_ok=True)
scpfile = Path(scpfile)
scpfile.parent.mkdir(parents=True, exist_ok=True)
self.fscp = scpfile.open("w", encoding="utf-8")
self.data = {}
[docs] def get_path(self, key):
return self.data[key]
def __setitem__(self, key, value):
assert isinstance(value, np.ndarray), type(value)
p = self.dir / f"{key}.npy"
p.parent.mkdir(parents=True, exist_ok=True)
np.save(str(p), value)
self.fscp.write(f"{key} {p}\n")
# Store the file path
self.data[key] = str(p)
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
[docs] def close(self):
self.fscp.close()
[docs]class NpyScpReader(collections.abc.Mapping):
"""Reader class for a scp file of numpy file.
Examples:
key1 /some/path/a.npy
key2 /some/path/b.npy
key3 /some/path/c.npy
key4 /some/path/d.npy
...
>>> reader = NpyScpReader('npy.scp')
>>> array = reader['key1']
"""
@typechecked
def __init__(self, fname: Union[Path, str]):
self.fname = Path(fname)
self.data = read_2columns_text(fname)
[docs] def get_path(self, key):
return self.data[key]
def __getitem__(self, key) -> np.ndarray:
p = self.data[key]
return np.load(p)
def __contains__(self, item):
return item
def __len__(self):
return len(self.data)
def __iter__(self):
return iter(self.data)
[docs] def keys(self):
return self.data.keys()