Data Preparation

dataset_root_dir/version=0/
├── corpus=mls/
│   ├── split=train/
│   │   ├── language=deu_Latn/
│   │   │   └── part-*.parquet
│   │   └── language=fra_Latn/
│   │       └── part-*.parquet
│   └── split=dev/
│       └── ...
└── corpus=fleurs/
    └── ...

text: string

audio_bytes: list<element: int8>
  child 0, element: int8

audio_size: int64

corpus: dictionary<values=string, indices=int32, ordered=0>

split: dictionary<values=string, indices=int32, ordered=0>

language: dictionary<values=string, indices=int32, ordered=0>

import ray
from datasets import load_dataset
from workflows.dataprep.audio_tools import AudioTableProcessor, map_to_target_schema
from workflows.dataprep.text_tools import text_normalize

# Initialize Ray
ray.init()

# Load dataset
ds = load_dataset(
    "mozilla-foundation/common_voice_11_0",
    "en",
    split="train"
)

# Convert to Ray dataset
ray_ds = ray.data.from_huggingface(ds)

# Define processing pipeline
def process_batch(batch):
    # Text normalization
    batch["text"] = [
        text_normalize(text, iso_code="en")
        for text in batch["sentence"]
    ]
    
    # Audio processing
    audio_processor = AudioTableProcessor()
    batch = audio_processor(batch)
    
    # Map to target schema
    batch = map_to_target_schema(
        batch,
        split="train",
        corpus="common_voice"
    )
    
    # Add language column
    batch["language"] = ["eng_Latn"] * len(batch)
    
    return batch

# Process and write
ray_ds.map_batches(process_batch).write_parquet(
    "/path/to/output/version=0",
    partition_cols=["corpus", "split", "language"],
    row_group_size=100
)

import pandas as pd
import pyarrow as pa
from pathlib import Path
from workflows.dataprep.audio_tools import AudioTableProcessor
from workflows.dataprep.text_tools import text_normalize

# Prepare data
audio_files = list(Path("/path/to/audio").glob("*.wav"))
transcripts = [...]  # Load corresponding transcripts

# Create DataFrame
df = pd.DataFrame({
    "audio_path": audio_files,
    "transcript": transcripts
})

# Process
processor = AudioTableProcessor()

def process_row(row):
    # Read audio
    with open(row["audio_path"], "rb") as f:
        audio_bytes = list(f.read())
    
    # Normalize text
    text = text_normalize(row["transcript"], iso_code="en")
    
    return {
        "audio_bytes": audio_bytes,
        "audio_size": len(audio_bytes) * 16000,  # Approximate
        "text": text,
        "corpus": "my_corpus",
        "split": "train",
        "language": "eng_Latn"
    }

processed_df = df.apply(process_row, axis=1)

# Write parquet
table = pa.Table.from_pandas(processed_df)
pa.parquet.write_to_dataset(
    table,
    root_path="/path/to/output/version=0",
    partition_cols=["corpus", "split", "language"],
    row_group_size=100
)

from omnilingual_asr.datasets.impl.mixture_parquet_asr_dataset import MixtureParquetAsrDataset
from fairseq2.models.tokenizers.hub import load_tokenizer

# Create dataset
dataset = MixtureParquetAsrDataset.from_path(
    path="/path/to/dataset/version=0",
    name="my_asr_dataset"
)

# Load tokenizer
tokenizer = load_tokenizer("omniASR_tokenizer_v1")

# Create reader
reader = dataset.create_reader(
    split="train",
    tokenizer=tokenizer,
    gangs=gangs,
    dtype=torch.float32,
    storage_config=storage_config,
    task_config=task_config,
)

# Iterate through batches
for batches in reader:
    for batch in batches:
        # batch.source_seqs: audio features [batch, time, features]
        # batch.target_seqs: text tokens [batch, seq_len]
        # batch.source_seq_lens: audio sequence lengths
        # batch.target_seq_lens: text sequence lengths
        train_step(batch)

python -m workflows.dataprep.dataloader_example \
  --dataset_path="root_ds/all_asr/version=0" \
  --split="train" \
  --num_iterations=10

import pyarrow.parquet as pq

# Load dataset
ds = pq.ParquetDataset("/path/to/dataset/version=0")

# Check partitions
print("Partitions:", ds.partitions)
print("Files:", len(ds.files))

# Read sample
table = ds.read()
print("Schema:", table.schema)
print("Num rows:", len(table))
print("Sample:", table.to_pandas().head())

# Verify statistics
stats = pd.read_csv("/path/to/language_distribution_0.tsv", sep="\t")
print("\nLanguage distribution:")
print(stats)

@article{conneau2022fleurs,
  title={FLEURS: Few-shot Learning Evaluation of Universal Representations of Speech},
  author={Conneau, Alexis and Ma, Min and Khanuja, Simran and Zhang, Yu and Axelrod, Vera and Dalmia, Siddharth and Riesa, Jason and Rivera, Clara and Bapna, Ankur},
  journal={arXiv preprint arXiv:2205.12446},
  year={2022}
}

@article{pratap2020mls,
  title={MLS: A Large-Scale Multilingual Dataset for Speech Research},
  author={Pratap, Vineel and Xu, Qiantong and Sriram, Anuroop and Synnaeve, Gabriel and Collobert, Ronan},
  journal={arXiv preprint arXiv:2012.03411},
  year={2020}
}