"""
KsponSpeech is a large-scale spontaneous speech corpus of Korean.
This corpus contains 969 hours of open-domain dialog utterances,
spoken by about 2,000 native Korean speakers in a clean environment.
All data were constructed by recording the dialogue of two people
freely conversing on a variety of topics and manually transcribing the utterances.
The transcription provides a dual transcription consisting of orthography and pronunciation,
and disfluency tags for spontaneity of speech, such as filler words, repeated words, and word fragments.
The original audio data has a pcm extension.
During preprocessing, it is converted into a file in the flac extension and saved anew.
KsponSpeech is publicly available on an open data hub site of the Korea government.
The dataset must be downloaded manually.
For more details, please visit:
Dataset: https://aihub.or.kr/aihubdata/data/view.do?currMenu=115&topMenu=100&aihubDataSe=realm&dataSetSn=123
Paper: https://www.mdpi.com/2076-3417/10/19/6936
"""
import logging
import re
from concurrent.futures.thread import ThreadPoolExecutor
from pathlib import Path
from typing import Dict, List, Optional, Sequence, Tuple, Union
import soundfile as sf
from tqdm.auto import tqdm
from lhotse import fix_manifests, validate_recordings_and_supervisions
from lhotse.audio import Recording, RecordingSet
from lhotse.recipes.utils import manifests_exist, read_manifests_if_cached
from lhotse.supervision import SupervisionSegment, SupervisionSet
from lhotse.utils import Pathlike
KSPONSPEECH = (
"train",
"dev",
"eval_clean",
"eval_other",
)
def normalize(
raw_content: str,
normalize_text: str = "default",
) -> Tuple[str, str]:
"""
Normalizing KsponSpeech text datasets with '.trn' extension.
Perform the following processing.
1. Separate file name and text labeling from raw content using separator '::'.
2. Remove noise labeling characters. (e.g. `o/`, `b/`...)
3. Remove the actual pronunciation from the text labeling, Use the spelling content.
4. Remove other special characters and double spaces from text labeling.
:param raw_content: A raw text labeling content containing file name and text labeling.
:param normalize_text: str, the text normalization type. Available options: "default", "none".
:return: A tuple with file name and normalized text labeling.
"""
if len(raw_content) == 0:
return ""
original_content_id, content = raw_content.split(" :: ")
if normalize_text == "none":
return original_content_id, content
elif normalize_text == "default":
content = re.sub(r"[a-z]/", "", content)
content = re.sub(r"\((.*?)\)/\((.*?)\)", r"\1", content)
content = content.replace("*", "")
content = content.replace("+", "")
content = content.replace("/", "")
while " " in content:
content = content.replace(" ", " ")
return original_content_id, content.strip()
[docs]
def prepare_ksponspeech(
corpus_dir: Pathlike,
dataset_parts: Union[str, Sequence[str]] = "all",
output_dir: Optional[Pathlike] = None,
num_jobs: int = 1,
normalize_text: str = "default",
) -> Dict[str, Dict[str, Union[RecordingSet, SupervisionSet]]]:
"""
Returns the manifests which consist of the Recordings and Supervisions.
When all the manifests are available in the ``output_dir``, it will simply read and return them.
:param corpus_dir: Pathlike, the path of the data dir.
:param dataset_parts: string or sequence of strings representing dataset part names, e.g. 'train', 'test'.
By default we will infer which parts are available in ``corpus_dir``.
:param output_dir: Pathlike, the path where to write the manifests.
:param num_jobs: int, number of parallel threads used for 'parse_utterance' calls.
:param normalize_text: str, the text normalization type. Available options: "default", "none".
:return: a Dict whose key is the dataset part, and the value is Dicts with the keys 'audio' and 'supervisions'.
"""
corpus_dir = Path(corpus_dir)
assert corpus_dir.is_dir(), f"No such directory: {corpus_dir}"
if dataset_parts == "all":
dataset_parts = set(KSPONSPEECH)
elif isinstance(dataset_parts, str):
dataset_parts = [dataset_parts]
manifests = {}
if output_dir is not None:
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
# Maybe the manifests already exist: we can read them and save a bit of preparation time.
manifests = read_manifests_if_cached(
dataset_parts=dataset_parts, output_dir=output_dir
)
with ThreadPoolExecutor(num_jobs) as ex:
for part in tqdm(dataset_parts, desc="Dataset parts"):
logging.info(f"Processing KsponSpeech subset: {part}")
if manifests_exist(part=part, output_dir=output_dir):
logging.info(f"KsponSpeech subset: {part} already prepared - skipping.")
continue
recordings = []
supervisions = []
futures = []
trans_path = corpus_dir / f"{part}.trn"
with open(trans_path) as f:
for line in f:
futures.append(
ex.submit(
parse_utterance, corpus_dir, part, line, normalize_text
)
)
for future in tqdm(futures, desc="Processing", leave=False):
result = future.result()
if result is None:
continue
recording, segment = result
recordings.append(recording)
supervisions.append(segment)
recording_set = RecordingSet.from_recordings(recordings)
supervision_set = SupervisionSet.from_segments(supervisions)
recording_set, supervision_set = fix_manifests(
recording_set, supervision_set
)
validate_recordings_and_supervisions(recording_set, supervision_set)
if output_dir is not None:
supervision_set.to_file(
output_dir / f"ksponspeech_supervisions_{part}.jsonl.gz"
)
recording_set.to_file(
output_dir / f"ksponspeech_recordings_{part}.jsonl.gz"
)
manifests[part] = {
"recordings": recording_set,
"supervisions": supervision_set,
}
return manifests
def pcm_to_flac(
pcm_path: Union[str, Path],
flac_path: Union[str, Path],
sample_rate: Optional[int] = 16000,
channels: Optional[int] = 1,
bit_depth: Optional[int] = 16,
) -> Path:
# typecasting
pcm_path = Path(pcm_path)
flac_path = Path(flac_path)
data, _ = sf.read(
pcm_path,
channels=channels,
samplerate=sample_rate,
format="RAW",
subtype="PCM_16",
)
sf.write(flac_path, data, sample_rate, format="FLAC")
return flac_path
def parse_utterance(
corpus_dir: Pathlike,
part: str,
line: str,
normalize_text: str = "default",
) -> Optional[Tuple[Recording, SupervisionSegment]]:
corpus_dir = Path(corpus_dir)
audio_path, normalized_line = normalize(line, normalize_text)
if "eval" in part:
audio_path = audio_path.split("/", maxsplit=1)[1]
audio_path = corpus_dir / audio_path
recording_id = audio_path.stem
# Create the Recording first
if not audio_path.is_file():
logging.warning(f"No such file: {audio_path}")
return None
flac_path = audio_path.with_suffix(".flac")
flac_path = pcm_to_flac(audio_path, flac_path)
recording = Recording.from_file(flac_path, recording_id=recording_id)
# Then, create the corresponding supervisions
segment = SupervisionSegment(
id=recording_id,
recording_id=recording_id,
start=0.0,
duration=recording.duration,
channel=0,
language="Korean",
text=normalized_line,
)
return recording, segment