Calculation Model Instances

CalcModelInstances(perfdb)

Class used for handling calculation model instances. Can be accessed via perfdb.calcmodels.instances.files.

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/calcmodel_instances.py

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Class used for handling calculation model instances.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.
    """
    super().__init__(perfdb)

    from .calcmodel_instance_files import CalcModelInstanceFiles

    # * subclasses

    self.files = CalcModelInstanceFiles(perfdb)

delete(calc_model_name, calc_model_type_name)

Deletes a calculation model

Parameters:

• calc_model_name

  (str) –

  Name of the calculation model

• calc_model_type_name

  (str) –

  Type of the calculation model

Source code in echo_postgres/calcmodel_instances.py

@validate_call
def delete(
    self,
    calc_model_name: str,
    calc_model_type_name: str,
) -> None:
    """Deletes a calculation model

    Parameters
    ----------
    calc_model_name : str
        Name of the calculation model
    calc_model_type_name : str
        Type of the calculation model
    """
    # getting id of the calc model type
    model_type_ids = self._perfdb.calcmodels.types.get_ids()
    if calc_model_type_name not in model_type_ids:
        raise ValueError(f"calc_model_type_name {calc_model_type_name} does not exist in the database")

    query = sql.SQL(
        "DELETE FROM performance.calculation_models WHERE name = {name} AND calc_model_type_id = {calc_model_type_id}",
    ).format(
        name=sql.Literal(calc_model_name),
        calc_model_type_id=sql.Literal(model_type_ids[calc_model_type_name]),
    )

    with self._perfdb.conn.reconnect() as conn:
        # deleting
        result = conn.execute(query)

    logger.debug(f"Deleted {result.rowcount} rows from performance.calculation_models")

get(calcmodel_types=None, output_type='dict')

Gets all calculation model instances definitions with detailed information.

The most useful keys/columns returned are:

• calc_model_type_id
• calc_model_type_name
• id
• display_name
• description
• calc_options

Parameters:

• calcmodel_types

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

  Names of the calculation model types to filter the results. If None, no filter is applied. 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, Any]] –

  In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}

• DataFrame –

  In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]

Source code in echo_postgres/calcmodel_instances.py

@validate_call
def get(
    self,
    calcmodel_types: list[str] | None = None,
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | DataFrame:
    """Gets all calculation model instances definitions with detailed information.

    The most useful keys/columns returned are:

    - calc_model_type_id
    - calc_model_type_name
    - id
    - display_name
    - description
    - calc_options

    Parameters
    ----------
    calcmodel_types : list[str] | None, optional
        Names of the calculation model types to filter the results. If None, no filter is applied.
        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, Any]]
        In case output_type is "dict", returns a dictionary in the format {name: {attribute: value, ...}, ...}
    DataFrame
        In case output_type is "DataFrame", returns a DataFrame with the following format: index = name, columns = [attribute, ...]
    """
    # checking arguments
    where = self._check_get_args(calcmodel_types)
    if output_type not in ["dict", "DataFrame"]:
        raise ValueError(f"output_type must be one of ['dict', 'DataFrame'], not {output_type}")

    query = [sql.SQL("SELECT * FROM performance.v_calculation_models "), where, sql.SQL(" ORDER BY name")]
    query = sql.Composed(query)

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query, post_convert="pyarrow")
    df = df.set_index("name")

    return df.to_dict(orient="index") if output_type == "dict" else df

get_ids(calcmodel_types=None)

Gets all calculation model instances and their respective ids.

Parameters:

• calcmodel_types

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

  Names of the calculation model types to filter the results. If None, no filter is applied. By default None

Returns:

• dict[str, int] –

  Dictionary with all calculation model instances and their respective ids in the format {calcmodel_type: id, ...}.

Source code in echo_postgres/calcmodel_instances.py

@validate_call
def get_ids(self, calcmodel_types: list[str] | None = None) -> dict[str, int]:
    """Gets all calculation model instances and their respective ids.

    Parameters
    ----------
    calcmodel_types : list[str] | None, optional
        Names of the calculation model types to filter the results. If None, no filter is applied.
        By default None

    Returns
    -------
    dict[str, int]
        Dictionary with all calculation model instances and their respective ids in the format {calcmodel_type: id, ...}.
    """
    # checking arguments
    where = self._check_get_args(calcmodel_types)

    query = [sql.SQL("SELECT name, id FROM performance.v_calculation_models "), where, sql.SQL(" ORDER BY name")]
    query = sql.Composed(query)

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query)

    return df.set_index("name").to_dict()["id"]

insert(calc_model_name, calc_model_type_name, display_name=None, description=None, calc_options=None, on_conflict='ignore')

Inserts a calculation model into the database.

Parameters:

• calc_model_name

  (str) –

  Name of the calculation model.

• calc_model_type_name

  (str) –

  Name of the calculation model type.

• display_name

  (str | None, default: None ) –

  Display name of the calculation model. By default, None.

• description

  (str | None, default: None ) –

  Description of the calculation model. By default, None.

• calc_options

  (dict[str, Any] | None, default: None ) –

  Options of the calculation model. By default, None.

• on_conflict

  (Literal['ignore', 'update'], default: 'ignore' ) –

  What to do in case of conflict. Can be one of ["ignore", "update"]. By default "ignore""

Source code in echo_postgres/calcmodel_instances.py

@validate_call
def insert(
    self,
    calc_model_name: str,
    calc_model_type_name: str,
    display_name: str | None = None,
    description: str | None = None,
    calc_options: dict[str, Any] | None = None,
    on_conflict: Literal["ignore", "update"] = "ignore",
) -> None:
    """Inserts a calculation model into the database.

    Parameters
    ----------
    calc_model_name : str
        Name of the calculation model.
    calc_model_type_name : str
        Name of the calculation model type.
    display_name : str | None
        Display name of the calculation model. By default, None.
    description : str | None
        Description of the calculation model. By default, None.
    calc_options : dict[str, Any] | None
        Options of the calculation model. By default, None.
    on_conflict : Literal["ignore", "update"], optional
        What to do in case of conflict. Can be one of ["ignore", "update"].
        By default "ignore""
    """
    # getting id of the calc model type
    model_type_ids = self._perfdb.calcmodels.types.get_ids()
    if calc_model_type_name not in model_type_ids:
        raise ValueError(f"calc_model_type_name {calc_model_type_name} does not exist in the database")

    # creating DataFrame to be inserted
    df = DataFrame(
        data={
            "name": [calc_model_name],
            "calc_model_type_id": [model_type_ids[calc_model_type_name]],
            "display_name": [display_name],
            "description": [description],
            "calc_options": [calc_options],
        },
    )
    self._perfdb.conn.pandas_to_sql(
        df,
        table_name="calculation_models",
        schema="performance",
        if_exists="append" if on_conflict == "ignore" else on_conflict,
        ignore_index=True,
        conflict_cols=["name"],
    )

    logger.debug(f"Calculation model '{calc_model_name}' with type '{calc_model_type_name}' inserted")