Function open_datastore_object_with_binary

nemo/utils/data_utils.py:183–234

Open a datastore object and return a file-like object. Args: path: path to an object num_retries: number of retries if the get command fails with ais binary, as AIS Python SDK has its own retry mechanism Returns: File-like object that supports read()

(path: str, num_retries: int = 5)

def open_datastore_object_with_binary(path: str, num_retries: int = 5):
    """Open a datastore object and return a file-like object.

    Args:
        path: path to an object
        num_retries: number of retries if the get command fails with ais binary,
            as AIS Python SDK has its own retry mechanism

    Returns:
        File-like object that supports read()
    """

    if is_datastore_path(path):
        endpoint = ais_endpoint()
        if endpoint is None:
            raise RuntimeError(f'AIS endpoint not set, cannot resolve {path}')

        binary = ais_binary()

        if not binary:
            raise RuntimeError(
                f"AIS binary is not found, cannot resolve {path}. "
                "Please either install it or install Lhotse with `pip install lhotse`.\n"
                "Lhotse's native open_best supports AIS Python SDK, "
                "which is the recommended way to operate with the data from AIStore.\n"
                "See AIS binary installation instructions at "
                "https://github.com/NVIDIA/aistore?tab=readme-ov-file#install-from-release-binaries.\n"
            )

        cmd = [binary, 'get', path, '-']

        done = False

        for _ in range(num_retries):
            with subprocess.Popen(
                cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False  # bytes mode
            ) as proc:
                stream = proc.stdout
                if stream.peek(1):
                    done = True
                    return stream

        if not done:
            with subprocess.Popen(
                cmd, shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=False
            ) as proc:
                error = proc.stderr.read().decode("utf-8", errors="ignore").strip()
            raise ValueError(
                f"{path} couldn't be opened with AIS binary "
                f"after {num_retries} attempts because of the following exception: {error}"
            )
    return None


def open_best(path: str, mode: str = "rb"):