Modules

buildstock_fetch.main.fetch_bldg_ids(product, release_year, weather_file, release_version, state, upgrade_id)

Fetch a list of Building ID's

Provided a state, returns a list of building ID's for that state.

Parameters:

Name	Type	Description	Default
product	str	The product type (e.g., 'resstock', 'comstock')	required
release_year	str	The release year (e.g., '2021', '2022')	required
weather_file	str	The weather file type (e.g., 'tmy3')	required
release_version	str	The release version number (e.g., '1')	required
state	str	The state to fetch building ID's for.	required

Returns:

Type	Description
list[BuildingID]	A list of building ID's for the given state.

Source code in buildstock_fetch/main.py

def fetch_bldg_ids(
    product: str, release_year: str, weather_file: str, release_version: str, state: str, upgrade_id: str
) -> list[BuildingID]:
    """Fetch a list of Building ID's

    Provided a state, returns a list of building ID's for that state.

    Args:
        product: The product type (e.g., 'resstock', 'comstock')
        release_year: The release year (e.g., '2021', '2022')
        weather_file: The weather file type (e.g., 'tmy3')
        release_version: The release version number (e.g., '1')
        state: The state to fetch building ID's for.

    Returns:
        A list of building ID's for the given state.
    """

    if product == "resstock":
        product_str = "res"
    elif product == "comstock":
        product_str = "com"
    else:
        raise InvalidProductError(product)

    release_name = f"{product_str}_{release_year}_{weather_file}_{release_version}"
    if not _validate_release_name(release_name):
        raise InvalidReleaseNameError(release_name)

    # Read the specific partition that matches our criteria
    partition_path = (
        METADATA_DIR
        / f"product={product}"
        / f"release_year={release_year}"
        / f"weather_file={weather_file}"
        / f"release_version={release_version}"
        / f"state={state}"
    )

    # Check if the partition exists
    if not partition_path.exists():
        return []

    # Read the parquet files in the specific partition
    df = pl.read_parquet(str(partition_path))

    # No need to filter since we're already reading the specific partition
    filtered_df = df

    # Convert the filtered data to BuildingID objects
    building_ids = []
    for row in filtered_df.iter_rows(named=True):
        building_id = BuildingID(
            bldg_id=int(row["bldg_id"]),
            release_number=release_version,
            release_year=release_year,
            res_com=product,
            weather=weather_file,
            upgrade_id=upgrade_id,
            state=state,
        )
        building_ids.append(building_id)

    return building_ids

buildstock_fetch.main.fetch_bldg_data(bldg_ids, file_type, output_dir, max_workers=5)

Download building data for a given list of building ids

Downloads the data for the given building ids and returns list of paths to the downloaded files.

Parameters:

Name	Type	Description	Default
bldg_ids	list[BuildingID]	A list of BuildingID objects to download data for.	required

Returns:

Type	Description
tuple[list[Path], list[str]]	A list of paths to the downloaded files.

Source code in buildstock_fetch/main.py

def fetch_bldg_data(
    bldg_ids: list[BuildingID], file_type: tuple[str, ...], output_dir: Path, max_workers: int = 5
) -> tuple[list[Path], list[str]]:
    """Download building data for a given list of building ids

    Downloads the data for the given building ids and returns list of paths to the downloaded files.

    Args:
        bldg_ids: A list of BuildingID objects to download data for.

    Returns:
        A list of paths to the downloaded files.
    """
    file_type_obj = _parse_requested_file_type(file_type)
    console = Console()

    downloaded_paths: list[Path] = []
    failed_downloads: list[str] = []

    # Calculate total files to download
    total_files = len(bldg_ids)
    if file_type_obj.metadata:
        total_files += 1  # Add metadata file
    if file_type_obj.load_curve_15min:
        total_files += len(bldg_ids)  # Add 15-minute load curve files
    if file_type_obj.load_curve_annual:
        total_files += len(bldg_ids)  # Add annual load curve files

    console.print(f"\n[bold blue]Starting download of {total_files} files...[/bold blue]")

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TaskProgressColumn(),
        TextColumn("•"),
        DownloadColumn(),
        TextColumn("•"),
        TransferSpeedColumn(),
        TextColumn("•"),
        TimeRemainingColumn(),
        console=console,
        transient=False,
    ) as progress:
        # Download building data if requested.
        if file_type_obj.hpxml or file_type_obj.schedule:
            _download_building_data_parallel(
                bldg_ids, file_type_obj, output_dir, max_workers, progress, downloaded_paths, failed_downloads, console
            )

        # Get metadata if requested. Only one building is needed to get the metadata.
        if file_type_obj.metadata:
            _download_metadata_single(bldg_ids, output_dir, progress, downloaded_paths)

        # Get 15 min load profile timeseries if requested.
        if file_type_obj.load_curve_15min:
            _download_15min_load_curves_parallel(
                bldg_ids, output_dir, max_workers, progress, downloaded_paths, failed_downloads, console
            )

        # Get annual load curve if requested.
        if file_type_obj.load_curve_annual:
            _download_annual_load_curves_parallel(
                bldg_ids, output_dir, max_workers, progress, downloaded_paths, failed_downloads, console
            )

    _print_download_summary(downloaded_paths, failed_downloads, console)

    return downloaded_paths, failed_downloads