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, weather_states=None)

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,
    weather_states: Union[list[str], None] = None,
) -> 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()

    # Initialize weather_states to empty list if None
    if weather_states is None:
        weather_states = []

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

    # Calculate total files to download
    total_files = 0
    if file_type_obj.metadata:
        unique_metadata_urls = _resolve_unique_metadata_urls(bldg_ids)
        total_files += len(unique_metadata_urls)  # 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_hourly:
        total_files += len(bldg_ids)  # Add hourly load curve files
    if file_type_obj.load_curve_daily:
        total_files += len(bldg_ids)  # Add daily load curve files
    if file_type_obj.load_curve_monthly:
        total_files += len(bldg_ids)  # Add monthly load curve files
    if file_type_obj.load_curve_annual:
        total_files += len(bldg_ids)  # Add annual load curve files
    if file_type_obj.weather:
        available_bldg_ids = [bldg_id for bldg_id in bldg_ids if bldg_id.state in weather_states]
        total_files += len(available_bldg_ids) * len(weather_states)  # Add weather map 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:
        _execute_downloads(
            file_type_obj,
            bldg_ids,
            output_dir,
            max_workers,
            progress,
            downloaded_paths,
            failed_downloads,
            console,
            weather_states,
        )

        # TODO: add EV related files
        # TODO: Write a function for downloading EV related files from SB's s3 bucket.
        # It should dynamically build the download url based on the release_name + state combo.
        # Make sure to follow the directory structure for downloading the files.
        if file_type_obj.trip_schedules:
            _download_trip_schedules_data(bldg_ids, output_dir, downloaded_paths)

    _print_download_summary(downloaded_paths, failed_downloads, console)

    return downloaded_paths, failed_downloads