Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def foo(a: Union[str, int]):
assert check_argument_types()
def foo(a: FooGeneric[str]):
assert check_argument_types()
def read_2column_text(path: Union[Path, str]) -> Dict[str, str]:
"""
Examples:
wav.scp:
key1 /some/path/a.wav
key2 /some/path/b.wav
>>> read_2column_text('wav.scp')
{'key1': '/some/path/a.wav', 'key2': '/some/path/b.wav'}
"""
assert check_argument_types()
data = {}
with Path(path).open("r") as f:
for linenum, line in enumerate(f, 1):
sps = line.rstrip().split(maxsplit=1)
if len(sps) != 2:
raise RuntimeError(
f"scp file must have two or more columns: "
f"{line} ({path}:{linenum})"
)
k, v = sps
if k in data:
raise RuntimeError(f"{k} is duplicated ({path}:{linenum})")
data[k] = v.rstrip()
assert check_return_type(data)
return data
max-pooling.
highway_depth: Depth of the highway layer.
rnn_size: The size of the encoder's hidden state. Note
that the actual encoder output state size will be
twice as long because it is the result of
concatenation of forward and backward hidden states.
filters: Specification of CNN filters. It is a list of tuples
specifying the filter size and number of channels.
Keyword arguments:
dropout_keep_prob: The dropout keep probability
(default 1.0)
"""
ModelPart.__init__(self, name, reuse, save_checkpoint, load_checkpoint,
initializers)
check_argument_types()
self.input_sequence = input_sequence
self.segment_size = segment_size
self.highway_depth = highway_depth
self.rnn_size = rnn_size
self.filters = filters
self.dropout_keep_prob = dropout_keep_prob
self.use_noisy_activations = use_noisy_activations
if dropout_keep_prob <= 0. or dropout_keep_prob > 1.:
raise ValueError(
("Dropout keep probability must be "
"in (0; 1], was {}").format(dropout_keep_prob))
if rnn_size <= 0:
raise ValueError("RNN size must be a positive integer.")
def __init__(self, lm: AbsLM, vocab_size: int, ignore_id: int = 0):
assert check_argument_types()
super().__init__()
self.lm = lm
self.sos = vocab_size - 1
self.eos = vocab_size - 1
# ignore_id may be assumed as 0, shared with CTC-blank symbol for ASR.
self.ignore_id = ignore_id
It writes the given configuration in a temporary directory and replaces ``%(port)s`` in the
configuration with an ephemeral TCP port.
If no configuration is given a default configuration is used where only anonymous
authentication is defined and the anonymous user has all privileges on everything.
One websocket transport is defined, listening on ``localhost``.
The Crossbar process is automatically terminated and temporary directory deleted when the host
process terminates.
:param config: YAML configuration for crossbar (for ``config.yaml``)
:return: the automatically selected port
"""
check_argument_types()
# Get the next available TCP port
port = get_next_free_tcp_port()
# Write the configuration file
tempdir = Path(mkdtemp())
atexit.register(shutil.rmtree, str(tempdir))
config_file = tempdir / 'config.yaml'
with config_file.open('w') as f:
f.write(config % {'port': port})
launch_crossbar(tempdir)
return port
def __init__(
self,
vocab_size: int,
encoder_output_size: int,
rnn_type: str = "lstm",
num_layers: int = 1,
hidden_size: int = 320,
sampling_probability: float = 0.0,
dropout: float = 0.0,
context_residual: bool = False,
replace_sos: bool = False,
num_encs: int = 1,
att_conf: dict = get_default_kwargs(build_attention_list),
):
# FIXME(kamo): The parts of num_spk should be refactored more more more
assert check_argument_types()
if rnn_type not in {"lstm", "gru"}:
raise ValueError(f"Not supported: rnn_type={rnn_type}")
super().__init__()
eprojs = encoder_output_size
self.dtype = rnn_type
self.dunits = hidden_size
self.dlayers = num_layers
self.context_residual = context_residual
self.sos = vocab_size - 1
self.eos = vocab_size - 1
self.odim = vocab_size
self.sampling_probability = sampling_probability
self.dropout = dropout
self.num_encs = num_encs
def mapply(self, fn, args, kwargs):
try:
memo = typeguard._CallMemo(fn, args=args, kwargs=kwargs)
typeguard.check_argument_types(memo)
except TypeError as exc:
print(exc)
raise XMLRPCInvalidParamTypes(exc)
return super().mapply(fn, args, kwargs)
def average_nbest_models(
cls,
output_dir: Path,
reporter: Reporter,
best_model_criterion: Sequence[Sequence[str]],
nbest: int,
) -> None:
assert check_argument_types()
# 1. Get nbests: List[Tuple[str, str, List[Tuple[epoch, value]]]]
nbest_epochs = [
(ph, k, reporter.sort_epochs_and_values(ph, k, m)[:nbest])
for ph, k, m in best_model_criterion
if reporter.has(ph, k)
]
_loaded = {}
for ph, cr, epoch_and_values in nbest_epochs:
# Note that len(epoch_and_values) doesn't always equal to nbest.
op = output_dir / f"{ph}.{cr}.ave_{len(epoch_and_values)}best.pt"
logging.info(
f"Averaging {len(epoch_and_values)}best models: "
f'criterion="{ph}.{cr}": {op}'
)
def __init__(self,
output_series: str,
decoder: SupportedDecoder,
postprocess: Postprocessor = None) -> None:
check_argument_types()
super().__init__(output_series, decoder)
self.postprocess = postprocess
self.vocabulary = self.decoder.vocabulary