{ "python.pythonPath": "C:\Users\BiGCARE\anaconda3\envs\sv2tts_korean\python.exe" } from encoder.data_objects.speaker_verification_dataset import SpeakerVerificationDataset from encoder.data_objects.speaker_verification_dataset import SpeakerVerificationDataLoader
import random
class RandomCycler: """ Creates an internal copy of a sequence and allows access to its items in a constrained random order. For a source sequence of n items and one or several consecutive queries of a total of m items, the following guarantees hold (one implies the other): - Each item will be returned between m // n and ((m - 1) // n) + 1 times. - Between two appearances of the same item, there may be at most 2 * (n - 1) other items. """
def __init__(self, source):
if len(source) == 0:
raise Exception("Can't create RandomCycler from an empty collection")
self.all_items = list(source)
self.next_items = []
def sample(self, count: int):
shuffle = lambda l: random.sample(l, len(l))
out = []
while count > 0:
if count >= len(self.all_items):
out.extend(shuffle(list(self.all_items)))
count -= len(self.all_items)
continue
n = min(count, len(self.next_items))
out.extend(self.next_items[:n])
count -= n
self.next_items = self.next_items[n:]
if len(self.next_items) == 0:
self.next_items = shuffle(list(self.all_items))
return out
def __next__(self):
return self.sample(1)[0]
import numpy as np
from typing import List from encoder.data_objects.speaker import Speaker
class SpeakerBatch: def init(self, speakers: List[Speaker], utterances_per_speaker: int, n_frames: int): self.speakers = speakers self.partials = {s: s.random_partial(utterances_per_speaker, n_frames) for s in speakers}
# Array of shape (n_speakers * n_utterances, n_frames, mel_n), e.g. for 3 speakers with
# 4 utterances each of 160 frames of 40 mel coefficients: (12, 160, 40)
self.data = np.array([frames for s in speakers for _, frames, _ in self.partials[s]])
from encoder.data_objects.random_cycler import RandomCycler from encoder.data_objects.speaker_batch import SpeakerBatch from encoder.data_objects.speaker import Speaker from encoder.params_data import partials_n_frames from torch.utils.data import Dataset, DataLoader from pathlib import Path
TODO: improve with a pool of speakers for data efficiency
class SpeakerVerificationDataset(Dataset): def init(self, datasets_root: Path): self.root = datasets_root speaker_dirs = [f for f in self.root.glob("*") if f.is_dir()] if len(speaker_dirs) == 0: raise Exception("No speakers found. Make sure you are pointing to the directory " "containing all preprocessed speaker directories.") self.speakers = [Speaker(speaker_dir) for speaker_dir in speaker_dirs] self.speaker_cycler = RandomCycler(self.speakers)
def __len__(self):
return int(1e10)
def __getitem__(self, index):
return next(self.speaker_cycler)
def get_logs(self):
log_string = ""
for log_fpath in self.root.glob("*.txt"):
with log_fpath.open("r") as log_file:
log_string += "".join(log_file.readlines())
return log_string
class SpeakerVerificationDataLoader(DataLoader): def init(self, dataset, speakers_per_batch, utterances_per_speaker, sampler=None, batch_sampler=None, num_workers=0, pin_memory=False, timeout=0, worker_init_fn=None): self.utterances_per_speaker = utterances_per_speaker
super().__init__(
dataset=dataset,
batch_size=speakers_per_batch,
shuffle=False,
sampler=sampler,
batch_sampler=batch_sampler,
num_workers=num_workers,
collate_fn=self.collate,
pin_memory=pin_memory,
drop_last=False,
timeout=timeout,
worker_init_fn=worker_init_fn
)
def collate(self, speakers):
return SpeakerBatch(speakers, self.utterances_per_speaker, partials_n_frames)
from encoder.data_objects.random_cycler import RandomCycler
from encoder.data_objects.utterance import Utterance from pathlib import Path
Contains the set of utterances of a single speaker
class Speaker: def init(self, root: Path): self.root = root self.name = root.name self.utterances = None self.utterance_cycler = None
def _load_utterances(self):
with self.root.joinpath("_sources.txt").open("r") as sources_file:
sources = [l.split(",") for l in sources_file]
sources = {frames_fname: wave_fpath for frames_fname, wave_fpath in sources}
self.utterances = [Utterance(self.root.joinpath(f), w) for f, w in sources.items()]
self.utterance_cycler = RandomCycler(self.utterances)
def random_partial(self, count, n_frames):
"""
Samples a batch of <count> unique partial utterances from the disk in a way that all
utterances come up at least once every two cycles and in a random order every time.
:param count: The number of partial utterances to sample from the set of utterances from
that speaker. Utterances are guaranteed not to be repeated if <count> is not larger than
the number of utterances available.
:param n_frames: The number of frames in the partial utterance.
:return: A list of tuples (utterance, frames, range) where utterance is an Utterance,
frames are the frames of the partial utterances and range is the range of the partial
utterance with regard to the complete utterance.
"""
if self.utterances is None:
self._load_utterances()
utterances = self.utterance_cycler.sample(count)
a = [(u,) + u.random_partial(n_frames) for u in utterances]
return a