Skip to content

Component Instances - Attributes

ComponentInstances(perfdb)

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

Parameters:

  • perfdb

    (PerfDB) –

    Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/component_instances.py
def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Class used for handling component instances.

    Parameters
    ----------
    perfdb : PerfDB
        Top level object carrying all functionality and the connection handler.
    """
    super().__init__(perfdb)

    from .component_instance_attributes import ComponentInstanceAttributes
    from .component_instances_history import ComponentInstancesHistory
    from .component_instances_latest import ComponentInstancesLatest

    # * subclasses

    self.attributes = ComponentInstanceAttributes(perfdb)
    self.history = ComponentInstancesHistory(perfdb)
    self.latest = ComponentInstancesLatest(perfdb)

get(component_ids=None, component_serial_numbers=None, component_batches=None, component_models=None, component_manufacturers=None, component_types=None, filter_type='and', get_attributes=False, output_type='dict')

Gets all component instances.

The most useful keys/columns returned are:

  • id
  • serial_number
  • batch
  • component_model_id
  • component_model_name
  • component_type_id
  • component_type_name
  • manufacturer_id
  • manufacturer_name

Parameters:

  • 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.

  • filter_type

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

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

  • get_attributes

    (bool, default: False ) –

    If True, will also get the attributes of the component models.

  • output_type

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

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

Returns:

  • list[dict[str, Any]]

    List with the component instances. Each element is a dictionary with the keys being the column names.

  • DataFrame

    DataFrame with the component instances. Index can be ignored.

Source code in echo_postgres/component_instances.py
@validate_call
def get(
    self,
    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,
    filter_type: Literal["and", "or"] = "and",
    get_attributes: bool = False,
    output_type: Literal["dict", "DataFrame"] = "dict",
) -> list[dict[str, Any]] | DataFrame:
    """Gets all component instances.

    The most useful keys/columns returned are:

    - id
    - serial_number
    - batch
    - component_model_id
    - component_model_name
    - component_type_id
    - component_type_name
    - manufacturer_id
    - manufacturer_name

    Parameters
    ----------
    component_ids : list[int] | None
        List of component ids to filter the results. By default None.
    component_serial_numbers : list[str] | None
        List of component serial numbers to filter the results. By default None.
    component_batches : list[str] | None
        List of component batches to filter the results. By default None.
    component_models : list[str] | None
        List of component model names to filter the results. By default None.
    component_manufacturers : list[str] | None
        List of component manufacturer names to filter the results. By default None.
    component_types : list[str] | None
        List of component type names to filter the results. By default None.
    filter_type : Literal["and", "or"]
        How to treat multiple filters. Can be one of ["and", "or"].
    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 "dict"

    Returns
    -------
    list[dict[str, Any]]
        List with the component instances. Each element is a dictionary with the keys being the column names.
    DataFrame
        DataFrame with the component instances. Index can be ignored.
    """
    where = self._check_get_args(
        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,
        filter_type=filter_type,
    )

    query = [
        sql.SQL("SELECT * FROM performance.v_component_instances "),
        where,
        sql.SQL(" ORDER BY id"),
    ]
    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("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,
        )
        if attrs.empty:
            logger.debug("No attributes found")
        else:
            # 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)

    return df.to_dict(orient="records") if output_type == "dict" else df

get_ids(component_ids=None, component_serial_numbers=None, component_batches=None, component_models=None, component_manufacturers=None, component_types=None, filter_type='and')

Gets all component instances ids.

Parameters:

  • 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.

  • filter_type

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

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

Returns:

  • list[int]

    List with the component ids.

Source code in echo_postgres/component_instances.py
@validate_call
def get_ids(
    self,
    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,
    filter_type: Literal["and", "or"] = "and",
) -> list[int]:
    """Gets all component instances ids.

    Parameters
    ----------
    component_ids : list[int] | None
        List of component ids to filter the results. By default None.
    component_serial_numbers : list[str] | None
        List of component serial numbers to filter the results. By default None.
    component_batches : list[str] | None
        List of component batches to filter the results. By default None.
    component_models : list[str] | None
        List of component model names to filter the results. By default None.
    component_manufacturers : list[str] | None
        List of component manufacturer names to filter the results. By default None.
    component_types : list[str] | None
        List of component type names to filter the results. By default None.
    filter_type : Literal["and", "or"]
        How to treat multiple filters. Can be one of ["and", "or"].

    Returns
    -------
    list[int]
        List with the component ids.
    """
    # checking inputs
    where = self._check_get_args(
        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,
        filter_type=filter_type,
    )

    query = [
        sql.SQL("SELECT id FROM performance.v_component_instances "),
        where,
        sql.SQL(" ORDER BY id"),
    ]
    query = sql.Composed(query)

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

    return df["id"].tolist()