Function predownload_toolchains

ci/util/toolchain_downloader.py:203–293

Pre-download all toolchains from a platformio.ini file. Args: platformio_ini_path: Path to platformio.ini file cache_dir: Cache directory (defaults to ~/.platformio/global_cache) max_workers: Maximum number of concurrent downloads Returns: Dictionary ma

(
    platformio_ini_path: Path,
    cache_dir: Optional[Path] = None,
    max_workers: int = 4,
)

def predownload_toolchains(
    platformio_ini_path: Path,
    cache_dir: Optional[Path] = None,
    max_workers: int = 4,
) -> dict[str, ToolchainDownloadResult]:
    """
    Pre-download all toolchains from a platformio.ini file.

    Args:
        platformio_ini_path: Path to platformio.ini file
        cache_dir: Cache directory (defaults to ~/.platformio/global_cache)
        max_workers: Maximum number of concurrent downloads

    Returns:
        Dictionary mapping package names to download results
    """
    if cache_dir is None:
        cache_dir = Path.home() / ".platformio" / "global_cache"

    cache_dir.mkdir(parents=True, exist_ok=True)

    logger.info(f"🔍 Extracting toolchains from: {platformio_ini_path}")
    logger.info(f"📁 Cache directory: {cache_dir}")

    # Extract all packages
    packages = extract_toolchains_from_platformio_ini(platformio_ini_path)

    if not packages:
        logger.warning("No packages found in platformio.ini")
        return {}

    logger.info(f"Found {len(packages)} total packages")

    # Deduplicate by name and version
    unique_packages: dict[str, PackageInfo] = {}
    for package in packages:
        key = f"{package.name}:{package.version or 'latest'}"
        if key not in unique_packages:
            unique_packages[key] = package

    logger.info(f"Processing {len(unique_packages)} unique packages")

    # Download toolchains concurrently
    results: dict[str, ToolchainDownloadResult] = {}
    downloader = ResumableDownloader(chunk_size=8192, max_retries=5)

    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(download_toolchain, package, cache_dir, downloader): package
            for package in unique_packages.values()
        }

        for future in as_completed(futures):
            package = futures[future]
            try:
                result = future.result()
                results[package.name] = result