KPI Availability Subtypes¶
KpiAvailabilitySubtypes(perfdb)
¶
Class used for handling availability subtypes. Can be accessed via perfdb.kpis.availability.subtypes.
Parameters:
Source code in echo_postgres/kpi_availability_subtypes.py
Python
def __init__(self, perfdb: e_pg.PerfDB) -> None:
"""Class used for handling availability subtypes. Can be accessed via `perfdb.kpis.availability.subtypes`.
Parameters
----------
perfdb : PerfDB
Top level object carrying all functionality and the connection handler.
"""
super().__init__(perfdb)
from .kpi_availability_subtype_categories import KpiAvailabilitySubtypeCategories
# * subclasses
self.categories = KpiAvailabilitySubtypeCategories(perfdb)
get(availability_types=None, get_categories=False, output_type='dict')
¶
Gets all availability subtypes definitions with detailed information.
The most useful keys/columns returned are:
- id
- description
- object_ids
- object_names
Parameters:
-
(availability_types¶list[str] | None, default:None) –Availability types to filter the results. If None, no filter is applied. By default None
-
(get_categories¶bool, default:False) –Whether to get the categories associated with the availability subtypes. If True, the categories will be added to the output.
-
(output_type¶Literal['dict', 'DataFrame', 'pl.DataFrame'], default:'dict') –Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"] By default "dict"
Returns:
-
dict[str, dict[str, Any]]–In case output_type is "dict", returns a dictionary in the format {availability_type: {availability_subtype: {attribute: value, ...}, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a pandas DataFrame with the following format: index = MultiIndex[availability_type_name, name], columns = [attribute, ...]
-
DataFrame–In case output_type is "pl.DataFrame", returns a Polars DataFrame
Source code in echo_postgres/kpi_availability_subtypes.py
Python
@validate_call
def get(
self,
availability_types: list[str] | None = None,
get_categories: bool = False,
output_type: Literal["dict", "DataFrame", "pl.DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | pd.DataFrame | pl.DataFrame:
"""Gets all availability subtypes definitions with detailed information.
The most useful keys/columns returned are:
- id
- description
- object_ids
- object_names
Parameters
----------
availability_types : list[str] | None, optional
Availability types to filter the results. If None, no filter is applied.
By default None
get_categories : bool, optional
Whether to get the categories associated with the availability subtypes. If True, the categories will be added to the output.
output_type : Literal["dict", "DataFrame", "pl.DataFrame"], optional
Output type of the data. Can be one of ["dict", "DataFrame", "pl.DataFrame"]
By default "dict"
Returns
-------
dict[str, dict[str, Any]]
In case output_type is "dict", returns a dictionary in the format {availability_type: {availability_subtype: {attribute: value, ...}, ...}, ...}
pd.DataFrame
In case output_type is "DataFrame", returns a pandas DataFrame with the following format: index = MultiIndex[availability_type_name, name], columns = [attribute, ...]
pl.DataFrame
In case output_type is "pl.DataFrame", returns a Polars DataFrame
"""
# validating inputs
where = self._check_get_args(availability_types=availability_types)
query = [sql.SQL("SELECT * FROM performance.v_availability_subtypes"), where, sql.SQL(" ORDER BY name")]
query = sql.Composed(query)
df = self._perfdb.conn.read_to_polars(query)
if not get_categories:
return convert_output(
df,
output_type=output_type,
index_col=["availability_type_name", "name"],
nest_by_index=True,
)
# get_categories requires pandas manipulation before output conversion
if output_type == "pl.DataFrame":
return df
df_pd = df.to_pandas(use_pyarrow_extension_array=True).set_index(["availability_type_name", "name"])
df_pd["categories"] = NA
categories: dict[str, dict[str, dict[str, dict[str, Any]]]] = self._perfdb.kpis.availability.subtypes.categories.get(
availability_types=availability_types,
)
for availability_type, availability_type_vals in categories.items():
for availability_subtype, availability_subtype_vals in availability_type_vals.items():
if (availability_type, availability_subtype) not in df_pd.index:
raise ValueError(
f"Availability subtype {availability_subtype} from availability type {availability_type} not found in the database",
)
df_pd.at[(availability_type, availability_subtype), "categories"] = availability_subtype_vals # noqa: PD008
if output_type == "DataFrame":
return df_pd
# converting to dict
result = df_pd.to_dict(orient="index")
final_result = {}
for (availability_type, availability_subtype), values in result.items():
final_result.setdefault(availability_type, {})[availability_subtype] = values
return final_result
get_ids(availability_types=None)
¶
Gets all availability subtypes and their respective ids.
Parameters:
-
(availability_types¶list[str] | None, default:None) –Availability types to filter the results. If None, no filter is applied. By default None
Returns:
-
dict[str, dict[str, int]]–Dictionary with all availability subtypes and their respective ids in the format (availability_type_name: {name: id, ...}, ...}.
Source code in echo_postgres/kpi_availability_subtypes.py
Python
@validate_call
def get_ids(
self,
availability_types: list[str] | None = None,
) -> dict[str, dict[str, int]]:
"""Gets all availability subtypes and their respective ids.
Parameters
----------
availability_types : list[str] | None, optional
Availability types to filter the results. If None, no filter is applied.
By default None
Returns
-------
dict[str, dict[str, int]]
Dictionary with all availability subtypes and their respective ids in the format (availability_type_name: {name: id, ...}, ...}.
"""
# validating inputs
where = self._check_get_args(availability_types)
query = [
sql.SQL("SELECT id, name, availability_type_name FROM performance.v_availability_subtypes"),
where,
sql.SQL(" ORDER BY name"),
]
query = sql.Composed(query)
df = self._perfdb.conn.read_to_polars(query)
result = {}
for row in df.iter_rows(named=True):
avail_type = row["availability_type_name"]
subtype_name = row["name"]
result.setdefault(avail_type, {})[subtype_name] = row["id"]
return result