Resource Assessments Values

ResourceAssessmentValues(perfdb)

Class used for handling resource assessment values. Can be accessed via perfdb.resourceassessments.values.

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/perfdb_root.py

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Base class that all subclasses should inherit from.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.

    """
    self._perfdb: e_pg.PerfDB = perfdb

get(spe_names=None, resource_type_names=None, resource_assessment_names=None, resource_assessment_types=None, companies=None, default=None, filter_type='and', output_type='dict')

Gets all resource assessment values definitions with detailed information.

The most useful keys/columns returned are:

• resource_assessment_type_name
• company_name
• p50
• std_longterm
• std_1year
• std_1month
• std_1day
• reference_date
• is_default

Parameters:

• spe_names

  (list[str] | None, default: None ) –

  List of SPE names to filter the query. If None, no filter will be applied. By default None

• resource_type_names

  (list[str] | None, default: None ) –

  List of resource type names to filter the query. If None, no filter will be applied. By default None

• resource_assessment_names

  (list[str] | None, default: None ) –

  List of resource assessment values to filter the query. If None, no filter will be applied. By default None

• resource_assessment_types

  (list[str] | None, default: None ) –

  List of resource assessment types to filter the query. If None, no filter will be applied. By default None

• companies

  (list[str] | None, default: None ) –

  List of companies to filter the query. If None, no filter will be applied. By default None

• filter_type

  (Literal['and', 'or'], default: 'and' ) –

  Type of filter to apply. Can be one of ["and", "or"]. By default "and"

• default

  (bool | None, default: None ) –

  If True, only default values will be returned. If False, only non-default values will be returned. If None, all values will be returned. By default None

• output_type

  (Literal['dict', 'DataFrame'], default: 'dict' ) –

  Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"

Returns:

• dict[str, dict[str, dict[str, Any]]] –

  In case output_value is "dict", returns a dictionary in the format {resource_assessment_name: {spe_name: {resource_type_name: {attribute: value, ...}, ...}, ...}, ...}

• DataFrame –

  In case output_value is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[resource_assessment_name, spe_name, resource_type_name], columns = [attribute, ...]

Source code in echo_postgres/resourceassessment_values.py

@validate_call
def get(
    self,
    spe_names: list[str] | None = None,
    resource_type_names: list[str] | None = None,
    resource_assessment_names: list[str] | None = None,
    resource_assessment_types: list[str] | None = None,
    companies: list[str] | None = None,
    default: bool | None = None,
    filter_type: Literal["and", "or"] = "and",
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, dict[str, Any]]] | DataFrame:
    """Gets all resource assessment values definitions with detailed information.

    The most useful keys/columns returned are:

    - resource_assessment_type_name
    - company_name
    - p50
    - std_longterm
    - std_1year
    - std_1month
    - std_1day
    - reference_date
    - is_default

    Parameters
    ----------
    spe_names : list[str] | None
        List of SPE names to filter the query. If None, no filter will be applied. By default None
    resource_type_names : list[str] | None
        List of resource type names to filter the query. If None, no filter will be applied. By default None
    resource_assessment_names : list[str] | None
        List of resource assessment values to filter the query. If None, no filter will be applied. By default None
    resource_assessment_types : list[str] | None
        List of resource assessment types to filter the query. If None, no filter will be applied. By default None
    companies : list[str] | None
        List of companies to filter the query. If None, no filter will be applied. By default None
    filter_type : Literal["and", "or"], optional
        Type of filter to apply. Can be one of ["and", "or"]. By default "and"
    default : bool | None, optional
        If True, only default values will be returned. If False, only non-default values will be returned. If None, all values will be returned. By default None
    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "dict"

    Returns
    -------
    dict[str, dict[str, dict[str, Any]]]
        In case output_value is "dict", returns a dictionary in the format {resource_assessment_name: {spe_name: {resource_type_name: {attribute: value, ...}, ...}, ...}, ...}
    DataFrame
        In case output_value is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[resource_assessment_name, spe_name, resource_type_name], columns = [attribute, ...]
    """
    if output_type not in ["dict", "DataFrame"]:
        raise ValueError(f"output_type must be one of ['dict', 'DataFrame'], not {output_type}")

    # validating arguments and building WHERE clause
    where = self._check_get_args(
        spe_names=spe_names,
        resource_type_names=resource_type_names,
        resource_assessment_names=resource_assessment_names,
        resource_assessment_types=resource_assessment_types,
        companies=companies,
        default=default,
        filter_type=filter_type,
    )

    query = [sql.SQL("SELECT * FROM performance.v_resource_assessment_values ")]
    if where:
        query.append(where)
    query.append(sql.SQL(" ORDER BY resource_assessment_name, spe_name, resource_type_name"))
    query = sql.Composed(query)

    with self._perfdb.conn.reconnect() as conn:
        # using post_convert="pyarrow" to avoid issues with arrays being converted to strings when getting data
        df = conn.read_to_pandas(query=query, post_convert="pyarrow")
    df = df.set_index(["resource_assessment_name", "spe_name", "resource_type_name"])

    if output_type == "DataFrame":
        return df

    # converting to Dict
    result = df.to_dict(orient="index")
    final_result = {}
    for (resource_assessment_name, spe_name, resource_type_name), data in result.items():
        if resource_assessment_name not in final_result:
            final_result[resource_assessment_name] = {}
        if spe_name not in final_result[resource_assessment_name]:
            final_result[resource_assessment_name][spe_name] = {}
        final_result[resource_assessment_name][spe_name][resource_type_name] = data

    return final_result