Subcomponent Models¶
SubcomponentModels(perfdb)
¶
Class used for handling subcomponent models. Can be accessed via perfdb.components.subcomponents.models.
Parameters:
Source code in echo_postgres/subcomponent_models.py
def __init__(self, perfdb: e_pg.PerfDB) -> None:
"""Class used for handling subcomponent models. Can be accessed via `perfdb.components.subcomponents.models`.
Parameters
----------
perfdb : PerfDB
Top level object carrying all functionality and the connection handler.
"""
super().__init__(perfdb)
from .subcomponent_model_attributes import SubcomponentModelAttributes
# * subclasses
self.attributes = SubcomponentModelAttributes(perfdb)
get(component_types=None, subcomponent_models=None, subcomponent_types=None, manufacturers=None, filter_type='and', get_attributes=False, output_type='dict')
¶
Gets all subcomponent models definitions with detailed information.
The most useful keys/columns returned are:
- id
- description
- component_type_id
- subcomponent_type_id
- manufacturer_id
- manufacturer_name
Parameters:
-
(component_types¶list[str] | None, default:None) –Name of the component types to filter the results.
-
(subcomponent_types¶list[str] | None, default:None) –Name of the subcomponent types to filter the results, by default None
-
(subcomponent_models¶list[str] | None, default:None) –Name of the subcomponent models to filter the results, by default None
-
(manufacturers¶list[str] | None, default:None) –Name of the manufacturers to filter the results, by default None
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"]. By default "and"
-
(get_attributes¶bool, default:False) –If True, will also get the attributes of the subcomponent models.
-
(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 {component_type: {subcomponents_type: {subcomponent_model: {attribute: value, ...}, ...}, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[component_type_name, subcomponents_type_name, name], columns = [attribute, ...]
Source code in echo_postgres/subcomponent_models.py
@validate_call
def get(
self,
component_types: list[str] | None = None,
subcomponent_models: list[str] | None = None,
subcomponent_types: list[str] | None = None,
manufacturers: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
get_attributes: bool = False,
output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, int]:
"""Gets all subcomponent models definitions with detailed information.
The most useful keys/columns returned are:
- id
- description
- component_type_id
- subcomponent_type_id
- manufacturer_id
- manufacturer_name
Parameters
----------
component_types : list[str] | None
Name of the component types to filter the results.
subcomponent_types : list[str] | None, optional
Name of the subcomponent types to filter the results, by default None
subcomponent_models : list[str] | None, optional
Name of the subcomponent models to filter the results, by default None
manufacturers : list[str] | None, optional
Name of the manufacturers to filter the results, by default None
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"]. By default "and"
get_attributes : bool, optional
If True, will also get the attributes of the subcomponent models.
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 {component_type: {subcomponents_type: {subcomponent_model: {attribute: value, ...}, ...}, ...}, ...}
DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[component_type_name, subcomponents_type_name, name], columns = [attribute, ...]
"""
where = self._check_get_args(
component_types=component_types,
subcomponent_models=subcomponent_models,
subcomponent_types=subcomponent_types,
filter_type=filter_type,
manufacturers=manufacturers,
)
query = [
sql.SQL("SELECT * FROM performance.v_subcomponent_models "),
where,
sql.SQL(" ORDER BY component_type_name, subcomponent_type_name, name"),
]
query = sql.Composed(query)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query)
# getting attributes
if get_attributes:
# names of the subcomponent models
got_subcomponent_models = df["name"].tolist()
attrs: DataFrame = self._perfdb.components.subcomponents.models.attributes.get(
subcomponent_models=got_subcomponent_models,
output_type="DataFrame",
values_only=False,
)
if attrs.empty:
logger.debug("No attributes found for the subcomponent models")
else:
# pivot the attributes
attrs = attrs.reset_index(drop=False).pivot(
index="subcomponent_model_id",
columns="attribute_name",
values="attribute_value",
)
# merging the attributes with the subcomponent models
df = df.set_index("id")
df = df.merge(attrs, left_index=True, right_index=True, how="left")
df = df.reset_index(drop=False)
df = df.set_index(["component_type_name", "subcomponent_type_name", "name"])
if output_type == "DataFrame":
return df
result = df.to_dict(orient="index")
final_result = {}
for (component_type_name, subcomponent_type_name, name), values in result.items():
if component_type_name not in final_result:
final_result[component_type_name] = {}
if subcomponent_type_name not in final_result[component_type_name]:
final_result[component_type_name][subcomponent_type_name] = {}
final_result[component_type_name][subcomponent_type_name][name] = values
return final_result
get_ids(component_types=None, subcomponent_types=None, subcomponent_models=None, manufacturers=None, filter_type='and')
¶
Method used to get all subcomponent models and their respective ids.
Parameters:
-
(component_types¶list[str] | None, default:None) –Name of the component types to filter the results.
-
(subcomponent_types¶list[str] | None, default:None) –Name of the subcomponent types to filter the results, by default None
-
(subcomponent_models¶list[str] | None, default:None) –Name of the subcomponent models to filter the results, by default None
-
(manufacturers¶list[str] | None, default:None) –Name of the manufacturers to filter the results, by default None
-
(filter_type¶Literal['and', 'or'], default:'and') –How to treat multiple filters. Can be one of ["and", "or"]. By default "and"
Returns:
-
dict[str, dict[str, dict[str, int]]]–Dictionary with all subcomponents types and their respective ids in the format {component_type: {subcomponents_type: {subcomponent_model: id, ...}, ...}, ...}.
Source code in echo_postgres/subcomponent_models.py
@validate_call
def get_ids(
self,
component_types: list[str] | None = None,
subcomponent_types: list[str] | None = None,
subcomponent_models: list[str] | None = None,
manufacturers: list[str] | None = None,
filter_type: Literal["and", "or"] = "and",
) -> dict[str, dict[str, dict[str, int]]]:
"""Method used to get all subcomponent models and their respective ids.
Parameters
----------
component_types : list[str] | None
Name of the component types to filter the results.
subcomponent_types : list[str] | None, optional
Name of the subcomponent types to filter the results, by default None
subcomponent_models : list[str] | None, optional
Name of the subcomponent models to filter the results, by default None
manufacturers : list[str] | None, optional
Name of the manufacturers to filter the results, by default None
filter_type : Literal["and", "or"], optional
How to treat multiple filters. Can be one of ["and", "or"]. By default "and"
Returns
-------
dict[str, dict[str, dict[str, int]]]
Dictionary with all subcomponents types and their respective ids in the format {component_type: {subcomponents_type: {subcomponent_model: id, ...}, ...}, ...}.
"""
# checking inputs
where = self._check_get_args(
component_types=component_types,
subcomponent_models=subcomponent_models,
subcomponent_types=subcomponent_types,
filter_type=filter_type,
manufacturers=manufacturers,
)
query = [
sql.SQL("SELECT component_type_name, subcomponent_type_name, name, id FROM performance.v_subcomponent_models "),
where,
sql.SQL(" ORDER BY component_type_name, subcomponent_type_name, name"),
]
query = sql.Composed(query)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query)
df = df.set_index(["component_type_name", "subcomponent_type_name", "name"])
result = df.to_dict(orient="index")
final_result = {}
for (component_type_name, subcomponent_type_name, name), values in result.items():
if component_type_name not in final_result:
final_result[component_type_name] = {}
if subcomponent_type_name not in final_result[component_type_name]:
final_result[component_type_name][subcomponent_type_name] = {}
final_result[component_type_name][subcomponent_type_name][name] = values["id"]
return final_result