Subcomponent Instances

SubcomponentInstances(perfdb)

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

Parameters:

• perfdb

  (PerfDB) –

  Top level object carrying all functionality and the connection handler.

Source code in echo_postgres/subcomponent_instances.py

Python

def __init__(self, perfdb: e_pg.PerfDB) -> None:
    """Class used for handling subcomponent instances. Can be accessed via `perfdb.components.subcomponents.instances`.

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

    from .subcomponent_instance_attributes import SubcomponentInstanceAttributes
    from .subcomponent_instances_history import SubcomponentInstancesHistory
    from .subcomponent_instances_latest import SubcomponentInstancesLatest

    # * subclasses

    self.attributes = SubcomponentInstanceAttributes(perfdb)
    self.history = SubcomponentInstancesHistory(perfdb)
    self.latest = SubcomponentInstancesLatest(perfdb)

edit(event_id, subcomponent_type, subcomponent_model=None, subcomponent_serial_number=None)

Edits a subcomponent instance. This method is used for testing purposes and should not be used in production.

Parameters:

• event_id

  (int) –

  The id of the event to which the subcomponent instance belongs.

• subcomponent_type

  (str) –

  The name of the subcomponent type.

• subcomponent_model

  (str | None, default: None ) –

  The name of the subcomponent model.

• subcomponent_serial_number

  (str | None, default: None ) –

  The serial number of the subcomponent instance.

Source code in echo_postgres/subcomponent_instances.py

Python

@validate_call
def edit(
    self,
    event_id: int,
    subcomponent_type: str,
    subcomponent_model: str | None = None,
    subcomponent_serial_number: str | None = None,
) -> None:
    """Edits a subcomponent instance. This method is used for testing purposes and should not be used in production.

    Parameters
    ----------
    event_id : int
        The id of the event to which the subcomponent instance belongs.
    subcomponent_type : str
        The name of the subcomponent type.
    subcomponent_model : str | None
        The name of the subcomponent model.
    subcomponent_serial_number : str | None
        The serial number of the subcomponent instance.
    """
    args = {
        "event_id": (event_id, True, int),
        "subcomponent_type": (subcomponent_type, True, str),
        "subcomponent_model": (subcomponent_model, False, str),
        "subcomponent_serial_number": (subcomponent_serial_number, False, str),
    }
    query = sql.SQL(
        """
        SELECT * FROM performance.fn_edit_subcomponent(
        _event_id=>{event_id},
        _subcomponent_type=>{subcomponent_type},
        _subcomponent_model=>{subcomponent_model},
        _subcomponent_serial_number=>{subcomponent_serial_number},
        )
        """,
    ).format(
        event_id=sql.Literal(event_id),
        subcomponent_type=sql.Literal(subcomponent_type),
        subcomponent_model=sql.Literal(subcomponent_model),
        subcomponent_serial_number=sql.Literal(subcomponent_serial_number),
    )
    self._perfdb.conn.execute(query, skip_retry=True)

    changed_params = {k: v for k, v in args.items() if v[0] is not None}
    logger.debug(
        f"Edited subcomponent instance for event id {event_id} and type {subcomponent_type}. Changed parameters: {changed_params}",
    )

get(subcomponent_ids=None, subcomponent_serial_numbers=None, subcomponent_models=None, subcomponent_manufacturers=None, subcomponent_types=None, component_types=None, filter_type='and', get_attributes=False, output_type='dict')

Gets all subcomponent instances.

The most useful keys/columns returned are:

• id
• serial_number
• subcomponent_model_id
• subcomponent_model_name
• subcomponent_type_id
• subcomponent_type_name
• manufacturer_id
• manufacturer_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_models

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

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

• subcomponent_manufacturers

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

  List of subcomponent manufacturer 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.

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

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

• list[dict[str, Any]] –

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

• DataFrame –

  In case output_type is "DataFrame", returns a pandas DataFrame with the subcomponent instances.

• DataFrame –

  In case output_type is "pl.DataFrame", returns a Polars DataFrame

Source code in echo_postgres/subcomponent_instances.py

Python

@validate_call
def get(
    self,
    subcomponent_ids: list[int] | None = None,
    subcomponent_serial_numbers: list[str] | None = None,
    subcomponent_models: list[str] | None = None,
    subcomponent_manufacturers: list[str] | None = None,
    subcomponent_types: 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", "pl.DataFrame"] = "dict",
) -> list[dict[str, Any]] | pd.DataFrame | pl.DataFrame:
    """Gets all subcomponent instances.

    The most useful keys/columns returned are:

    - id
    - serial_number
    - subcomponent_model_id
    - subcomponent_model_name
    - subcomponent_type_id
    - subcomponent_type_name
    - manufacturer_id
    - manufacturer_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_models : list[str] | None
        List of subcomponent model names to filter the results. By default None.
    subcomponent_manufacturers : list[str] | None
        List of subcomponent manufacturer 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.
    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 subcomponent models.
    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
    -------
    list[dict[str, Any]]
        List with the subcomponent instances. Each element is a dictionary with the keys being the column names.
    pd.DataFrame
        In case output_type is "DataFrame", returns a pandas DataFrame with the subcomponent instances.
    pl.DataFrame
        In case output_type is "pl.DataFrame", returns a Polars DataFrame
    """
    where = self._check_get_args(
        subcomponent_ids=subcomponent_ids,
        subcomponent_serial_numbers=subcomponent_serial_numbers,
        subcomponent_models=subcomponent_models,
        subcomponent_manufacturers=subcomponent_manufacturers,
        subcomponent_types=subcomponent_types,
        component_types=component_types,
        filter_type=filter_type,
    )

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

    df = self._perfdb.conn.read_to_polars(query)

    # getting attributes
    if get_attributes:
        # names of the subcomponent models
        got_subcomponent_instances = df["id"].to_list()
        attrs: pl.DataFrame = self._perfdb.components.subcomponents.instances.attributes.get(
            subcomponent_ids=got_subcomponent_instances,
            output_type="pl.DataFrame",
            values_only=True,
        )
        # pivot the attributes
        attrs = attrs.pivot(index="subcomponent_instance_id", on="attribute_name", values="attribute_value")
        # merging the attributes with the subcomponent models
        df = df.join(attrs, left_on="id", right_on="subcomponent_instance_id", how="left")

    return convert_output(df, output_type, orient="records")

get_ids(subcomponent_ids=None, subcomponent_serial_numbers=None, subcomponent_models=None, subcomponent_manufacturers=None, subcomponent_types=None, component_types=None, filter_type='and')

Gets all subcomponent instances ids.

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_models

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

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

• subcomponent_manufacturers

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

  List of subcomponent manufacturer 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.

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

Source code in echo_postgres/subcomponent_instances.py

Python

@validate_call
def get_ids(
    self,
    subcomponent_ids: list[int] | None = None,
    subcomponent_serial_numbers: list[str] | None = None,
    subcomponent_models: list[str] | None = None,
    subcomponent_manufacturers: list[str] | None = None,
    subcomponent_types: list[str] | None = None,
    component_types: list[str] | None = None,
    filter_type: Literal["and", "or"] = "and",
) -> list[int]:
    """Gets all subcomponent instances ids.

    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_models : list[str] | None
        List of subcomponent model names to filter the results. By default None.
    subcomponent_manufacturers : list[str] | None
        List of subcomponent manufacturer 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.
    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 subcomponent ids.
    """
    # checking inputs
    where = self._check_get_args(
        subcomponent_ids=subcomponent_ids,
        subcomponent_serial_numbers=subcomponent_serial_numbers,
        subcomponent_models=subcomponent_models,
        subcomponent_manufacturers=subcomponent_manufacturers,
        subcomponent_types=subcomponent_types,
        component_types=component_types,
        filter_type=filter_type,
    )

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

    df = self._perfdb.conn.read_to_polars(query)

    return df["id"].to_list()