Function compute_file_hash

modelscope/utils/file_utils.py:381–421

Compute SHA256 hash with automatic optimization for large files. For file paths larger than async_threshold_mb, uses async double-buffered I/O to overlap disk reads with hash computation. For smaller files or non-path inputs (bytes, BinaryIO), falls back to synchronous hashing. Arg

(
    file_path_or_obj: Union[str, Path, bytes, BinaryIO],
    buffer_size_mb: Optional[int] = 16,
    async_threshold_mb: int = 32,
    tqdm_desc: Optional[str] = '[Calculating]',
    disable_tqdm: Optional[bool] = True,
)

def compute_file_hash(
    file_path_or_obj: Union[str, Path, bytes, BinaryIO],
    buffer_size_mb: Optional[int] = 16,
    async_threshold_mb: int = 32,
    tqdm_desc: Optional[str] = '[Calculating]',
    disable_tqdm: Optional[bool] = True,
) -> dict:
    """Compute SHA256 hash with automatic optimization for large files.

    For file paths larger than async_threshold_mb, uses async double-buffered
    I/O to overlap disk reads with hash computation. For smaller files or
    non-path inputs (bytes, BinaryIO), falls back to synchronous hashing.

    Args:
        file_path_or_obj: File path, bytes, or file-like object.
        buffer_size_mb: Read buffer size in MB. Default 16MB for NAS optimization.
        async_threshold_mb: File size threshold for async mode. Default 32MB.
        tqdm_desc: Progress bar description.
        disable_tqdm: Whether to disable progress bar.

    Returns:
        dict with keys: file_path_or_obj, file_hash, file_size.
    """
    # Use async mode for large files accessed by path
    if isinstance(file_path_or_obj, (str, Path)):
        file_size = os.path.getsize(str(file_path_or_obj))
        if file_size >= async_threshold_mb * 1024 * 1024:
            return _get_file_hash_async(
                file_path_or_obj,
                buffer_size_mb=buffer_size_mb,
                tqdm_desc=tqdm_desc,
                disable_tqdm=disable_tqdm,
            )

    # Fallback to synchronous hashing
    return get_file_hash(
        file_path_or_obj,
        buffer_size_mb=buffer_size_mb,
        tqdm_desc=tqdm_desc,
        disable_tqdm=disable_tqdm,
    )


def is_relative_path(url_or_filename: str) -> bool: