Subcomponent Instances - Attributes

SubcomponentInstanceAttributes(perfdb)

Class used for handling subcomponent instances attributes. Can be accessed via perfdb.components.subcomponents.instances.attributes.

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

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

get(subcomponent_ids=None, subcomponent_serial_numbers=None, subcomponent_batches=None, subcomponent_models=None, subcomponent_types=None, attribute_names=None, filter_type='and', output_type='dict', values_only=False)

Method to get the attributes of the given subcomponent instances.

The most useful keys/columns returned are:

• subcomponent_instance_id
• subcomponent_instance_name
• subcomponent_type_id
• subcomponent_type_name
• subcomponent_model_id
• subcomponent_model_name
• attribute_id
• attribute_name
• attribute_value
• data_type_name

Parameters:

• subcomponent_ids

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

  List of subcomponent ids to filter the results. By default None.

• subcomponent_serial_numbers

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

  List of subcomponent serial numbers to filter the results. By default None.

• subcomponent_batches

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

  List of subcomponent batches to filter the results. By default None.

• subcomponent_models

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

  List of subcomponent model names to filter the results. By default None.

• subcomponent_types

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

  List of subcomponent type names to filter the results. By default None.

• attribute_names

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

  List of attribute names to filter the results. If set to None will get all. By default None

• filter_type

  (Literal['and', 'or'], default: 'and' ) –

  How to treat multiple filters. Can be one of ["and", "or"]. By default "and"

• output_type

  (Literal['dict', 'DataFrame'], default: 'dict' ) –

  Output type of the data. Can be one of ["dict", "DataFrame"] By default "dict"

• values_only

  (bool, default: False ) –

  If set to True, will only return the values of the attributes, skipping display_name, id, etc.

Returns:

• dict[str, dict[str, Any | dict[str, Any]]] –

  In case output_type is "dict", returns a dictionary in the format {subcomponent_instance_id: {attribute_name: {attribute: value, ...}, ...}, ...} If values_only is set to True, returns a dictionary in the format {subcomponent_instance_id: {attribute_name: value, ...}, ...}

• DataFrame –

  In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[subcomponent_instance_id, attribute_name], columns = [attribute, ...] If values_only is set to True, returns a DataFrame with the following format: index = MultiIndex[subcomponent_instance_id, attribute_name], columns = ["attribute_value"]

Source code in echo_postgres/subcomponent_instance_attributes.py

