utils.ditto_utils

Utility functions for dealing with SMART DS dataset.

Examples:

>>> from erad import ditto_utils
>>> ditto_utils.download_smartds_data('P4R',  '.')

create_networkx_from_ditto(output_path, file_name, **kwargs)

Creates networkx graph from OpenDSS model using Ditto.

Parameters:

Name	Type	Description	Default
output_path	str	Path to store the networkx data in json file format	required
file_name	str	JSON file name used to export the network	required
kwargs	dict	Keyword arguments accepted by Ditto	{}

Source code in erad\utils\ditto_utils.py

def create_networkx_from_ditto(
    output_path: str, file_name: str, **kwargs
) -> None:
    """Creates networkx graph from OpenDSS model using Ditto.

    Args:
        output_path (str): Path to store the networkx
            data in json file format
        file_name (str): JSON file name used to export
            the network
        kwargs (dict): Keyword arguments accepted
            by Ditto
    """
    try:
        output_path = Path(output_path)
        return _create_networkx_from_ditto(output_path, file_name, **kwargs)
    finally:
        for file_path in output_path.iterdir():
            if file_path.suffix == ".adjlist":
                file_path.unlink(missing_ok=True)

create_networkx_from_json(json_file_path)

Returns networkx graph from JSON file.

Source code in erad\utils\ditto_utils.py

def create_networkx_from_json(json_file_path: str):
    """Returns networkx graph from JSON file."""
    content = read_file(json_file_path)
    return json_graph.adjacency_graph(content)

download_aws_dir(bucket, path, target, unsigned=True, **kwargs)

Utility function download data from AWS S3 directory.

Parameters:

Name	Type	Description	Default
bucket	str	Name of the bucket.	required
path	str	S3 bucket prefix	required
target	str	Path for downloading the data	required
unsigned	bool	Indicate whether to use credential or not	True
kwargs	dict	Keyword arguments accepted by boto3.client	{}

Source code in erad\utils\ditto_utils.py

@timeit
def download_aws_dir(
    bucket: str, path: str, target: str, unsigned=True, **kwargs
) -> None:
    """Utility function download data from AWS S3 directory.

    Args:
        bucket (str): Name of the bucket.
        path (str): S3 bucket prefix
        target (str): Path for downloading the data
        unsigned (bool): Indicate whether to use credential or not
        kwargs (dict): Keyword arguments accepted by `boto3.client`
    """

    target = Path(target)
    if unsigned:
        client = boto3.client("s3", config=Config(signature_version=UNSIGNED))
    else:
        if kwargs:
            client = boto3.client("s3", **kwargs)
        else:
            client = boto3.client("s3")

    # Handle missing / at end of prefix
    if not path.endswith("/"):
        path += "/"

    paginator = client.get_paginator("list_objects_v2")
    for result in paginator.paginate(Bucket=bucket, Prefix=path):

        # Download each file individually
        for key in result["Contents"]:

            # Calculate relative path
            rel_path = key["Key"][len(path) :]

            # Skip paths ending in /
            if not key["Key"].endswith("/"):
                local_file_path = target / rel_path
                local_file_path.parent.mkdir(parents=True, exist_ok=True)
                client.download_file(bucket, key["Key"], str(local_file_path))

download_smartds_data(smartds_region, output_path='./smart_ds_downloads', year=2018, area='SFO', s3_bucket_name='oedi-data-lake', folder_name='opendss_no_loadshapes', cache_folder='cache')

Utility function to download SMARTDS data from AWS S3 bucket.

Parameters:

Name	Type	Description	Default
smartds_region	str	SMARTDS region name	required
output_path	str	Path for downloaded data	'./smart_ds_downloads'
year	int	Valid year input for downloading the data	2018
area	str	Valid SMARTDS area	'SFO'
s3_bucket_name	str	S3 bucket name storing the SMARTDS data	'oedi-data-lake'
folder_name	str	S3 bucket folder to download	'opendss_no_loadshapes'
cache_folder	str	Folder path for caching the results	'cache'

Raises:

Type	Description
SMARTDSInvalidInput	Raises this error if year and/or area provided is not valid.

Returns:

Name	Type	Description
str	str	Folder path containing downloaded data.

Source code in erad\utils\ditto_utils.py

@timeit
def download_smartds_data(
    smartds_region: str,
    output_path: str = "./smart_ds_downloads",
    year: int = 2018,
    area: str = "SFO",
    s3_bucket_name: str = "oedi-data-lake",
    folder_name: str = "opendss_no_loadshapes",
    cache_folder: str = "cache",
) -> str:
    """Utility function to download SMARTDS data from AWS S3 bucket.

    Args:
        smartds_region (str): SMARTDS region name
        output_path (str): Path for downloaded data
        year (int): Valid year input for downloading the data
        area (str): Valid SMARTDS area
        s3_bucket_name (str): S3 bucket name storing the SMARTDS data
        folder_name (str): S3 bucket folder to download
        cache_folder (str): Folder path for caching the results

    Raises:
        SMARTDSInvalidInput: Raises this error if year and/or area
            provided is not valid.

    Returns:
        str: Folder path containing downloaded data.
    """
    if year not in SMARTDS_VALID_YEARS or area not in SMARTDS_VALID_AREAS:
        raise SMARTDSInvalidInput(
            f"Not valid input! year= {year} area={area}, \
            valid_years={SMARTDS_VALID_YEARS}, valid_areas={SMARTDS_VALID_AREAS}"
        )

    output_path = Path(output_path)
    cache_folder = Path(cache_folder)

    output_path.mkdir(exist_ok=True)
    cache_folder.mkdir(exist_ok=True)

    cache_key = (
        f"{smartds_region}__{year}__{area}__{s3_bucket_name}_{folder_name}"
    )
    cache_data_folder = cache_folder / cache_key
    output_folder = output_path / cache_key

    if cache_data_folder.exists():
        logger.info(f"Cache hit for {cache_data_folder}")
        shutil.copytree(cache_data_folder, output_folder, dirs_exist_ok=True)

    else:
        logger.info(
            f"Cache missed reaching to AWS for downloading the data ..."
        )
        output_folder.mkdir(exist_ok=True)
        prefix = f"SMART-DS/v1.0/{year}/{area}/{smartds_region}/scenarios/base_timeseries/{folder_name}/"
        download_aws_dir(s3_bucket_name, prefix, output_folder)
        shutil.copytree(output_folder, cache_data_folder, dirs_exist_ok=False)

    logger.info(f"Check the folder {output_folder} for downloaded data")
    return output_folder