| | import csv |
| | import random |
| | from collections import defaultdict |
| | from pathlib import Path |
| |
|
| | import click |
| | import yaml |
| |
|
| |
|
| | |
| | @click.command(help='Randomly select test samples') |
| | @click.argument( |
| | 'config', |
| | type=click.Path(file_okay=True, dir_okay=False, resolve_path=True, writable=True, path_type=Path), |
| | metavar="CONFIG" |
| | ) |
| | @click.option( |
| | '--rel_path', |
| | type=click.Path(file_okay=False, dir_okay=True, resolve_path=True, path_type=Path), |
| | default=None, |
| | help='Path that is relative to the paths mentioned in the config file.' |
| | ) |
| | @click.option( |
| | '--min', '_min', |
| | show_default=True, |
| | type=click.IntRange(min=1), |
| | default=10, |
| | help='Minimum number of test samples.' |
| | ) |
| | @click.option( |
| | '--max', '_max', |
| | show_default=True, |
| | type=click.IntRange(min=1), |
| | default=20, |
| | help='Maximum number of test samples (note that each speaker will have at least one test sample).' |
| | ) |
| | @click.option( |
| | '--per_speaker', |
| | show_default=True, |
| | type=click.IntRange(min=1), |
| | default=4, |
| | help='Expected number of test samples per speaker.' |
| | ) |
| | def select_test_set(config, rel_path, _min, _max, per_speaker): |
| | assert _min <= _max, 'min must be smaller or equal to max' |
| | with open(config, 'r', encoding='utf8') as f: |
| | hparams = yaml.safe_load(f) |
| |
|
| | spk_map = None |
| | spk_ids = hparams['spk_ids'] |
| | speakers = hparams['speakers'] |
| | raw_data_dirs = list(map(Path, hparams['raw_data_dir'])) |
| | assert isinstance(speakers, list), 'Speakers must be a list' |
| | assert len(speakers) == len(raw_data_dirs), \ |
| | 'Number of raw data dirs must equal number of speaker names!' |
| | if not spk_ids: |
| | spk_ids = list(range(len(raw_data_dirs))) |
| | else: |
| | assert len(spk_ids) == len(raw_data_dirs), \ |
| | 'Length of explicitly given spk_ids must equal the number of raw datasets.' |
| | assert max(spk_ids) < hparams['num_spk'], \ |
| | f'Index in spk_id sequence {spk_ids} is out of range. All values should be smaller than num_spk.' |
| |
|
| | spk_map = {} |
| | path_spk_map = defaultdict(list) |
| | for ds_id, (spk_name, raw_path, spk_id) in enumerate(zip(speakers, raw_data_dirs, spk_ids)): |
| | if spk_name in spk_map and spk_map[spk_name] != spk_id: |
| | raise ValueError(f'Invalid speaker ID assignment. Name \'{spk_name}\' is assigned ' |
| | f'with different speaker IDs: {spk_map[spk_name]} and {spk_id}.') |
| | spk_map[spk_name] = spk_id |
| | path_spk_map[spk_id].append((ds_id, rel_path / raw_path if rel_path else raw_path)) |
| |
|
| | training_cases = [] |
| | for spk_raw_dirs in path_spk_map.values(): |
| | training_case = [] |
| | |
| | for ds_id, raw_data_dir in spk_raw_dirs: |
| | with open(raw_data_dir / 'transcriptions.csv', 'r', encoding='utf8') as f: |
| | reader = csv.DictReader(f) |
| | for row in reader: |
| | if (raw_data_dir / 'wavs' / f'{row["name"]}.wav').exists(): |
| | training_case.append(f'{ds_id}:{row["name"]}') |
| | training_cases.append(training_case) |
| |
|
| | test_prefixes = [] |
| | total = min(_max, max(_min, per_speaker * len(training_cases))) |
| | quotient, remainder = total // len(training_cases), total % len(training_cases) |
| | if quotient == 0: |
| | test_counts = [1] * len(training_cases) |
| | else: |
| | test_counts = [quotient + 1] * remainder + [quotient] * (len(training_cases) - remainder) |
| | for i, count in enumerate(test_counts): |
| | test_prefixes += sorted(random.sample(training_cases[i], count)) |
| | if not hparams['test_prefixes'] or click.confirm('Overwrite existing test prefixes?', abort=False): |
| | hparams['test_prefixes'] = test_prefixes |
| | hparams['num_valid_plots'] = len(test_prefixes) |
| | with open(config, 'w', encoding='utf8') as f: |
| | yaml.dump(hparams, f, sort_keys=False) |
| | print('Test prefixes saved.') |
| | else: |
| | print('Test prefixes not saved, aborted.') |
| |
|
| | if __name__ == '__main__': |
| | select_test_set() |
| |
|
