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
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'], 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 {availability_type: {availability_subtype: {attribute: value, ...}, ...}, ...}
-
DataFrame–In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[availability_type_name, name], columns = [attribute, ...]
Source code in echo_postgres/kpi_availability_subtypes.py
@validate_call
def get(
self,
availability_types: list[str] | None = None,
get_categories: bool = False,
output_type: Literal["dict", "DataFrame"] = "dict",
) -> dict[str, dict[str, Any]] | 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"], 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 {availability_type: {availability_subtype: {attribute: value, ...}, ...}, ...}
DataFrame
In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[availability_type_name, name], columns = [attribute, ...]
"""
# 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)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query, post_convert="pyarrow")
df = df.set_index(["availability_type_name", "name"])
if get_categories:
df["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.index:
raise ValueError(
f"Availability subtype {availability_subtype} from availability type {availability_type} not found in the database",
)
df.at[(availability_type, availability_subtype), "categories"] = availability_subtype_vals # noqa: PD008
if output_type == "DataFrame":
return df
# converting to dict
result = df.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
@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)
with self._perfdb.conn.reconnect() as conn:
df = conn.read_to_pandas(query)
result = df.set_index(["availability_type_name", "name"]).to_dict()["id"]
final_result = {}
for (availability_type, availability_subtype), id_ in result.items():
final_result.setdefault(availability_type, {})[availability_subtype] = id_
return final_result