sockeye/data_io.py [218:449]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - return bucket_batch_sizes def calculate_length_statistics(source_iterables: Sequence[Iterable[Any]], target_iterables: Sequence[Iterable[Any]], max_seq_len_source: int, max_seq_len_target: int) -> 'LengthStatistics': """ Returns mean and standard deviation of target-to-source length ratios of parallel corpus. :param source_iterables: Source sequence readers. :param target_iterables: Target sequence readers. :param max_seq_len_source: Maximum source sequence length. :param max_seq_len_target: Maximum target sequence length. :return: The number of sentences as well as the mean and standard deviation of target to source length ratios. """ mean_and_variance = OnlineMeanAndVariance() for sources, targets in parallel_iter(source_iterables, target_iterables): source_len = len(sources[0]) target_len = len(targets[0]) if source_len > max_seq_len_source or target_len > max_seq_len_target: continue length_ratio = target_len / source_len mean_and_variance.update(length_ratio) return LengthStatistics(mean_and_variance.count, mean_and_variance.mean, mean_and_variance.std) def analyze_sequence_lengths(sources: List[str], targets: List[str], vocab_sources: List[vocab.Vocab], vocab_targets: List[vocab.Vocab], max_seq_len_source: int, max_seq_len_target: int) -> 'LengthStatistics': train_sources_sentences, train_targets_sentences = create_sequence_readers(sources, targets, vocab_sources, vocab_targets) length_statistics = calculate_length_statistics(train_sources_sentences, train_targets_sentences, max_seq_len_source, max_seq_len_target) logger.info("%d sequences of maximum length (%d, %d) in '%s' and '%s'.", length_statistics.num_sents, max_seq_len_source, max_seq_len_target, sources[0], targets[0]) logger.info("Mean training target/source length ratio: %.2f (+-%.2f)", length_statistics.length_ratio_mean, length_statistics.length_ratio_std) return length_statistics def are_none(sequences: Sequence[Sized]) -> bool: """ Returns True if all sequences are None. """ if not sequences: return True return all(s is None for s in sequences) def are_token_parallel(sequences: Sequence[Sized]) -> bool: """ Returns True if all sequences in the list have the same length. """ if not sequences or len(sequences) == 1: return True else: return all(len(s) == len(sequences[0]) for s in sequences) class DataStatisticsAccumulator: def __init__(self, buckets: List[Tuple[int, int]], vocab_source: Optional[Dict[str, int]], vocab_target: Dict[str, int], length_ratio_mean: float, length_ratio_std: float) -> None: self.buckets = buckets num_buckets = len(buckets) self.length_ratio_mean = length_ratio_mean self.length_ratio_std = length_ratio_std if vocab_source is not None: self.unk_id_source = vocab_source[C.UNK_SYMBOL] self.size_vocab_source = len(vocab_source) else: self.unk_id_source = None self.size_vocab_source = 0 self.unk_id_target = vocab_target[C.UNK_SYMBOL] self.size_vocab_target = len(vocab_target) self.num_sents = 0 self.num_discarded = 0 self.num_tokens_source = 0 self.num_tokens_target = 0 self.num_unks_source = 0 self.num_unks_target = 0 self.max_observed_len_source = 0 self.max_observed_len_target = 0 self._mean_len_target_per_bucket = [OnlineMeanAndVariance() for _ in range(num_buckets)] self._length_ratio_per_bucket = [OnlineMeanAndVariance() for _ in range(num_buckets)] def sequence_pair(self, source: List[int], target: List[int], bucket_idx: Optional[int]): if bucket_idx is None: self.num_discarded += 1 return source_len = len(source) target_len = len(target) length_ratio = target_len / (source_len if source_len else 1.) self._mean_len_target_per_bucket[bucket_idx].update(target_len) self._length_ratio_per_bucket[bucket_idx].update(length_ratio) self.num_sents += 1 self.num_tokens_source += source_len self.num_tokens_target += target_len self.max_observed_len_source = max(source_len, self.max_observed_len_source) self.max_observed_len_target = max(target_len, self.max_observed_len_target) if self.unk_id_source is not None: self.num_unks_source += source.count(self.unk_id_source) self.num_unks_target += target.count(self.unk_id_target) @property def mean_len_target_per_bucket(self) -> List[Optional[float]]: return [mean_and_variance.mean if mean_and_variance.count > 0 else None for mean_and_variance in self._mean_len_target_per_bucket] @property def length_ratio_stats_per_bucket(self) -> List[Tuple[Optional[float], Optional[float]]]: return [(mean_and_variance.mean, mean_and_variance.std) if mean_and_variance.count > 0 else (None, None) for mean_and_variance in self._length_ratio_per_bucket] @property def statistics(self): num_sents_per_bucket = [mean_and_variance.count for mean_and_variance in self._mean_len_target_per_bucket] return DataStatistics(num_sents=self.num_sents, num_discarded=self.num_discarded, num_tokens_source=self.num_tokens_source, num_tokens_target=self.num_tokens_target, num_unks_source=self.num_unks_source, num_unks_target=self.num_unks_target, max_observed_len_source=self.max_observed_len_source, max_observed_len_target=self.max_observed_len_target, size_vocab_source=self.size_vocab_source, size_vocab_target=self.size_vocab_target, length_ratio_mean=self.length_ratio_mean, length_ratio_std=self.length_ratio_std, buckets=self.buckets, num_sents_per_bucket=num_sents_per_bucket, average_len_target_per_bucket=self.mean_len_target_per_bucket, length_ratio_stats_per_bucket=self.length_ratio_stats_per_bucket) def create_shards(source_fnames: List[str], target_fnames: List[str], num_shards: int, output_prefix: str) -> Tuple[List[Tuple[Tuple[str, ...], Tuple[str, ...]]], bool]: """ Assign source/target sentence pairs to shards at random. :param source_fnames: The path to the source text (and optional token-parallel factor files). :param target_fnames: The path to the target text (and optional token-parallel factor files). :param num_shards: The total number of shards. :param output_prefix: The prefix under which the shard files will be created. :return: List of tuples of source (and source factor) file names and target (and target factor) file names for each shard and a flag of whether the returned file names are temporary and can be deleted. """ if num_shards == 1: return [(tuple(source_fnames), tuple(target_fnames))], True os.makedirs(output_prefix, exist_ok=True) sources_shard_fnames = [[os.path.join(output_prefix, C.SHARD_SOURCE % i) + ".%d" % f for i in range(num_shards)] for f in range(len(source_fnames))] targets_shard_fnames = [[os.path.join(output_prefix, C.SHARD_TARGET % i) + ".%d" % f for i in range(num_shards)] for f in range(len(target_fnames))] with ExitStack() as exit_stack: sources_shards = [[exit_stack.enter_context(smart_open(f, mode="wb")) for f in sources_shard_fnames[i]] for i in range(len(source_fnames))] targets_shards = [[exit_stack.enter_context(smart_open(f, mode="wb")) for f in targets_shard_fnames[i]] for i in range(len(target_fnames))] source_readers = [exit_stack.enter_context(smart_open(f, mode="rb")) for f in source_fnames] target_readers = [exit_stack.enter_context(smart_open(f, mode="rb")) for f in target_fnames] random_shard_iter = iter(lambda: random.randrange(num_shards), None) for (sources, targets), random_shard_index in zip(parallel_iter(source_readers, target_readers, True, False), random_shard_iter): random_shard_index = cast(int, random_shard_index) for i, line in enumerate(sources): file = sources_shards[i][random_shard_index] file.write(line) for i, line in enumerate(targets): file = targets_shards[i][random_shard_index] file.write(line) sources_shard_fnames_by_shards = zip(*sources_shard_fnames) targets_shard_fnames_by_shards = zip(*targets_shard_fnames) return list(zip(sources_shard_fnames_by_shards, targets_shard_fnames_by_shards)), False class RawParallelDatasetLoader: """ Loads a data set of variable-length parallel source/target sequences into buckets of NDArrays. :param buckets: Bucket list. :param eos_id: End-of-sentence id. :param pad_id: Padding id. :param eos_id: Unknown id. :param skip_blanks: Whether to skip blank lines. :param dtype: Data type. :param shift_target_factors: If true, shift secondary target factors (i>1) to the right. Target factor shifting: Data I/O sequence: f1: A B C fs: a b c Target sequence: f1: A B C fs: a b Label sequence: f1: A B C fs: a b c """ def __init__(self, buckets: List[Tuple[int, int]], eos_id: int, pad_id: int, skip_blanks: bool = True, - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - sockeye/data_io_pt.py [199:431]: - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - return bucket_batch_sizes def calculate_length_statistics(source_iterables: Sequence[Iterable[Any]], target_iterables: Sequence[Iterable[Any]], max_seq_len_source: int, max_seq_len_target: int) -> 'LengthStatistics': """ Returns mean and standard deviation of target-to-source length ratios of parallel corpus. :param source_iterables: Source sequence readers. :param target_iterables: Target sequence readers. :param max_seq_len_source: Maximum source sequence length. :param max_seq_len_target: Maximum target sequence length. :return: The number of sentences as well as the mean and standard deviation of target to source length ratios. """ mean_and_variance = OnlineMeanAndVariance() for sources, targets in parallel_iter(source_iterables, target_iterables): source_len = len(sources[0]) target_len = len(targets[0]) if source_len > max_seq_len_source or target_len > max_seq_len_target: continue length_ratio = target_len / source_len mean_and_variance.update(length_ratio) return LengthStatistics(mean_and_variance.count, mean_and_variance.mean, mean_and_variance.std) def analyze_sequence_lengths(sources: List[str], targets: List[str], vocab_sources: List[vocab.Vocab], vocab_targets: List[vocab.Vocab], max_seq_len_source: int, max_seq_len_target: int) -> 'LengthStatistics': train_sources_sentences, train_targets_sentences = create_sequence_readers(sources, targets, vocab_sources, vocab_targets) length_statistics = calculate_length_statistics(train_sources_sentences, train_targets_sentences, max_seq_len_source, max_seq_len_target) logger.info("%d sequences of maximum length (%d, %d) in '%s' and '%s'.", length_statistics.num_sents, max_seq_len_source, max_seq_len_target, sources[0], targets[0]) logger.info("Mean training target/source length ratio: %.2f (+-%.2f)", length_statistics.length_ratio_mean, length_statistics.length_ratio_std) return length_statistics def are_none(sequences: Sequence[Sized]) -> bool: """ Returns True if all sequences are None. """ if not sequences: return True return all(s is None for s in sequences) def are_token_parallel(sequences: Sequence[Sized]) -> bool: """ Returns True if all sequences in the list have the same length. """ if not sequences or len(sequences) == 1: return True else: return all(len(s) == len(sequences[0]) for s in sequences) class DataStatisticsAccumulator: def __init__(self, buckets: List[Tuple[int, int]], vocab_source: Optional[Dict[str, int]], vocab_target: Dict[str, int], length_ratio_mean: float, length_ratio_std: float) -> None: self.buckets = buckets num_buckets = len(buckets) self.length_ratio_mean = length_ratio_mean self.length_ratio_std = length_ratio_std if vocab_source is not None: self.unk_id_source = vocab_source[C.UNK_SYMBOL] self.size_vocab_source = len(vocab_source) else: self.unk_id_source = None self.size_vocab_source = 0 self.unk_id_target = vocab_target[C.UNK_SYMBOL] self.size_vocab_target = len(vocab_target) self.num_sents = 0 self.num_discarded = 0 self.num_tokens_source = 0 self.num_tokens_target = 0 self.num_unks_source = 0 self.num_unks_target = 0 self.max_observed_len_source = 0 self.max_observed_len_target = 0 self._mean_len_target_per_bucket = [OnlineMeanAndVariance() for _ in range(num_buckets)] self._length_ratio_per_bucket = [OnlineMeanAndVariance() for _ in range(num_buckets)] def sequence_pair(self, source: List[int], target: List[int], bucket_idx: Optional[int]): if bucket_idx is None: self.num_discarded += 1 return source_len = len(source) target_len = len(target) length_ratio = target_len / (source_len if source_len else 1.) self._mean_len_target_per_bucket[bucket_idx].update(target_len) self._length_ratio_per_bucket[bucket_idx].update(length_ratio) self.num_sents += 1 self.num_tokens_source += source_len self.num_tokens_target += target_len self.max_observed_len_source = max(source_len, self.max_observed_len_source) self.max_observed_len_target = max(target_len, self.max_observed_len_target) if self.unk_id_source is not None: self.num_unks_source += source.count(self.unk_id_source) self.num_unks_target += target.count(self.unk_id_target) @property def mean_len_target_per_bucket(self) -> List[Optional[float]]: return [mean_and_variance.mean if mean_and_variance.count > 0 else None for mean_and_variance in self._mean_len_target_per_bucket] @property def length_ratio_stats_per_bucket(self) -> List[Tuple[Optional[float], Optional[float]]]: return [(mean_and_variance.mean, mean_and_variance.std) if mean_and_variance.count > 0 else (None, None) for mean_and_variance in self._length_ratio_per_bucket] @property def statistics(self): num_sents_per_bucket = [mean_and_variance.count for mean_and_variance in self._mean_len_target_per_bucket] return DataStatistics(num_sents=self.num_sents, num_discarded=self.num_discarded, num_tokens_source=self.num_tokens_source, num_tokens_target=self.num_tokens_target, num_unks_source=self.num_unks_source, num_unks_target=self.num_unks_target, max_observed_len_source=self.max_observed_len_source, max_observed_len_target=self.max_observed_len_target, size_vocab_source=self.size_vocab_source, size_vocab_target=self.size_vocab_target, length_ratio_mean=self.length_ratio_mean, length_ratio_std=self.length_ratio_std, buckets=self.buckets, num_sents_per_bucket=num_sents_per_bucket, average_len_target_per_bucket=self.mean_len_target_per_bucket, length_ratio_stats_per_bucket=self.length_ratio_stats_per_bucket) def create_shards(source_fnames: List[str], target_fnames: List[str], num_shards: int, output_prefix: str) -> Tuple[List[Tuple[Tuple[str, ...], Tuple[str, ...]]], bool]: """ Assign source/target sentence pairs to shards at random. :param source_fnames: The path to the source text (and optional token-parallel factor files). :param target_fnames: The path to the target text (and optional token-parallel factor files). :param num_shards: The total number of shards. :param output_prefix: The prefix under which the shard files will be created. :return: List of tuples of source (and source factor) file names and target (and target factor) file names for each shard and a flag of whether the returned file names are temporary and can be deleted. """ if num_shards == 1: return [(tuple(source_fnames), tuple(target_fnames))], True os.makedirs(output_prefix, exist_ok=True) sources_shard_fnames = [[os.path.join(output_prefix, C.SHARD_SOURCE % i) + ".%d" % f for i in range(num_shards)] for f in range(len(source_fnames))] targets_shard_fnames = [[os.path.join(output_prefix, C.SHARD_TARGET % i) + ".%d" % f for i in range(num_shards)] for f in range(len(target_fnames))] with ExitStack() as exit_stack: sources_shards = [[exit_stack.enter_context(smart_open(f, mode="wb")) for f in sources_shard_fnames[i]] for i in range(len(source_fnames))] targets_shards = [[exit_stack.enter_context(smart_open(f, mode="wb")) for f in targets_shard_fnames[i]] for i in range(len(target_fnames))] source_readers = [exit_stack.enter_context(smart_open(f, mode="rb")) for f in source_fnames] target_readers = [exit_stack.enter_context(smart_open(f, mode="rb")) for f in target_fnames] random_shard_iter = iter(lambda: random.randrange(num_shards), None) for (sources, targets), random_shard_index in zip(parallel_iter(source_readers, target_readers, True, False), random_shard_iter): random_shard_index = cast(int, random_shard_index) for i, line in enumerate(sources): file = sources_shards[i][random_shard_index] file.write(line) for i, line in enumerate(targets): file = targets_shards[i][random_shard_index] file.write(line) sources_shard_fnames_by_shards = zip(*sources_shard_fnames) targets_shard_fnames_by_shards = zip(*targets_shard_fnames) return list(zip(sources_shard_fnames_by_shards, targets_shard_fnames_by_shards)), False class RawParallelDatasetLoader: """ Loads a data set of variable-length parallel source/target sequences into buckets of tensors. :param buckets: Bucket list. :param eos_id: End-of-sentence id. :param pad_id: Padding id. :param eos_id: Unknown id. :param skip_blanks: Whether to skip blank lines. :param dtype: Data type. :param shift_target_factors: If true, shift secondary target factors (i>1) to the right. Target factor shifting: Data I/O sequence: f1: A B C fs: a b c Target sequence: f1: A B C fs: a b Label sequence: f1: A B C fs: a b c """ def __init__(self, buckets: List[Tuple[int, int]], eos_id: int, pad_id: int, skip_blanks: bool = True, - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -