Component Instances - Latest

ComponentInstancesLatest(perfdb)

Class used for handling the latest installed component instances. Can be accessed via perfdb.components.instances.latest.

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(object_names=None, component_ids=None, component_serial_numbers=None, component_batches=None, component_models=None, component_manufacturers=None, component_types=None, locations=None, filter_type='and', get_attributes=False, output_type='DataFrame')

Gets components instances latest (including dates when they were installed and removed).

The most useful keys/columns returned are:

• object_id
• object_name
• component_type_id
• component_type_name
• manufacturer_id
• manufacturer_name
• component_model_id
• component_model_name
• location_id
• location_name
• component_instance_id
• serial_number
• batch
• installation_date
• duration

Parameters:

• object_names

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

  List of object names to filter the results. By default None.

• component_ids

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

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

• component_serial_numbers

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

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

• component_batches

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

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

• component_models

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

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

• component_manufacturers

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

  List of component manufacturer names to filter the results. By default None.

• component_types

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

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

• locations

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

  List of locations 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 component models.

• output_type

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

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

Returns:

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

  Dictionary on the format {"object_name": {"component_type_name": {"location_name": {"attribute": value, ...}, ...}, ...}, ...}

• DataFrame –

  DataFrame with the latest component instances. Index is a MultiIndex with levels "object_name", "component_type_name", "location_name"

Source code in echo_postgres/component_instances_latest.py

@validate_call
def get(
    self,
    object_names: list[str] | None = None,
    component_ids: list[int] | None = None,
    component_serial_numbers: list[str] | None = None,
    component_batches: list[str] | None = None,
    component_models: list[str] | None = None,
    component_manufacturers: list[str] | None = None,
    component_types: list[str] | None = None,
    locations: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
    get_attributes: bool = False,
    output_type: Literal["dict", "DataFrame"] = "DataFrame",
) -> dict[str, dict[str, dict[str, Any]]] | DataFrame:
    """Gets components instances latest (including dates when they were installed and removed).

    The most useful keys/columns returned are:

    - object_id
    - object_name
    - component_type_id
    - component_type_name
    - manufacturer_id
    - manufacturer_name
    - component_model_id
    - component_model_name
    - location_id
    - location_name
    - component_instance_id
    - serial_number
    - batch
    - installation_date
    - duration

    Parameters
    ----------
    object_names : list[str] | None, optional
        List of object names to filter the results. By default None.
    component_ids : list[int] | None, optional
        List of component ids to filter the results. By default None.
    component_serial_numbers : list[str] | None, optional
        List of component serial numbers to filter the results. By default None.
    component_batches : list[str] | None, optional
        List of component batches to filter the results. By default None.
    component_models : list[str] | None, optional
        List of component model names to filter the results. By default None.
    component_manufacturers : list[str] | None, optional
        List of component manufacturer names to filter the results. By default None.
    component_types : list[str] | None, optional
        List of component type names to filter the results. By default None.
    locations : list[str] | None, optional
        List of locations 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 component models.
    output_type : Literal["dict", "DataFrame"], optional
        Output type of the data. Can be one of ["dict", "DataFrame"]
        By default "DataFrame"

    Returns
    -------
    dict[str, dict[str, dict[str, Any]]]
        Dictionary on the format {"object_name": {"component_type_name": {"location_name": {"attribute": value, ...}, ...}, ...}, ...}
    DataFrame
        DataFrame with the latest component instances. Index is a MultiIndex with levels "object_name", "component_type_name", "location_name"
    """
    where = self._check_get_args(
        object_names=object_names,
        component_ids=component_ids,
        component_serial_numbers=component_serial_numbers,
        component_batches=component_batches,
        component_models=component_models,
        component_manufacturers=component_manufacturers,
        component_types=component_types,
        locations=locations,
        filter_type=filter_type,
    )

    query = [
        sql.SQL("SELECT * FROM performance.v_component_instance_latest "),
        where,
        sql.SQL(" ORDER BY object_name, component_type_name, location_name"),
    ]
    query = sql.Composed(query)

    with self._perfdb.conn.reconnect() as conn:
        df = conn.read_to_pandas(query)

    # getting attributes
    if get_attributes:
        df = df.set_index("component_instance_id")
        # names of the component models
        got_component_instances = df.index.tolist()
        attrs: DataFrame = self._perfdb.components.instances.attributes.get(
            component_ids=got_component_instances,
            output_type="DataFrame",
            values_only=True,
        )
        # pivot the attributes
        attrs = attrs.reset_index(drop=False).pivot(index="component_instance_id", columns="attribute_name", values="attribute_value")
        # merging the attributes with the component models
        df = df.merge(attrs, left_index=True, right_index=True, how="left")
        # resetting the index
        df = df.reset_index(drop=False)

    # defining object_name, component_type_name and location_name as the index
    df = df.set_index(["object_name", "component_type_name", "location_name"])

    if output_type == "dict":
        result = df.to_dict(orient="index")
        final_result = {}
        for (object_name, component_type_name, location_name), values in result.items():
            if object_name not in final_result:
                final_result[object_name] = {}
            if component_type_name not in final_result[object_name]:
                final_result[object_name][component_type_name] = {}
            final_result[object_name][component_type_name][None if isna(location_name) else location_name] = values
        return final_result

    return df