Calculation Model Instances - Files Definitions¶
CalcModelInstanceFileDefinitions(perfdb)
¶
Class used for handling calculation model instance file definitions. Can be accessed via perfdb.calcmodels.instances.files.definitions.
Parameters:
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
delete(calculation_model_name, file_name)
¶
Deletes a calculation model instance file definition.
Parameters:
-
(calculation_model_name¶str) –Name of the calculation model instance.
-
(file_name¶str) –Name of the file.
Source code in echo_postgres/calcmodel_instance_file_definitions.py
@validate_call
def delete(
self,
calculation_model_name: str,
file_name: str,
) -> None:
"""Deletes a calculation model instance file definition.
Parameters
----------
calculation_model_name : str
Name of the calculation model instance.
file_name : str
Name of the file.
"""
# getting file_def_id
file_def_ids = self.get_ids(calcmodels=[calculation_model_name], file_names=[file_name])
if calculation_model_name not in file_def_ids or file_name not in file_def_ids[calculation_model_name]:
return
query = sql.SQL(
"DELETE FROM performance.calculation_model_files_def WHERE file_def_id = {file_def_id}",
).format(
file_def_id=sql.Literal(file_def_ids[calculation_model_name][file_name]),
)
with self._perfdb.conn.reconnect() as conn:
# deleting
result = conn.execute(query)
logger.debug(f"Deleted {result.rowcount} rows from performance.calculation_model_files_def")
get(calcmodel_types=None, calcmodels=None, file_names=None, filter_type='and', model_as_regex=False, output_type='dict')
¶
Gets all calculation model instance file definitions with detailed information.
The most useful keys/columns returned are:
- file_def_id
- description
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
-
(calcmodels¶list[str] | None, default:None) –Names of the calculation model instances to filter the results. If None, no filter is applied. By default None
-
(file_names¶list[str] | None, default:None) –Names of the files to filter the results. If None, no filter is applied. By default None
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"]. By default "and"
-
(model_as_regex¶bool, default:False) –If True, calcmodel_types and calcmodels filters will be treated as regex. Can only be used if a single value is passed to each filter. By default False
-
(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_type is "dict", returns a dictionary in the format {calculation_model_name: {file_name: {attribute: value, ...}, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[calculation_model_name, file_name], columns = [attribute, ...]
Source code in echo_postgres/calcmodel_instance_file_definitions.py
@validate_call
def get(
self,
calcmodel_types: list[str] | None = None,
calcmodels: list[str] | None = None,
file_names: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
model_as_regex: bool = False,
output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, dict[str, Any]]] | DataFrame:
"""Gets all calculation model instance file definitions with detailed information.
The most useful keys/columns returned are:
- file_def_id
- description
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
calcmodels : list[str] | None, optional
Names of the calculation model instances to filter the results. If None, no filter is applied.
By default None
file_names: list[str] | None, optional
Names of the files to filter the results. If None, no filter is applied.
By default None
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"].
By default "and"
model_as_regex : bool, optional
If True, calcmodel_types and calcmodels filters will be treated as regex. Can only be used if a single value is passed to each filter.
By default False
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_type is "dict", returns a dictionary in the format {calculation_model_name: {file_name: {attribute: value, ...}, ...}, ...}
DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[calculation_model_name, file_name], columns = [attribute, ...]
"""
# checking arguments
where = self._check_get_args(
calcmodel_types=calcmodel_types,
calcmodels=calcmodels,
file_names=file_names,
filter_type=filter_type,
model_as_regex=model_as_regex,
)
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_model_files_def "),
where,
sql.SQL(" ORDER BY calculation_model_name, file_name"),
]
query = sql.Composed(query)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query)
df = df.set_index(["calculation_model_name", "file_name"])
if output_type == "DataFrame":
return df
# removing not wanted columns
df = df.drop(columns=["calc_model_id", "calc_model_type_id", "calc_model_type_name"], errors="ignore")
result = df.to_dict(orient="index")
final_result = {}
for (calc_model, file_name), file_def in result.items():
if calc_model not in final_result:
final_result[calc_model] = {}
final_result[calc_model][file_name] = file_def
return final_result
get_ids(calcmodel_types=None, calcmodels=None, file_names=None, filter_type='and', model_as_regex=False)
¶
Gets all calculation model instance files 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
-
(calcmodels¶list[str] | None, default:None) –Names of the calculation model instances to filter the results. If None, no filter is applied. By default None
-
(file_names¶list[str] | None, default:None) –Names of the files to filter the results. If None, no filter is applied. By default None
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"]. By default "and"
-
(model_as_regex¶bool, default:False) –If True, calcmodel_types and calcmodels filters will be treated as regex. Can only be used if a single value is passed to each filter. By default False
Returns:
-
dict[str, dict[str, int]]–Dictionary with all calculation model instance files definitions and their respective ids in the format {calc_model: {file_name: id, ...}, ...}.
Source code in echo_postgres/calcmodel_instance_file_definitions.py
@validate_call
def get_ids(
self,
calcmodel_types: list[str] | None = None,
calcmodels: list[str] | None = None,
file_names: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
model_as_regex: bool = False,
) -> dict[str, dict[str, int]]:
"""Gets all calculation model instance files 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
calcmodels : list[str] | None, optional
Names of the calculation model instances to filter the results. If None, no filter is applied.
By default None
file_names: list[str] | None, optional
Names of the files to filter the results. If None, no filter is applied.
By default None
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"].
By default "and"
model_as_regex : bool, optional
If True, calcmodel_types and calcmodels filters will be treated as regex. Can only be used if a single value is passed to each filter.
By default False
Returns
-------
dict[str, dict[str, int]]
Dictionary with all calculation model instance files definitions and their respective ids in the format {calc_model: {file_name: id, ...}, ...}.
"""
# checking arguments
where = self._check_get_args(
calcmodel_types=calcmodel_types,
calcmodels=calcmodels,
file_names=file_names,
filter_type=filter_type,
model_as_regex=model_as_regex,
)
query = [
sql.SQL("SELECT calculation_model_name, file_name, file_def_id FROM performance.v_calculation_model_files_def "),
where,
sql.SQL(" ORDER BY calculation_model_name, file_name"),
]
query = sql.Composed(query)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query)
df = df.set_index(["calculation_model_name", "file_name"])
result = df.to_dict()["file_def_id"]
final_result = {}
for (calc_model, file_name), file_def_id in result.items():
if calc_model not in final_result:
final_result[calc_model] = {}
final_result[calc_model][file_name] = file_def_id
return final_result
insert(calculation_model_name, file_name, description=None, on_conflict='ignore')
¶
Inserts a calculation model instance file definition into the database.
Parameters:
-
(calculation_model_name¶str) –Name of the calculation model instance.
-
(file_name¶str) –Name of the file.
-
(description¶str | None, default:None) –Description of the file. 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_instance_file_definitions.py
@validate_call
def insert(
self,
calculation_model_name: str,
file_name: str,
description: str | None = None,
on_conflict: Literal["ignore", "update"] = "ignore",
) -> None:
"""Inserts a calculation model instance file definition into the database.
Parameters
----------
calculation_model_name : str
Name of the calculation model instance.
file_name : str
Name of the file.
description : str | None, optional
Description of the file. 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 calculation model id
calc_model_ids = self._perfdb.calcmodels.instances.get_ids()
if calculation_model_name not in calc_model_ids:
raise ValueError(f"Calculation model '{calculation_model_name}' does not exist in database")
# creating DataFrame to be inserted
df = DataFrame(
data={
"calculation_model_id": [calc_model_ids[calculation_model_name]],
"file_name": [file_name],
"description": [description],
},
)
self._perfdb.conn.pandas_to_sql(
df,
table_name="calculation_model_files_def",
schema="performance",
if_exists="append" if on_conflict == "ignore" else on_conflict,
ignore_index=True,
conflict_cols=["calculation_model_id", "file_name"],
)
logger.debug(f"Calculation model instance file definition '{calculation_model_name}' with file '{file_name}' inserted")