Function _parse_data_specifier

espnet2/speechlm/dataloader/iterator.py:348–404

Parse data specifier strings into structured tuples. Args: task_data_factors: Space-separated unregistered data specifiers. Format: "task:name:data_json[:factor]" Example: "asr:librispeech:train.json:2.0" task_registered_data_factors: Space-separated regi

(
    task_data_factors: str,
    task_registered_data_factors: str,
)

def _parse_data_specifier(
    task_data_factors: str,
    task_registered_data_factors: str,
) -> Tuple[List[Tuple], List[Tuple]]:
    """Parse data specifier strings into structured tuples.

    Args:
        task_data_factors: Space-separated unregistered data specifiers.
            Format: "task:name:data_json[:factor]"
            Example: "asr:librispeech:train.json:2.0"
        task_registered_data_factors: Space-separated registered data
            specifiers. Format: "task:name[:factor]"
            Example: "tts:ljspeech:1.5"

    Returns:
        Tuple of (unregistered_list, registered_list) where:
        - unregistered_list: List of (task, name, data_json, factor)
        - registered_list: List of (task, name, factor)

    Raises:
        ValueError: If specifier format is invalid.
    """
    cache_unregistered = []
    if task_data_factors.strip():
        for entry in task_data_factors.split():
            parts = entry.split(":")
            if len(parts) == 4:
                task, name, data_json, factor = parts
                factor = float(factor)
            elif len(parts) == 3:
                task, name, data_json = parts
                factor = 1.0
            else:
                raise ValueError(
                    f"Invalid unregistered specifier '{entry}'. "
                    f"Expected format: 'task:name:data_json[:factor]'"
                )
            cache_unregistered.append((task, name, data_json, factor))

    cache_registered = []
    if task_registered_data_factors.strip():
        for entry in task_registered_data_factors.split():
            parts = entry.split(":")
            if len(parts) == 3:
                task, name, factor = parts
                factor = float(factor)
            elif len(parts) == 2:
                task, name = parts
                factor = 1.0
            else:
                raise ValueError(
                    f"Invalid registered specifier '{entry}'. "
                    f"Expected format: 'task:name[:factor]'"
                )
            cache_registered.append((task, name, factor))

    return cache_unregistered, cache_registered