@validate_call
def get(
    self,
    subcomponent_ids: list[int] | None = None,
    subcomponent_serial_numbers: list[str] | None = None,
    subcomponent_batches: list[str] | None = None,
    subcomponent_models: list[str] | None = None,
    subcomponent_types: list[str] | None = None,
    attribute_names: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
    output_type: Literal["dict", "DataFrame"] = "dict",
    values_only: bool = False,
) -> dict[str, dict[str, Any | dict[str, Any]]] | DataFrame:
    """Method to get the attributes of the given subcomponent instances.

    The most useful keys/columns returned are:

    - subcomponent_instance_id
    - subcomponent_instance_name
    - subcomponent_type_id
    - subcomponent_type_name
    - subcomponent_model_id
    - subcomponent_model_name
    - attribute_id
    - attribute_name
    - attribute_value
    - data_type_name

    Parameters
    ----------
    subcomponent_ids : list[int] | None
        List of subcomponent ids to filter the results. By default None.
    subcomponent_serial_numbers : list[str] | None
        List of subcomponent serial numbers to filter the results. By default None.
    subcomponent_batches : list[str] | None
        List of subcomponent batches to filter the results. By default None.
    subcomponent_models : list[str] | None
        List of subcomponent model names to filter the results. By default None.
    subcomponent_types : list[str] | None
        List of subcomponent type names to filter the results. By default None.
    attribute_names : list[str] | None, optional
        List of attribute names to filter the results. If set to None will get all. By default None
    filter_type : Literal["and", "or"], optional
        How to treat multiple filters. Can be one of ["and", "or"].
        By default "and"
    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "dict"
    values_only : bool, optional
        If set to True, will only return the values of the attributes, skipping display_name, id, etc.

    Returns
    -------
    dict[str, dict[str, Any | dict[str, Any]]]
        In case output_type is "dict", returns a dictionary in the format {subcomponent_instance_id: {attribute_name: {attribute: value, ...}, ...}, ...}
        If values_only is set to True, returns a dictionary in the format {subcomponent_instance_id: {attribute_name: value, ...}, ...}
    DataFrame
        In case output_type is "DataFrame", returns a DataFrame with the following format: index = MultiIndex[subcomponent_instance_id, attribute_name], columns = [attribute, ...]
        If values_only is set to True, returns a DataFrame with the following format: index = MultiIndex[subcomponent_instance_id, attribute_name], columns = ["attribute_value"]
    """
    # checking if all subcomponent instances are valid
    if subcomponent_ids:
        existing_instances = self._perfdb.components.subcomponents.instances.get_ids(subcomponent_ids=subcomponent_ids)
        if missing_instances := set(subcomponent_ids) - set(existing_instances):
            raise ValueError(f"The following subcomponent instances do not exist: {missing_instances}")

    # building the query
    query = [
        sql.SQL(
            "SELECT {values} FROM performance.v_subcomponent_instance_attributes",
        ).format(
            values=sql.SQL(
                "subcomponent_instance_id, attribute_name, attribute_value, data_type_name",
            )
            if values_only
            else sql.SQL("*"),
        ),
    ]
    where = []
    if subcomponent_ids:
        where.append(
            sql.SQL(" subcomponent_instance_id IN ({subcomponent_ids}) ").format(
                subcomponent_ids=sql.SQL(",").join(sql.Literal(value) for value in subcomponent_ids),
            ),
        )
    if subcomponent_serial_numbers:
        where.append(
            sql.SQL(" serial_number IN ({subcomponent_serial_numbers}) ").format(
                subcomponent_serial_numbers=sql.SQL(",").join(sql.Literal(value) for value in subcomponent_serial_numbers),
            ),
        )
    if subcomponent_batches:
        where.append(
            sql.SQL(" batch IN ({subcomponent_batches}) ").format(
                subcomponent_batches=sql.SQL(",").join(sql.Literal(value) for value in subcomponent_batches),
            ),
        )
    if subcomponent_models:
        where.append(
            sql.SQL(" subcomponent_model_name IN ({subcomponent_models}) ").format(
                subcomponent_models=sql.SQL(",").join(sql.Literal(value) for value in subcomponent_models),
            ),
        )
    if subcomponent_types:
        where.append(
            sql.SQL(" subcomponent_type_name IN ({subcomponent_types}) ").format(
                subcomponent_types=sql.SQL(",").join(sql.Literal(value) for value in subcomponent_types),
            ),
        )
    if attribute_names:
        where.append(
            sql.SQL(" attribute_name IN ({attribute_names}) ").format(
                attribute_names=sql.SQL(",").join(sql.Literal(an) for an in attribute_names),
            ),
        )
    if where:
        where = sql.SQL(f" {filter_type.upper()} ").join(where)
        query.append(sql.SQL(" WHERE "))
        query.append(where)
    query.append(sql.SQL(" ORDER BY subcomponent_instance_id, attribute_name"))
    query = sql.Composed(query)

    # executing the query
    with self._perfdb.conn.reconnect() as conn:
        # setting attribute_value as subcomponent to avoid casting json column as string
        df = conn.read_to_pandas(query, post_convert="pyarrow")

    # casting the attribute values
    df = cast_attributes(df=df, index_cols=["subcomponent_instance_id"])

    df = df.set_index(["subcomponent_instance_id", "attribute_name"])

    # returning the result
    if output_type == "dict":
        # dropping unwanted columns
        if values_only:
            df = df["attribute_value"]
            output = df.to_dict()
        else:
            output = df[["attribute_id", "attribute_value", "data_type_id", "data_type_name", "modified_date"]].to_dict(orient="index")
        # converting dict where the keys are tuples {(key1, key2): value}, to a dict where the keys are strings like {key1: {key2: value}}
        new_output = {}
        for (om, an), values in output.items():
            if om not in new_output:
                new_output[om] = {}
            new_output[om][an] = values
        return new_output
    if output_type == "DataFrame" and values_only:
        df = df[["attribute_value"]].copy()

    return df